# Databases > Declaring, migrating and querying Postgres 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: ```python -- orders/db.py -- from encore.storage import postgres db = postgres.Database("orders", migrations="./migrations") ``` ```sql -- 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](#using-django). ## 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 `_.up.sql`: - `` 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. - `` is a short description of the change. - Files that don't match the pattern are ignored. A matching `_.down.sql` file can revert a migration. Encore never runs down migrations by itself; run them with `encore db migrate --database --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. ```sql -- 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. > **Important** > > During a deployment, code from the previous release may still be running against the migrated schema. Keep migrations compatible with that code: add a column before the code uses it, and remove it only after no running code reads it. ### 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: ```shell # 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: ```python 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. ```python 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: ```python 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: | Postgres type | Python type | |---|---| | `boolean` | `bool` | | `smallint`, `integer`, `bigint` | `int` | | `real`, `double precision` | `float` | | `text`, `varchar`, enums | `str` | | `bytea` | `bytes` | | `json`, `jsonb` | `dict`, `list` or another JSON value | | `numeric` | `str` | | `timestamp`, `timestamptz` | `str`, in RFC 3339 format | | `date`, `time` | `str` | | `uuid`, `inet`, `cidr` | `str` | | arrays | `list` | | `NULL` | `None` | ## Transactions `begin()` starts a transaction. As an async context manager, the transaction commits when the block exits normally, and rolls back if it raises: ```python 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: ```python 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: ```python 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: | Method | Access recorded for the service | |---|---| | `db.read()` | Read | | `db.write()` | Read and write | | `db.admin()` | Read, write and admin | 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: ```python 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: ```python 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)] ``` > **Note** > > For now, the access level controls the permissions Encore records, not what the handle can do: all handles share the database's connection pool and credentials, and a `read()` handle can still write. `read(primary=True)` is accepted, but reads aren't routed to replicas yet, so it has no effect. ## 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: ```python -- 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. ```python 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: ```python -- storage/resources.py -- from encore.storage import postgres # Django manages the schema through its own migrations. db = postgres.Database("django") ``` ```python -- 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: ```shell 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: ```python -- 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()) ``` ```shell encore exec -- uv run python seed.py ``` **At startup, only locally**, by checking the environment with [`encore.app_meta()`](/docs/py/develop/metadata) in your startup hook. Make the inserts idempotent, since they run on every start: ```python 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](/docs/py/develop/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](/docs/cli/cli-reference) for all `encore db` commands. ## Errors Database errors raise `EncoreError`, a subclass of `RuntimeError` that you can import from `encore`: ```python 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.