Metadata-Version: 2.3
Name: encore-dev
Version: 2.0.0a16
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**.

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:

```sh
python -m pip install encore-dev
```

Or add it to a uv project:

```sh
uv add encore-dev
```

## Declare resources

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

## Service URLs

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

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

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.

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

## 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.

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

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

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

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