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

Question

I have editors and articles. Many editors may be related to many articles and many articles may have many editors at same time.

My DB tables are

• Article

id	subject	text
1	New Year Holidays	In this year... etc etc etc

• Editor

id	name	email
1	John Smith	some@email

• EditorArticleRelation

editor_id	article_id
1	1

My models are

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'))
)

My schemas are

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

My crud

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

My route

@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)

What I want →

I want to post data for article creation API and automatically resolve and add editor relations, or raise error if some of editors doesn't exist:

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

Questions are:

1. How to correctly set crud operations (HOW TO SET EDITORS HERE place)?
2. How to correctly set create/read schemas and relation fields (especially WHAT I NEED TO SET HERE place)?
3. How to correctly set route code (especially WHAT I NEED TO SET HERE place)?
4. If here is no possible to resolve relations automatically, what place will be better to resolve relations (check if editor exists, etc)? route or crud?
5. Maybe my way is bad at all? If you know any examples how to handle many-to-many relations with pydantic and sqlalchemy, any information will be welcome

Answer 1

Not sure if my solution is most effective, but I did it by this way:

• route (same as in question):

...
@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)
...

• schema (same as in question):

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

• crud (solution is here):

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

Any better ideas are welcome

Answer 2

A solution I found.

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