The FastAPI–MySQL Stack That Survives Past the Demo

When writing a FastAPI backend, connecting to a database isn't just about writing a single connection string.

A MySQL integration that can be maintained long-term usually needs to solve these problems simultaneously:

Which Python driver actually connects to MySQL
How the ORM maps Python classes to database tables
How each HTTP request gets an independent database session
How write operations commit transactions and roll back on failure
As the project grows, where to place database connections, models, routes, and business logic

Let's look at the complete flow first.

This diagram can be broken down into two main lines:

Application Startup
  -> Read DATABASE_URL
  -> Create Engine
  -> Create SessionLocal
  -> Optional: Create tables, write initialization data

API Request
  -> FastAPI Router receives request
  -> Depends injects database Session
  -> Service handles business rules
  -> Repository / ORM executes database reads/writes
  -> commit or rollback
  -> Close Session after request ends

FastAPI is not directly responsible for connecting to MySQL. It handles HTTP, parameter validation, dependency injection, and response returns.

The components that actually interact with MySQL are:

FastAPI
  -> SQLAlchemy
  -> PyMySQL
  -> MySQL

1. Clarifying the Collaboration Between Them

If you only remember one relationship, remember this chain:

User initiates HTTP request
  -> FastAPI finds the corresponding route function
  -> Depends calls get_db to create a Session
  -> Route function passes Session to Service
  -> Service handles business rules and transactions
  -> Repository uses Session to write SQLAlchemy queries
  -> SQLAlchemy translates ORM queries into SQL
  -> PyMySQL sends SQL to MySQL
  -> MySQL executes SQL and returns results
  -> SQLAlchemy converts results into Python objects
  -> FastAPI converts response objects into JSON and returns to frontend

In this relationship, each layer has a clear division of responsibilities.

FastAPI
  -> Manages HTTP, not underlying database connections

Depends / get_db
  -> Manages the database Session lifecycle for each request

SQLAlchemy Engine
  -> Manages database entry point and connection pool

SQLAlchemy Session
  -> Manages queries, writes, commits, and rollbacks within a single business operation

SQLAlchemy ORM Model
  -> Manages mapping between Python classes and database tables

PyMySQL
  -> Manages low-level communication between the Python process and the MySQL service

MySQL
  -> Actually stores and queries data

2. Required Third-Party Dependencies

A synchronous FastAPI + MySQL project needs at least these dependencies:

uv add "fastapi[standard]" sqlalchemy pymysql pydantic-settings

If using pip:

pip install "fastapi[standard]" sqlalchemy pymysql pydantic-settings

These packages handle different levels of responsibility.

1. FastAPI

FastAPI is the web framework, responsible for:

Defining HTTP interfaces
Parsing path parameters, query parameters, and request bodies
Using Pydantic for data validation
Injecting dependencies via Depends
Automatically generating OpenAPI documentation

Let's clarify two terms first.

Routing refers to "which function should handle a given HTTP request." For example, GET /users/1 is handled by the get_user function.

Dependency Injection means "whatever objects the route function needs, FastAPI prepares them before calling the function." A database Session is well-suited for dependency injection because each request needs an independent Session, and the Session must be closed after the request ends.

The way a database connection enters FastAPI is usually not by manually creating a connection inside the route function, but through dependency injection:

from typing import Annotated

from fastapi import Depends
from sqlalchemy.orm import Session


# SessionDep is a type alias:
# It tells FastAPI that whenever a route parameter is annotated as SessionDep,
# execute get_db() first and pass the resulting database Session in.
SessionDep = Annotated[Session, Depends(get_db)]

This way, the route only declares "I need a database Session," while the creation and closure of that Session are handled by a unified dependency function.

2. SQLAlchemy

SQLAlchemy is the database toolkit and ORM.

It is responsible for:

Creating the database connection entry point Engine
Managing the connection pool
Creating request-level Sessions
Defining ORM models
Generating SQL
Executing CRUD operations
Managing transaction commits and rollbacks

ORM stands for Object Relational Mapping.

It solves this problem:

In Python code, you operate on objects.
In the database, you operate on tables and rows.
The ORM handles the conversion between the two.

For example, there is a User class in Python and a users table in the database. The ORM maps User.username to the users.username field. This allows you to write Python objects and query expressions instead of concatenating SQL strings throughout your business logic.

When using SQLAlchemy, business code typically does not directly concatenate SQL strings but operates on Python classes:

# select(User) means query the User model corresponding to the users table
# where(...) means append query conditions
# scalar(...) means fetch only one ORM object result
user = db.scalar(select(User).where(User.username == username))

This line of code will ultimately be translated into SQL by SQLAlchemy and passed to the underlying database driver for execution.

3. PyMySQL

PyMySQL is the Python driver for MySQL.

It is the low-level library truly responsible for communicating with the MySQL service. SQLAlchemy itself is not a MySQL driver; it needs a DBAPI driver to connect to a specific database.

A driver can be understood as a database "adapter." Different databases speak differently; MySQL, PostgreSQL, and SQLite each have their own protocols and specifics. SQLAlchemy handles generating SQL and managing the ORM, but actually opening a network connection, logging into MySQL, sending SQL, and reading results requires a driver.

DBAPI is a set of interface specifications that database drivers in Python follow. PyMySQL implements this specification, so SQLAlchemy can connect to MySQL through it.

When connecting to MySQL, the SQLAlchemy URL is generally written as:

mysql+pymysql://username:password@host:port/database_name?charset=utf8mb4

The pymysql part here tells SQLAlchemy:

Database type is MySQL
Underlying driver uses PyMySQL

If the MySQL account uses an authentication method requiring RSA, you can install:

uv add "pymysql[rsa]"

Or:

pip install "PyMySQL[rsa]"

For most local development, just pymysql is sufficient.

4. pydantic-settings

pydantic-settings is used to read environment variables and .env configurations.

The database connection string, whether to print SQL, and whether to auto-create tables should not be hardcoded but placed in environment variables.

Typical .env:

DATABASE_URL="mysql+pymysql://root:[email protected]:3306/todo_api?charset=utf8mb4"
DATABASE_ECHO=false
DATABASE_AUTO_CREATE=false

Reading configuration:

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # Tell pydantic-settings to read configuration from the .env file
    # env_file_encoding helps avoid encoding issues with Chinese or special characters
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # Required config: the application should fail to start if DATABASE_URL is missing
    database_url: str

    # Whether to print SQL. Can be set to true for debugging, usually kept false during normal development
    database_echo: bool = False

    # Whether to auto-create tables on startup. Can be true for local demos, generally false for formal projects
    database_auto_create: bool = False


# Create a global configuration object. Other modules read config via settings.database_url
settings = Settings()

5. Alembic

Alembic is not a required dependency for connecting to MySQL, but formal projects usually need it.

It handles database migrations:

uv add alembic

Why is a migration tool needed?

Because after a project goes live, table structure changes cannot rely on casually running Base.metadata.create_all(). Production environments need migration files that are trackable, reversible, and reviewable.

Remember this distinction first:

Local learning / early demos
  -> can use create_all()

Team projects / production environments
  -> should use Alembic to manage migrations

3. How to Write the Connection String

The complete format for a MySQL connection string:

mysql+pymysql://username:password@host:port/database_name?charset=utf8mb4

Example:

DATABASE_URL="mysql+pymysql://root:[email protected]:3306/todo_api?charset=utf8mb4"

Breaking it down:

mysql+pymysql://root:[email protected]:3306/todo_api?charset=utf8mb4
                |    |        |         |    |        |
                user password host     port db_name  charset

A few points need special attention.

First, it's recommended to explicitly write mysql+pymysql://, not just mysql://.

mysql+pymysql:// is more explicit: MySQL dialect plus PyMySQL driver. In team projects, don't rely on implicit defaults, otherwise you'll easily run into "it works on my machine but not on others" issues when switching environments.

Second, the recommended charset is utf8mb4.

utf8mb4 fully supports Unicode, including emoji. There's no need for new projects to use MySQL's older utf8.

Third, if the password contains special characters, URL encoding is needed.

For example, if the password contains @, #, /, :, the connection string might be parsed incorrectly. A safer approach is to avoid these characters in passwords, or to URL-encode them when generating the connection string.

4. Creating the Engine

Engine is SQLAlchemy's database entry point.

It can be understood as:

Engine = Database connection entry point + Connection pool

The connection pool here refers to a set of reusable database connections.

If every request reconnected to MySQL and then disconnected, the cost would be high. The connection pool maintains some connections in advance; when a request comes in, it borrows one, and returns it after use. This allows the backend service to handle a large number of requests more stably.

Engine itself is not a specific query, nor a connection for a specific request. It's more like an application-level database entry point, typically created once at application startup and then reused by the entire application.

Basic implementation:

from sqlalchemy import create_engine

from app.core.config import settings


engine = create_engine(
    # SQLAlchemy uses this URL to determine database type, driver, username, password, host, and database name
    settings.database_url,

    # echo=True prints SQL to the console, very useful for troubleshooting
    echo=settings.database_echo,

    # Check if the connection is still alive before fetching it from the pool
    # MySQL might actively disconnect after long idle periods; this parameter reduces disconnection errors
    pool_pre_ping=True,
)

Common parameters:

echo=True: Print SQL, suitable for debugging
echo=False: Don't print SQL, suitable for normal development
pool_pre_ping=True: Check if the connection is usable before fetching it from the pool, reducing issues caused by MySQL disconnecting idle connections

The MySQL server might close connections that have been idle for too long. If the backend service continues to use this invalidated connection, it will fail when executing SQL. pool_pre_ping=True allows SQLAlchemy to perform a lightweight check before using the connection, reconnecting if the connection is found to be unusable.

5. Creating SessionLocal

Session is the ORM's operational context.

It can be understood as:

Session = A database read/write context + Transaction boundary

The name Session is easily misunderstood. It's not the Session from browser login state, nor is it a user session.

In SQLAlchemy, Session represents a database operation context: it knows which objects you've queried, which objects you've modified, which objects are ready to be added, and whether to finally commit the transaction.

SessionLocal is a "Session factory." It's not a specific Session, but a callable object used to create Sessions. Each incoming request creates a new Session through it.

Creating a Session factory:

from sqlalchemy.orm import sessionmaker


SessionLocal = sessionmaker(
    # Sessions created by this factory all use the previously created engine
    bind=engine,

    # Don't auto-flush before queries, reducing implicit database write timing
    autoflush=False,

    # Don't auto-commit transactions. Must manually db.commit() after write operations
    autocommit=False,

    # Object fields remain readable after commit, won't expire immediately
    expire_on_commit=False,
)

Parameter meanings:

bind=engine: Which Engine this Session factory uses
autoflush=False: Don't automatically flush changes before queries, reducing implicit behavior for beginners
autocommit=False: Don't auto-commit transactions; write operations must explicitly commit
expire_on_commit=False: Object fields can still be read directly after commit

Usage:

from sqlalchemy import select


# with handles closing the Session after the code block ends
# Closing a Session doesn't mean closing the MySQL service, but returning the connection to the pool
with SessionLocal() as db:
    # select(User) generates a SQLAlchemy query object for querying the users table
    # scalars(...) means return ORM objects, not raw row objects
    users = db.scalars(select(User)).all()

When with exits, it closes the Session and returns the connection to the connection pool.

6. Defining ORM Models

ORM models use Python classes to describe database tables.

A model refers to "the representation of a database table in Python code." Each row of data in the table, when read out, can become a Python object. The object's fields correspond to the database table's columns.

First, define a unified Base:

from sqlalchemy.orm import DeclarativeBase


# All ORM models inherit from Base
# SQLAlchemy collects all table structures through Base.metadata
class Base(DeclarativeBase):
    pass

Then define business models:

from datetime import datetime

from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class User(Base):
    # __tablename__ specifies which database table this model corresponds to
    __tablename__ = "users"

    # primary_key=True indicates primary key
    # autoincrement=True means MySQL automatically generates incrementing IDs
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # unique=True means duplicates are not allowed at the database level
    # nullable=False means this column cannot be null
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(191), unique=True, nullable=False)

    # default=1 is a Python-side default. If status is not passed when creating a User, it defaults to 1
    status: Mapped[int] = mapped_column(Integer, nullable=False, default=1)

    # server_default=func.now() means the database server generates the default time
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

This code represents:

User class
  -> users table

User.username
  -> users.username field

User.email
  -> users.email field

Mapped[...] and mapped_column(...) are the declarative style recommended in SQLAlchemy 2.x. Type annotations not only help the editor but also help the ORM understand field types more clearly.

7. Executing CRUD Operations

1. Query

Query all users:

from sqlalchemy import select
from sqlalchemy.orm import Session


def list_users(db: Session) -> list[User]:
    return list(db.scalars(select(User)).all())

Query by primary key:

from sqlalchemy.orm import Session


def get_user(db: Session, user_id: int) -> User | None:
    return db.get(User, user_id)

Query by condition:

from sqlalchemy import select
from sqlalchemy.orm import Session


def get_user_by_username(db: Session, username: str) -> User | None:
    return db.scalar(select(User).where(User.username == username))

Paginated query:

from sqlalchemy import func, select
from sqlalchemy.orm import Session


def page_users(db: Session, page: int, page_size: int) -> dict[str, object]:
    # Page 1 skips 0 records, page 2 skips page_size records
    offset = (page - 1) * page_size

    # First query the total count, to show the frontend how many records there are
    total = db.scalar(select(func.count()).select_from(User)) or 0

    # Then query the current page's data
    items = db.scalars(
        select(User)
        .order_by(User.id.desc())
        .offset(offset)
        .limit(page_size)
    ).all()

    return {
        "items": list(items),
        "total": total,
        "page": page,
        "page_size": page_size,
    }

2. Create

from sqlalchemy.orm import Session


def create_user(db: Session, username: str, email: str) -> User:
    # This only creates the ORM object in Python memory, not yet written to the database
    user = User(
        username=username,
        email=email,
        status=1,
    )

    # Add to the current Session, indicating this record is ready to be inserted
    db.add(user)

    # The INSERT is only actually executed after committing the transaction
    db.commit()

    # Refresh the object to get database-generated fields like id, created_at
    db.refresh(user)

    return user

Key points:

db.add(user): Adds the object to the current Session
db.commit(): Commits the transaction, actually writing to the database
db.refresh(user): Refreshes the object from the database, getting auto-incremented IDs, default values, etc.

3. Update

from sqlalchemy.orm import Session


def update_user_email(db: Session, user_id: int, email: str) -> User | None:
    # db.get is suitable for querying by primary key
    user = db.get(User, user_id)
    if user is None:
        return None

    # After modifying ORM object fields, SQLAlchemy records this change
    user.email = email

    # SQLAlchemy generates an UPDATE statement on commit
    db.commit()
    db.refresh(user)

    return user

As long as the object was queried from the current Session, modifying a field and executing commit() will cause SQLAlchemy to generate the corresponding UPDATE.

4. Delete

from sqlalchemy.orm import Session


def delete_user(db: Session, user_id: int) -> bool:
    user = db.get(User, user_id)
    if user is None:
        return False

    # Mark this ORM object for deletion
    db.delete(user)

    # SQLAlchemy generates a DELETE statement on commit
    db.commit()

    return True

When deleting data, consider foreign key constraints, related tables, business rules, and whether physical deletion is allowed. In real projects, data like users, orders, and payment records are usually not casually physically deleted.

8. How to Handle Transactions

Database write operations must have clear transaction boundaries.

The most common approach is:

from sqlalchemy.orm import Session


def create_user_with_role(db: Session, user: User, role: Role) -> User:
    try:
        # Place two create actions in the same transaction
        db.add(user)
        db.add(role)

        # Only commit the transaction if both objects are written successfully
        db.commit()
        db.refresh(user)
        return user
    except Exception:
        # If any step fails, undo all database changes in this transaction
        db.rollback()
        raise

The transaction rule is simple:

All succeed
  -> commit

Fail midway
  -> rollback

You can also use the begin() context:

# begin automatically manages the transaction:
# Normal exit auto-commits, exception auto-rollbacks
with SessionLocal.begin() as db:
    db.add(user)
    db.add(role)

begin() auto-commits on normal exit and auto-rollbacks on exception.

In business interfaces, it's more common to pass the Session to a Service or Repository, then commit after a clear business action is completed. This reads more clearly and makes it easier to handle business exceptions.

9. Injecting Database Session in FastAPI

FastAPI recommends using dependency injection to manage request-level resources.

The dependency function for a database Session can be written like this:

from collections.abc import Generator

from sqlalchemy.orm import Session

from app.db.session import SessionLocal


def get_db() -> Generator[Session, None, None]:
    # Create a new database Session for each incoming request
    db = SessionLocal()
    try:
        # yield passes db to the route function for use
        # After the route function finishes executing, code continues to finally
        yield db
    finally:
        # Close the Session after the request ends, returning the connection to the pool
        db.close()

The behavior of this code is:

Request enters
  -> Create a Session

Route function executes
  -> Use this Session to query or write to the database

Request ends
  -> finally closes the Session

Usage in routes:

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from app.dependencies.database import get_db
from app.modules.users.service import UserService


# prefix means all interfaces under this router start with /users
# tags is used to group OpenAPI documentation
router = APIRouter(prefix="/users", tags=["users"])

# Declare a reusable database dependency type
# Later, route parameters written as db: SessionDep will automatically get the Session provided by get_db
SessionDep = Annotated[Session, Depends(get_db)]


@router.get("/{user_id}")
def get_user(user_id: int, db: SessionDep) -> dict[str, object]:
    # The route layer doesn't directly write database queries, but calls the Service
    user = UserService.get_user(db, user_id)
    if user is None:
        # Service returns None, route layer converts it to HTTP 404
        raise HTTPException(status_code=404, detail="User not found")

    return {
        "id": user.id,
        "username": user.username,
        "email": user.email,
    }

Here, the route layer only does three things:

Receives HTTP parameters
Calls business logic
Returns HTTP response

It doesn't handle creating database connections, nor does it directly pile up many SQLAlchemy queries.

10. Building a Maintainable Project Structure from Scratch

When the project is small, a single main.py can run it.

As the project grows, it's not advisable to cram configuration, database connections, models, routes, and business logic together. A more stable structure is:

todo-api/
├── app/
│   ├── main.py
│   ├── api/
│   │   └── v1/
│   │       └── router.py
│   ├── core/
│   │   └── config.py
│   ├── db/
│   │   ├── base.py
│   │   ├── models.py
│   │   └── session.py
│   ├── dependencies/
│   │   └── database.py
│   └── modules/
│       └── users/
│           ├── models.py
│           ├── schemas.py
│           ├── repository.py
│           ├── service.py
│           └── router.py
├── migrations/
├── tests/
├── .env
└── pyproject.toml

The responsibilities of each layer must be clear.

app/core/config.py
  -> Reads environment variables

app/db/session.py
  -> Creates Engine, SessionLocal, init_database

app/db/base.py
  -> Defines SQLAlchemy Base

app/db/models.py
  -> Unified import of all ORM models, convenient for create_all or Alembic to discover models

app/dependencies/database.py
  -> Defines get_db

app/modules/users/models.py
  -> User table ORM model

app/modules/users/schemas.py
  -> Request and response bodies for user interfaces

app/modules/users/repository.py
  -> User-related database reads/writes

app/modules/users/service.py
  -> User business rules

app/modules/users/router.py
  -> User HTTP interfaces

The core principle of this structure is:

Common infrastructure goes in app/core, app/db, app/dependencies
Business code goes in app/modules/{module_name}

11. How to Write the Core Files

1. app/core/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # Specify the .env file as the local configuration source
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # Application name, displayed in the API documentation
    app_name: str = "Todo API"

    # MySQL connection string, e.g., mysql+pymysql://root:[email protected]:3306/todo_api
    database_url: str

    # Whether to print SQL
    database_echo: bool = False

    # Whether to auto-create tables on startup via create_all
    database_auto_create: bool = False


# Global configuration object. Other modules only read from it, no manual env var parsing in business code
settings = Settings()

2. app/db/base.py

from sqlalchemy.orm import DeclarativeBase


# Base class inherited by all ORM models
# Later, models like User, Todo, Role will all be registered in Base.metadata
class Base(DeclarativeBase):
    pass

3. app/modules/users/models.py

from datetime import datetime

from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class User(Base):
    # users is the actual table name in the database
    __tablename__ = "users"

    # id is the primary key, auto-generated by MySQL
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # username and email both require uniqueness and cannot be null
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(191), unique=True, nullable=False)

    # status can be used to represent business states like enabled, disabled
    status: Mapped[int] = mapped_column(Integer, nullable=False, default=1)

    # created_at is written by the database with the current time
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

4. app/db/models.py

from app.modules.users.models import User

# Unified export of models, convenient for one-time import elsewhere
__all__ = ["User"]

This file looks simple but is very important.

SQLAlchemy only knows which tables exist in Base.metadata after the model classes are imported by Python. Centrally importing models avoids the problem of "clearly wrote the model, but create_all or Alembic can't find the table."

5. app/db/session.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.config import settings
from app.db.base import Base
from app.db import models


engine = create_engine(
    # Read connection string from config, avoiding hardcoding credentials in code
    settings.database_url,

    # Control whether to print SQL
    echo=settings.database_echo,

    # Prevent errors on the first query of a request after MySQL idle connection expires
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(
    # Sessions created by SessionLocal are all bound to this engine
    bind=engine,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False,
)


def init_database() -> None:
    # Only auto-create tables when explicitly enabled
    # Formal projects usually use Alembic, not relying on create_all
    if settings.database_auto_create:
        Base.metadata.create_all(bind=engine)

from app.db import models is not to directly use the models variable in code, but to ensure all ORM models are imported, allowing Base.metadata to collect the table structures.

6. app/dependencies/database.py

from collections.abc import Generator

from sqlalchemy.orm import Session

from app.db.session import SessionLocal


def get_db() -> Generator[Session, None, None]:
    # Create a Session for the current request
    db = SessionLocal()
    try:
        # Pass the Session to the FastAPI route function
        yield db
    finally:
        # Whether the interface succeeds or fails, finally close the Session
        db.close()

7. app/modules/users/schemas.py

from datetime import datetime

from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    # When creating a user, the frontend must pass username and email
    username: str
    email: EmailStr


class UserRead(BaseModel):
    # User fields returned to the frontend
    id: int
    username: str
    email: EmailStr
    status: int
    created_at: datetime

If using EmailStr, you need to install the email validation dependency:

uv add "pydantic[email]"

Or:

pip install "pydantic[email]"

If you don't want to add this dependency, you can first change EmailStr to a regular str.

8. app/modules/users/repository.py

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.modules.users.models import User


class UserRepository:
    @staticmethod
    def get_by_id(db: Session, user_id: int) -> User | None:
        # Repository only cares about how the database queries
        return db.get(User, user_id)

    @staticmethod
    def get_by_username(db: Session, username: str) -> User | None:
        # Query a single user by unique username
        return db.scalar(select(User).where(User.username == username))

    @staticmethod
    def create(db: Session, username: str, email: str) -> User:
        # Here only add, not commit
        # commit is handled uniformly by the Service after the complete business action ends
        user = User(username=username, email=email)
        db.add(user)
        return user

Repository only handles database reads/writes, doesn't write HTTP exceptions, and doesn't care how the interface returns.

9. app/modules/users/service.py

from sqlalchemy.orm import Session

from app.modules.users.models import User
from app.modules.users.repository import UserRepository
from app.modules.users.schemas import UserCreate


class UserService:
    @staticmethod
    def get_user(db: Session, user_id: int) -> User | None:
        # Service can directly reuse Repository
        return UserRepository.get_by_id(db, user_id)

    @staticmethod
    def create_user(db: Session, data: UserCreate) -> User:
        # Business rule: username cannot be duplicated
        exists = UserRepository.get_by_username(db, data.username)
        if exists is not None:
            raise ValueError("Username already exists")

        try:
            # First execute the database create action
            user = UserRepository.create(
                db,
                username=data.username,
                email=data.email,
            )

            # Commit the transaction only after the complete business action succeeds
            db.commit()
            db.refresh(user)
            return user
        except Exception:
            # If an error occurs before or during commit, rollback this transaction
            db.rollback()
            raise

Service handles business rules and transaction boundaries.

For example, "username cannot be duplicated" is a business rule and should be in the Service. commit and rollback are also better placed at the point where a business action is completed.

10. app/modules/users/router.py

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.dependencies.database import get_db
from app.modules.users.schemas import UserCreate, UserRead
from app.modules.users.service import UserService


# This file only defines the HTTP interfaces for the users module
router = APIRouter(prefix="/users", tags=["users"])
SessionDep = Annotated[Session, Depends(get_db)]


@router.get("/{user_id}", response_model=UserRead)
def get_user(user_id: int, db: SessionDep) -> UserRead:
    # 1. Route layer receives user_id
    # 2. Passes db and user_id to Service
    user = UserService.get_user(db, user_id)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    # ORM object converted to response model, then converted to JSON by FastAPI
    return UserRead.model_validate(user, from_attributes=True)


@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
def create_user(data: UserCreate, db: SessionDep) -> UserRead:
    try:
        # data has already been validated by Pydantic, handed to Service for business processing
        user = UserService.create_user(db, data)
    except ValueError as exc:
        # Business exceptions converted to HTTP status codes at the route layer
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(exc),
        ) from exc

    return UserRead.model_validate(user, from_attributes=True)

The route layer should be thin.

It should care about HTTP semantics, such as:

What path
What method
What status code
What is the request body
What is the response body
What HTTP error do business exceptions translate to

It should not be stuffed with database queries.

11. app/api/v1/router.py

from fastapi import APIRouter

from app.modules.users.router import router as users_router


api_router = APIRouter(prefix="/api/v1")

# Mount the users module's interfaces under /api/v1
api_router.include_router(users_router)

12. app/main.py

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api.v1.router import api_router
from app.core.config import settings
from app.db.session import init_database


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Execute initialization logic once on application startup
    init_database()

    # Before yield is startup logic, after yield you can write shutdown logic
    yield


app = FastAPI(
    title=settings.app_name,
    # lifespan manages logic for application startup and shutdown phases
    lifespan=lifespan,
)


# Register the main router
app.include_router(api_router)

Startup:

uv run fastapi dev app/main.py

You can also use Uvicorn:

uv run uvicorn app.main:app --reload

12. How a Create User Request Flows Through the Entire Process

Now let's string the previous files together and look at a real request:

POST /api/v1/users
Content-Type: application/json

{
  "username": "copyer",
  "email": "[email protected]"
}

Its execution order in the backend is as follows.

1. app/main.py
   FastAPI app has already include_router(api_router)

2. app/api/v1/router.py
   /api/v1 prefix matches, main router continues to pass the request to users_router

3. app/modules/users/router.py
   POST /users matches the create_user route function

4. FastAPI discovers the route parameter has db: SessionDep
   So it first executes get_db()

5. app/dependencies/database.py
   get_db() creates a new database Session via SessionLocal()

6. router.py
   create_user(data, db) begins execution
   data is the Pydantic-validated UserCreate
   db is the SQLAlchemy Session provided by get_db()

7. service.py
   UserService.create_user(db, data) executes business rules
   First checks if username is duplicated

8. repository.py
   UserRepository.get_by_username(db, data.username)
   Uses SQLAlchemy to query the User table

9. SQLAlchemy
   Translates select(User).where(...) into SELECT SQL

10. PyMySQL
    Sends the SELECT SQL to MySQL

11. MySQL
    Executes the query, returns whether this user already exists

12. service.py
    If username doesn't exist, calls Repository to create the User object

13. repository.py
    db.add(user)
    Marks user as ready to be inserted, but not yet actually written to the database

14. service.py
    db.commit()
    Commits the transaction, SQLAlchemy generates INSERT SQL

15. PyMySQL
    Sends the INSERT SQL to MySQL

16. MySQL
    Writes to the users table, generates auto-increment id and default time

17. service.py
    db.refresh(user)
    Refreshes the newly generated database fields back into the Python object

18. router.py
    Converts the ORM object to UserRead

19. FastAPI
    Converts UserRead to JSON response

20. get_db()
    Request ends, finally executes db.close()
    Session closed, connection returned to the connection pool

In this chain, the most important thing is not to mix responsibilities.

Router
  -> Only handles HTTP input/output

Service
  -> Handles business rules and transactions

Repository
  -> Handles database read/write details

Session
  -> Records the current database operations and handles commit / rollback

Engine
  -> Provides database connection entry point and connection pool

PyMySQL
  -> Actually communicates with MySQL

If creating a user fails, the flow becomes:

Error occurs during Service execution
  -> except catches the exception
  -> db.rollback()
  -> Undoes the changes already prepared for writing in this transaction
  -> raise continues to throw the exception
  -> Router or global exception handler converts the exception to an HTTP response
  -> get_db finally closes the Session

So commit, rollback, and close are three different actions:

commit
  -> Confirms writing to the database

rollback
  -> Undoes the current transaction

close
  -> Closes the current Session, returns the connection to the connection pool

Don't use close as a substitute for rollback, and don't assume data is already written to the database after add. The actual database write happens at commit.

13. Synchronous or Asynchronous

This article uses synchronous SQLAlchemy + PyMySQL.

That is to say:

SQLAlchemy create_engine
  -> PyMySQL
  -> Synchronous database calls

This approach is suitable for most beginner projects and general backend systems; the code is simple, there are plenty of resources, and problems are easy to troubleshoot.

If you want a fully asynchronous chain, the dependencies become a different set:

SQLAlchemy create_async_engine
  -> asyncmy or aiomysql
  -> AsyncSession
  -> async def routes

Don't mix synchronous Sessions with asynchronous Engines, and don't forcefully change the database layer to asynchronous just because FastAPI supports async def. Writing the synchronous version clearly first is more important than piling on asynchronous concepts from the start.

14. Common Mistakes

1. Forgetting to Commit the Transaction

Only writing:

db.add(user)

Data will not actually be written to the database. Write operations need to execute:

db.commit()

If you also need to get the auto-increment ID or database default values, then execute:

db.refresh(user)

2. Not Rolling Back After an Exception

After a write operation fails, you should:

db.rollback()

Otherwise, the current Session might be in a failed state, and subsequent use will cause errors.

3. Creating an Engine Directly in a Route

Don't write:

@router.get("/users")
def list_users():
    engine = create_engine(...)

Engine should be created once at application startup, reusing the connection pool. Only create and close Sessions per request.

4. Hardcoding Database Passwords in Code

Don't hardcode the connection string in Python files.

A better way:

.env
  -> Local development configuration

Production environment variables
  -> Production configuration

.env is usually not committed to Git.

5. Models Not Being Imported

If tables are not generated after calling Base.metadata.create_all(bind=engine), a common reason is that the model classes haven't been imported.

The solution is to create a unified model import file:

from app.modules.users.models import User

__all__ = ["User"]

And import it before initializing the database.

Comments

Top 1 from juejin.cn, machine-translated. The original thread is authoritative.

爬虫技术实验室

About FastAPI connecting to databases, I also started by writing the synchronous version following the docs. Later, when building a proxy gateway, I discovered a big pitfall 😂 The routing layer used async, but the database operations still used synchronous SQLAlchemy sessions, meaning every DB query was blocking the event loop. After switching everything to async session + aiomysql, with the same concurrency, the response time dropped by almost half. But the dependency injection pattern mentioned in the article is indeed the most comfortable — each request gets its own independent session, automatically closed when the request ends, much cleaner than manually managing sessions back in my Flask days.