🌐 masrikdahir.com · GitHub

📋 Table of Contents

• ✨ Features
• 🏗️ Architecture
• 📁 Project Structure
• ⚙️ Prerequisites
• 🚀 Quick Start
• 📖 Usage
• 🔧 Configuration
• 🧪 Testing
• 🔄 CI/CD
• 📝 Changelog
• 🤝 Contributing
• 📄 License

✨ Features

• 🌐 135 Service Modules — Typed wrappers for S3, SQS, DynamoDB, Lambda, Bedrock, ECS, and 100+ more AWS services with 7,300+ method wrappers
• ⚡ Full Async Parity — Every sync function has a native async def counterpart in aws_util.aio.* using a custom aiohttp engine (no monkey-patching)
• 🛡️ Structured Exception Hierarchy — AwsThrottlingError, AwsNotFoundError, AwsPermissionError, etc. replace generic RuntimeError while staying backward-compatible
• 📦 Pydantic Result Models — Every API response is returned as an immutable BaseModel, not a raw dict
• 🔄 TTL Client Cache — Bounded LRU cache (64 entries, 15-min TTL) auto-rotates STS credentials in long-running processes
• 🔌 Multi-Service Orchestration — Pre-built modules for blue/green deployments, data lake management, cross-account operations, disaster recovery, and CI/CD pipelines
• 🛠️ Lambda Middleware — Idempotency, batch processing, timeout guards, cold-start tracking, and feature flags out of the box
• 🔐 Security & Compliance — Least-privilege analysis, secret rotation, data masking, VPC auditing, WAF management, and compliance snapshots
• 📊 Observability — Structured logging, X-Ray tracing, EMF metrics, CloudWatch alarms, and automated dashboard generation

🏗️ Architecture

graph TB
    subgraph App["🖥️ Your Application"]
        Sync["Sync API<br/>aws_util.*"]
        Async["Async API<br/>aws_util.aio.*"]
    end

    subgraph Core["⚙️ Core Layer"]
        Client["_client.py<br/>TTL LRU Cache"]
        Engine["aio/_engine.py<br/>aiohttp Transport"]
        Errors["exceptions.py<br/>Error Hierarchy"]
    end

    subgraph Services["☁️ AWS Services (135 modules)"]
        S3["S3 · SQS · SNS"]
        DB["DynamoDB · RDS · ElastiCache"]
        Compute["Lambda · ECS · EKS · Batch"]
        AI["Bedrock · SageMaker · Rekognition"]
        Security["IAM · KMS · Secrets Manager"]
        More["+ 100 more services"]
    end

    subgraph Orchestration["🔗 Multi-Service Orchestration"]
        Deploy["Deployment & Blue/Green"]
        DataLake["Data Lake & ETL"]
        Resilience["Resilience & Circuit Breaker"]
        Observe["Observability & Monitoring"]
    end

    Sync -->|boto3| Client
    Async -->|aiohttp + SigV4| Engine
    Client --> Services
    Engine --> Services
    Errors -.->|classifies| Services
    Services --> Orchestration

📁 Project Structure

📦 aws-util/
├── 📁 src/aws_util/              # 138 sync modules
│   ├── 🔑 __init__.py            # Top-level convenience imports
│   ├── ⚙️ _client.py             # TTL-aware boto3 client factory
│   ├── 🚨 exceptions.py          # Structured error hierarchy
│   ├── 📁 aio/                   # 137 async counterparts
│   │   ├── ⚙️ _engine.py         # Native aiohttp async engine
│   │   └── 📄 *.py               # One async module per service
│   ├── 📄 s3.py                   # S3 operations
│   ├── 📄 dynamodb.py             # DynamoDB operations
│   ├── 📄 lambda_.py              # Lambda management
│   ├── 📄 bedrock.py              # Bedrock AI services
│   ├── 📄 blue_green.py           # Blue/green deployments
│   ├── 📄 data_lake.py            # Data lake management
│   ├── 📄 observability.py        # Monitoring & logging
│   └── 📄 ... (125+ more)        # All other service modules
├── 📁 tests/                      # 276 test files, 100% coverage
├── 📁 .github/workflows/          # 20 CI/CD workflows
├── ⚙️ pyproject.toml              # Build config & task runner
├── ⚙️ Pipfile                     # Dependency management
└── 📖 README.md

graph LR
    Root["📦 aws-util"] --> Src["📁 src/aws_util"]
    Root --> Tests["🧪 tests (276 files)"]
    Root --> GH["📁 .github/workflows"]
    Src --> SyncMods["📄 135 sync modules"]
    Src --> AIO["📁 aio/"]
    AIO --> AsyncMods["📄 137 async modules"]
    AIO --> Engine["⚙️ _engine.py"]
    SyncMods -->|"1:1 parity"| AsyncMods

⚙️ Prerequisites

Before you begin, make sure you have the following installed:

💡 Tip: Configure AWS credentials via aws configure or environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION).

🚀 Quick Start

1. Clone the repository

git clone https://github.com/Masrik-Dahir/aws-util-python.git
cd aws-util-python

2. Install dependencies

pip install -e .[dev]
# Or with pipenv:
PIPENV_IGNORE_VIRTUALENVS=1 pipenv install --dev

3. Install from PyPI (library usage)

pip install aws-util

4. Use it — Sync

from aws_util.s3 import upload_file, download_bytes, list_objects
from aws_util.dynamodb import get_item, put_item
from aws_util.sqs import send_message, receive_messages

# Upload a file to S3
upload_file("my-bucket", "data/report.csv", "/tmp/report.csv")

# Query DynamoDB
item = get_item("users-table", {"pk": "user#123"})

# Send an SQS message
send_message("my-queue", {"event": "order.placed", "id": "abc"})

5. Use it — Async

from aws_util.aio.s3 import upload_file, download_bytes
from aws_util.aio.dynamodb import get_item, put_item

async def main():
    await upload_file("my-bucket", "key.csv", "/tmp/data.csv")
    item = await get_item("users-table", {"pk": "user#123"})

📖 Usage

Placeholder Resolution (SSM + Secrets Manager)

from aws_util import retrieve

# Resolves SSM parameters and Secrets Manager secrets
db_url = retrieve("{{ssm:/prod/db/url}}")
api_key = retrieve("{{secret:prod/api-key}}")

Multi-Service Orchestration

from aws_util.deployer import deploy_lambda_with_config, deploy_ecs_from_ecr
from aws_util.data_pipeline import run_glue_then_query, export_query_to_s3_json
from aws_util.security_ops import audit_public_s3_buckets, rotate_iam_access_key
from aws_util.blue_green import ecs_blue_green_deployer

Lambda Middleware

from aws_util.lambda_middleware import (
    idempotent_handler, batch_processor, middleware_chain,
    lambda_timeout_guard, cold_start_tracker, lambda_response,
)

Event-Driven Orchestration

from aws_util.event_orchestration import (
    create_eventbridge_rule, saga_orchestrator, fan_out_fan_in,
    create_schedule, run_workflow,
)

Resilience & Error Handling

from aws_util.resilience import circuit_breaker, retry_with_backoff, dlq_monitor_and_alert
from aws_util.exceptions import AwsThrottlingError, AwsNotFoundError

try:
    item = get_item("table", {"pk": "key"})
except AwsNotFoundError:
    print("Resource not found")
except AwsThrottlingError:
    print("Rate limited — retrying")

Observability

from aws_util.observability import (
    StructuredLogger, create_xray_trace, emit_emf_metric,
    create_lambda_alarms, generate_lambda_dashboard,
)

Security & Compliance

from aws_util.security_compliance import (
    least_privilege_analyzer, secret_rotation_orchestrator,
    data_masking_processor, compliance_snapshot,
)

Cost Optimization

from aws_util.cost_optimization import (
    lambda_right_sizer, unused_resource_finder,
    concurrency_optimizer, cost_attribution_tagger,
)

AI/ML Pipelines

from aws_util.ai_ml_pipelines import (
    bedrock_serverless_chain, s3_document_processor,
    image_moderation_pipeline, translation_pipeline,
)

All 135 Service Modules

🔧 Configuration

The library uses boto3's standard credential chain. Credentials can also come from IAM roles, instance profiles, SSO, or environment variables.

🧪 Testing

# Run all tests
task test

# Run tests with coverage report
task test-cov

# Run mutation testing
task mutation-test

# Run coverage with 100% enforcement
task coverage

# Lint and format
task lint

# Type checking
task typecheck

# Full pre-release check (lint + test + coverage + typecheck)
task prepare

Test Stats

🔄 CI/CD

This project uses 20 GitHub Actions workflows for comprehensive automation.

Pipeline Flow

flowchart LR
    PR[Pull Request] --> Lint[🔍 Lint]
    PR --> Type[📝 Typecheck]
    Lint --> Test[🧪 Test]
    Type --> Test
    Test --> Cov[📊 Coverage 100%]
    Test --> Mut[🧬 Mutation]
    Cov --> |merge to master| Build[📦 Build]
    Build --> |tag v*| Publish[🚀 PyPI]

All checks must pass before merging. See .github/workflows/ for full configuration.

📝 Changelog

See CHANGELOG.md for the full history.

🤝 Contributing

Contributions are welcome! Please follow these steps:

1. Fork the repository
2. Create a feature branch: git checkout -b feat/amazing-feature
3. Commit your changes: git commit -m 'feat: add amazing feature'
4. Push to the branch: git push origin feat/amazing-feature
5. Open a Pull Request

Development Setup

git clone https://github.com/Masrik-Dahir/aws-util-python.git
cd aws-util-python
pip install -e .[dev]
task prepare   # lint + test + coverage + typecheck

Commit Convention

This project uses Conventional Commits:

Please ensure all tests pass and coverage stays at 100% before opening a PR.

📄 License

Distributed under the MIT License. See LICENSE for more information.