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.51.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.51.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.51.0-py3-none-any.whl | Python 3 | none | any | Details |
Release files / amgi_sqs_event_source_mapping-0.51.0-py3-none-any.whl
| Download URL | amgi_sqs_event_source_mapping-0.51.0-py3-none-any.whl |
|---|---|
| Size | 7.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
b1e47d1e2937a90b9621d726fafd6b72bcb498d8f5c0311105b18472eaef9687
|
|
BLAKE2b-256 checksum How to use checksums |
7399134740ffcb566972f6ff15601f3025415b25a01a8f71dca5bd2db2239732
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.5 {"installer":{"name":"uv","version":"0.13.5","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}
|
Security for amgi-sqs-event-source-mapping 0.51.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
