This article focuses on practical generic CRUD patterns with FastAPI and SQLAlchemy 2.0. It explains the differences between synchronous and asynchronous implementations, transaction boundaries, connection pool configuration, and common error handling. The core pain points are technology selection anxiety, async pitfalls, and uncontrolled ORM queries. Keywords: FastAPI, SQLAlchemy 2.0, async CRUD.
Technical Specification Snapshot
| Parameter | Description |
|---|---|
| Primary Language | Python 3 |
| Web Framework | FastAPI |
| ORM | SQLAlchemy 2.0 |
| Sync Protocol/Driver | PostgreSQL + psycopg2 / MySQL sync drivers |
| Async Protocol/Driver | PostgreSQL + asyncpg / MySQL + aiomysql |
| Transaction Pattern | session.commit() / session.begin() |
| Article Popularity | Blog园 technical article, original views: 31 |
| Core Dependencies | fastapi, sqlalchemy, asyncpg, aiomysql |
SQLAlchemy 2.0 Has Unified the CRUD Mental Model in FastAPI
After integrating SQLAlchemy 2.0 into FastAPI, the biggest change is not that there are “more APIs,” but that the usage patterns for synchronous and asynchronous workflows are now much more aligned. Model definitions, query construction, transaction boundaries, pagination, and relationship loading can all reuse nearly the same approach.
For developers, the real question is not whether async is “more advanced,” but whether the current workload is a high-concurrency, I/O-bound scenario and whether the team can maintain an end-to-end async stack.
SQLAlchemy 2.0 Declarative Models Are Better Suited to Typed Development
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True) # Primary key
name: Mapped[str] = mapped_column(String(50)) # Username
version: Mapped[int] = mapped_column(default=1) # Optimistic lock versionThis example shows the Mapped + mapped_column style recommended by SQLAlchemy 2.0. It provides clearer type hints and works better with IDEs and static analysis.
The Difference Between Sync and Async Engine Configuration Primarily Lives in the Driver Layer
Synchronous mode usually works well for low-concurrency projects where the team is comfortable with blocking calls. Asynchronous mode is better suited to services that must handle multiple kinds of I/O at the same time, such as databases, external APIs, and WebSockets. The most important difference between them lies in the engine creation function and the database driver.
The Engine and Session Factory Should Be Clearly Layered from the Start
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import sessionmaker
# Synchronous engine
sync_engine = create_engine(
"postgresql://user:pass@localhost/db",
pool_size=20,
max_overflow=10,
pool_pre_ping=True, # Prevent dead connections from being reused
)
SyncSessionLocal = sessionmaker(bind=sync_engine)
# Asynchronous engine
async_engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
pool_size=20,
max_overflow=10,
pool_pre_ping=True, # Connection pre-check is also recommended for async workloads
)
AsyncSessionLocal = async_sessionmaker(
async_engine,
class_=AsyncSession,
expire_on_commit=False,
)This configuration highlights a critical point: async is not achieved by adding await to a synchronous URL. You must switch to a driver that truly supports asynchronous operation, such as asyncpg.
The Core Logic of Generic CRUD Is Nearly Identical in Both Modes
The CRUD object lifecycle does not change: create the object, add it to the session, commit the transaction, and refresh the instance. The real differences between synchronous and asynchronous workflows are the invocation style and whether you need to explicitly await the operation.
Database-Generated Fields Should Be Explicitly Refreshed After Record Creation
from sqlalchemy.ext.asyncio import AsyncSession
# Synchronous version
def create_user_sync(session, name: str):
user = User(name=name)
session.add(user)
session.commit()
session.refresh(user) # Refresh auto-generated fields such as the ID
return user
# Asynchronous version
async def create_user_async(session: AsyncSession, name: str):
user = User(name=name)
session.add(user)
await session.commit()
await session.refresh(user) # Async refresh must be awaited
return userThis example shows the most common insert pattern. The key detail is not add(), but the post-commit refresh() call and the required await in async code.
Bulk Writes Work Best Inside Explicit Transaction Boundaries
async def bulk_create_users(session: AsyncSession, names: list[str]):
async with session.begin():
session.add_all([User(name=n) for n in names]) # Bulk insert inside a transactionThis snippet emphasizes that write operations should preferably live inside session.begin() so you get automatic commit behavior and rollback on exceptions.
Queries Should Fully Move to select() with Explicit Relationship Loading
SQLAlchemy 2.0 no longer encourages the query() style. The recommended approach is to build statements with select() and execute them through session.execute(). That makes query syntax almost identical across sync and async code and makes it easier to abstract a repository layer.
Pagination, Sorting, and Fuzzy Search Can Be Composed in a Single Statement
from sqlalchemy import select
async def search_users(session: AsyncSession, keyword: str, page: int, size: int):
stmt = (
select(User)
.where(User.name.ilike(f"%{keyword}%")) # Fuzzy search
.order_by(User.id.desc()) # Descending order
.limit(size) # Page size
.offset((page - 1) * size) # Pagination offset
)
result = await session.execute(stmt)
return result.scalars().all()This example shows that select() is sufficient for most back-office list queries, and it is more expressive than the old chained ORM query style.
Relationship Data Must Be Preloaded Proactively in Async Scenarios
from sqlalchemy import select
from sqlalchemy.orm import selectinload
stmt = (
select(User)
.options(selectinload(User.posts)) # Preload posts to avoid lazy-loading errors
.where(User.id == 1)
)The key point here is avoiding MissingGreenlet. In the async ORM, lazy loading is often not merely slow; it fails outright.
Complex Queries Are Ultimately Assembled from Subqueries, Aggregations, and Joins
Once requirements move into analytical territory, such as “search active users and include their post count,” do not rely on ORM magic to do everything automatically. A more stable approach is to aggregate in a subquery first, then combine it with the main table using outerjoin.
User and Post Statistics Can Be Extended Safely Through a Subquery
from sqlalchemy import select, func, and_, or_, desc
async def search_users_with_post_count(session: AsyncSession, keyword: str, page: int = 1, size: int = 10):
post_count_sub = (
select(Post.author_id, func.count(Post.id).label("post_count"))
.group_by(Post.author_id)
.subquery()
)
stmt = (
select(User, func.coalesce(post_count_sub.c.post_count, 0).label("post_count"))
.outerjoin(post_count_sub, User.id == post_count_sub.c.author_id)
.where(
and_(
User.is_active == True,
or_(
User.username.ilike(f"%{keyword}%"),
User.email.ilike(f"%{keyword}%"),
),
)
)
.order_by(desc("post_count"))
.limit(size)
.offset((page - 1) * size)
)
result = await session.execute(stmt)
return result.all()This example demonstrates a common enterprise query pattern: subquery aggregation, null handling, composed conditions, alias-based sorting, and pagination.
Updates and Deletes Should Prioritize Transaction Consistency and Concurrency Safety
For simple updates, you can load an object first and then change its attributes. For bulk updates and state transitions, however, direct update() expressions are usually a better fit. If the business logic has a risk of concurrent overwrites, add a version field to implement optimistic locking.
Bulk Updates and Optimistic Locking Map Naturally to SQL Expressions
from sqlalchemy import update
async def deactivate_users(session: AsyncSession, user_ids: list[int]):
async with session.begin():
stmt = (
update(User)
.where(User.id.in_(user_ids))
.values(active=False) # Bulk deactivate users
)
result = await session.execute(stmt)
return result.rowcountThis pattern fits bulk state changes well and avoids the extra ORM overhead of loading each object individually.
stmt = (
update(User)
.where(User.id == uid, User.version == old_version) # Version check
.values(name=new_name, version=old_version + 1) # Atomically increment the version
)This pattern is used for optimistic locking and helps prevent data overwrites in concurrent write scenarios.
Session Management in FastAPI Should Keep the Lifecycle Strictly Aligned with the Request
In FastAPI, the recommended approach is to manage database sessions through dependency injection. Create one session per request and close it when the response completes. Then add session.begin() inside business logic for write operations to create a second layer of protection.
Request-Scoped Session Dependencies Are Critical to Connection Pool Stability
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit() # Commit after a successful request
except Exception:
await session.rollback() # Roll back on exceptions
raise
finally:
await session.close() # Always release the connectionThis pattern standardizes connection release, rollback on failure, and commit behavior so that long-running transactions do not exhaust the connection pool.
The Principle for Coexisting Sync and Async Code Is Clear Boundaries, Not Mixed Usage
During legacy upgrades, it is common for synchronous and asynchronous code to coexist. The correct approach is not to pass mixed session types into the same function, but to define clear module boundaries. When an async route must call synchronous logic, use a thread pool to isolate blocking behavior.
Running Synchronous Database Calls in a Thread Pool Prevents Event Loop Blocking
import asyncio
def sync_get_user(user_id: int):
return {"id": user_id}
async def async_endpoint():
user = await asyncio.to_thread(sync_get_user, user_id=1) # Run sync code in a worker thread
return userThis example shows a safe compatibility path for async endpoints that must call older synchronous logic. However, it should remain a transition strategy rather than the long-term architectural default.
Most Common Failures Come from Lazy Loading, Transaction Leaks, and Incorrect Driver Selection
The three most common issues in async CRUD are these: forgetting to await refresh(), triggering MissingGreenlet by accessing lazy-loaded relationships, and exhausting the connection pool because transactions were never closed. These are not fundamentally framework problems. They are boundary-management problems.
AI Visual Analysis
AI Visual Insight: This image is a page-level sharing prompt animation that guides users to open the share entry in the upper-right corner of the blog page. It does not include database schema details, API workflows, ORM mappings, or performance metrics, so it does not directly explain any implementation details of FastAPI or SQLAlchemy.
The Technology Choice Should Be Driven by Throughput Characteristics and Team Maintenance Cost
If your service mostly performs straightforward database reads and writes and concurrency pressure is limited, choose the synchronous approach first because the debugging and operational cost is lower. If the system is I/O-intensive from the beginning and the database driver, cache, and messaging stack all support async, then a fully asynchronous stack is worth the investment.
Do not treat async as a performance silver bullet. What truly determines CRUD quality is disciplined model design, clear transaction boundaries, thoughtful query design, connection pool configuration, and robust error-handling strategy.
FAQ
FAQ 1: Must a FastAPI Project Use Async from Day One?
Not necessarily. If the application mainly performs ordinary database reads and writes and the team is more comfortable with synchronous programming, the sync model is more stable. The event loop advantages of async become obvious only in high-concurrency, I/O-bound scenarios.
FAQ 2: Why Does Async SQLAlchemy Frequently Raise MissingGreenlet?
Because code accesses lazy-loaded relationship attributes inside an async session. The solution is to explicitly preload relationships during the query with selectinload() or joinedload() instead of relying on automatic runtime loading.
FAQ 3: How Can You Prevent the Database Connection Pool from Being Exhausted?
The key is to shorten transaction lifecycles, consistently manage sessions through dependency injection, roll back immediately on exceptions, and add explicit transaction boundaries around write operations. You should also enable pool_pre_ping=True to prevent dead connections from being reused.
Core Summary: This article systematically reconstructs the CRUD implementation path for FastAPI + SQLAlchemy 2.0. It covers synchronous and asynchronous model definitions, engine configuration, transaction management, query optimization, common pitfalls, and coexistence strategies, helping developers make pragmatic trade-offs among performance, complexity, and maintainability.