| Name | abs-utils JSON |
| Version |
0.3.1
JSON |
| download |
| home_page | None |
| Summary | AutoBridge Systems Utility Library |
| upload_time | 2025-07-10 11:35:33 |
| maintainer | None |
| docs_url | None |
| author | AutoBridgeSystems |
| requires_python | <4.0,>=3.13 |
| license | None |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Utils Package
This package provides utility functions and classes for common operations in the application, including logging and Azure Service Bus integration.
## Installation
The package is managed using Poetry. To install dependencies:
```bash
poetry install
```
## Features
### Logger
A simple logging utility that provides a standardized way to set up logging across the application.
#### Usage
```python
from abs_utils.logger import setup_logger
# Create a logger instance
logger = setup_logger("my_module")
# Use the logger
logger.info("This is an info message")
logger.error("This is an error message")
```
### Azure Service Bus Integration
Provides functionality to interact with Azure Service Bus for message queuing and event handling.
#### AzureServiceBus Class
A wrapper class for Azure Service Bus operations.
```python
from abs_utils.azure_service_bus import AzureServiceBus
# Initialize the service bus client
service_bus = AzureServiceBus(
connection_string="your_connection_string",
queue_name="your_queue_name"
)
# Send a message
await service_bus.send({"key": "value"})
```
#### Event Decorator
A decorator that automatically sends events to Azure Service Bus after function execution. This is particularly useful for tracking entity operations in your application.
```python
from fastapi import APIRouter, Depends, Request
from dependency_injector.wiring import inject, Provide
from abs_utils.azure_service_bus import azure_event_decorator, AzureServiceBus
router = APIRouter()
@router.post("/{entity_name}/records")
@inject
@azure_event_decorator(event_type="record_created")
async def create_record(
entity_name: str,
data: dict,
request: Request,
azure_service_bus: AzureServiceBus = Depends(Provide[Container.azure_service_bus]),
service: YourService = Depends(Provide[Container.yourService])
):
# Your function logic here
return await service.create(data, entity_name)
```
The decorator automatically creates and sends an event payload with the following structure:
```json
{
"event_id": "uuid",
"event_type": "record_created",
"entity_name": "your_entity_name",
"entity_id": "record_id",
"payload": {
// Your data payload
},
"user": {
"id": "user_id",
"uuid": "user_uuid",
"email": "user_email",
"name": "user_name"
}
}
```
Key features of the decorator:
1. Automatically captures the entity name from the route parameters
2. Extracts user information from the request state
3. Generates a unique event ID for each event
4. Handles errors gracefully with logging
5. Works seamlessly with FastAPI dependency injection
## Dependencies
- Python 3.x
- azure-servicebus
- fastapi
- logging
- dependency-injector
## Contributing
When adding new utilities to this package:
1. Place new modules in the appropriate subdirectory under `abs_utils/`
2. Add proper documentation and type hints
3. Update this README with usage examples
4. Add tests for new functionality
Raw data
{
"_id": null,
"home_page": null,
"name": "abs-utils",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.13",
"maintainer_email": null,
"keywords": null,
"author": "AutoBridgeSystems",
"author_email": "info@autobridgesystems.com",
"download_url": "https://files.pythonhosted.org/packages/ff/d6/9f1bd2337a857d065265023197676c1673c4faab822431be1ec701fbe043/abs_utils-0.3.1.tar.gz",
"platform": null,
"description": "# Utils Package\n\nThis package provides utility functions and classes for common operations in the application, including logging and Azure Service Bus integration.\n\n## Installation\n\nThe package is managed using Poetry. To install dependencies:\n\n```bash\npoetry install\n```\n\n## Features\n\n### Logger\n\nA simple logging utility that provides a standardized way to set up logging across the application.\n\n#### Usage\n\n```python\nfrom abs_utils.logger import setup_logger\n\n# Create a logger instance\nlogger = setup_logger(\"my_module\")\n\n# Use the logger\nlogger.info(\"This is an info message\")\nlogger.error(\"This is an error message\")\n```\n\n### Azure Service Bus Integration\n\nProvides functionality to interact with Azure Service Bus for message queuing and event handling.\n\n#### AzureServiceBus Class\n\nA wrapper class for Azure Service Bus operations.\n\n```python\nfrom abs_utils.azure_service_bus import AzureServiceBus\n\n# Initialize the service bus client\nservice_bus = AzureServiceBus(\n connection_string=\"your_connection_string\",\n queue_name=\"your_queue_name\"\n)\n\n# Send a message\nawait service_bus.send({\"key\": \"value\"})\n```\n\n#### Event Decorator\n\nA decorator that automatically sends events to Azure Service Bus after function execution. This is particularly useful for tracking entity operations in your application.\n\n```python\nfrom fastapi import APIRouter, Depends, Request\nfrom dependency_injector.wiring import inject, Provide\nfrom abs_utils.azure_service_bus import azure_event_decorator, AzureServiceBus\n\nrouter = APIRouter()\n\n@router.post(\"/{entity_name}/records\")\n@inject\n@azure_event_decorator(event_type=\"record_created\")\nasync def create_record(\n entity_name: str,\n data: dict,\n request: Request,\n azure_service_bus: AzureServiceBus = Depends(Provide[Container.azure_service_bus]),\n service: YourService = Depends(Provide[Container.yourService])\n):\n # Your function logic here\n return await service.create(data, entity_name)\n```\n\nThe decorator automatically creates and sends an event payload with the following structure:\n```json\n{\n \"event_id\": \"uuid\",\n \"event_type\": \"record_created\",\n \"entity_name\": \"your_entity_name\",\n \"entity_id\": \"record_id\",\n \"payload\": {\n // Your data payload\n },\n \"user\": {\n \"id\": \"user_id\",\n \"uuid\": \"user_uuid\",\n \"email\": \"user_email\",\n \"name\": \"user_name\"\n }\n}\n```\n\nKey features of the decorator:\n1. Automatically captures the entity name from the route parameters\n2. Extracts user information from the request state\n3. Generates a unique event ID for each event\n4. Handles errors gracefully with logging\n5. Works seamlessly with FastAPI dependency injection\n\n## Dependencies\n\n- Python 3.x\n- azure-servicebus\n- fastapi\n- logging\n- dependency-injector\n\n## Contributing\n\nWhen adding new utilities to this package:\n1. Place new modules in the appropriate subdirectory under `abs_utils/`\n2. Add proper documentation and type hints\n3. Update this README with usage examples\n4. Add tests for new functionality\n",
"bugtrack_url": null,
"license": null,
"summary": "AutoBridge Systems Utility Library",
"version": "0.3.1",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "e1652fdaa6b7b31448c9cabc23fb7dccdfd2977e66c7b4d01ce61e5d39937b52",
"md5": "206f865fd7175c9d033273c17fa97aa8",
"sha256": "694e54a2a209a1df5709760fbd0b8919e176724bf3f381f5c882ca092c6bc5b4"
},
"downloads": -1,
"filename": "abs_utils-0.3.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "206f865fd7175c9d033273c17fa97aa8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.13",
"size": 11139,
"upload_time": "2025-07-10T11:35:32",
"upload_time_iso_8601": "2025-07-10T11:35:32.529074Z",
"url": "https://files.pythonhosted.org/packages/e1/65/2fdaa6b7b31448c9cabc23fb7dccdfd2977e66c7b4d01ce61e5d39937b52/abs_utils-0.3.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "ffd69f1bd2337a857d065265023197676c1673c4faab822431be1ec701fbe043",
"md5": "5cd3c2d11da6ae2a951285360e3acaa3",
"sha256": "18947557479742570e0c08d8da8874093ee856ec6b15363b323e52b12bbcd73c"
},
"downloads": -1,
"filename": "abs_utils-0.3.1.tar.gz",
"has_sig": false,
"md5_digest": "5cd3c2d11da6ae2a951285360e3acaa3",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.13",
"size": 7870,
"upload_time": "2025-07-10T11:35:33",
"upload_time_iso_8601": "2025-07-10T11:35:33.628051Z",
"url": "https://files.pythonhosted.org/packages/ff/d6/9f1bd2337a857d065265023197676c1673c4faab822431be1ec701fbe043/abs_utils-0.3.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-10 11:35:33",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "abs-utils"
}
