How to Connect to MySQL in FastAPI: Simple Guide

To connect FastAPI to MySQL, use SQLAlchemy with an async driver like asyncmy or aiomysql. Configure the database URL, create an async engine, and use AsyncSession to interact with the database asynchronously within your FastAPI app.

Syntax

Here is the basic syntax to connect FastAPI to MySQL using SQLAlchemy with async support:

• DATABASE_URL: The connection string to your MySQL database.
• create_async_engine: Creates an async engine for database communication.
• AsyncSession: Manages async database sessions.
• sessionmaker: Factory to create sessions.
• Depends: FastAPI dependency injection to provide sessions to routes.

python

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends

DATABASE_URL = "mysql+asyncmy://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)

async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

app = FastAPI()

Example

This example shows a FastAPI app connecting to MySQL asynchronously, creating a simple user table, and adding a user.

python

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String

DATABASE_URL = "mysql+asyncmy://user:password@localhost/testdb"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

app = FastAPI()

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post("/users/")
async def create_user(name: str, session: AsyncSession = Depends(get_session)):
    new_user = User(name=name)
    session.add(new_user)
    await session.commit()
    await session.refresh(new_user)
    return {"id": new_user.id, "name": new_user.name}

Output

POST /users/ with JSON {"name": "Alice"} returns {"id": 1, "name": "Alice"}

Common Pitfalls

• Using synchronous MySQL drivers like mysqlclient or pymysql with async code causes blocking and errors.
• For async support, use drivers like asyncmy or aiomysql with the correct SQLAlchemy URL prefix mysql+asyncmy:// or mysql+aiomysql://.
• Not using AsyncSession and mixing sync sessions leads to runtime errors.
• Forgetting to await async calls like session.commit() causes unexpected behavior.

python

## Wrong way (sync driver with async code):
# DATABASE_URL = "mysql+pymysql://user:password@localhost/db"
# engine = create_async_engine(DATABASE_URL)  # This will fail

## Right way (async driver):
# DATABASE_URL = "mysql+asyncmy://user:password@localhost/db"
# engine = create_async_engine(DATABASE_URL)

## Wrong session usage:
# session = Session(engine)  # sync session in async code

## Right session usage:
# async_session = sessionmaker(engine, class_=AsyncSession)
# async with async_session() as session:
#     await session.commit()

Quick Reference

Summary tips for connecting FastAPI to MySQL:

• Use asyncmy or aiomysql for async MySQL drivers.
• Set DATABASE_URL with prefix mysql+asyncmy:// or mysql+aiomysql://.
• Create async engine with create_async_engine.
• Use AsyncSession from sqlalchemy.ext.asyncio.
• Always await async database calls.
• Use FastAPI Depends to inject sessions into routes.

Key Takeaways

Use async MySQL drivers like asyncmy or aiomysql with SQLAlchemy for FastAPI.

Create an async engine and AsyncSession to handle database operations asynchronously.

Always await async calls like session.commit() to avoid errors.

Inject database sessions into FastAPI routes using Depends for clean code.

Avoid mixing synchronous drivers or sessions in async FastAPI apps.