fastapi + sqlalchemy + pydantic → 如何处理多对多关系

Question

我有 editor 和 article。许多编辑可能与许多文章相关，许多文章可能同时有多个编辑。

我的数据库表是

文章

id	主题	文字
1	新年假期	今年...等等等等等等

编辑器

id	姓名	电子邮件
1	约翰·史密斯	一些@电子邮件

EditorArticleRelation

editor_id	article_id
1	1

我的模特是

from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

from database import Base

class Editor(Base):
    __tablename__ = "editor"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(32), unique=False, index=False, nullable=True)
    email = Column(String(115), unique=True, index=True)
    articles = relationship("Article",
                    secondary=EditorArticleRelation,
                    back_populates="articles",
                    cascade="all, delete")

class Article(Base):
    __tablename__ = "article"

    id = Column(Integer, primary_key=True, index=True)
    subject = Column(String(32), unique=True, index=False)
    text = Column(String(256), unique=True, index=True, nullable=True)
    editors = relationship("Editor",
                    secondary=EditorArticleRelation,
                    back_populates="editors",
                    cascade="all, delete")

EditorArticleRelation = Table('editorarticlerelation', Base.metadata,
    Column('editor_id', Integer, ForeignKey('editor.id')),
    Column('article_id', Integer, ForeignKey('article.id'))
)

我的架构是

from typing import Optional, List
from pydantic import BaseModel

class EditorBase(BaseModel):
    name: Optional[str]
    email: str

class EditorCreate(EditorBase):
    pass

class Editor(EditorBase):
    id: int

    class Config:
        orm_mode = True

class ArticleBase(BaseModel):
    subject: str
    text: str

class ArticleCreate(ArticleBase):
    # WHAT I NEED TO SET HERE???
    editor_ids: List[int] = []

class Article(ArticleBase):
    id: int
    editors: List[Editor] = []

    class Config:
        orm_mode = True

我的垃圾

def create_article(db: Session, article_data: schema.ArticleCreate):
    db_article = model.Article(subject=article_data.subject, text=article_data.text, ??? HOW TO SET EDITORS HERE ???)
    db.add(db_article)
    db.commit()
    db.refresh(db_article)
    return db_article

我的路线

@app.post("/articles/", response_model=schema.Article)
def create_article(article_data: schema.ArticleCreate, db: Session = Depends(get_db)):
    db_article = crud.get_article_by_name(db, name=article_data.name)
    if db_article:
        raise HTTPException(status_code=400, detail="article already registered")
    if len(getattr(article_data, 'editor_ids', [])) > 0:
        ??? WHAT I NEED TO SET HERE???
    return crud.create_article(db=db, article_data=article_data)

我想要的→

我想post文章创建数据API并自动解析和添加编辑关系，或者如果某些编辑不存在则报错：

{
  "subject": "Fresh news"
  "text": "Today is ..."
  "editor_ids": [1, 2, ...]
}

问题是：

如何正确设置crud操作（HOW TO SET EDITORS HERE处）？
如何正确设置create/read模式和关系字段（尤其是WHAT I NEED TO SET HERE地方）？
如何正确设置路由代码（尤其是WHAT I NEED TO SET HERE地方）？
如果这里无法自动解析关系，那么在什么地方解析关系比较好（检查编辑器是否存在等）？路线还是路况？
也许我的方法根本不好？如果您知道如何处理 pydantic 和 sqlalchemy 的多对多关系的任何示例，欢迎提供任何信息

Answer 1

不确定我的解决方案是否最有效，但我是这样做的：

路线（与问题相同）：

...
@app.post("/articles/", response_model=schema.Article)
def create_article(article_data: schema.ArticleCreate, db: Session = Depends(get_db)):
    db_article = crud.get_article_by_name(db, name=article_data.name)
    if db_article:
        raise HTTPException(status_code=400, detail="article already registered")
    return crud.create_article(db=db, article_data=article_data)
...

架构（与问题相同）：

...
class ArticleCreate(ArticleBase):
    editor_ids: List[int] = []
...

crud（解决方案在这里）：

def create_article(db: Session, article_data: schema.ArticleCreate):
    db_article = model.Article(subject=article_data.subject, text=article_data.text)
    if (editors := db.query(model.Editor).filter(model.Editor.id.in_(article_data.editor_ids))).count() == len(endpoint_data.topic_ids):
        db_article.topics.extend(editors)
    else:
        # even if at least one editor is not found, an error is raised
        # if existence is not matter you can skip this check and add relations only for existing data
        raise HTTPException(status_code=404, detail="editor not found")
    db.add(db_article)
    db.commit()
    db.refresh(db_article)
    return db_article

欢迎提出更好的意见

Answer 2

我找到的解决方案。

def create_user_groups(db: Session, user_groups: schemas.UserGroupsBase):
    db_user = db.query(models.User).filter(models.User.id == user_groups.id_user).first()
    db_group = db.query(models.Group).filter(models.Group.id == user_groups.id_group).first()

    if not db_user and db_group:
        raise HTTPException(status_code=409, detail="User or Group not found in system.")

    db_user.groups.append(db_group)

    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

fastapi + sqlalchemy + pydantic → 如何处理多对多关系

fastapi + sqlalchemy + pydantic → how to process many-to-many relations

python

sqlalchemy

pydantic

fastapi

我想要的→

问题是：