FastAPI_

An Appwrite native MySQL database is a standard MySQL engine, so FastAPI with SQLAlchemy 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 page and use the SQLAlchemy ORM, the FastAPI dependency system, and Alembic the same way you would against any self-hosted MySQL server.

Install dependencies

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

Set the connection string

Fetch the connection string with the API. 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:

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

This guide uses the 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:

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 page for TLS and network access controls.

Define a model

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:

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:

alembic init -t async migrations

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

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, and let max_overflow absorb short bursts:

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) 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 on port 6033 instead, and avoid keeping another application pool on top of it:

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

And in the engine setup:

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 page for the trade-offs.

Use a branch for previews and CI

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