Metadata-Version: 2.4
Name: nimble-aws
Version: 0.2.0
Summary: Asynchronous, typed adapters for AWS services, tuned for Lambda, ECS and Glue: one cached client per service, batch APIs and bounded concurrency.
License-Expression: MIT
Author: Lucas Tiense Blazzi
Author-email: lucast.blazzi@gmail.com
Requires-Python: >=3.11,<4
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Provides-Extra: lambda
Requires-Dist: aiobotocore (>=3.10,<4)
Requires-Dist: aws-lambda-powertools (>=3.23,<4) ; extra == "lambda"
Requires-Dist: boto3 (>=1.40,<2)
Requires-Dist: pydantic-settings (>=2.4,<3)
Requires-Dist: typing-extensions (>=4.12)
Project-URL: Repository, https://github.com/lucasblazzi/nimble
Description-Content-Type: text/markdown

# Nimble-AWS

Asynchronous, typed adapters for AWS services, built on [aiobotocore](https://github.com/aio-libs/aiobotocore)
and tuned for code that runs inside AWS: Lambda, ECS and Glue.

- **One client per service, reused.** Clients (and their HTTP/TLS connection pools) are created once per
  process and event loop, not on every call.
- **Single and batch methods for every operation.** `get_object` / `get_object_batch`,
  `put_item` / `put_item_batch`, `send_message` / `send_message_batch` ... Batch methods use the native
  batch APIs where AWS has one (DynamoDB `BatchWriteItem`/`BatchGetItem`, SQS `SendMessageBatch`, SNS
  `PublishBatch`, S3 `DeleteObjects`, Secrets Manager `BatchGetSecretValue` ...), retry unprocessed
  entries and run with bounded concurrency.
- **No silent data loss.** Partial failures come back in a `BatchResult` and raise `PartialBatchError`
  by default.
- **Dynamic and typed.** Every method forwards extra `**kwargs` to the boto3 method it wraps, typed with
  the TypedDicts in `nimble_aws.pytypes`, generated from the botocore service models (the same source as
  the [boto3 reference](https://docs.aws.amazon.com/boto3/latest/reference/services/)).

## Installation

```bash
pip install nimble-aws            # adapters
pip install "nimble-aws[lambda]"  # + Lambda handler helpers (aws-lambda-powertools)
```

Python 3.11+ (Glue 5.x runs 3.11; Lambda supports up to 3.14).

## Usage

### Lambda

Create adapters at module level and run coroutines with `nimble_aws.run`, which keeps one event loop alive
for the whole execution environment. Warm invocations reuse the open clients.
Do **not** use `asyncio.run` per invocation: it closes the loop and the connections with it.

```python
from nimble_aws import run
from nimble_aws.adapter import DynamoDB, S3

s3 = S3()
ddb = DynamoDB()


async def handle(event):
    raw = await s3.get_object("my-bucket", event["key"], encoding="utf-8")
    await ddb.put_item("my-table", {"pk": event["key"], "size": len(raw)})
    return {"ok": True}


def lambda_handler(event, context):
    return run(handle(event))
```

With the handler helpers:

```python
from nimble_aws.handler import LambdaHandler, SQSEvent, event_source


class Handler(LambdaHandler):
    @event_source(data_class=SQSEvent)
    async def handle_event(self, event: SQSEvent):
        failed = []
        for record in event.records:
            ...
        return SQSEvent.batch_item_failures(failed)

    def generate_response(self, body):
        return body


lambda_handler = Handler.entrypoint(settings)
```

### Glue and other synchronous code

```python
from nimble_aws import run
from nimble_aws.adapter import S3

result = run(S3().get_object_batch("bucket", keys, encoding="utf-8"))
```

`run` also works inside notebooks (Glue interactive sessions), where an event loop is already running.

### ECS / long-running services

Use the adapters inside your own event loop and close the clients on shutdown:

```python
from nimble_aws import close_clients


async def on_sigterm():
    await close_clients()
```

### Batch methods

```python
from nimble_aws import PartialBatchError
from nimble_aws.adapter import SQS

sqs = SQS()
result = await sqs.send_message_batch(queue_url, [{"id": i} for i in range(1000)])
result.results  # one entry per input, in order (None where it failed)
result.failed  # [BatchFailure(index, item, code, message), ...]

try:
    await sqs.send_message_batch(queue_url, messages)
except PartialBatchError as error:
    retry = [f.item for f in error.result.failed]

# or decide yourself
result = await sqs.send_message_batch(queue_url, messages, raise_on_failure=False)
```

### Dynamic parameters

Any parameter of the underlying boto3 method can be passed as a keyword argument, and anything the adapter
does not wrap is one call away:

```python
await s3.put_object("bucket", "a.json", body, ContentType="application/json", Tagging="team=data")
await ddb.query("table", {"kind": "audio"}, index="kind-index", filters={"deleted": False}, ScanIndexForward=False)
await s3.call("get_bucket_location", Bucket="bucket")  # any operation
async for page in s3.paginate("list_object_versions", Bucket="bucket"):
    ...
```

## Configuration

| Variable | Default | |
|---|---|---|
| `NIMBLE_AWS_REGION` / `AWS_REGION` / `AWS_DEFAULT_REGION` / `REGION` | botocore default | Region of the adapters |
| `NIMBLE_AWS_MAX_CONCURRENCY` | 32 | Requests in flight per batch call |
| `NIMBLE_AWS_MAX_POOL_CONNECTIONS` | 64 | HTTP connections per client |
| `NIMBLE_AWS_CONNECT_TIMEOUT` / `NIMBLE_AWS_READ_TIMEOUT` | 3 s / 30 s | Lambda and Step Functions use longer read timeouts |
| `NIMBLE_AWS_RETRY_MODE` / `NIMBLE_AWS_MAX_ATTEMPTS` | `standard` / 3 | botocore retries |
| `NIMBLE_AWS_BATCH_MAX_ATTEMPTS` | 5 | Retries of unprocessed DynamoDB batch entries |
| `NIMBLE_AWS_SECRET_CACHE_TTL` | 300 s | Secrets Manager cache (0 disables) |

Every adapter also accepts `region=`, `config=` (an `AioConfig`), `endpoint_url=`, `max_concurrency=` and
`clients=` (an `AwsClients` cache) in its constructor.

## Adapters

| Adapter | Single | Batch |
|---|---|---|
| `S3` | `get_object`, `head_object`, `put_object`, `delete_object`, `list_objects`, `generate_presigned_url` | `*_batch` of each (`delete_object_batch` uses `DeleteObjects`) |
| `DynamoDB` | `get_item`, `put_item`, `delete_item`, `update_item`, `query`, `scan`, `describe_table` | `get_item_batch` (`BatchGetItem`), `put_item_batch`/`delete_item_batch` (`BatchWriteItem`), `update_item_batch`, `query_batch` |
| `SQS` | `send_message`, `receive_message`, `delete_message` | `send_message_batch`, `receive_message_batch`, `delete_message_batch` |
| `SNS` | `publish` | `publish_batch` |
| `EventBridge` | `put_event` | `put_event_batch` |
| `Lambda` | `invoke` | `invoke_batch` |
| `StepFunctions` | `start_sync_execution`, `start_execution` | `*_batch` of each |
| `Glue` | `get_jobs`, `get_job_runs`, `start_job_run` | `get_job_runs_batch`, `start_job_run_batch` |
| `SecretsManager` | `get_secret` (cached) | `get_secret_batch` (`BatchGetSecretValue`) |
| `RdsData` | `execute_statement`, `query`, `transaction()` | `execute_statement_batch` (`BatchExecuteStatement`) |
| `ApiGateway` | `post_to_connection`, `get_api_key` | `post_to_connection_batch` |
| `AppSync` | `list_api_keys`, `get_api_key`, `get_graphql_api`, `get_graphql_url` | |
| `Cognito` | `get_user`, `get_user_attributes` | |

## Types (`nimble_aws.pytypes`)

- `pytypes.<service>`: `<Op>RequestTypeDef`, `<Op>OptionsTypeDef` (the `**kwargs` of the adapter method),
  `<Op>ResponseTypeDef` and every nested shape, for each boto3 method the adapters call. Docstrings link to
  the boto3 reference page of the method. Regenerate after upgrading botocore:
  `python scripts/generate_pytypes.py`.
- `pytypes.common`: `BatchResult`, `BatchFailure`, `ResponseMetadataTypeDef`.
- `pytypes.models`: `OutgoingMessage`, `ReceivedMessage`, `PutObjectEntry`, `ListObjectsResult`,
  `ItemUpdate`, `QuerySpec`, `EventEntry`, `ExecutionRequest`, `JobRunRequest` ...

## Development

```bash
poetry install --with dev --extras lambda
poetry run pytest           # runs against a local moto server, no AWS account needed
poetry run ruff check .
poetry run pyright
```

## Migrating from 0.1

| 0.1 | 0.2 |
|---|---|
| `S3().get_files(bucket, paths, decode=True)` | `(await S3().get_object_batch(bucket, keys, encoding="utf-8")).results` |
| `put_file`, `put_files`, `put_files_with_content_type`, `put_files_with_metadata` | `put_object(...)` / `put_object_batch(bucket, [{"key", "body", "ContentType": ..., "Metadata": ...}])` |
| `list_files(bucket, prefixes)` | `list_objects_batch(bucket, prefixes, delimiter="/")` (all pages) |
| `head_files`, `generate_pre_signed_url(s)` | `head_object_batch`, `generate_presigned_url(_batch)` |
| `DynamoDB.put_items` / `delete_items` / `get_items` | `put_item_batch` / `delete_item_batch` / `get_item_batch` |
| `get_item` returned `{}` when missing | returns `None` |
| `update_items(table, items, key_fields)` | `update_item(table, key, attributes)` / `update_item_batch` |
| `index_query(table, index, filters)` | `query(table, {field: value}, index=name, filters=...)` |
| `scan_dynamo(table)` | `scan(table, segments=4)` |
| `SQS.publish_messages_to_queue` | `send_message_batch` |
| `read_ten_messages` / `read_messages` / `read_all_messages` | `receive_message` / `receive_message_batch(max_messages=...)`, returning `ReceivedMessage` (call `.json()`) |
| `SNS.publish_messages_to_topic(_with_attributes)` | `publish_batch(topic, messages or OutgoingMessage(...))` |
| `EventBridge.publish_events` | `put_event_batch` |
| `StepFunctions.run_sync_execution(s)` | `start_sync_execution(_batch)` (raises on failed executions) |
| `Glue.get_jobs_runs` | `get_job_runs_batch` |
| `AppSync.list_keys` / `get_key` / `get_domain` | `list_api_keys` / `get_api_key` / `get_graphql_url` |
| `RdsData.execute_statement(cluster_arn, secret_arn, database, sql)` | `RdsData(cluster_arn, secret_arn, database).execute_statement(sql, {"id": 1})` |
| `region=` on every method | `region=` on the adapter constructor |
| `nimble_aws.adapter.config.env` | `nimble_aws.get_settings()` |

## License

MIT. See [LICENSE](../LICENSE).

