---
layout: article
title: FastAPI
description: Use FastAPI and SQLAlchemy 2.x with an Appwrite native MySQL database. Configure the async asyncmy engine, pass TLS through connect_args, inject a session per request, and run Alembic migrations against the direct database port.
---

An Appwrite native MySQL database is a standard MySQL engine, so [FastAPI](https://fastapi.tiangolo.com/) with [SQLAlchemy](https://docs.sqlalchemy.org/) and an async MySQL driver works against it with no Appwrite-specific runtime code. You point `create_async_engine` at the connection string from the [connections](/docs/products/databases/mysql/connections) page and use the SQLAlchemy ORM, the FastAPI dependency system, and Alembic the same way you would against any self-hosted MySQL server.

**Before you start**

You'll need a native MySQL database in a `ready` state and its credentials. See [native MySQL databases](/docs/products/databases/mysql) to create one and [connections](/docs/products/databases/mysql/connections) to retrieve the connection string. The primary user is `admin`, and the database name is generated for each database.

# Install dependencies

```bash
pip install "fastapi[standard]" "sqlalchemy[asyncio]" asyncmy alembic
```

# Set the connection string

Fetch the connection string with the [API](/docs/products/databases/mysql/connections#credentials). Put it in your environment, never commit it. SQLAlchemy's asyncmy dialect uses the `mysql+asyncmy://` scheme, so swap the leading `mysql://` for it and add `charset=utf8mb4`:

```env
DATABASE_URL="mysql+asyncmy://admin:<password>@db-<hash>.<region>.appwrite.center:3306/<database>?charset=utf8mb4"
DATABASE_SSL=true
```

This guide uses the [asyncmy](https://github.com/long2ice/asyncmy) driver. SQLAlchemy also supports the async `mysql+aiomysql://` dialect, but the snippets below use `mysql+asyncmy://`.

# Create the async engine

Cloud connections require TLS. Build an `ssl.SSLContext` and pass it through `connect_args`; use `DATABASE_SSL=false` only for a local MySQL server that does not offer TLS:

```python
import ssl

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

connect_args = (
    {"ssl": ssl.create_default_context()}
    if settings.database_ssl
    else {}
)

engine = create_async_engine(
    settings.database_url,
    connect_args=connect_args,
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
```

The server certificate is signed by a well-known public CA, so `ssl.create_default_context()` validates the certificate chain against the system trust store. See the [Network](/docs/products/databases/mysql/network-security) page for TLS and network access controls.

# Define a model

```python
from datetime import datetime

from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "fastapi_users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
```

SQLAlchemy renders the integer primary key as `AUTO_INCREMENT` for MySQL. The explicit `String(255)` keeps the unique email index within MySQL's indexed column limits.

# Inject a session per request

FastAPI's dependency system gives each request its own `AsyncSession` and closes it when the request finishes. Define a dependency that yields a session, then annotate path operations with it:

```python
from typing import Annotated

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_session():
    async with SessionLocal() as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

@app.post("/users")
async def create_user(email: str, session: SessionDep):
    user = User(email=email)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

@app.get("/users")
async def list_users(session: SessionDep):
    result = await session.scalars(select(User).order_by(User.created_at.desc()))
    return result.all()
```

Create the engine once at module scope and reuse it for the whole process, the pool lives inside it. Don't open a new engine per request.

# Run migrations

Generate Alembic's async scaffold, which opens the connection asynchronously and hands a sync connection to the migration context:

```bash
alembic init -t async migrations
```

Point `target_metadata` at `Base.metadata` in `migrations/env.py`, then autogenerate and apply:

```bash
alembic revision --autogenerate -m "init"
alembic upgrade head
```

Review the generated migration before applying it. If the database contains tables that Alembic should not manage, configure Alembic's include filters or run the migration against a branch so autogenerate only emits changes for your FastAPI app.

Run migrations against the **direct** engine port (`3306`), not the pooler. DDL needs a session-level connection, and Alembic's async template already uses a `NullPool`, so a fresh connection is opened and closed per run. The primary `admin` user owns the default database and can run schema changes.

# Pool sizing

A long-running `uvicorn` server holds a SQLAlchemy pool for its lifetime, so connect to the **direct** engine port (`3306`) with a sized pool. Keep `pool_size` multiplied by the number of server processes under the connection limit of your [specification](/docs/products/databases/mysql#specifications), and let `max_overflow` absorb short bursts:

```python
engine = create_async_engine(
    settings.database_url,
    connect_args=connect_args,
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)
```

When the same app runs on a serverless platform (for example an Appwrite [function](/docs/products/functions)) where each invocation is a fresh instance, that can fan out into more backend connections than the engine allows. Route runtime traffic through the [connection pooler](/docs/products/databases/mysql/connection-pooling) on port `6033` instead, and avoid keeping another application pool on top of it:

```env
DATABASE_URL="mysql+asyncmy://admin:<password>@db-<hash>.<region>.appwrite.center:6033/<database>?charset=utf8mb4"
DATABASE_SSL=true
```

And in the engine setup:

```python
from sqlalchemy import NullPool

engine = create_async_engine(
    settings.database_url,
    connect_args=connect_args,
    poolclass=NullPool,
    pool_pre_ping=True,
)
```

The pooler defaults to **transaction mode**, which returns the backend connection to the pool after each transaction. Use **session mode** if your application relies on session-level state such as user variables, temporary tables, or server-side prepared statements, see the [pooler](/docs/products/databases/mysql/connection-pooling#modes) page for the trade-offs.

# Use a branch for previews and CI

[Branches](/docs/products/databases/mysql/branches) are instant, isolated copies of a database with their own hostname and connection string. They're ideal for running migrations against throwaway data in a pull-request preview or an integration-test job:

1. Create a branch from the API and read its `connectionString`.
2. Rewrite the scheme to `mysql+asyncmy://`, add `charset=utf8mb4`, and export it as `DATABASE_URL`.
3. Run `alembic upgrade head` against the branch's direct port, then your test suite.
4. Delete the branch when the job finishes.

Because a branch starts from a storage snapshot, the schema and data match the source database at branch time, so migrations run against representative data without touching production.

# Related

- [Connections](/docs/products/databases/mysql/connections): Retrieve credentials and rotate the primary password.
- [Connection pooler](/docs/products/databases/mysql/connection-pooling): Pool modes, ports, and read/write splitting for serverless workloads.
- [Branches](/docs/products/databases/mysql/branches): Ephemeral database copies for preview environments and CI.
- [Network](/docs/products/databases/mysql/network-security): TLS modes, certificate verification, mTLS, and IP allowlists.