Nimble-AWS

Asynchronous, typed adapters for AWS services, built on 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).

Installation

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.

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:

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

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:

from nimble_aws import close_clients


async def on_sigterm():
    await close_clients()

Batch methods

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:

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

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

Adapters

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

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

License

MIT. See LICENSE.