Metadata-Version: 2.5
Name: 021-python-clients
Version: 0.1.0
Summary: Observed clients for Redis, SQS, QStash, S3, HTTP, email and KMS, built on 021-python-observability
Project-URL: Repository, https://github.com/from021-io/python-clients
Project-URL: Changelog, https://github.com/from021-io/python-clients/blob/main/CHANGELOG.md
Project-URL: Contract, https://github.com/from021-io/python-clients/blob/main/CONTRACT.md
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: email,httpx,kms,observability,opentelemetry,qstash,redis,resend,s3,ses,sqs
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: 021-python-observability<0.4,>=0.3
Requires-Dist: httpx<1,>=0.28
Requires-Dist: opentelemetry-api<2,>=1.44
Provides-Extra: aws
Requires-Dist: boto3<2,>=1.40; extra == 'aws'
Requires-Dist: cryptography>=44; extra == 'aws'
Provides-Extra: httpx2
Requires-Dist: httpx2<3,>=2; extra == 'httpx2'
Provides-Extra: redis
Requires-Dist: redis<9,>=6; extra == 'redis'
Description-Content-Type: text/markdown

# 021-python-clients

Clients for the external services our Python services call, with the observability of
[021-python-observability](https://pypi.org/project/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.

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

| Module | Extra | What it gives |
| --- | --- | --- |
| `clients_021.http` | none | `HttpClient`: one httpx client per dependency, timeouts, retries on idempotent methods |
| `clients_021.httpx2` | `httpx2` | `ObservedTransport` for SDKs built on httpx2 (Anthropic, OpenAI) |
| `clients_021.redis` | `redis` | `RedisClient`: JSON with TTL, one-time reads, windowed counters, ping |
| `clients_021.queue.qstash` | none | `QStashClient`: JSON publishing, deduplication, delivery options and forwarded trace context |
| `clients_021.queue.sqs` | `aws` | `SqsClient`: send, receive, delete, visibility, trace carried in the message |
| `clients_021.s3` | `aws` | `S3Client`: put, get, head, delete, presigned URLs |
| `clients_021.email` | none | `EmailMessage`, the `EmailSender` protocol, `ResendSender` (Resend's HTTP API) |
| `clients_021.email.ses` | `aws` | `SesSender`: the same protocol over Amazon SES v2 |
| `clients_021.kms` | `aws` | `KmsClient`: encrypt, decrypt and data keys bound to an encryption context; AES-256-GCM envelopes |

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. Import queue clients explicitly:

```python
from clients_021.queue.sqs import SqsClient
from clients_021.queue.qstash import QStashClient
```

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.

## 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`).
- A counter and a duration histogram: `db.client.operation.*` (Redis, shared with SQL),
  `messaging.client.operation.*` (SQS, QStash, email), `storage.client.operation.*` (S3),
  `crypto.client.operation.*` (KMS), `http.client.request.*` (HTTP, and the HTTP request
  under QStash and Resend). Labels are bounded; see [CONTRACT.md](CONTRACT.md).
- Redis, SQS, QStash, S3, 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

```python
from clients_021 import DependencyUnavailable, HttpClient
from clients_021.redis import RedisClient
from clients_021.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")
```

### QStash

```python
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](https://upstash.com/docs/qstash/features/deduplication).

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](https://upstash.com/docs/qstash/howto/signature).

### Email

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

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

```python
from clients_021.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](CONTRACT.md).

### SDKs on httpx2

For an SDK on httpx2:

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

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

```bash
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](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](https://upstash.com/docs/qstash/howto/local-development),
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](https://github.com/getmoto/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.
