Metadata-Version: 2.3
Name: encore-dev
Version: 2.0.0a7
Summary: Encore SDK for Python — infrastructure resources
Author: Encore
License: MPL-2.0
Classifier: Programming Language :: Python :: 3
Requires-Dist: opentelemetry-api>=1.27
Requires-Dist: opentelemetry-semantic-conventions>=0.62b1
Requires-Dist: typing-extensions>=4.12
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# Encore SDK for Python

The Python SDK for [Encore](https://encore.dev) — **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).

```python
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.

```python
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.

```python
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:

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```sh
uv build sdk/py/infra
```

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

```sh
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:

```sh
# 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:

```sh
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`:

```sh
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:

```sh
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:

```sh
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:

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

To check formatting without changing files:

```sh
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:

```python
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:

```python
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](https://google.github.io/styleguide/pyguide.html#docstrings)
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:

```sh
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:

```sh
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:

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

## Object storage

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

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```sh
cargo xtask build-py-runtime
```

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

```sh
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.
