SQLAlchemy Many-to-Many Relationships in a FastAPI POST Endpoint

Overview

You have posts and tags. A client sends POST /posts with a title and a list of tag names, and you need to save the post, reuse tags that already exist, create the ones that do not, and return a clean response. The usual first attempt creates duplicate tags or fails with a unique constraint error.

This guide shows a working pattern using SQLAlchemy 2.0 style models and Pydantic v2.

1. Model the Association Table

A many-to-many relationship needs a join table. For a plain link with no extra columns, a Table with a composite primary key is enough.

from sqlalchemy import Column, ForeignKey, String, Table
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


post_tags = Table(
    'post_tags',
    Base.metadata,
    Column('post_id', ForeignKey('posts.id', ondelete='CASCADE'), primary_key=True),
    Column('tag_id', ForeignKey('tags.id', ondelete='CASCADE'), primary_key=True),
)


class Tag(Base):
    __tablename__ = 'tags'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)


class Post(Base):
    __tablename__ = 'posts'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    tags: Mapped[list[Tag]] = relationship(secondary=post_tags)

The unique constraint on Tag.name is what protects you from duplicates at the database level. The composite primary key stops the same tag being linked to a post twice.

If the link needs its own data, such as who added the tag or when, use an association object model instead of a bare table.

2. Separate Input and Output Schemas

Do not reuse the ORM model or one schema for everything. Input should contain only what the client may set. Output should include generated fields.

from pydantic import BaseModel, ConfigDict, Field


class PostCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    tags: list[str] = Field(default_factory=list, max_length=20)


class TagRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str


class PostRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    tags: list[TagRead]

from_attributes=True lets Pydantic v2 build the response from ORM objects. In Pydantic v1 the equivalent setting is orm_mode.

3. Load Existing Tags, Create Missing Ones

Normalize the input first, then query once for all names, and create only the tags that are missing.

from sqlalchemy import select
from sqlalchemy.orm import Session


def get_or_create_tags(db: Session, names: list[str]) -> list[Tag]:
    cleaned = {n.strip().lower() for n in names if n.strip()}
    if not cleaned:
        return []

    existing = db.scalars(select(Tag).where(Tag.name.in_(cleaned))).all()
    by_name = {t.name: t for t in existing}

    for name in cleaned - by_name.keys():
        tag = Tag(name=name)
        db.add(tag)
        by_name[name] = tag

    return list(by_name.values())

Using a set removes duplicates inside the request itself, which is a frequent cause of unique violations. A single IN query avoids one query per tag.

4. The Endpoint

from fastapi import Depends, FastAPI, status

app = FastAPI()


@app.post('/posts', response_model=PostRead, status_code=status.HTTP_201_CREATED)
def create_post(payload: PostCreate, db: Session = Depends(get_db)):
    post = Post(title=payload.title)
    post.tags = get_or_create_tags(db, payload.tags)
    db.add(post)
    db.commit()
    db.refresh(post)
    return post

Assigning to post.tags makes SQLAlchemy insert the join rows when it flushes. Because the response model is PostRead, FastAPI serializes only the declared fields.

Here get_db is the per-request session dependency. If you use an async session, the same logic applies with await on execution, commit, and refresh, and you must load the relationship explicitly (for example with selectinload) before serializing, since lazy loading does not work in async contexts.

5. The Race Condition You Should Still Handle

Two requests can create the same new tag at the same time. Both pass the "does it exist" check, and one then fails on the unique constraint. Catch IntegrityError, roll back, and retry the operation once, or use your database's upsert, such as PostgreSQL INSERT ... ON CONFLICT DO NOTHING, then re-select. Rely on the constraint, not the check, for correctness.

6. Common Mistakes

• Creating Tag(name=...) for every name without checking, which produces duplicates or errors.
• Returning the ORM model directly without a response schema, which can leak fields.
• Forgetting ondelete behavior on the join table, which leaves orphan rows or blocks deletes.
• Accepting unbounded tag lists. Validate the list length and tag length.
• Case and whitespace differences creating near-duplicate tags.

Conclusion

Use an association table with a unique tag name, normalize and deduplicate input, query existing tags in one pass, and return a dedicated response schema. Let the database constraint be the final guard against races.

If you want help modeling relationships or cleaning up an existing FastAPI data layer, I can help. Contact me.

Additional Resources

• SQLAlchemy: Basic Relationship Patterns
• FastAPI: Response Model
• Pydantic documentation