> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-detect-table-modification.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Use FastAPI and SQLAlchemy with ClickHouse Managed Postgres

> Connect a FastAPI app to ClickHouse Managed Postgres with async SQLAlchemy and asyncpg, run Alembic migrations, and serve CRUD endpoints over verified TLS

export const BetaBadge = ({link, galaxyTrack, galaxyEvent}) => {
  if (link) {
    return <a href={link} target="_blank" rel="noopener noreferrer" className="betaBadge" onClick={galaxyTrack && galaxyEvent ? galaxyOnClick(galaxyEvent) : undefined}>
                <span>Beta</span>
            </a>;
  }
  return <a href="https://clickhouse.com/docs/reference/settings/beta-and-experimental-features#beta-features" className="betaBadge">
            <span>Beta feature</span>
        </a>;
};

<BetaBadge link="https://clickhouse.com/cloud/postgres" galaxyTrack={true} galaxyEvent="docs.managed-postgres.guides-fastapi-beta" />

[FastAPI](https://fastapi.tiangolo.com/) is an async Python web framework, and [SQLAlchemy](https://www.sqlalchemy.org/) is the most widely used Python ORM. In this guide, you define a typed SQLAlchemy model, create its table with an [Alembic](https://alembic.sqlalchemy.org/) migration, and serve create, read, update, and delete endpoints with FastAPI. The app uses the [asyncpg](https://magicstack.github.io/asyncpg/) driver, and every connection uses TLS with full certificate verification.

<h2 id="prerequisites">
  Prerequisites
</h2>

* [uv](https://docs.astral.sh/uv/) and Python 3.11 or later. This guide was tested with Python 3.13, FastAPI 0.142, SQLAlchemy 2.1, asyncpg 0.31, Alembic 1.20, and Uvicorn 0.54.
* A ClickHouse Cloud account
* [`psql`](https://www.postgresql.org/download/), to create the database. You can also run the `CREATE DATABASE` statement in the [SQL console](/integrations/connectors/sql-clients/sql-console).

<h2 id="create-service">
  Create a ClickHouse Managed Postgres service
</h2>

In the ClickHouse Cloud console, click **New service** and select **Postgres**. The instance is ready in a few minutes. See the [quickstart](/products/managed-postgres/quickstart) for a walkthrough.

<h2 id="connection-details">
  Get your connection details
</h2>

Click **Connect** in the left sidebar of your service. The modal shows your username, password, server, and port, and it has a **Directly** / **via PgBouncer** toggle.

This guide uses both connections:

* **PgBouncer (port `6432`)** for the app. SQLAlchemy keeps a pool of connections in every Uvicorn worker, and the bundled [PgBouncer](/products/managed-postgres/connection#pgbouncer) lets many workers and replicas share a small number of Postgres connections. It runs in transaction pooling mode.
* **Direct (port `5432`)** for Alembic. Migrations run DDL, can run for a long time, and may depend on session state, so they should talk to Postgres directly.

Turn on **Use SSL** and click **Download CA certificate**. You can also download it from **Settings → CA Certificate**. The certificate is unique to your instance, so the driver can use it to verify that it's talking to your server.

With **Use SSL** on, the URL in the modal ends in `sslmode=verify-full&sslrootcert=...`. asyncpg doesn't read these parameters, so copy only the password and host from the modal. The app passes the CA certificate in code instead (see [Configure the connection](#configure-connection)).

<h2 id="project-setup">
  Set up the project
</h2>

Create a project and install FastAPI, Uvicorn, SQLAlchemy with its asyncio extra, asyncpg, and Alembic:

```bash theme={null}
uv init --bare fastapi-managed-postgres && cd fastapi-managed-postgres
uv add fastapi uvicorn "sqlalchemy[asyncio]" asyncpg alembic
mkdir app && touch app/__init__.py
```

<Tip>
  **Adding to an existing app?**

  Skip the `uv init` and `mkdir` commands. In your project, run `uv add "sqlalchemy[asyncio]" asyncpg alembic`, then continue with [Configure the connection](#configure-connection). If your app uses SQLite through async SQLAlchemy:

  * Change your `create_async_engine` call to match the one in `app/db.py`. In `migrations/env.py`, import `Base` and your models from your own modules.
  * Remove `Base.metadata.create_all` from startup. Alembic creates your existing tables in Postgres, but it doesn't copy data from SQLite.
</Tip>

<h2 id="configure-connection">
  Configure the connection
</h2>

Move the CA certificate you downloaded into the project directory and rename it to `ca-certificate.pem`.

Create a database for the app. Replace `<PASSWORD>` and the host with the values from the **Connect** modal:

```bash theme={null}
psql "postgresql://postgres:<PASSWORD>@your-instance.pg.clickhouse.cloud:5432/postgres?sslmode=verify-full&sslrootcert=ca-certificate.pem" -c "CREATE DATABASE guide_fastapi;"
```

```text theme={null}
CREATE DATABASE
```

Create a `.env` file in the project root, or add these variables to your existing one. The two URLs differ only in the port, use the `postgresql+asyncpg` scheme, and have no SSL parameters:

```bash title=".env" theme={null}
# The app connects through PgBouncer (port 6432)
DATABASE_URL="postgresql+asyncpg://postgres:<PASSWORD>@your-instance.pg.clickhouse.cloud:6432/guide_fastapi"
# Alembic migrations connect directly to Postgres (port 5432)
DIRECT_DATABASE_URL="postgresql+asyncpg://postgres:<PASSWORD>@your-instance.pg.clickhouse.cloud:5432/guide_fastapi"
DATABASE_CA_CERT="ca-certificate.pem"
```

`DATABASE_CA_CERT` is resolved relative to the directory you run commands from. If your password contains special characters such as `@`, `/`, or `#`, URL-encode them. Keep `.env` out of version control.

Create `app/db.py`. It builds the TLS settings, the engine, and a session factory that the rest of the app shares:

```python title="app/db.py" theme={null}
import os
import ssl

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase


def ssl_context() -> ssl.SSLContext:
    # Verifies the certificate chain against your instance's CA and checks the hostname
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.load_verify_locations(os.environ["DATABASE_CA_CERT"])
    return context


# The app uses the PgBouncer connection (port 6432)
engine = create_async_engine(
    os.environ["DATABASE_URL"],
    connect_args={"ssl": ssl_context()},
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass
```

An `ssl.SSLContext` created with `PROTOCOL_TLS_CLIENT` requires a valid certificate and checks the hostname, which is the equivalent of `sslmode=verify-full`. SQLAlchemy passes it to asyncpg through `connect_args`.

<Warning>
  **Pass the CA certificate in `connect_args`**

  * Without an `ssl` argument, asyncpg uses TLS if the server offers it, but it doesn't verify the certificate.
  * Don't copy `sslmode` or `sslrootcert` from the Connect modal into the URL. SQLAlchemy passes URL parameters to asyncpg as keyword arguments, and the connection fails with `TypeError: connect() got an unexpected keyword argument 'sslmode'`.
  * On Python 3.13 and later, a context from `ssl.create_default_context` fails with `certificate verify failed: Missing Authority Key Identifier`, because that function enables strict X.509 checks. Create the context with `ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)` as shown.
  * If the CA certificate is wrong, the connection fails with `certificate verify failed: unable to get local issuer certificate`. If the hostname doesn't match, for example when you connect by IP address, it fails with `certificate verify failed: IP address mismatch`.
</Warning>

<Note>
  If you'd rather keep the TLS settings in the URL, or your app uses synchronous SQLAlchemy sessions, use the [psycopg](https://www.psycopg.org/psycopg3/) driver instead of asyncpg. It's built on libpq and accepts the parameters from the Connect modal, for example `postgresql+psycopg://...:5432/guide_fastapi?sslmode=verify-full&sslrootcert=ca-certificate.pem`, with both `create_engine` and `create_async_engine`.
</Note>

<h2 id="model">
  Define the model
</h2>

Create `app/models.py` with a typed SQLAlchemy model:

```python title="app/models.py" theme={null}
from datetime import datetime

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

from app.db import Base


class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    done: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
```

<h2 id="migrations">
  Run migrations
</h2>

Create an Alembic environment from the async template:

```bash theme={null}
uv run alembic init -t async migrations
```

Replace the contents of the generated `migrations/env.py`. The new version loads your models for autogenerate and connects directly to Postgres with the same TLS settings as the app:

```python title="migrations/env.py" theme={null}
import asyncio
import os
from logging.config import fileConfig

from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import create_async_engine

import app.models  # noqa: F401 (registers the models on Base.metadata)
from app.db import Base, ssl_context

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_migrations() -> None:
    # Migrations use the direct connection (port 5432)
    engine = create_async_engine(
        os.environ["DIRECT_DATABASE_URL"],
        poolclass=pool.NullPool,
        connect_args={"ssl": ssl_context()},
    )
    async with engine.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await engine.dispose()


asyncio.run(run_migrations())
```

Alembic ignores the placeholder `sqlalchemy.url` in `alembic.ini`, because `env.py` builds the engine itself.

Generate a migration from the model, then apply it:

```bash theme={null}
uv run --env-file .env alembic revision --autogenerate -m "create todos"
uv run --env-file .env alembic upgrade head
```

```text theme={null}
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.plugins] setting up autogenerate plugin alembic.autogenerate.schemas
INFO  [alembic.runtime.plugins] setting up autogenerate plugin alembic.autogenerate.tables
INFO  [alembic.runtime.plugins] setting up autogenerate plugin alembic.autogenerate.types
INFO  [alembic.runtime.plugins] setting up autogenerate plugin alembic.autogenerate.constraints
INFO  [alembic.runtime.plugins] setting up autogenerate plugin alembic.autogenerate.defaults
INFO  [alembic.runtime.plugins] setting up autogenerate plugin alembic.autogenerate.comments
INFO  [alembic.autogenerate.compare.tables] Detected added table 'todos'
Generating /path/to/fastapi-managed-postgres/migrations/versions/9b8d29f05361_create_todos.py ...  done
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 9b8d29f05361, create todos
```

Review the generated file in `migrations/versions` and commit it with your code. Alembic records the applied revision in the `alembic_version` table. Each time you change a model, run `revision --autogenerate` and `upgrade head` again.

<h2 id="endpoints">
  Add the API endpoints
</h2>

Create `app/main.py`. Each request gets its own `AsyncSession` through a FastAPI dependency, and the engine's pool is closed when the app shuts down:

```python title="app/main.py" theme={null}
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import SessionLocal, engine
from app.models import Todo


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session


Session = Annotated[AsyncSession, Depends(get_session)]


class TodoCreate(BaseModel):
    title: str


class TodoUpdate(BaseModel):
    title: str | None = None
    done: bool | None = None


class TodoOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    done: bool
    created_at: datetime


async def get_todo_or_404(session: AsyncSession, todo_id: int) -> Todo:
    todo = await session.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo


@app.get("/todos", response_model=list[TodoOut])
async def list_todos(session: Session):
    result = await session.scalars(select(Todo).order_by(Todo.id))
    return result.all()


@app.post("/todos", response_model=TodoOut, status_code=201)
async def create_todo(body: TodoCreate, session: Session):
    todo = Todo(title=body.title)
    session.add(todo)
    await session.commit()
    return todo


@app.get("/todos/{todo_id}", response_model=TodoOut)
async def read_todo(todo_id: int, session: Session):
    return await get_todo_or_404(session, todo_id)


@app.patch("/todos/{todo_id}", response_model=TodoOut)
async def update_todo(todo_id: int, body: TodoUpdate, session: Session):
    todo = await get_todo_or_404(session, todo_id)
    for field, value in body.model_dump(exclude_unset=True).items():
        setattr(todo, field, value)
    await session.commit()
    return todo


@app.delete("/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: int, session: Session) -> None:
    await session.delete(await get_todo_or_404(session, todo_id))
    await session.commit()
```

SQLAlchemy runs the statements of each session in one transaction, which ends at `session.commit`. When the session closes, any uncommitted work is rolled back. On `INSERT`, SQLAlchemy reads the generated `id` and `created_at` back with `RETURNING`, so `create_todo` doesn't need an extra query.

<h2 id="verify">
  Run and verify
</h2>

Start the app:

```bash theme={null}
uv run --env-file .env uvicorn app.main:app --port 3112
```

```text theme={null}
INFO:     Started server process [86235]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:3112 (Press CTRL+C to quit)
```

In a second terminal, create two todos, mark the first one done, and delete the second:

```bash theme={null}
curl -X POST localhost:3112/todos -H 'content-type: application/json' -d '{"title": "Try FastAPI"}'
curl -X POST localhost:3112/todos -H 'content-type: application/json' -d '{"title": "Write a migration"}'
curl -X PATCH localhost:3112/todos/1 -H 'content-type: application/json' -d '{"done": true}'
curl -X DELETE localhost:3112/todos/2
```

```text theme={null}
{"id":1,"title":"Try FastAPI","done":false,"created_at":"2026-10-01T14:20:39.315984Z"}
{"id":2,"title":"Write a migration","done":false,"created_at":"2026-10-01T14:20:39.520672Z"}
{"id":1,"title":"Try FastAPI","done":true,"created_at":"2026-10-01T14:20:39.315984Z"}
```

The `DELETE` request returns `204 No Content`. Requesting a todo that doesn't exist returns `404`:

```bash theme={null}
curl -i localhost:3112/todos/2
```

```text theme={null}
HTTP/1.1 404 Not Found
date: Thu, 01 Oct 2026 14:20:39 GMT
server: uvicorn
content-length: 27
content-type: application/json

{"detail":"Todo not found"}
```

Open [http://localhost:3112/todos](http://localhost:3112/todos) in your browser to list the remaining todos, or [http://localhost:3112/docs](http://localhost:3112/docs) to try the endpoints in FastAPI's interactive API docs.

To confirm that the direct connection uses verified TLS, query `pg_stat_ssl` with the same TLS settings as the app:

```bash theme={null}
uv run --env-file .env python -c "
import asyncio, os
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from app.db import ssl_context

async def main():
    engine = create_async_engine(os.environ['DIRECT_DATABASE_URL'], connect_args={'ssl': ssl_context()})
    async with engine.connect() as conn:
        result = await conn.execute(text('SELECT ssl, version FROM pg_stat_ssl WHERE pid = pg_backend_pid()'))
        print(result.one())
    await engine.dispose()

asyncio.run(main())
"
```

```text theme={null}
(True, 'TLSv1.3')
```

To see the data in the console, open **SQL console** in the left sidebar of your service, expand `guide_fastapi` and then `public`, and click the `todos` table. The `alembic_version` table next to it holds the migration history. The `todos` table contains the remaining row:

| id | title | done | created\_at |
| - | - | - | - |
| 1 | Try FastAPI | true | 2026-10-01 14:20:39.315984+00 |

Stop the app with <kbd>Ctrl</kbd>+<kbd>C</kbd>.

<h2 id="pgbouncer">
  Using PgBouncer
</h2>

The app works through PgBouncer with the default asyncpg and SQLAlchemy settings:

* asyncpg runs every query as a named, protocol-level prepared statement, and SQLAlchemy caches up to 100 of them per connection. The bundled PgBouncer supports these statements in transaction pooling mode, and asyncpg frees them with a protocol-level `Close` message rather than SQL `DEALLOCATE`. You don't need `statement_cache_size=0` or a custom `prepared_statement_name_func`. If you turn off asyncpg's statement cache anyway, also add `prepared_statement_cache_size=0` to the URL to turn off SQLAlchemy's cache.
* Session settings made with `SET`, such as `search_path`, don't carry over to the next transaction and can leak to other clients. Use `SET LOCAL` inside a transaction instead.
* `pg_stat_ssl` reports the connection between PgBouncer and Postgres, so the TLS check above returns `(False, None)` through port `6432`. The app's connection to PgBouncer is still verified: a wrong CA certificate fails in the same way.
* Alembic doesn't take advisory locks and runs each `upgrade` in one transaction, so it also works through PgBouncer. The direct connection is still the better choice for long migrations.

<Warning>
  **Changing a column's type**

  PgBouncer keeps the prepared statements on its Postgres connections. After a migration changes the type of a column that a query returns, such as `integer` to `bigint`, that query fails through PgBouncer with `InvalidCachedStatementError: cached statement plan is invalid due to a database schema or configuration change`, even from new connections, until PgBouncer replaces its Postgres connections. To recover, close PgBouncer's Postgres connections for the database over the direct connection, as the same user the app connects as. PgBouncer connects to Postgres locally, so its connections have no `client_addr`; the query leaves direct connections alone. PgBouncer opens new connections, and each pooled app connection may fail one more time before SQLAlchemy refreshes its cache:

  ```sql theme={null}
  SELECT pg_terminate_backend(pid) FROM pg_stat_activity
  WHERE datname = 'guide_fastapi' AND usename = current_user
    AND client_addr IS NULL AND backend_type = 'client backend'
    AND pid <> pg_backend_pid();
  ```

  Adding a column doesn't cause this, because SQLAlchemy lists the columns it selects.
</Warning>

<h2 id="next-steps">
  Next steps
</h2>

* [Connection](/products/managed-postgres/connection): connection strings, PgBouncer, and TLS
* [Settings](/products/managed-postgres/settings): Postgres and PgBouncer parameters, such as `max_connections`
* [Read replicas](/products/managed-postgres/read-replicas): send read-only queries to a replica with a second engine
* [Security](/products/managed-postgres/security): IP access lists and private networking