python - 如何使用 eve-sqlalchemy 更新项目关系

Question

我创建了一个简单的 eve-sqlalchemy 应用程序，它具有 User 和 Role 资源，以模仿 an example on their webpage 。 User 类可以具有多个角色。下面提供了完整的示例代码。

我希望能够POST具有角色的用户，或PATCH用户的角色。我一直无法找到一种方法来做到这一点而不遇到错误。

看起来eve-sqlalchemy将User.roles的类型强制为整数，这是Role类的主键 id。如果我将 POST 或 PATCH 请求设置为使 roles 字段为整数，eve-sqlalchemy 会提示self.driver.app.config['SOURCES'] 中不存在 userroles 关联表。虽然我可以进行这些更改(即将 userroles 表设为声明性 ORM 并将其注册到装饰器)，但我不确定这是否是正确的方法。

总而言之，eve-sqlalchemy 希望我如何POST 具有角色列表的用户 ，或PATCH用户的现有角色？

服务器:

from eve import Eve
from eve.utils import config

from sqlalchemy import create_engine, Table, Column, String, Integer, ForeignKey, func, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

from eve_sqlalchemy.decorators import registerSchema
from eve_sqlalchemy import SQL
from eve_sqlalchemy.validation import ValidatorSQL


ID_FIELD = 'id'
SQLALCHEMY_DATABASE_URI = 'sqlite:////tmp/eve-sqla-test.db'


config.ID_FIELD = ID_FIELD
config.ITEM_LOOKUP_FIELD = ID_FIELD


Base = declarative_base()


userroles_table = Table('userroles', Base.metadata,
    Column('user_id', Integer, ForeignKey("users.id"), primary_key=True),
    Column('role_id', Integer, ForeignKey("roles.id"), primary_key=True)
)


class CommonColumns(Base):
    __abstract__ = True
    _created = Column(DateTime, default=func.now())
    _updated = Column(DateTime, default=func.now(), onupdate=func.now())
    _etag = Column(String(40))


class Role(CommonColumns):
    __tablename__ = 'roles'
    id = Column(Integer, primary_key=True, autoincrement=True)
    role = Column(String, unique=True, nullable=False)


class User(CommonColumns):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    login = Column(String, unique=True, nullable=False)
    roles = relationship("Role", backref="users", secondary=userroles_table)


def create_entries():
    '''Creates test entries for querying with eve'''
    engine = create_engine(SQLALCHEMY_DATABASE_URI)

    Base.metadata.bind = engine
    Base.metadata.create_all()

    SessionMaker = sessionmaker(bind=engine)
    s = SessionMaker()

    u1 = User(login='user1')
    u1.roles.append(Role(role='admin'))

    u2 = User(login='user2')
    u2.roles.append(Role(role='user'))

    s.add_all((u1, u2))
    s.commit()


def run():
    '''Runs the eve server'''

    for table in (User, Role):
        registerSchema(table.__tablename__)(table)

    users = User._eve_schema[User.__tablename__]
    users.update({
        'item_methods': ['GET', 'PATCH', 'DELETE'],
        'resource_methods': ['GET', 'POST'],
    })

    roles = Role._eve_schema[Role.__tablename__]
    roles.update({
        'resource_methods': ['GET', 'POST'],
    })

    DOMAIN = {
        'users': users,
        'roles': roles,
    }

    SETTINGS = {
        'SQLALCHEMY_DATABASE_URI': SQLALCHEMY_DATABASE_URI,
        'SQLALCHEMY_TRACK_MODIFICATIONS': False,
        'ID_FIELD': ID_FIELD,
        'ITEM_LOOKUP_FIELD': ID_FIELD,
        'DOMAIN': DOMAIN,
    }

    app = Eve(validator=ValidatorSQL, data=SQL, settings=SETTINGS)

    db = app.data.driver
    Base.metadata.bind = db.engine
    db.Model = Base

    app.run(debug=True)


if __name__ == '__main__':
    '''Test area'''

    #create_entries()
    run()

请求:

import json

import requests


u4 = {
   'login': 'user4',
}


# users get works
r = requests.get('http://localhost:5000/users')
print(r.text)


# user creation works
r = requests.post('http://localhost:5000/users', json=u4)
print(r.text)


# roles get works
r = requests.get('http://localhost:5000/roles')
print(r.text)


# user creation with roles fail
u5 = {
   'login': 'user5',
   'roles': [1,]
}

r = requests.post('http://localhost:5000/users', json=u5)
print(r.text)  # {"_issues": {"roles": ["field 'roles' could not be coerced", "must be of integer type"]}, "_status": "ERR"}


# user patch with role fails
r = requests.get('http://localhost:5000/users/1')
patch_headers = {"If-Match": r.json()['_etag'], 'Content-type': 'application/json; charset=utf-8'}

r = requests.patch('http://localhost:5000/users/1', headers=patch_headers, json={'roles': [1,]})
print(r.text)  # {"_issues": {"roles": ["field 'roles' could not be coerced", "must be of integer type"]}, "_status": "ERR"}

Answer 1

我最终做的是:

1. 将 userroles_table 从表转换为声明性类 UserRole。
2. 使用 eve_sqlalchemy 注册 UserRole
3. 将 UserRole 添加到 DOMAIN。

然后可以通过新的资源端点user_roles直接查询或修改关联表。 eve_sqlalchemy 似乎在生成架构时隐藏了外键，因此包含与外键的关系非常重要。