separation of concerns in fastapi + sqlalchemy by encapsulating reusable queries in the model and keep services and routes thin<\/figcaption><\/figure><\/div>\n\n\nWhat we should avoid is putting session-bound methods directly on the model, such as:<\/p>\n\n\n\n
@classmethod\nasync def get_published(cls, session: AsyncSession):\n return (await session.scalars(select(cls).where(cls.status == \"published\"))).all()<\/code><\/pre>\n\n\n\nThat starts mixing model definition with I\/O and transaction concerns. If a model method executes the session internally and returns a raw list of objects, you lose composability, it means you can not apply more filter on get_published()<\/code>. Returning a statement is usually the cleaner boundary, and it keeps the code easier to test and maintain.<\/p>\n\n\n\n3. Avoid Two Different Meanings Of “No String Value”<\/h2>\n\n\n\n
One of the most common data quality problems in APIs is allowing both NULL<\/code> and \"\"<\/code> to mean “missing text”. Pick one representation and normalize input consistently.<\/p>\n\n\n\nIn SQLAlchemy<\/strong>, using None<\/code> \/ NULL<\/code> is usually the cleanest choice for optional string data:<\/p>\n\n\n\nclass CustomerCreate(BaseModel):\n middle_name: str | None = None\n\n @field_validator(\"middle_name\")\n @classmethod\n def normalize_middle_name(cls, value: str | None) -> str | None:\n if value is None:\n return None\n value = value.strip()\n return value or None\n\n\nclass Customer(Base):\n __tablename__ = \"customers\"\n\n id: Mapped[int] = mapped_column(primary_key=True)\n middle_name: Mapped[str | None] = mapped_column(String(100), nullable=True)<\/code><\/pre>\n\n\n\nFor fields where empty string is a meaningful value, keep nullable=False<\/code> and store \"\"<\/code> consistently. The important part is not which convention you choose. The important part is avoiding both at the same time.<\/p>\n\n\n\nFor optional unique string fields, NULL<\/code> is usually the better choice than \"\"<\/code>.<\/p>\n\n\n\n4. Keep Alembic Migrations Clean<\/h2>\n\n\n\n
FastAPI + SQLAlchemy<\/strong> projects usually rely on Alembic for schema changes. A common mistake is letting migration history become noisy and chaotic during active feature development.<\/p>\n\n\n\nFor example, one feature branch might generate these revisions:<\/p>\n\n\n\n
001_add_order_table.py\n002_fix_order_table.py\n003_fix_order_table_again.py\n004_add_missing_index.py<\/code><\/pre>\n\n\n\nBefore merging a private branch, squash all migration generated in the brnach into one clean revision. However, there exist cases where the migration should be applied in more than one phase, in that cases having more than one migration file is inevitable. <\/p>\n\n\n\n
Typical flow:<\/p>\n\n\n\n
alembic revision --autogenerate -m \"add orders table\"\nalembic upgrade head<\/code><\/pre>\n\n\n\nBest practice:<\/p>\n\n\n\n
\n- keep branch-local migrations tidy<\/li>\n\n\n\n
- write descriptive revision names<\/li>\n\n\n\n
- do not rewrite migrations already used by teammates or production<\/li>\n\n\n\n
- prefer one clean migration per coherent change instead of five correction migrations in a row<\/li>\n<\/ul>\n\n\n\n
Alembic does not have the same built-in “squash migrations” workflow Django has, so cleanup usually means reorganizing revisions before they become shared history.<\/p>\n\n\n\n
5. Use Two Identifiers: Private PK And Public UUID<\/h2>\n\n\n\n
In API-driven applications, it is often useful to keep a compact internal primary key and expose a public UUID externally.<\/p>\n\n\n\n
Example:<\/p>\n\n\n\n
class Order(Base):\n __tablename__ = \"orders\"\n\n id: Mapped[int] = mapped_column(BigInteger, primary_key=True)\n public_id: Mapped[uuid.UUID] = mapped_column(\n Uuid,\n default=uuid.uuid4,\n unique=True,\n nullable=False,\n index=True,\n )\n status: Mapped[str] = mapped_column(String(30), nullable=False)<\/code><\/pre>\n\n\n\nThen your public routes can use public_id<\/code>:<\/p>\n\n\n\n@app.get(\"\/orders\/{public_id}\")\nasync def get_order(public_id: uuid.UUID, session: SessionDep):\n stmt = select(Order).where(Order.public_id == public_id)\n return await session.scalar(stmt)<\/code><\/pre>\n\n\n\nThis pattern gives us:<\/p>\n\n\n\n
\n- small and efficient internal joins on integer keys<\/li>\n\n\n\n
- stable public identifiers<\/li>\n\n\n\n
- less leakage of record counts and ordering<\/li>\n<\/ul>\n\n\n\n
Also note the use of SQLAlchemy’s backend-agnostic Uuid<\/code> type. In SQLAlchemy 2.x, you no longer need custom GUID hacks for common cross-database UUID storage.<\/p>\n\n\n\nAlthough there are valid distributed-system cases for UUID or ULID primary keys, for many business apps, integer primary key plus public UUID is a very pragmatic default.<\/p>\n\n\n\n
6. Use SQL Expressions Instead Of Processing Rows In Python<\/h2>\n\n\n\n
If the database can do the filtering, comparison, aggregation, or update, let it do it.<\/p>\n\n\n\n
Bad practice:<\/p>\n\n\n\n
students = []\nresult = await session.scalars(select(Student))\nfor student in result:\n if student.math_score > student.english_score:\n students.append(student)<\/code><\/pre>\n\n\n\nBetter:<\/p>\n\n\n\n
stmt = select(Student).where(Student.math_score > Student.english_score)\nstudents = (await session.scalars(stmt)).all()<\/code><\/pre>\n\n\n\nSQLAlchemy <\/strong>gives you rich expression tools such as:<\/p>\n\n\n\n\nfunc<\/code> for SQL functions<\/li>\n\n\n\n- subqueries and CTEs<\/li>\n\n\n\n
- SQL expressions in updates<\/li>\n\n\n\n
- relationship loader options<\/li>\n<\/ul>\n\n\n\n
Another useful example is atomic updates:<\/p>\n\n\n\n
post = await session.get(Post, post_id)\npost.view_count = Post.view_count + 1\nawait session.commit()<\/code><\/pre>\n\n\n\nLetting the database express the change is safer than reading a value in Python, incrementing it, and writing it back later.<\/p>\n\n\n\n
7. Add A Generic pk Property On Your Base Model<\/h2>\n\n\n\n
One of the nicest ideas in Django models <\/strong>is obj.pk<\/code>. It lets you refer to the primary key generically without coupling your code to a specific column name like id<\/code>, uuid<\/code>, or invoice_number<\/code>.<\/p>\n\n\n\nYou can get the same convenience in SQLAlchemy <\/strong>with a hybrid property on your declarative base:<\/p>\n\n\n\nfrom sqlalchemy import inspect\nfrom sqlalchemy.ext.hybrid import hybrid_property\nfrom sqlalchemy.orm import DeclarativeBase\n\n\nclass Base(DeclarativeBase):\n @classmethod\n def _get_pk_field(cls):\n mapper = inspect(cls)\n fields = mapper.primary_key\n\n if len(fields) != 1:\n raise ValueError(f\"{cls.__name__} has composite primary key\")\n\n return fields[0]\n\n @hybrid_property\n def pk(self):\n pk_field = type(self)._get_pk_field()\n return getattr(self, pk_field.key)\n\n @pk.expression\n def pk(cls):\n return cls._get_pk_field()<\/code><\/pre>\n\n\n\nThis has the same benefits Django developers enjoy:<\/p>\n\n\n\n
\n- your code stays generic even when different models use different primary-key names<\/li>\n\n\n\n
pk<\/code> expresses intent better than hardcoding id<\/code><\/li>\n\n\n\n- the same API works on instances and inside SQL expressions<\/li>\n\n\n\n
- schema refactors are less painful because calling code does not care what the PK column is called<\/li>\n<\/ul>\n\n\n\n
Example model:<\/p>\n\n\n\n
class Todo(Base):\n __tablename__ = \"todos\"\n\n id: Mapped[int] = mapped_column(primary_key=True)\n title: Mapped[str] = mapped_column(String(200))<\/code><\/pre>\n\n\n\nInstance access:<\/p>\n\n\n\n
todo = Todo(title=\"Write the post\")\ntodo.pk<\/code><\/pre>\n\n\n\nClass-level SQL expression:<\/p>\n\n\n\n
Todo.pk<\/code><\/pre>\n\n\n\nComparison expression:<\/p>\n\n\n\n
Todo.pk == 1<\/code><\/pre>\n\n\n\nSelecting the primary key generically:<\/p>\n\n\n\n
select(Todo.pk)<\/code><\/pre>\n\n\n\nFiltering by primary key without hardcoding id<\/code>:<\/p>\n\n\n\nstmt = select(Todo).where(Todo.pk == 1)\ntodo = await session.scalar(stmt)<\/code><\/pre>\n\n\n\nWhen the primary key is not named id<\/code>:<\/p>\n\n\n\nclass Invoice(Base):\n __tablename__ = \"invoices\"\n\n invoice_number: Mapped[str] = mapped_column(String(50), primary_key=True)\n amount_cents: Mapped[int]<\/code><\/pre>\n\n\n\nNow the calling code stays consistent:<\/p>\n\n\n\n
stmt = select(Invoice).where(Invoice.pk == \"INV-2026-0001\")\ninvoice = await session.scalar(stmt)<\/code><\/pre>\n\n\n\nsession.get()<\/strong><\/code> is still a great API for direct primary-key retrieval when you already have the PK value in hand. But adding Base.pk<\/code> gives your models the same expressive, generic primary-key interface that makes Django code pleasant to write and maintain. <\/p>\n\n\n\nOne improvement would be to define two reusable base classes: a primary base class without an id<\/code> field, which still provides the pk<\/code> property described above, and a second base class that inherits from it and adds id<\/code> as the default primary key. This gives each model the flexibility to either use the conventional integer id<\/code> primary key or define its own primary key, such as a UUID, slug, or another natural key.<\/p>\n\n\n\nfrom sqlalchemy import inspect\nfrom sqlalchemy.ext.hybrid import hybrid_property\nfrom sqlalchemy.orm import DeclarativeBase\n\n\nclass Base(DeclarativeBase):\n @classmethod\n def _get_pk_field(cls):\n mapper = inspect(cls)\n fields = mapper.primary_key\n\n if len(fields) != 1:\n raise ValueError(f\"{cls.__name__} has composite primary key\")\n\n return fields[0]\n\n @hybrid_property\n def pk(self):\n pk_field = type(self)._get_pk_field()\n return getattr(self, pk_field.key)\n\n @pk.expression\n def pk(cls):\n return cls._get_pk_field()\n\n\nclass BaseWithId(Base):\n __abstract__ = True\n\n id: Mapped[int] = mapped_column(primary_key=True)\n\n\nclass User(BaseWithId):\n __tablename__ = \"users\"\n\n name: Mapped[str]\n\n\nclass Country(Base):\n __tablename__ = \"countries\"\n\n code: Mapped[str] = mapped_column(primary_key=True)\n name: Mapped[str]\n<\/code><\/pre>\n\n\n\nIn this example, User<\/code> uses the default integer id<\/code>, while Country<\/code> uses code<\/code> as its primary key. Both models can still access their primary-key value through the shared pk<\/code> property.<\/p>\n\n\n\n8. Name Models And Relationships Clearly<\/h2>\n\n\n\n
Naming matters because ORM code is read far more often than it is written. A few practical conventions:<\/p>\n\n\n\n
\n- class names should be singular:
User<\/code>, Order<\/code>, BlogPost<\/code><\/strong><\/li>\n\n\n\n- table names should be predictable:
users<\/code>, orders<\/code>, blog_posts<\/code><\/strong><\/li>\n\n\n\n- scalar relationships should be singular:
author<\/code>, profile<\/code>, customer<\/code><\/strong><\/li>\n\n\n\n- collection relationships should be plural:
posts<\/code>, items<\/code>, roles<\/code><\/strong><\/li>\n<\/ul>\n\n\n\nExample:<\/p>\n\n\n\n
class Author(Base):\n __tablename__ = \"authors\"\n\n id: Mapped[int] = mapped_column(primary_key=True)\n name: Mapped[str] = mapped_column(String(100))\n posts: Mapped[list[\"BlogPost\"]] = relationship(back_populates=\"author\")\n\n\nclass BlogPost(Base):\n __tablename__ = \"blog_posts\"\n\n id: Mapped[int] = mapped_column(primary_key=True)\n author_id: Mapped[int] = mapped_column(ForeignKey(\"authors.id\"))\n title: Mapped[str] = mapped_column(String(200))\n\n author: Mapped[Author] = relationship(back_populates=\"posts\")<\/code><\/pre>\n\n\n\nThis is clearer than ambiguous names like post_data<\/code>, author_ref<\/code>, or blogpostlist<\/code>.<\/p>\n\n\n\n9. Use A yield<\/code> Dependency With async_sessionmaker<\/code><\/h2>\n\n\n\nFastAPI’s documentation recommends using dependencies with yield<\/code> when you need setup and teardown around a request. For database sessions, that maps perfectly to SQLAlchemy sessions: create the session before yield<\/code>, and make sure it is closed after the request finishes.<\/p>\n\n\n\nFor modern async SQLAlchemy code, the cleanest version is:<\/p>\n\n\n\n
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine\n\n\nengine = create_async_engine(\n DATABASE_URL,\n pool_size=10,\n max_overflow=20,\n pool_timeout=30,\n pool_pre_ping=True,\n)\n\nSessionLocal = async_sessionmaker(engine, expire_on_commit=False)\n\n\nasync def get_session():\n async with SessionLocal() as session:\n yield session<\/code><\/pre>\n\n\n\nIt is better to use FastAPI’s yield<\/code> dependency pattern for request-scoped setup and cleanup. And use SQLAlchemy’s async_sessionmaker<\/code>, which is the dedicated SQLAlchemy 2.x factory for AsyncSession<\/code>. It’s clearer and more modern than using generic sessionmaker(..., class_=AsyncSession)<\/code><\/p>\n\n\n\nBenefits:<\/p>\n\n\n\n
\n- each request gets its own session<\/li>\n\n\n\n
- the session is closed automatically after the request<\/li>\n\n\n\n
- the code is small and easy to understand<\/li>\n<\/ul>\n\n\n\n
If you prefer manual cleanup, this is also valid:<\/p>\n\n\n\n
async def get_session():\n session = SessionLocal()\n try:\n yield session\n finally:\n await session.close()<\/code><\/pre>\n\n\n\nBut async with SessionLocal() as session:<\/strong><\/mark><\/code> is usually cleaner and less error-prone.<\/p>\n\n\n\nAlso, an AsyncSession<\/code> should not be shared across concurrent tasks. The safe mental model is one session per request, and if you spawn concurrent tasks, each task should have its own session.<\/p>\n\n\n\n10. Keep Transaction Boundaries Explicit<\/h2>\n\n\n\n
This is one of the most important SQLAlchemy <\/strong>habits to get right in FastAPI apps: understand when you need flush()<\/code> and when you need commit()<\/code>.<\/p>\n\n\n\n\nflush()<\/strong><\/code> sends pending SQL to the database inside the current transaction<\/li>\n\n\n\ncommit()<\/code> <\/strong>permanently finishes the transaction<\/li>\n<\/ul>\n\n\n\nMany bugs come from calling commit()<\/code><\/strong> too early just to get a generated primary key or to make a later insert work.<\/p>\n\n\n\nBad:<\/p>\n\n\n\n
@app.post(\"\/orders\")\nasync def create_order(payload: OrderCreate, session: SessionDep):\n order = Order(customer_id=payload.customer_id, status=\"pending\")\n session.add(order)\n await session.commit()\n\n item = OrderItem(order_id=order.id, sku=payload.sku, quantity=payload.quantity)\n session.add(item)\n await session.commit()\n return order<\/code><\/pre>\n\n\n\nGood:<\/p>\n\n\n\n
@app.post(\"\/orders\")\nasync def create_order(payload: OrderCreate, session: SessionDep):\n order = Order(customer_id=payload.customer_id, status=\"pending\")\n session.add(order)\n await session.flush()\n\n item = OrderItem(order_id=order.id, sku=payload.sku, quantity=payload.quantity)\n session.add(item)\n\n await session.commit()\n await session.refresh(order)\n return order<\/code><\/pre>\n\n\n\nIt is better because:<\/p>\n\n\n\n
\n- one request owns one transaction<\/li>\n\n\n\n
- you still get
order.id<\/strong><\/code> after
![\"Keep](\"https:\/\/mshaeri.com\/blog\/wp-content\/uploads\/2026\/05\/image-4-1024x611.png\" "\\"Keep")
<\/a>![\"separation](\"https:\/\/mshaeri.com\/blog\/wp-content\/uploads\/2026\/05\/image-2-1024x394.png\" "\\"separation")
<\/a>