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

Question

我有editors 和articles。许多编辑可能与许多文章相关，并且许多文章可能同时有许多编辑。

\n

我的数据库表是

\n

\n
• 文章
\n

\n

\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n

ID	主题	文本
1	新年假期	今年...等等等等

\n

\n

\n
• 编辑
\n

\n

\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n

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

\n

\n

\n
• 编辑文章关系
\n

\n

\n\n\n\n\n\n\n\n\n\n\n\n\n\n

编辑ID	文章编号
1	1

\n

\n

我的模型是

\n

from sqlalchemy import Boolean, Column, Integer, String, ForeignKey\nfrom sqlalchemy.orm import relationship\n\nfrom database import Base\n\nclass Editor(Base):\n    __tablename__ = "editor"\n\n    id = Column(Integer, primary_key=True, index=True)\n    name = Column(String(32), unique=False, index=False, nullable=True)\n    email = Column(String(115), unique=True, index=True)\n    articles = relationship("Article",\n                    secondary=EditorArticleRelation,\n                    back_populates="articles",\n                    cascade="all, delete")\n\nclass Article(Base):\n    __tablename__ = "article"\n\n    id = Column(Integer, primary_key=True, index=True)\n    subject = Column(String(32), unique=True, index=False)\n    text = Column(String(256), unique=True, index=True, nullable=True)\n    editors = relationship("Editor",\n                    secondary=EditorArticleRelation,\n                    back_populates="editors",\n                    cascade="all, delete")\n\nEditorArticleRelation = Table(\'editorarticlerelation\', Base.metadata,\n    Column(\'editor_id\', Integer, ForeignKey(\'editor.id\')),\n    Column(\'article_id\', Integer, ForeignKey(\'article.id\'))\n)\n

\n

我的模式是

\n

from typing import Optional, List\nfrom pydantic import BaseModel\n\nclass EditorBase(BaseModel):\n    name: Optional[str]\n    email: str\n\nclass EditorCreate(EditorBase):\n    pass\n\nclass Editor(EditorBase):\n    id: int\n\n    class Config:\n        orm_mode = True\n\nclass ArticleBase(BaseModel):\n    subject: str\n    text: str\n\nclass ArticleCreate(ArticleBase):\n    # WHAT I NEED TO SET HERE???\n    editor_ids: List[int] = []\n\nclass Article(ArticleBase):\n    id: int\n    editors: List[Editor] = []\n\n    class Config:\n        orm_mode = True\n

\n

我的粗鲁

\n

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

\n

我的路线

\n

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

\n

我想要什么\xe2\x86\x92

\n

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

\n

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

\n

问题是：

\n

\n
1. 如何正确设置增删改查操作（HOW TO SET EDITORS HERE地方）？
\n
2. 如何正确设置创建/读取模式和关系字段（尤其是WHAT I NEED TO SET HERE位置）？
\n
3. 如何正确设置路线代码（尤其是WHAT I NEED TO SET HERE地点）？
\n
4. 如果这里无法自动解决关系，那么什么地方可以更好地解决关系（检查编辑器是否存在等）？路线还是粗鲁？
\n
5. 也许我的方法根本就不好？pydantic如果您知道如何处理与和的多对多关系的任何示例sqlalchemy，欢迎提供任何信息
\n

\n

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] = []
...

• 增删改查（解决方案在这里）：

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

有更好的想法欢迎提出