Encore SDK for Python

The Python SDK for Encore — infrastructure resources only.

Declare infrastructure (SQL databases, pub/sub topics & subscriptions, object storage buckets, Redis databases, cron jobs, metrics, secrets) as module-level variables. Encore discovers these declarations, provisions the resources, and connects your application to them at runtime.

Services are declared by the language-agnostic encore.service.json file, not in Python. Use the SDK with your preferred Python web framework.

Installation

Requires Python 3.9 or later. Install from PyPI with pip:

python -m pip install encore-dev

Or add it to a uv project:

uv add encore-dev

Declare resources

from encore import pubsub, secrets
from encore.storage import postgres

db = postgres.Database("orders", migrations="./migrations")
events = pubsub.Topic("order-events", exactly_once=True)
api_key = secrets.Secret[str]("ApiKey")


async def place_order(order):
    await db.exec("INSERT INTO orders (id) VALUES ($1)", order.id)
    await events.publish({"id": order.id})

Service URLs

Use base_url to look up a service's configured base URL:

from encore.services import base_url

order = f"{base_url('orders')}/orders/123"

The returned string has no trailing slash: a call's path goes after it. Unknown service names raise ValueError. Lookup reads the runtime configuration and makes no network request.

Looking up a service's URL makes the calling service depend on it, which Encore reads from the code: pass the name as a string literal, call base_url directly rather than passing it around, and call it in a service.

Postgres queries and handles

Declare postgres.Database("orders", migrations="./migrations") at module scope. migrations is optional and keyword-only. Use Database.named("orders") to reference a database declared elsewhere.

query(sql, *params) returns a Cursor. Use async for to stream rows, await query.one() to fetch exactly one row (raising EncoreError on no rows or more than one), await query.first() to fetch the first row or None, or await query.all() to collect all rows in a list. Each query can be consumed once. Reusing it or mixing helpers with iteration raises EncoreError, including after partial iteration or an error. Helpers close their cursors automatically. For an early exit from streaming, use async with db.query(...) as query:. Entering the context does not execute SQL. Exiting it closes the query.

exec(sql, *params) returns an ExecResult with rows_affected. Rows are dictionaries keyed by column name. Values are bound with $1, $2, etc.

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


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

DatabaseRef, Connection, and Transaction implement the nominal Queryable interface. Database exposes the same query methods but deliberately does not implement Queryable. Encore rejects passing or aliasing the declaration; select read(), write(), or admin() before handing it to another function. Direct database operations infer writer permissions. Reference creation records reader, writer, or admin permissions at that call site.

Handles share the database's pool. A read() handle runs on read-only connections, so its writes fail with SQLSTATE 25006 and its transactions are read-only; it reads from the database's read-only pool on a replica if the runtime config has one, unless primary=True is set.

async def rename_order(order_id: int, name: str):
    async with db.acquire() as conn:
        async with conn.begin() as tx:
            result = await tx.exec(
                "UPDATE orders SET name = $1 WHERE id = $2", name, order_id
            )
        return result.rows_affected

Both begin() and acquire() can be awaited or used as async context managers. Transaction contexts commit on a clean exit and roll back on exceptions; connection contexts return the connection to its pool. A connection cannot be used or closed while its transaction is active. Closed handles reject queries.

Streaming queries borrow their connection or transaction until exhausted or closed. If you stop early, close the stream explicitly before reusing its owner:

async def first_order():
    async with db.acquire() as conn:
        async with conn.query("SELECT * FROM orders ORDER BY id") as rows:
            async for row in rows:
                return row

Pub/Sub payloads

Use pubsub.Subscription[T] to choose how payloads are decoded. Encore validates the type argument when subscriptions start. Omitting it selects ordinary JSON decoding. Handler annotations guide type checking but do not select the decoder.

from pydantic import BaseModel


class Order(BaseModel):
    id: str


async def process_order(message: pubsub.Message[Order]) -> None:
    print(message.data.id)


processed = pubsub.Subscription[Order](events, "process-order", handler=process_order)

Supported payloads are JSON values, bytes, and top-level Pydantic v2 models (when Pydantic is installed). pubsub.JSON and pubsub.Payload describe these public types. Use Subscription[pubsub.JSON] for plain JSON: decoding returns ordinary Python values without validating or constructing nested types. Use a Pydantic model when fields need validation or nested models need construction.

Subscription[bytes] returns the payload unchanged. Unparameterized Subscription decodes ordinary JSON. Dataclasses, model unions, models nested in plain JSON containers, and string forward references are not supported; wrap nested models in a top-level Pydantic model and pass that class directly. Original bytes remain available as message.raw_data. Payloads are decoded once before the handler is called; decoding errors propagate for normal retry handling.

Application-owned shutdown

Without a hook, Encore shuts down non-cooperatively as the interpreter exits: after the server has drained, it drains Pub/Sub and flushes telemetry. On import it makes SIGTERM exit the interpreter, unless a handler is already set or the process runs tests: with 0 once the shutdown completed in time, else 1.

Use cooperative shutdown when application tasks may still use Encore resources. The following hook accepts a framework-specific callback that stops new application work and waits for existing work to finish:

from typing import Awaitable, Callable

from encore import shutdown


async def on_shutdown(
    stop_and_drain_application: Callable[[], Awaitable[None]],
) -> int:
    shutdown.begin(cooperative=True)

    await stop_and_drain_application()
    shutdown.mark_application_drained()

    result = await shutdown.completed()
    return result.code()

Connect this hook to the host's shutdown lifecycle. Keep its event loop and the Encore runtime alive until cleanup completes. The host should enforce its own overall shutdown deadline and decide when to terminate the process.

Call begin() before awaiting phase helpers. With cooperative shutdown, signal application draining before awaiting cleanup() or completed(); otherwise Encore waits for that signal until its deadline expires.

If application work has already drained, use non-cooperative shutdown:

from encore import shutdown


async def stop_encore() -> int:
    shutdown.begin()  # cooperative=False
    result = await shutdown.completed()
    return result.code()

begin() returns immediately. All callers share the same shutdown process, and the first call chooses the cooperative mode. Repeated calls do not restart cleanup, change the mode, or reset the deadline. Cancelling an individual waiter does not cancel background cleanup.

Phases and observation

Phase	Meaning
NOT_STARTED	Shutdown has not started.
STARTED	Pub/Sub intake has been signalled to stop. Encore is waiting for message handlers and, in cooperative mode, application draining.
CLEANUP	Encore has stopped waiting for draining and is performing resource cleanup.
COMPLETED	The cleanup task has finished, including if it failed. Check the result or exception.

phase() reads the current phase synchronously. await started() and await cleanup() wait until shutdown has reached or passed their respective phases. They return immediately if the phase has already passed.

await completed() waits for actual cleanup completion, including after the deadline. These waits are independent: callers do not need to await every phase in sequence. Reading phase() is valid before begin(); waiting for a phase before begin() raises ShutdownError.

CLEANUP closes SQL, Redis, and object-storage resources concurrently, then exports final metrics and flushes tracing. Retaining a database, cache client, bucket, or object handle does not keep its connections usable after cleanup. New operations fail; interrupted object downloads report an error rather than returning a partial body as success, and Redis subscriptions end.

Drain application work before signalling mark_application_drained(). Forced cleanup cancels Encore-owned resource I/O even if application tasks are still running. External clients created from connection URLs are owned by your application. Cloud SDKs release their HTTP pools when their clients are dropped; best-effort S3 multipart abort requests may outlive completed() and require a running executor to finish.

Deadline and forced cleanup

The core deadline defaults to five seconds, configured by deployment.graceful_shutdown.total in the runtime configuration. It starts when shutdown is first begun and covers both draining and cleanup.

When the deadline expires:

1. Encore records Timeout in the shared state.
2. If draining is still in progress, Encore stops waiting for it and enters CLEANUP, even without an application-drained signal.
3. Cleanup continues. completed() returns only after the cleanup task finishes.

The timeout is sticky: finishing later does not change it to success. It is not a hard upper bound on how long completed() can wait. The controller does not exit the process when its deadline expires.

To skip draining before the deadline, request forced cleanup:

from encore import shutdown


async def force_remaining_cleanup() -> int:
    # Shutdown must already have been started with begin().
    shutdown.force_cleanup()
    await shutdown.cleanup()  # Observe the transition; cleanup is not done yet.
    result = await shutdown.completed()
    return result.code()

force_cleanup() returns immediately and is idempotent. It raises ShutdownError before begin(). Once resource cleanup has started, forcing it again has no effect. Forcing alone does not record Timeout: successful cleanup before the deadline still returns SUCCESS.

Both the deadline and explicit forcing abandon the drain wait; they do not kill active handlers. Those handlers may continue using resources while cleanup runs. The host must account for this when choosing when to force cleanup.

Results and errors

Python outcome	Meaning
ShutdownResult.SUCCESS	Cleanup finished before the deadline. Explicit forcing may have skipped draining. code() returns 0.
ShutdownResult.TIMEOUT	The deadline expired before cleanup finished. The returned result still means the cleanup task has now finished. code() returns 1.
ShutdownError	The core reported cleanup failure or progress could not be observed, including waiting before begin().

The SDK maps native Failed to ShutdownError; it does not return the public FAILED enum member from completed(). Native progress errors are translated to ShutdownError with the original EncoreError retained as their cause. An unrecognized native enum value raises ValueError.

If cleanup panics after the deadline has already recorded Timeout, the core logs the failure and preserves Timeout as the reported result. A timeout is not evidence that all cleanup operations eventually succeeded.

Resource initialization

Importing the SDK synchronously initializes the Encore runtime. Run applications through Encore so runtime configuration is available before imports. Initialization failures surface during import, before normal application startup completes.

Buckets, SQL databases, pub/sub topics, and secrets store their declarations and resolve native handles on first use, caching the results with thread-safe initialization. These lookups are synchronous against the already initialized runtime. Importing application modules requires the configured runtime and native extension. Use encore test to run application tests with Encore-managed resources.

Initialize independently in each worker process. Forking an already initialized native runtime is not supported by this startup model.

Metrics

Metrics are OpenTelemetry instruments of the meter encore, declared at module scope in three scopes (Counter, SharedCounter, GlobalCounter, and so on for UpDownCounter, Gauge and Histogram), each with a …Group form whose labels are a TypedDict or a dataclass:

from dataclasses import dataclass

from encore import metrics


@dataclass(frozen=True)
class Labels:
    route: str


latency = metrics.HistogramGroup[Labels]("request_duration", unit="s")

latency.with_labels(Labels("/orders")).record(0.025)

A service's own metric (Counter, HistogramGroup, …) is exported as <service>.<name>: request_duration of the service shop is shop.request_duration. Shared and global ones keep their name, and are declared once: other modules import the declaration, and code in another language binds one by name, as in metrics.CounterGroup[Labels].named("responses_total"). Hand a metric to other code as a reference, latency.ref().

Label keys are snake_case; values are strings, ints and bools, anything else a TypeError. unit, description and a histogram's buckets must be literals: Encore's parser reads them.

OpenTelemetry

Encore records its spans and metrics through its own providers and installs no OpenTelemetry globals, leaving an app's own setup alone. For the app's spans and instruments, and instrumentation packages', install Encore's providers or pass them to an instrumentation:

from opentelemetry import metrics, trace

from encore.metrics import meter_provider
from encore.tracing import tracer_provider

trace.set_tracer_provider(tracer_provider())
metrics.set_meter_provider(meter_provider())

Object storage

Declare a bucket at module scope, then use its async methods:

from encore.storage import objects

uploads = objects.Bucket("uploads")


async def example() -> None:
    await uploads.upload(
        "hello.txt",
        b"Hello!",
        content_type="text/plain",
        if_not_exists=True,
    )
    attrs = await uploads.attrs("hello.txt")
    present = await uploads.exists("hello.txt")
    part = await uploads.download("hello.txt", offset=0, length=5)
    async for entry in uploads.list(prefix="hello", limit=10):
        print(entry.name, entry.size)
    download_url = await uploads.signed_download_url("hello.txt", ttl=3600)
    await uploads.remove("hello.txt")

attrs, exists, download, and remove accept a keyword-only version. signed_upload_url and signed_download_url return URL strings (str); TTL is in seconds, defaults to one hour, and must be between 1 and 604800. public_url is synchronous and requires a public bucket. Bucket.named(name) looks up an existing bucket configured for the service.

Storage failures raise ObjectsError (an EncoreError subclass), with ObjectNotFound, PreconditionFailed, and InvalidArgument for specific cases. exists returns False for missing objects and raises on other storage failures. If leaving a listing early, use contextlib.aclosing(bucket.list(...)) to close the native stream promptly.

For large objects, upload_stream(name, chunks) consumes an asynchronous iterable of bytes with backpressure. It supports the same content_type and if_not_exists options as upload. The native pipe is limited to 64 KiB, and each write across the native boundary is at most 64 KiB. S3 multipart uploads retain at most four 8 MiB part payloads and apply backpressure while waiting for the provider; completed parts retain only their metadata. Other providers may use their own upload buffers. The input iterable owns its chunks, so avoid producing the entire object at once. Producer exceptions and cancellation abort the upload. The caller owns closing an input generator that needs cleanup; cancellation cannot undo an upload the provider has already committed.

download_stream yields provider-sized byte chunks and supports version, offset, and length. It reads on demand. Use aclosing when a loop might stop early, including when piping a download into another upload:

from contextlib import aclosing


async def copy_large_object() -> None:
    async with aclosing(uploads.download_stream("source.bin")) as chunks:
        await uploads.upload_stream(
            "copy.bin", chunks, content_type="application/octet-stream"
        )

Redis databases

Declare a database at module scope and call commands through its families:

from encore.storage.redis import Database, ReadOnlyDatabaseRef

sessions = Database("sessions")


async def save_user():
    await sessions.string.set("user:1", b"hello", ex=60)
    return await sessions.string.get("user:1")  # bytes | None


async def load(reader: ReadOnlyDatabaseRef):
    return await reader.string.get("user:1")


async def handle_request():
    return await load(sessions.read())

Database(name) has no persistence or eviction options yet. Creating a database, selecting a family, and creating a handle defer native lookup until first use. Direct resource calls must keep the family and command together, as in sessions.string.get(...), so Encore can determine their permissions. To pass access to another function or keep a family interface in a variable, obtain a read() or write() handle first.

Families are the properties generic, string, bitmap, geo, hash, json, list, set, sorted_set, stream, and pubsub. Read handles expose only read operations, and the native runtime also rejects writes through them. read(primary=True) requests primary access; the current runtime uses the primary for all reads, including when this flag is false.

Keys and hash/stream field names are str inputs. Stored values, members, and message payloads accept Value: bytes, str, int, float, or an implementation of RedisValue. Strings are encoded as UTF-8, integers as decimal text, floats as ASCII text using repr(), and bytes are sent unchanged. Booleans are not supported as stored values. Floats include nan and infinities, which can be stored and retrieved but may be rejected by Redis numeric commands.

Reads return bytes by default. Select a type with command[str](...), command[int](...), command[float](...), or a custom value class to decode stored values. The return type follows the requested type:

await sessions.string.set("visits", 42)
visits = await sessions.string.get[int]("visits")  # int | None

await sessions.string.set("balance", 1.25)
balance = await sessions.string.get[float]("balance")  # float | None

await sessions.list.rpush("names", "Ada", "Grace")
names = await sessions.list.lrange[str]("names", 0, -1)  # list[str]

RedisValue is a structural protocol: no base class or registration is required. Implement encode(self) -> bytes and a class method decode(cls, data: bytes) that returns an instance of the class:

import json
from dataclasses import dataclass


@dataclass
class User:
    name: str

    def encode(self) -> bytes:
        return json.dumps({"name": self.name}).encode("utf-8")

    @classmethod
    def decode(cls, data: bytes) -> "User":
        return cls(**json.loads(data))


await sessions.string.set("user:1", User("Ada"))
user = await sessions.string.get[User]("user:1")  # User | None
await sessions.hash.hset("profiles", ("first", User("Ada")))
users = await sessions.hash.hgetall[User]("profiles")  # dict[bytes, User]

The same conversion applies to list elements, hash values, set/sorted-set/geo members, stream field values, and pubsub payloads. For pubsub, use subscribe[User](...) or psubscribe[User](...); Message.payload then has that type. Database keys, hash/stream field names, stream IDs, channel names, cursors, scores, and other metadata retain their existing types. A custom value class is not recorded in Redis: callers must choose the matching decoder.

Missing values remain None, and empty collections remain empty. Malformed UTF-8, invalid integers, and custom codec errors raise instead of becoming None. Reads that also mutate data (for example getdel, lpop, or set with get=True) execute before decoding: a decoding error does not undo the command. set[User](..., get=True) decodes the previous value; without get=True, set only accepts the default bytes response mode.

Multi-value operations accept variadic arguments, with tuples for grouped values, for example db.hash.hset("user", ("name", b"Ada")) and db.string.mget("first", "second"). Unpack existing collections with *items. Most commands require at least one item; geo.geohash, geo.geopos, and json.get allow none. json.mget requires a keyword-only path, and sorted_set.zinterstore/zunionstore keep weights as a keyword sequence. Commands with structured, option-dependent replies (including streams and geo searches) retain the native RESP structure: lists/dictionaries containing bytes, numbers, and None. With withscores=True, sorted-set ranges return (member: bytes, score: float) pairs.

string.set() returns None normally; with get=True it returns the previous value. With nx=True or xx=True and no get, it returns b"OK" if the write happened and None otherwise.

RedisJSON methods accept Python JSON values and decode document replies:

await sessions.json.set("profile", "$", {"name": "Ada", "active": True})
profiles = await sessions.json.get("profile", "$")

JSONPath replies preserve arrays of matches. JSON requires a server with RedisJSON support; the local emulator does not currently implement it.

Redis channel subscriptions are separate from Encore's pubsub.Topic API. Use an async context manager to release the dedicated connection on exit:

async def listen():
    async with await sessions.pubsub.subscribe("events") as messages:
        async for message in messages:
            print(message.channel, message.payload)  # both bytes

psubscribe() accepts glob patterns. Subscription.receive() returns a Message or None after closure; close() interrupts pending receives and is safe to repeat. Leaving the context also closes the subscription when the body raises or is cancelled.

db.connection_string (also available on write handles) returns the direct server URL and grants broad read-write access. External clients using that URL do not apply Encore's key prefixes or tracing. Read-only handles do not expose this URL.