021-python-clients

Clients for the external services our Python services call, with the observability of 021-python-observability built in. A service uses these instead of calling redis, boto3, httpx or an email SDK directly, so every dependency call gets the same span, the same metrics under a stable dependency name, and the same failure type.

uv add "021-python-clients[redis,aws,httpx2,vercel-blob]"   # pick the extras you use

Every client takes the service's Observability (from setup_observability) and a dependency name, which becomes the dependency label of its metrics.

Organization

Providers with the same purpose live under one capability package: email/ contains Resend and SES; queue/ contains SQS and QStash; storage/ contains S3 and Vercel Blob; cache/ contains Redis; crypto/ contains KMS. HTTP transports remain at the root. Import providers explicitly:

from clients_021.queue.sqs import SqsClient
from clients_021.queue.qstash import QStashClient
from clients_021.storage.s3 import S3Client
from clients_021.cache.redis import RedisClient
from clients_021.crypto.kms import KmsClient

SQS supports polling, receipts and visibility changes; QStash delivers to an HTTP endpoint. Their provider APIs remain separate inside queue/. Email providers share EmailSender because both implement the same send operation. A shared package does not imply that every provider implements an interchangeable consumption protocol. Importing queue or QStash does not load the optional AWS SDK.

The 0.1.0 imports clients_021.redis, clients_021.s3 and clients_021.kms remain compatible aliases for the same classes and result types. Importing a capability package does not load its optional SDKs.

What every call gets

• A CLIENT span named <dependency> <operation> (HTTP <dependency> for HTTP), with the system attribute (db.system.name=redis, messaging.system=aws_sqs|resend|aws_ses, rpc.service=S3|KMS, rpc.system=vercel-blob).
• A counter and a duration histogram: db.client.operation.* (Redis, shared with SQL), messaging.client.operation.* (SQS, QStash, email), storage.client.operation.* (S3, Vercel Blob), crypto.client.operation.* (KMS), http.client.request.* (HTTP, and the HTTP request under QStash and Resend). Labels are bounded; see CONTRACT.md.
• Redis, SQS, QStash, S3, Vercel Blob, email and KMS emit a dependency.call_failed warning on failure, with dependency, operation and error type only. HTTP transports emit spans and metrics.
• DependencyUnavailable(dependency, operation, reason) when the service timed out, refused the connection, throttled or failed with a 5xx. Map it to a 503. Errors that mean the request is wrong (access denied, a missing bucket, a wrong Redis type) keep their original type: they are bugs, not outages.

Never in this library's spans, metrics or logs: keys, URLs, bodies, message ids or object names; email addresses, subjects, bodies or tags; KMS key ids, encryption contexts, plaintext, ciphertext or data keys. Use static dependency names such as identity-provider or uploads; never derive them from a tenant, user or request. All clients apply observability's identifier rule (1–96 letters, digits, _, . or -, starting with a letter or digit).

Examples

from clients_021 import DependencyUnavailable, HttpClient
from clients_021.cache.redis import RedisClient
from clients_021.storage.s3 import S3Client
from clients_021.queue.sqs import SqsClient

redis = RedisClient.from_url(obs, settings.redis_url)
await redis.set_json("session:abc", {"user": "ana"}, ttl_seconds=1800)
used = await redis.incr_window("ratelimit:ip:1.2.3.4:28735", by=1, ttl_seconds=61)

queue = SqsClient.create(obs, settings.queue_url, region="us-east-1")
await queue.send(body, attributes={"type": "notes.created.v1"})
for message in await queue.receive(wait_seconds=20):
    with obs.tracer.start_as_current_span("handle", context=message.trace_context()):
        ...
    await queue.delete(message.receipt)

files = S3Client.create(obs, "service-uploads", region="us-east-1")
await files.put("reports/2026-10.pdf", data, content_type="application/pdf")
url = files.presign_get("reports/2026-10.pdf", expires_seconds=300)

http = HttpClient(obs, timeout_seconds=10)
response = await http.request("GET", jwks_url, dependency="identity-provider")

Vercel Blob

from clients_021.storage.vercel_blob import VercelBlobClient

files = VercelBlobClient(
    obs,
    token=settings.blob_token,
    store_id=settings.blob_store_id,
    access="private",  # Required; must match the store. Public is also supported.
    dependency="documents",
    timeout_seconds=10,
)
try:
    uploaded = await files.put("reports/report.pdf", pdf_bytes, content_type="application/pdf")
    content = await files.get(uploaded.url)  # bytes, or None for a missing object
    info = await files.head(uploaded.url)    # metadata, or None for a missing object
    if info is not None:
        await files.delete(uploaded.url, if_match=info.etag)
finally:
    await files.aclose()

Uses the official Python Blob SDK, vercel-blob>=0.5,<0.6, installed with the vercel-blob extra. This is the current functional SDK API; it does not depend on the earlier AsyncBlobClient API or the full vercel bundle. Credentials and store access are explicit; no environment or default SDK session is used for requests. A read-write token's store must match store_id (the store_ prefix is allowed).

put returns BlobUpload (the SDK's PutResult); head returns BlobInfo (its HeadResult). These are provider-specific types, not a common S3 protocol. Uploads require a pathname; reads, metadata and deletes accept a pathname or a Vercel delivery URL. The SDK validates those URLs and does not follow redirects. get buffers the entire body in memory.

Each operation has one attempt and a finite total deadline, including body consumption. The client reuses its connection pool across calls and closes it through aclose(), including an injected httpx2.AsyncBaseTransport. Call aclose() only after in-flight calls finish. Calls after close fail locally. SDK sessions are scoped to each operation so concurrent clients keep their credentials separate. No trace headers are sent by default.

allow_overwrite=False and add_random_suffix=False by default. Both can be explicitly enabled, along with cache_control_max_age. A timeout may follow a successful upload: retrying with a random suffix can create another object; overwriting a stable pathname can race with other writers. Deleting a missing object succeeds. if_match guards a delete against a changed ETag. CDN caches can retain deleted content briefly.

429, 5xx, transport failures and deadlines become DependencyUnavailable. Authentication, missing stores, precondition failures, invalid inputs and malformed responses retain SDK exceptions; only a missing object becomes None for get/head. Cancellation propagates. The storage operation includes the complete request and response body; this client does not emit an additional HTTP metric. Signed URLs, multipart and streaming APIs stay outside this initial client.

QStash

from clients_021.queue.qstash import QStashClient

qstash = QStashClient(
    obs,
    token=settings.qstash_token,
    base_url=settings.qstash_url,  # the API URL for your selected region
    timeout_seconds=10,           # this service's publication request
)
try:
    published = await qstash.publish_json(
        settings.events_url,
        {"type": "notes.created.v1", "data": {"note_id": note_id}},
        deduplication_id=event_id,  # stable across retries of this logical event
        delivery_retries=3,
        delivery_timeout_seconds=60,
        delay_seconds=0,
        label="notes.created.v1",
        forward_headers={"X-Request-ID": request_id},
    )
    # published.message_id and published.deduplicated describe QStash acceptance.
finally:
    await qstash.aclose()

Create the client once per service lifecycle; the example closes it explicitly for completeness. publish_json targets one HTTP(S) URL and supports any JSON value, including null. Acknowledgement means queued, not delivered. Deduplication keys are required and preserved verbatim; QStash rejects keys containing :, so this client raises ValueError for those keys before making a request. Use a UUID or a stable hash of a composite business key instead of replacing punctuation, which can collapse different keys. Reuse the key after an uncertain publication outcome; scope it to the logical event and destination. QStash's deduplication window is finite, so consumers must still be idempotent. See Upstash's deduplication contract.

The publisher forwards the current trace to the destination by default. Use propagate=False for destinations that should not receive it. Caller-supplied forwarded trace headers are replaced or removed together. forward_headers uses ordinary header names; the client adds the provider prefix. These headers are sent to QStash for delivery, so only pass credentials intended for that destination.

The first slice covers publishing, not schedules, batches, URL groups or a web framework's receiving endpoint. Authenticate incoming deliveries using QStash signature verification.

Email

Behind one protocol, so a service can switch between Resend and SES by configuration:

from clients_021.email import EmailMessage, EmailRejected, EmailSender, ResendSender
from clients_021.email.ses import SesSender

email: EmailSender = (
    ResendSender(obs, settings.resend_api_key)
    if settings.email_provider == "resend"
    else SesSender.create(obs, region="us-east-1", configuration_set="transactional")
)
message = EmailMessage(
    sender="021 <hola@from021.io>",
    to=user.email,
    subject="Your sign-in code",
    text=f"Your code is {code}",
    tags={"category": "sign_in"},
)
try:
    await email.send(message, idempotency_key=f"sign-in:{attempt.id}")
except EmailRejected as rejected:  # the address, the domain or the credentials; not an outage
    log.warning("email.rejected", status=rejected.status, code=rejected.code)

EmailRejected carries the provider's status and error name (validation_error, MessageRejected, ...), never its message. Resend sends each request once; with an idempotency_key a retry after DependencyUnavailable within 24 hours cannot send twice. SES has no idempotency key (the argument is ignored): deduplicate in the caller when a retry must not send twice. Tags are for provider analytics; never put personal data in them.

KMS

The encryption context is required on every call, so a ciphertext only decrypts for the purpose it was made for:

from clients_021.crypto.kms import KmsClient

keys = KmsClient.create(obs, settings.session_key_arn, region="us-east-1")
context = {"purpose": "refresh-token", "tenant": tenant_id}
stored = await keys.encrypt(refresh_token, context=context)   # up to 4096 bytes
refresh_token = await keys.decrypt(stored, context=context)
envelope = await keys.seal(document, context=context)          # any size
document = await keys.unseal(envelope, context=context)

The context is logged by KMS in CloudTrail in plaintext: it names a purpose, never a secret. seal makes one GenerateDataKey call and encrypts locally with AES-256-GCM; unseal makes one Decrypt call. A wrong context, a tampered ciphertext or another key's ciphertext keep their botocore type (they are bugs or attacks, not outages); a tampered envelope raises cryptography's InvalidTag. The envelope layout is in CONTRACT.md.

SDKs on httpx2

For an SDK on httpx2:

from anthropic import AsyncAnthropic, DefaultAsyncHttpxClient
from clients_021.httpx2 import ObservedTransport

transport = ObservedTransport(obs, dependency="anthropic")
claude = AsyncAnthropic(http_client=DefaultAsyncHttpxClient(transport=transport))

It does not send traceparent to third-party APIs unless propagate=True. The transport preserves SDK exceptions so the SDK retains control of retries and error handling; it does not convert them to DependencyUnavailable.

Service integration

Create clients once during service startup and reuse their connection pools. Keep business rules in service adapters: queue envelopes, deduplication-key generation, storage integrity and key naming do not belong in this library. Inject the configured SDK client into the constructor when you need provider-specific options (for example S3 addressing style or credentials).

Close clients during shutdown, after stopping consumers and waiting for in-flight calls. aclose() closes the underlying client, including an injected one; do not share that SDK client with another owner. An AsyncExitStack makes cleanup explicit:

from contextlib import AsyncExitStack

async with AsyncExitStack() as resources:
    http = HttpClient(obs, timeout_seconds=10)
    resources.push_async_callback(http.aclose)
    redis = RedisClient.from_url(obs, settings.redis_url)
    resources.push_async_callback(redis.aclose)
    files = S3Client.create(obs, "service-uploads", region="us-east-1")
    resources.push_async_callback(files.aclose)
    queue = SqsClient.create(obs, settings.queue_url, region="us-east-1")
    resources.push_async_callback(queue.aclose)
    email = ResendSender(obs, settings.resend_api_key)
    resources.push_async_callback(email.aclose)
    keys = KmsClient.create(obs, settings.session_key_arn, region="us-east-1")
    resources.push_async_callback(keys.aclose)
    # Run the service here. Stop its background workers before leaving this scope.

Retry policies are explicit:

• HTTP retries GET, HEAD, OPTIONS, PUT and DELETE on network failures, timeouts, remote protocol/proxy connection failures and 502/503/504. retries=2 means at most three attempts. POST and PATCH run once. Invalid URLs/protocols and invalid request construction keep their original exception. HTTP status responses are returned for the caller to handle.
• Resend sends each request once (POST); send an idempotency_key and retry in the caller.
• AWS factories (SQS, S3, SES, KMS) use botocore standard retries with at most three total attempts. Injected SDK clients keep their configured policy. SQS consumers must tolerate duplicate messages; a timeout does not prove that a send or write was not applied. SQS allows ten message attributes including trace context; reserve two for traceparent and tracestate, which belong to the library.
• Redis keeps the injected SDK's retry policy. For non-idempotent operations such as take_json and incr_window, do not configure automatic replay after an ambiguous timeout. Authentication and authorization errors keep their Redis exception type.
• QStash sends each publication POST once. delivery_retries configures subsequent delivery by QStash, not retrying this request. HTTP 429 and 5xx become DependencyUnavailable; other rejected statuses and invalid acknowledgements raise QStashResponseError with a status code and no provider body. Delivery timeouts and delays use integer seconds and remain subject to the provider plan's limits.
• Vercel Blob sends each operation once, with a finite total deadline (10 seconds by default). An injected transport must also avoid automatic retries of uncertain writes.
• httpx2 SDKs own their retries; ObservedTransport adds no extra retry loop.

DependencyUnavailable signals a transient dependency failure, not permission to replay any operation. Map it at the service boundary and apply the operation's idempotency rules. Its message and structured fields are safe to log when dependency names are static. Its chained SDK exception may contain sensitive provider data; do not log exception chains.

S3, SQS, SES and KMS run the synchronous AWS SDK in worker threads. Cancelling the await does not stop an SDK request already running in its thread. Timeouts are per SDK/network attempt, not an overall deadline. Limit consumer concurrency and let in-flight work finish before closing the clients. Redis window counters require Redis server 7 or newer (EXPIRE NX).

HTTP transport spans end when response headers arrive. Body downloads and SDK streaming are outside that span; instrument the enclosing service/SDK operation when you need total latency or failures while consuming the body. S3 get buffers the entire object in memory; use a dedicated streaming adapter for large files.

Development

uv sync --all-extras
make check   # ruff, pyright strict, pytest (Redis at REDIS_URL, default redis://127.0.0.1:63799/15)

See CONTRIBUTING.md for the module layout and the checklist for adding clients. QStash protocol tests need no external service. To exercise actual delivery and deduplication, start the official local server, set its QSTASH_URL and QSTASH_TOKEN, and run uv run python tests/qstash_local.py. That smoke test accepts only a loopback QStash server and binds its destination dynamically.

SQS, S3, SES and KMS tests run against moto (plus botocore's Stubber where moto ignores a field or differs from AWS); Resend tests use httpx's MockTransport; Redis tests need a real Redis. Licensed under Apache-2.0.