1. The Layered Backend Architecture Blueprint
In production environments, cramming database queries, data validation, and business logic into route handlers leads to an unmaintainable codebase. I enforce a strict three-tier architecture: Routers (HTTP parsing, query params, status codes), Services (domain business logic, transactional boundaries, third-party integrations), and Repositories (database CRUD operations, query optimization).
This separation ensures that database queries are reusable across multiple endpoints, and unit tests can easily mock out the database tier to test complex business logic without spinning up external Docker databases.
By strictly separating HTTP transport concerns from business rules, you can reuse domain services in background Celery workers, CLI commands, or automated cron scripts without refactoring a single line of business logic.
app/
├── api/
│ ├── v1/
│ │ ├── endpoints/ # Routers: auth.py, orders.py, products.py
│ │ └── api_router.py
├── core/
│ ├── config.py # Pydantic BaseSettings with environment vars
│ ├── database.py # Async engine & sessionmaker
│ └── security.py # Passlib, JWT encoding/decoding
├── models/ # SQLAlchemy 2.0 Declarative ORM entities
├── schemas/ # Pydantic v2 Request/Response validation models
├── repositories/ # Database query execution classes
└── services/ # Business rule orchestration & transactions2. Asynchronous PostgreSQL with SQLAlchemy 2.0 and Asyncpg
SQLAlchemy 2.0 fundamentally modernized Python ORM patterns by unifying select() statements and introducing first-class asynchronous drivers. We pair SQLAlchemy with Asyncpg—the fastest PostgreSQL client library available for Python.
Proper connection pool configuration is crucial to prevent connection exhaustion under heavy mobile app usage. In my configurations, I specify pool_size=20 and max_overflow=10 with a pool_recycle of 1800 seconds to prevent stale connections from being terminated by cloud load balancers.
Using pool_pre_ping=True guarantees that before any query executes, the engine verifies that the underlying TCP connection is healthy, eliminating sporadic 500 Internal Server Errors caused by cloud network timeouts.
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.config import settings
engine = create_async_engine(
settings.POSTGRES_ASYNC_URI,
pool_size=20,
max_overflow=10,
pool_recycle=1800,
pool_pre_ping=True,
echo=False,
)
AsyncSessionLocal = async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
async def get_db_session() -> AsyncSession:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise3. Bulletproof Migrations with Alembic
Never allow ORM methods like Base.metadata.create_all() in production. Hardcoding table creation bypasses migration tracking, making schema rollbacks and blue-green deployments impossible.
Alembic tracks every schema modification in revision files checked into Git. By configuring alembic/env.py to import your SQLAlchemy Base metadata, running 'alembic revision --autogenerate -m create_users_table' automatically detects changes and generates deterministic migration scripts that execute safely during CI/CD deployments.
This allows multiple engineers to collaborate seamlessly across Git branches without encountering database schema conflicts, and enables instant rollbacks if a migration exhibits performance degradation in production.
4. Global Error Handling and Consistent Response Envelopes
Mobile applications need consistent error responses to display meaningful feedback to users. If one endpoint returns a string error and another returns an object with 'detail', client-side parsing breaks.
I implement a custom global exception handler that intercepts all HTTPException and RequestValidationError events, wrapping them in a standardized JSON error envelope containing an error code, user-friendly message, and field-specific validation errors.
This unified structure allows the mobile Flutter app to parse any backend failure into a clean UI dialog or toast notification with zero edge-case crashes.
5. Pagination, Filtering, and Performance Indexing
Never expose unpaginated list endpoints. An endpoint that returns 10,000 product records in a single query will consume all server memory and cause the mobile client to freeze. Always implement cursor-based or limit-offset pagination.
On the PostgreSQL side, add compound B-tree indexes for frequently filtered columns (e.g., status, created_at, user_id). Indexing transforms full-table scans of 500ms into sub-2ms index lookups, keeping mobile response times instantaneous even as databases grow into gigabytes.
For infinite-scroll mobile feeds, cursor-based pagination using encoded base64 timestamps outperforms traditional offset-limit pagination. Offsets become increasingly slow on deep pages (e.g., OFFSET 50000 requires PostgreSQL to scan 50,000 rows before returning 20), whereas cursor queries use indexed WHERE clauses that execute in constant time: WHERE created_at < :cursor ORDER BY created_at DESC LIMIT 20.
Final Thoughts
Building scalable REST APIs with FastAPI, Pydantic v2, and Async SQLAlchemy provides an exceptional balance of raw execution speed, strict type safety, and clean architecture. When you follow these production principles, your backend can scale effortlessly to millions of requests.
Key Takeaways
- Enforce a 3-tier architecture (Routers, Services, Repositories) for maintainability.
- Pair SQLAlchemy 2.0 with Asyncpg and configure connection pools with pool_pre_ping.
- Manage all database schema changes strictly via version-controlled Alembic migrations.
- Implement unified exception handlers to guarantee consistent JSON error envelopes for mobile apps.
- Add compound B-tree indexes on PostgreSQL tables to ensure sub-millisecond query execution.
Frequently Asked Questions
Why use Asyncpg instead of Psycopg2 with FastAPI?
Psycopg2 is synchronous and blocks Python's asyncio event loop during queries, crippling FastAPI's concurrency. Asyncpg is completely non-blocking and achieves up to 3x higher throughput on PostgreSQL benchmarks.
How do you handle N+1 query problems in Async SQLAlchemy?
Always use selectinload() or joinedload() from sqlalchemy.orm to eagerly fetch related entities in a single SQL query, preventing catastrophic N+1 query storms during list endpoints.
Should database sessions be committed in the repository or service layer?
Commit at the service layer or use a session dependency manager. Repositories should only stage changes (session.add()), allowing service methods to coordinate multiple repository actions within a single atomic transaction.
How do you manage database connection pooling in multi-worker environments?
Remember that connection pool limits apply per Uvicorn worker process. If pool_size=20 and you run 4 workers, your maximum active connections to PostgreSQL will be 80. Ensure PostgreSQL max_connections in postgresql.conf is sized appropriately.
Why is cursor pagination superior to offset-limit for mobile apps?
Cursor pagination eliminates the 'page drift' problem where new items inserted at the top of a feed shift rows and cause duplicate items to appear on subsequent infinite-scroll pages in the mobile app.
