Encore SDK for Python

The Python SDK for Encore — infrastructure resources only.

The distribution is named encore-dev; Python code imports it as encore.

Declare infrastructure (SQL databases, pub/sub topics & subscriptions, object storage buckets, Redis databases, cron jobs, metrics, secrets) as module-level variables. Encore's parser discovers them for provisioning and wiring, and the native runtime binding (encore_sdk_core over PyO3) connects them to the provisioned infrastructure at run time.

Services are declared by the language-agnostic encore.service.json file, not in Python — this SDK is infrastructure-only (the counterpart of the Go/TS/Rust SDKs).

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})

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 one row or None (rejecting multiple rows), 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).one()


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.

Like the Rust SDK, access selection currently controls parser permission inference. Handles still share one runtime pool; per-handle privilege enforcement and replica routing (including the read(primary=True) preference) are pending.

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

Native Postgres integration tests can be enabled with ENCORE_PYTHON_POSTGRES_TEST_HOST=host:port. They expect an isolated Postgres instance with database/user postgres and password encore-test, and create and remove a temporary schema.

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.

Packaging

encore-dev contains the Python SDK and is built with uv:

uv build sdk/py/infra

For release packaging, stage the SDK and stamp its version without changing the checked-in manifest:

cargo xtask pack-python --version v2.0.0-alpha.1

This writes encore_dev-2.0.0a1-py3-none-any.whl, the source distribution, and SHA-256 sidecars to dist/python/ (override with --output). Alpha and beta versions become aN and bN; nightly and development versions become .devN. The command requires uv and the native build prerequisites (including Protoc). It builds the native runtime to generate fresh type stubs, but packages only the pure Python SDK and does not publish to PyPI.

The Rust extension lives in the parent sdk/py/ directory and is packaged separately by Maturin. Its encore-runtime wheel is an internal artifact for the CLI, not a dependency of the public SDK:

# With Maturin installed:
cargo xtask build-py-runtime --release --wheel
# Optionally select the wheel's target (requires a suitable toolchain):
cargo xtask build-py-runtime --release --wheel --target aarch64-apple-darwin

SDK distributions are written to sdk/py/infra/dist/; native wheels go to sdk/py/dist/.

Building locally

From the repository root:

cargo xtask build-py-runtime

This stages _encore_runtime inside infra/encore/ for the SDK's local loader, and in runtimes/python/ for the development CLI. It also regenerates encore/internal/_native.pyi from the compiled extension. Edit native signatures in Rust, then rerun this command to update the stub. An editable install of the SDK can then load the native extension without setting PYTHONPATH:

uv run --project sdk/py/infra pytest sdk/py/infra/tests

The CLI sets ENCORE_PYTHON_RUNTIME_DIR for runs, tests, and exec commands. The SDK selects the native library using the running interpreter's build ABI: standard CPython uses the root abi3 library; free-threaded Python 3.14 uses cp314t/_encore_runtime.so (.pyd on Windows). This selection is unchanged if a free-threaded interpreter temporarily enables its GIL. Unsupported ABIs fail with a missing-runtime error instead of falling back to an incompatible library.

Build the local free-threaded variant separately:

cargo xtask build-py-runtime --python-runtime cp314t

The default builds abi3. Both variants can coexist. Native-only wheel builds (--wheel) continue to use maturin's interpreter selection instead.

ENCORE_PYTHON_RUNTIME_LIB overrides directory selection with an explicit library. An explicit library or directory is authoritative; failures do not fall back. Without either variable, the loader searches beside itself and in parent and dist/ directories using the same ABI layout.

Configure the development CLI's ENCORE_RUNTIMES_PATH to point at this checkout's runtimes/ directory. Published SDK wheels exclude native libraries; standalone use requires a runtime directory or explicit library from the CLI installation.

App builds install encore-dev and copy the runtime directory, preserving its ABI variants for selection by the deployment interpreter. An explicit library override is packaged separately and remains authoritative. Cross-platform runtime retrieval is not yet supported.

Resource initialization

Importing the SDK synchronously initializes the native Encore runtime, matching TypeScript. 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. SDK unit tests supply a minimal runtime configuration in tests/conftest.py; integration tests use encore test.

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

Linting and formatting

From the repository root, lint the SDK and its tests:

uv run --locked --project sdk/py/infra ruff check sdk/py/infra

Add --fix to apply safe fixes. The enabled rules check syntax/style errors, undefined and unused names, common bug patterns, and Google-style docstrings (E4, E7, E9, F, B, D).

From the repository root, format the SDK and its tests with Ruff:

uv run --project sdk/py/infra ruff format sdk/py/infra

To check formatting without changing files:

uv run --project sdk/py/infra ruff format --check sdk/py/infra

The generated _native.pyi is excluded from linting and formatting because it is owned by the PyO3 stub generator. It is still checked by ty.

Numeric counters and label groups

Counters default to unsigned 64-bit integers. Use Counter[float] for fractional increments and CounterGroup[Labels, float] for labeled float counters:

from dataclasses import dataclass

from encore import metrics


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


requests = metrics.CounterGroup[Labels]("requests")
usage = metrics.CounterGroup[Labels, float]("usage")
cpu_time = metrics.Counter[float]("cpu_time")

requests.with_labels(Labels("checkout")).increment()
usage.with_labels(Labels("checkout")).increment(0.125)
cpu_time.increment(0.025)

This syntax works on Python 3.9 and later. Put numeric type arguments on the constructor; a variable annotation alone does not select runtime storage. with_labels() returns Counter[int] or Counter[float] to the type checker. Existing Counter(name) and CounterGroup[Labels](name) keep their integer behavior. Gauge values already use floating-point storage.

Float increments must be finite and nonnegative. An update that would overflow the total raises ValueError without changing it. Double-precision totals can incur rounding, including loss of integer precision above 2**53. Integer and float counters cannot share a native series. Integer totals still wrap on 64-bit overflow. Calling increment() without a value adds one in either mode.

All metric groups accept dataclass instances as well as mappings, including TypedDicts. Fields must contain strings, booleans, or finite numbers. Nested objects are rejected. Numeric labels are rounded down to integer strings and booleans become "true" or "false". Names are normalized to snake_case; service is reserved. The label values are captured when binding, so later mutation does not change an existing series. GaugeGroup[Labels] and HistogramGroup[Labels] also provide typed label arguments.

Histogram metrics

Declare histograms at module scope to record distributions such as request duration or payload size. Record one observation for each measurement:

from typing import TypedDict

from encore import metrics


class Labels(TypedDict):
    route: str


latency = metrics.Histogram("request_duration", unit="seconds")
sizes = metrics.HistogramGroup[Labels]("response_size", unit="bytes")

latency.observe(0.025)
sizes.with_labels({"route": "/orders"}).observe(512)

The runtime maintains count, sum, and automatic exponential buckets without keeping individual measurements. Bucket widths grow by at most 10%. unit is optional metadata, must be a string literal when supplied, and does not convert values. Histogram groups normalize and cache labels like counter groups. Use labels with a small set of values: each combination retains a series for the process lifetime.

Prefer finite observations. As in Rust, NaN increments the count and makes the sum NaN without entering a bucket; infinities enter their own buckets.

Prometheus remote write exports the bucket distribution for approximate percentile queries. AWS, GCP, and Datadog currently export only <name>_count and <name>_sum, which support averages but not percentiles.

Documentation style

Follow the Google Python docstring guide for modules, classes, functions, methods, and Python examples in documentation. Start docstrings with a concise summary on the opening triple-double-quote line. End the summary with punctuation, keep it within 80 characters, and separate any further explanation with a blank line. Use imperative summaries for functions ("Return the current phase.") and noun phrases for classes and properties.

Use Args:, Returns:, Yields:, Raises:, and Attributes: as appropriate, with four-space indentation for entries and continuation lines. Document behavior callers need to know: units, defaults, side effects, resource ownership, and relevant exceptions. Parameter names must match the signature, including *args and **kwargs; type annotations usually make repeated type descriptions unnecessary. Document public data attributes in the class's Attributes: section and constructor arguments in __init__.

Simple APIs can use a one-line docstring when the summary and signature explain their behavior completely. Omit Returns: for functions returning only None; use Yields: for generators, describing each yielded item. For a decorated context manager, describe the context manager returned to the caller. Put module examples under Typical usage example:. Keep README prose and examples consistent with these API contracts; Markdown guides retain their normal heading structure.

Ruff enforces the Google docstring convention through the existing lint command and pre-commit hook. Test docstrings are optional; add them to explain unusual setup or behavior rather than repeating test names. Existing test docstrings still follow the same formatting rules. Review documentation for accuracy as well as formatting: Ruff cannot verify behavioral contracts.

Type checking

From the repository root, check the SDK, native stubs, and tests with ty:

uv run --project sdk/py/infra ty check --project sdk/py/infra

The checker uses the minimum Python version from requires-python and resolves dependencies from the SDK's virtual environment.

Pre-commit hooks

From the repository root, install the prek hooks once per checkout:

uv run --locked --project sdk/py/infra prek install

The hooks apply safe Ruff lint fixes, format the SDK, and check it with ty when relevant SDK files change. If a hook changes files, stage those changes and commit again. All hooks use the same locked tools as CI. To run them manually:

uv run --locked --project sdk/py/infra prek run --all-files

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.

Testing Postgres

Build the native runtime from the repository root:

cargo xtask build-py-runtime

Then run the Postgres tests from sdk/py/infra:

uv sync --locked
uv run pytest tests/test_sqldb.py tests/test_postgres_runner.py -v

The integration runner starts one disposable postgres:17 Docker container and removes it after the suite. To use an existing test server, set ENCORE_PYTHON_POSTGRES_TEST_HOST=localhost:5432. The server must provide a postgres database and user with password encore-test and schema creation privileges. Use a dedicated test server.

tests/test_postgres_runner.py launches tests/postgres_integration_suite.py in a separate process to isolate the native runtime configuration. The suite is collected explicitly by the runner and excluded from normal pytest discovery by its filename. Each test and Hypothesis example gets a fresh schema. The suite includes parameter round trips, generated transaction and cursor operation sequences, and synchronized cancellation regressions. Hypothesis uses deterministic generation and reports minimized failing examples. Sequences are repeatable for unchanged tests and the same Hypothesis version. Resource cleanup and bounded polling live in tests/postgres_test_support.py. These integration tests are run locally and are not executed by CI.