| Name | finalsa-sqs-client JSON |
| Version |
2.1.1
JSON |
| download |
| home_page | None |
| Summary | finalsa-sqs-client is a Python package for interacting with AWS SQS queues. |
| upload_time | 2025-08-19 02:21:26 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.10 |
| license | MIT License Copyright (c) 2021 Luis Diego Jiménez Delgado Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
aws
common
finalsa
lambda
utilities
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# finalsa-sqs-client
Thin, typed wrapper around AWS SQS using boto3 with a simple in-memory test implementation.
- Python 3.10+
- Adds W3C trace headers to all messages by default (traceparent, tracestate)
- Accepts dict or JSON string payloads
- Returns a convenient `SqsReponse` model with flattened message attributes
## Installation
Using pip or uv:
```bash
pip install finalsa-sqs-client
# or
uv add finalsa-sqs-client
```
Requires valid AWS credentials for the default boto3 client (env vars, config file, or IAM role).
## Quick start
```python
from finalsa.sqs.client import SqsServiceImpl, SqsException
svc = SqsServiceImpl()
# Resolve a queue URL by name (helper)
queue_url = svc.get_queue_url("my-queue")
# Send a JSON payload (dict) with optional custom attributes
svc.send_message(queue_url, {"hello": "world"}, {"tenant": "acme"})
# Receive up to 10 messages and delete them
messages = svc.receive_messages(queue_url, max_number_of_messages=10, wait_time_seconds=1)
for msg in messages:
print("body:", msg.body) # raw JSON string
print("attrs:", msg.message_attributes) # flattened: {"traceparent": "...", ...}
svc.delete_message(queue_url, msg.receipt_handle)
# Other helpers
arn = svc.get_queue_arn(queue_url)
attrs = svc.get_queue_attributes(queue_url)
```
Errors are raised as `SqsException` wrapping the underlying boto3 error when applicable.
## Message attributes and tracing
- Custom attributes can be passed as a simple dict of strings: `{"key": "value"}`.
- The client adds two trace headers automatically to every send:
- `traceparent`
- `tracestate`
- When receiving messages, attributes are returned flattened into a `dict[str, str]` on `SqsReponse.message_attributes`.
Example sending raw string JSON and custom attributes:
```python
svc.send_raw_message(queue_url, '{"a":1}', {"source": "ingestor"})
```
## In-memory test service
For local tests or offline development, use the built-in, in-memory implementation:
```python
from finalsa.sqs.client import SqsServiceTest
svc = SqsServiceTest()
svc.send_message("test-queue", {"x": 1})
msgs = svc.receive_messages("test-queue")
assert len(msgs) == 1
svc.delete_message("test-queue", msgs[0].receipt_handle)
```
Notes:
- `SqsServiceTest` keeps messages in process memory keyed by the queue URL/name you pass.
- It also injects the same trace attributes as the real client so behavior matches production closely.
## API overview
The primary entry points are exported from `finalsa.sqs.client`:
- `SqsService`: abstract interface
- `SqsServiceImpl`: boto3-backed implementation
- `SqsServiceTest`: in-memory implementation for tests
- `SqsException`: error wrapper
Key methods (see docstrings for details):
- `receive_messages(queue_url, max_number_of_messages=1, wait_time_seconds=1) -> list[SqsReponse]`
- `send_message(queue_url, payload: dict, message_attributes: dict | None = None) -> None`
- `send_raw_message(queue_url, data: dict | str, message_attributes: dict | None = None) -> None`
- `delete_message(queue_url, receipt_handle) -> None`
- `get_queue_arn(queue_url) -> str`
- `get_queue_attributes(queue_url) -> dict`
- `get_queue_url(queue_name) -> str`
Payload handling:
- Dict payloads are serialized with orjson (compact) before sending.
- String payloads are sent as-is.
## Credentials and configuration
`SqsServiceImpl` uses the default `boto3.client('sqs')`. Configure credentials/region via any standard boto3 method:
- Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION, ...)
- AWS config/credentials files
- IAM role when running in AWS
## Development
Run tests with uv (recommended):
```bash
uv sync
uv run pytest -q
```
Or with plain pip:
```bash
python -m venv .venv && source .venv/bin/activate
pip install -e .[test]
pytest -q
```
## License
MIT – see `LICENSE.md`.
Raw data
{
"_id": null,
"home_page": null,
"name": "finalsa-sqs-client",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "aws, common, finalsa, lambda, utilities",
"author": null,
"author_email": "Luis Jimenez <luis@finalsa.com>",
"download_url": "https://files.pythonhosted.org/packages/68/4c/b4e1d82e2a6c283a45ace1d34b28a537084277b4b3fa9e0715bf07339a58/finalsa_sqs_client-2.1.1.tar.gz",
"platform": null,
"description": "# finalsa-sqs-client\n\nThin, typed wrapper around AWS SQS using boto3 with a simple in-memory test implementation.\n\n- Python 3.10+\n- Adds W3C trace headers to all messages by default (traceparent, tracestate)\n- Accepts dict or JSON string payloads\n- Returns a convenient `SqsReponse` model with flattened message attributes\n\n\n## Installation\n\nUsing pip or uv:\n\n```bash\npip install finalsa-sqs-client\n# or\nuv add finalsa-sqs-client\n```\n\nRequires valid AWS credentials for the default boto3 client (env vars, config file, or IAM role).\n\n\n## Quick start\n\n```python\nfrom finalsa.sqs.client import SqsServiceImpl, SqsException\n\nsvc = SqsServiceImpl()\n\n# Resolve a queue URL by name (helper)\nqueue_url = svc.get_queue_url(\"my-queue\")\n\n# Send a JSON payload (dict) with optional custom attributes\nsvc.send_message(queue_url, {\"hello\": \"world\"}, {\"tenant\": \"acme\"})\n\n# Receive up to 10 messages and delete them\nmessages = svc.receive_messages(queue_url, max_number_of_messages=10, wait_time_seconds=1)\nfor msg in messages:\n\t\tprint(\"body:\", msg.body) # raw JSON string\n\t\tprint(\"attrs:\", msg.message_attributes) # flattened: {\"traceparent\": \"...\", ...}\n\t\tsvc.delete_message(queue_url, msg.receipt_handle)\n\n# Other helpers\narn = svc.get_queue_arn(queue_url)\nattrs = svc.get_queue_attributes(queue_url)\n```\n\nErrors are raised as `SqsException` wrapping the underlying boto3 error when applicable.\n\n\n## Message attributes and tracing\n\n- Custom attributes can be passed as a simple dict of strings: `{\"key\": \"value\"}`.\n- The client adds two trace headers automatically to every send:\n\t- `traceparent`\n\t- `tracestate`\n- When receiving messages, attributes are returned flattened into a `dict[str, str]` on `SqsReponse.message_attributes`.\n\nExample sending raw string JSON and custom attributes:\n\n```python\nsvc.send_raw_message(queue_url, '{\"a\":1}', {\"source\": \"ingestor\"})\n```\n\n\n## In-memory test service\n\nFor local tests or offline development, use the built-in, in-memory implementation:\n\n```python\nfrom finalsa.sqs.client import SqsServiceTest\n\nsvc = SqsServiceTest()\nsvc.send_message(\"test-queue\", {\"x\": 1})\nmsgs = svc.receive_messages(\"test-queue\")\nassert len(msgs) == 1\nsvc.delete_message(\"test-queue\", msgs[0].receipt_handle)\n```\n\nNotes:\n- `SqsServiceTest` keeps messages in process memory keyed by the queue URL/name you pass.\n- It also injects the same trace attributes as the real client so behavior matches production closely.\n\n\n## API overview\n\nThe primary entry points are exported from `finalsa.sqs.client`:\n\n- `SqsService`: abstract interface\n- `SqsServiceImpl`: boto3-backed implementation\n- `SqsServiceTest`: in-memory implementation for tests\n- `SqsException`: error wrapper\n\nKey methods (see docstrings for details):\n- `receive_messages(queue_url, max_number_of_messages=1, wait_time_seconds=1) -> list[SqsReponse]`\n- `send_message(queue_url, payload: dict, message_attributes: dict | None = None) -> None`\n- `send_raw_message(queue_url, data: dict | str, message_attributes: dict | None = None) -> None`\n- `delete_message(queue_url, receipt_handle) -> None`\n- `get_queue_arn(queue_url) -> str`\n- `get_queue_attributes(queue_url) -> dict`\n- `get_queue_url(queue_name) -> str`\n\nPayload handling:\n- Dict payloads are serialized with orjson (compact) before sending.\n- String payloads are sent as-is.\n\n\n## Credentials and configuration\n\n`SqsServiceImpl` uses the default `boto3.client('sqs')`. Configure credentials/region via any standard boto3 method:\n- Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION, ...)\n- AWS config/credentials files\n- IAM role when running in AWS\n\n\n## Development\n\nRun tests with uv (recommended):\n\n```bash\nuv sync\nuv run pytest -q\n```\n\nOr with plain pip:\n\n```bash\npython -m venv .venv && source .venv/bin/activate\npip install -e .[test]\npytest -q\n```\n\n\n## License\n\nMIT \u2013 see `LICENSE.md`.\n\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2021 Luis Diego Jim\u00e9nez Delgado Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.",
"summary": "finalsa-sqs-client is a Python package for interacting with AWS SQS queues.",
"version": "2.1.1",
"project_urls": {
"Homepage": "https://github.com/finalsa/finalsa-common-lambda"
},
"split_keywords": [
"aws",
" common",
" finalsa",
" lambda",
" utilities"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "fd66d68be70b38534248c5380a0bf2f87bc641ee2403e06c079fd1b6120144a0",
"md5": "4fc16042336165f02224adfd1228656c",
"sha256": "bce94a20c6b17726e88b4be2f61632f23eb20722aaec2ed8dc85a21bd8f3be81"
},
"downloads": -1,
"filename": "finalsa_sqs_client-2.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "4fc16042336165f02224adfd1228656c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 8958,
"upload_time": "2025-08-19T02:21:24",
"upload_time_iso_8601": "2025-08-19T02:21:24.712182Z",
"url": "https://files.pythonhosted.org/packages/fd/66/d68be70b38534248c5380a0bf2f87bc641ee2403e06c079fd1b6120144a0/finalsa_sqs_client-2.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "684cb4e1d82e2a6c283a45ace1d34b28a537084277b4b3fa9e0715bf07339a58",
"md5": "5840b84355c84fa2b274012ff8ab4e2e",
"sha256": "01c67e577d639d16dfa97209961d302ec0a4fc06e5def11ae95281c8da6cc1c1"
},
"downloads": -1,
"filename": "finalsa_sqs_client-2.1.1.tar.gz",
"has_sig": false,
"md5_digest": "5840b84355c84fa2b274012ff8ab4e2e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 6066,
"upload_time": "2025-08-19T02:21:26",
"upload_time_iso_8601": "2025-08-19T02:21:26.750156Z",
"url": "https://files.pythonhosted.org/packages/68/4c/b4e1d82e2a6c283a45ace1d34b28a537084277b4b3fa9e0715bf07339a58/finalsa_sqs_client-2.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-19 02:21:26",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "finalsa",
"github_project": "finalsa-common-lambda",
"github_not_found": true,
"lcname": "finalsa-sqs-client"
}
