Alien Python bindings

alienplatform gives async Python applications typed access to resources linked by Alien. Provider selection, short-lived credential refresh, error metadata, and cloud behavior stay in the shared Rust binding core.

The package is under development and is not published to a package registry yet.

Local development

Build an editable local-provider package from the repository root:

python -m venv .venv
source .venv/bin/activate
python -m pip install 'maturin>=1.9,<2' 'pytest>=8,<10' 'pytest-asyncio>=1,<2'
maturin develop --manifest-path crates/alien-bindings-python/Cargo.toml \
  --no-default-features --features local
pytest -q crates/alien-bindings-python/tests

The native extension uses Python's asyncio interface and runs Alien futures on its Tokio runtime. Resource factories are lazy: missing or malformed bindings fail when first used with a structured AlienError.

Resources

from alienplatform import ai, container, key, kv, postgres, queue, sandbox, storage, vault, worker
from sqlalchemy.ext.asyncio import create_async_engine

files = storage("files")
await files.put("reports/today.json", b"{}", condition="absent")
assert await files.get("reports/today.json") == b"{}"
assert await files.get_optional("reports/missing.json") is None

cache = kv("cache")
await cache.put("status", b"ready", condition="absent")

jobs = queue("jobs")
await jobs.send({"kind": "index"})

secrets = vault("secrets")
api_token = await secrets.get("api-token")

database = await postgres("database").connection()
engine = create_async_engine(**database.sqlalchemy_async_engine_kwargs())

service_url = await container("api").internal_url()

response = await worker("processor").invoke(
    method="POST", path="/jobs", body=b"{}", timeout_ms=30_000
)

ciphertext = await key("records").encrypt(b"private", {"tenant": "tenant-42"})

model = await ai("assistant").connection()
# OpenAI-compatible clients use model.openai_base_url. model.api_key is None
# for ambient Bedrock, Vertex, and Foundry resources because the embedded Rust
# gateway authenticates with the workload identity.

Do not log or render Postgres connection strings, passwords, vault values, or model credentials. Connection object representations redact credentials by default.

AI

ai(name).connection() returns a provider-neutral base URL. Managed Bedrock, Vertex, and Foundry resources start the same in-process Rust gateway used by the TypeScript SDK, so signing and short-lived credential refresh stay in the shared implementation. External OpenAI and Anthropic bindings return the projected key and provider endpoint. Pass openai_base_url to an OpenAI-compatible client.

Sandbox commands stream raw stdout and stderr frames with production-order sequence numbers:

runtime = sandbox("runtime")
instance, created = await runtime.get_or_create(tenant_key="tenant-42", timeout_ms=300_000)
stream = await runtime.run_command(
    instance.sandbox_id,
    "python",
    ["-c", "print('hello')"],
    timeout_ms=30_000,
)
try:
    async for frame in stream:
        if frame.data is not None:
            print(frame.data.decode(errors="replace"), end="")
        if frame.exit_code is not None and frame.exit_code != 0:
            raise RuntimeError(f"command exited with {frame.exit_code}")
finally:
    await stream.close()

Long-running commands can use start_job, poll_job, and cancel_job. Call capabilities() for new sessions, or capabilities(sandbox_id) for an existing session, before optional file, job, pause/resume, or isolation behavior. Azure sessions retain the transport they were created with across image upgrades and rollbacks.

PostgreSQL TLS

Postgres.connection() returns the provider-neutral connection parameters, a wire-protocol URL, CA certificates, and SSL mode. sqlalchemy_async_engine_kwargs() translates the URL and supplies the verified TLS context together so create_async_engine() cannot omit provider CA certificates. Lower-level integrations can use sqlalchemy_async_url() and ssl_context() directly.

Errors

AlienError includes stable code, context, retryable, internal, http_status_code, and hint fields. Branch on code rather than parsing the message. Errors from object storage are intentionally mapped without object paths or provider response bodies so application logs do not leak customer data.

Remote KV

from alienplatform import Bindings

bindings = await Bindings.for_remote_customer(
    project=project, external_id=external_id, token=token
)
await bindings.kv("check-cache").put("status", b'{"ready":true}', ttl_seconds=86400)

Use a backend token with write access and enable remoteAccess on the KV resource before applying setup. Remote credentials refresh in the Rust core. AWS and Azure scope access to the selected table. Firestore access covers the project's documents because IAM cannot isolate collections in the shared default database. Raw value limits are 408,576 bytes for DynamoDB, 783,360 bytes for Firestore, and 24,576 bytes for Azure/local.

Worker queue batches

Install the optional alienplatform[worker] dependencies and configure the Worker's queue trigger with batch: {maxMessages: 10, maxWindowSeconds: 2}. Register handlers using the logical queue resource ID:

import asyncio
from alienplatform import QueueBatch, QueueOutcome, on_queue_batch, run_worker

async def process(batch: QueueBatch) -> list[QueueOutcome]:
    outcomes = []
    for delivery in batch.deliveries:
        # Persist the application's work before returning success.
        await persist(delivery.message.payload)
        outcomes.append(QueueOutcome(delivery.delivery_id, "success"))
    return outcomes

on_queue_batch("jobs", process)
asyncio.run(run_worker())

Return exactly one success or retry for each delivery ID. Missing, duplicate, unknown, or invalid outcomes reject the batch before acknowledging any item. An exception retries the batch. The runtime owns acknowledgment; do not also acknowledge these deliveries through a queue binding. on_queue_message continues to handle triggers without batch configuration. "*" matches any otherwise unregistered source.

Batching is an explicit deployment opt-in. Deploy a runtime and SDK that support batch handlers when adding batch to a queue trigger. A manager upgrade alone preserves legacy queue routes and concurrency settings. Removing batch restores single-message infrastructure settings; deploy a single-message handler with that change. On AWS, enabling batching checks queue visibility against six times the Worker timeout plus the collection window. Live queues are increased when needed; a derived visibility increase for setup-owned queues returns setup-required before Worker updates. Rerun setup and complete its normal handoff before enabling batching; the controller also verifies installed visibility before changing the Worker. Disabling batching never shortens an existing queue visibility timeout.

A batch may contain fewer than maxMessages. The collection window is a maximum wait, not a periodic schedule or a minimum batch size. Messages are unordered and may be delivered again, including after an uncertain acknowledgment. Make work idempotent. Distinct deliveries can share a provider message ID; use delivery_id when constructing outcomes. Python handlers execute sequentially within each application process; this does not provide a distributed lock.

For applications that explicitly receive through a queue binding, ack_batch(receipts) returns one result per receipt and is not atomic. renew_lease(receipt, duration_seconds) requires 1–600 seconds. AWS, GCP, and local queues reset the remaining lease, so a smaller value intentionally shortens it. Azure renews the queue's configured lock duration; the requested value is a validated hint, not an exact duration guarantee.

Storage notifications in queues

Parse a binding QueueMessage or a Worker's QueueEvent (including delivery.message inside a batch) with the typed helper:

from alienplatform import storage_event_from_message

event = storage_event_from_message(message)
print(event.bucket_name, event.object_key, event.timestamp, event.etag)

The helper validates the portable camelCase JSON payload and returns Python attributes in snake_case, including a timezone-aware UTC timestamp and optional object metadata. Invalid JSON or schema fields raise AlienError with code INVALID_STORAGE_EVENT_MESSAGE. Parsing does not acknowledge a message. Use normal queue-binding acknowledgment or Worker outcomes after processing succeeds. Notifications are at least once and unordered; make processing idempotent.