amgi-sqs-event-source-mapping
amgi-sqs-event-source-mapping is an adaptor for AMGI applications to run in an SQS event source mapped Lambda.
Installation
pip install amgi-sqs-event-source-mapping==0.48.0
Example
This example uses AsyncFast:
from dataclasses import dataclass
from amgi_sqs_event_source_mapping import SqsEventSourceMappingHandler
from asyncfast import AsyncFast
app = AsyncFast()
@dataclass
class Order:
item_ids: list[str]
@app.channel("order-queue")
async def order_queue(order: Order) -> None:
# Makes an order
...
handler = SqsEventSourceMappingHandler(app)
What it does
- Converts SQS batch events into AMGI
message.receiveevents - Uses the SQS queue name as the AMGI message address
- Supports partial batch failures so only failed messages are retried
- Sends outbound messages back to SQS efficiently using batching
- Optionally manages application startup and shutdown via AMGI lifespan
- Verifies message integrity using the SQS-provided MD5 checksum and retries corrupted messages
Record handling
- Record bodies are passed to your app as bytes
- SQS record attributes become AMGI headers
- Records are only acknowledged when your app emits
message.ack - Records that are not acknowledged are treated as failures and will be retried
- Corrupted messages are detected automatically and retried
Lifespan
Lifespan support is enabled by default.
- Startup runs once per Lambda execution environment
- Shutdown is attempted when the environment is terminated
Shutdown handling relies on signal.SIGTERM, which is supported by Python 3.12 and later Lambda runtimes.
To use fully stateless, per-invocation behavior, disable lifespan:
handler = SqsEventSourceMappingHandler(app, lifespan=False)
Contact
For questions or suggestions, please contact jack.burridge@mail.com.
License
Copyright 2025 AMGI
1 maintainer
Metadata
1 maintainer
Release files for amgi-sqs-event-source-mapping 0.48.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| amgi_sqs_event_source_mapping-0.48.0-py3-none-any.whl | Python 3 | none | any | Details |
Release files / amgi_sqs_event_source_mapping-0.48.0-py3-none-any.whl
| Download URL | amgi_sqs_event_source_mapping-0.48.0-py3-none-any.whl |
|---|---|
| Size | 7.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
6ef5d9877cd0825689f31b7c054346f0250838dcbea588cf5a6b398ec6101857
|
|
BLAKE2b-256 checksum How to use checksums |
adce27311fea53b473d10e8600cc191c010e2bc71a4224fd3e54f7f6fa976229
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.0 {"installer":{"name":"uv","version":"0.13.0","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
