Databases

Encore treats Postgres databases as logical resources. You declare a database in your code, Encore creates it and applies its migrations, and the SDK gives you an async API to query it.

Creating a database

Declare a database with postgres.Database as a module-level variable in your service, and point it at a directory of migrations:

orders/db.py
from encore.storage import postgres

db = postgres.Database("orders", migrations="./migrations")

orders/migrations/1_create_orders.up.sql
CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    customer TEXT NOT NULL,
    total_cents BIGINT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

With this in place, encore run creates the orders database on a local Postgres cluster and applies the migrations before the service starts. Declaring the database doesn't open a connection: the SDK connects the first time you use it.

Encore reads the declaration statically, so:

• The declaration must be a module-level assignment, not inside a function, class or conditional.
• The name is the only positional argument, and must be a string literal. Database names are unique within the app.
• migrations is keyword-only, and must be a string literal or None. It's a slash-separated path to a subdirectory, relative to the file that declares the database, such as "./migrations". Absolute paths and .. aren't allowed, and the directory must exist.
• Leaving out migrations declares a database without Encore-managed migrations. Use this when something else owns the schema, such as Django's migrations.

Migrations

Encore applies each migration once, in order, and records the latest one it applied in the database's schema_migrations table.

Name each migration file <version>_<description>.up.sql:

• <version> is a number. Migrations run in ascending numeric order, so leading zeros are fine (0001_create_orders.up.sql). Each version can only be used once.
• <description> is a short description of the change.
• Files that don't match the pattern are ignored.

A matching <version>_<description>.down.sql file can revert a migration. Encore never runs down migrations by itself; run them with encore db migrate --database <name> --down.

Changing the schema

To change the schema, add a new migration with a higher version than every existing one. Don't edit a migration that has already been applied: Encore doesn't run it again. A new file with a lower version than the latest applied one isn't applied either.

orders/migrations/2_add_status.up.sql
ALTER TABLE orders ADD COLUMN status TEXT NOT NULL DEFAULT 'pending';

The next encore run applies it, and so does the next deployment.

When a migration fails

Encore runs each migration file as a single batch, so the statements of a failed migration are rolled back together, unless the file manages its own transactions. The database is then marked as dirty at the failed version, and Encore won't migrate it any further until you record the last good version yourself:

# Show the recorded version and whether it's dirty.
encore db version --database orders
# Migration 2 failed: record version 1 as the last applied one, without running anything.
encore db migrate --database orders --set=1

Then fix the migration and run the app again. If the very first migration failed, --set=-1 records that none has been applied.

encore db reset --database orders drops the local database and creates it again from its migrations, discarding its data.

Querying

Use query to read rows and exec to run statements. Parameters are passed positionally and referenced as $1, $2, and so on:

async def create_order(customer: str, total_cents: int) -> int:
    row = await db.query(
        "INSERT INTO orders (customer, total_cents) VALUES ($1, $2) RETURNING id",
        customer,
        total_cents,
    ).one()
    return row["id"]


async def mark_shipped(order_id: int) -> bool:
    result = await db.exec(
        "UPDATE orders SET status = 'shipped' WHERE id = $1", order_id
    )
    return result.rows_affected == 1

Rows are dictionaries keyed by column name. exec returns an ExecResult, whose rows_affected is the number of rows the statement affected; any rows it returns are discarded.

query returns a Cursor, which runs no SQL until you consume it. Consume it once, in one of three ways:

• await cursor.one() returns the single matching row, or None if there is none. It raises EncoreError if the query returns more than one row.
• await cursor.all() returns every row in a list.
• async for row in cursor streams the rows one at a time.

async def recent_orders(customer: str) -> list[dict]:
    return await db.query(
        "SELECT id, total_cents FROM orders WHERE customer = $1 ORDER BY id DESC LIMIT 10",
        customer,
    ).all()


async def total_for(customer: str) -> int:
    total = 0
    async for row in db.query(
        "SELECT total_cents FROM orders WHERE customer = $1", customer
    ):
        total += row["total_cents"]
    return total

Consuming a cursor a second time, or mixing the helpers with iteration, raises EncoreError. one() and all() close the cursor for you, and so does iterating to the end. If you stop iterating early, use the cursor as an async context manager so it closes on exit:

async def first_large_order():
    async with db.query("SELECT * FROM orders ORDER BY total_cents DESC") as rows:
        async for row in rows:
            if row["total_cents"] > 10_000:
                return row
    return None

Types

Parameters can be None, bool, int, float, str, bytes, bytearray, list, tuple and dict. Encore converts each to the type of the column or placeholder it's bound to:

• A str can be bound to text, uuid, numeric, inet, cidr and enum columns, and to dates and times: timestamptz takes an RFC 3339 string such as "2026-01-31T12:00:00Z", date takes "2026-01-31".
• Any value bound to a json or jsonb column is stored as JSON, including dict and list.
• Other Python types, such as datetime or Decimal, aren't supported as parameters yet. Convert them to strings first.

Rows come back as these Python types:

Transactions

begin() starts a transaction. As an async context manager, the transaction commits when the block exits normally, and rolls back if it raises:

async def add_orders(customer: str, totals: list[int]) -> int:
    async with db.begin() as tx:
        for total in totals:
            await tx.exec(
                "INSERT INTO orders (customer, total_cents) VALUES ($1, $2)",
                customer,
                total,
            )
        row = await tx.query(
            "SELECT count(*) AS n FROM orders WHERE customer = $1", customer
        ).one()
    return row["n"]

You can also await begin() and finish the transaction yourself with commit() or rollback(). The transaction can't be used after that, even if committing failed:

tx = await db.begin()
await tx.exec("INSERT INTO orders (customer, total_cents) VALUES ($1, $2)", "ada", 1200)
await tx.commit()

A transaction that is neither committed nor rolled back is rolled back when it's garbage collected. Close any open cursor before you commit or roll back.

Connections

query, exec and begin on the database take a connection from its pool for each call. To run several statements on the same connection, for example while holding a session-level advisory lock, acquire one with acquire(). The connection goes back to the pool when the block exits:

async def process_order(order_id: int) -> None:
    async with db.acquire() as conn:
        await conn.exec("SELECT pg_advisory_lock($1)", order_id)
        try:
            await conn.exec(
                "UPDATE orders SET status = 'processing' WHERE id = $1", order_id
            )
        finally:
            await conn.exec("SELECT pg_advisory_unlock($1)", order_id)

The connection returns to the pool as it is, so undo any session state you change on it.

A connection runs one operation at a time. A streaming cursor borrows its connection or transaction until it's exhausted or closed, so close it before you run the next query on the same connection. conn.begin() starts a transaction on the connection, which stays acquired after the transaction ends. Both acquire() and begin() can also be awaited instead of used with async with; close an awaited connection yourself with await conn.close().

Handles

Encore grants each service access to a database based on how the service's code uses it. Using the declaration directly (db.query(...), db.exec(...), db.begin(), db.acquire()) records read-write access for the service.

To hand the database to other code, such as a helper function or a framework dependency, take a handle first. Passing or aliasing the declaration itself is an error, because Encore couldn't tell what the other code does with it:

Each handle is a DatabaseRef, with the same query, exec, begin and acquire methods as the database. DatabaseRef, Connection and Transaction all implement postgres.Queryable, so a helper can accept any of them. The Database declaration deliberately doesn't, so a type checker catches passing it by mistake:

from encore.storage import postgres

from .db import db


async def find_order(conn: postgres.Queryable, order_id: int):
    return await conn.query("SELECT * FROM orders WHERE id = $1", order_id).one()


async def show_order(order_id: int):
    return await find_order(db.read(), order_id)

With FastAPI, a dependency can return a handle:

from typing import Annotated

from encore.storage import postgres
from fastapi import Depends

from db import db


async def sql_read() -> postgres.DatabaseRef:
    return db.read()


SqlReader = Annotated[postgres.DatabaseRef, Depends(sql_read)]

Using another service's database

By default, each service uses its own database. To use a database declared by another service, reference it by name with Database.named at module scope, instead of importing the other service's code:

reports/db.py
from encore.storage import postgres

orders_db = postgres.Database.named("orders")


async def completed_orders() -> int:
    row = await orders_db.query(
        "SELECT count(*) AS n FROM orders WHERE status = 'shipped'"
    ).one()
    return row["n"]

named takes the database's name as a string literal. The database is configured where it's declared: its migrations stay with the declaring service, which may be written in another language. Encore records how the reports service uses the database, exactly as for its own, and grants it access.

Connecting with other clients and ORMs

db.connection_string is a postgresql:// URL for connecting any Postgres client to the database. It points at a proxy that the Encore runtime runs inside your process, so it's only valid within a process that runs the SDK. Reading it records read-write access for the service.

import psycopg

from db import db


async def count_with_psycopg() -> int:
    async with await psycopg.AsyncConnection.connect(db.connection_string) as conn:
        cursor = await conn.execute("SELECT count(*) FROM orders")
        (count,) = await cursor.fetchone()
        return count

Queries made through the connection string bypass the SDK, so they don't show up in Encore's traces.

Using Django

Django's ORM can use an Encore database through Django's standard PostgreSQL backend and psycopg. Declare the database without migrations, since Django owns the schema, and configure DATABASES from the connection string:

storage/resources.py
from encore.storage import postgres

# Django manages the schema through its own migrations.
db = postgres.Database("django")

config/settings.py
from psycopg.conninfo import conninfo_to_dict

from storage.resources import db

connection = conninfo_to_dict(db.connection_string)
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": connection.pop("dbname"),
        "USER": connection.pop("user", ""),
        "PASSWORD": connection.pop("password", ""),
        "HOST": connection.pop("host", ""),
        "PORT": connection.pop("port", ""),
        "OPTIONS": connection,
    }
}

Encore creates the database but doesn't run Django's migrations. Run Django's management commands through encore exec, so the SDK gets its configuration:

encore exec -- uv run python manage.py makemigrations
encore exec -- uv run python manage.py migrate

Other ORMs and query builders, such as SQLAlchemy, connect the same way: give them the connection string, and manage their migrations with their own tools.

Inserting test and seed data

There are a few ways to put data in a database during development.

In a migration, if every environment should have the data, such as the rows of a lookup table.

With a script, run against your local database with encore exec. The script uses the SDK as your service does. Run it from the service's directory:

orders/seed.py
import asyncio

from db import db


async def main() -> None:
    for customer in ["ada", "grace"]:
        await db.exec(
            "INSERT INTO orders (customer, total_cents) VALUES ($1, $2)",
            customer,
            1000,
        )


asyncio.run(main())

encore exec -- uv run python seed.py

At startup, only locally, by checking the environment with encore.app_meta() in your startup hook. Make the inserts idempotent, since they run on every start:

import encore

from db import db


async def seed_local_data() -> None:
    if encore.app_meta().environment.cloud == "local":
        await db.exec(
            "INSERT INTO orders (id, customer, total_cents) VALUES (1, 'ada', 1000) "
            "ON CONFLICT (id) DO NOTHING"
        )

Tests get their own copy of each database. It starts from the migrated schema, including rows inserted by migrations, but not data from scripts or your local database. See Testing.

Inspecting databases

The Encore CLI connects to your local databases by their Encore name:

• encore db shell --database orders opens a psql shell.
• encore db conn-uri --database orders prints a connection URI for other tools.
• encore db query --database orders "SELECT count(*) FROM orders" runs one query and prints the result.
• encore db proxy runs a local proxy that forwards to all the app's databases.
• encore db list lists the app's databases.

See the CLI reference for all encore db commands.

Errors

Database errors raise EncoreError, a subclass of RuntimeError that you can import from encore:

from encore import EncoreError

from db import db


async def try_create(order_id: int) -> bool:
    try:
        await db.exec(
            "INSERT INTO orders (id, customer, total_cents) VALUES ($1, 'ada', 1000)",
            order_id,
        )
    except EncoreError:
        return False
    return True

EncoreError doesn't expose Postgres error codes yet, so the message is the only way to tell, say, a unique violation from a syntax error.