ACB Documentation: Main | Core Systems | Actions | Adapters
ACB is a modular Python framework for building asynchronous applications with pluggable components. It provides a collection of self-contained actions and flexible adapters that integrate with various systems, along with a dynamic configuration and dependency injection system.
- Modular Architecture: Mix and match components to build your application
- Asynchronous First: Built for high-performance async operations
- Pluggable Adapters: Swap implementations without changing your code
- Automatic Discovery: Components are automatically discovered and registered
- Configuration-Driven: Change behavior through configuration rather than code
- Type Safety: Built on Pydantic for validation and type safety
- Installation
- Quick Start
- Architecture Overview
- Core Components
- Use Cases
- Built-in Components
- Advanced Usage
- Documentation
- Acknowledgements
- License
- Projects Using ACB
- Contributing
Install ACB using pdm:
pdm add acb
ACB supports various optional dependencies for different adapters and functionality:
Feature Group | Components | Installation Command |
---|---|---|
Redis | Cache, NoSQL | pdm add "acb[redis]" |
SQL | Database (MySQL, PostgreSQL) | pdm add "acb[sql]" |
NoSQL | Database (MongoDB, Firestore, Redis) | pdm add "acb[nosql]" |
Storage | File storage (S3, GCS, Azure, local) | pdm add "acb[storage]" |
Logging | Loguru, structlog | pdm add "acb[logging]" |
Compression | gzip, brotli | pdm add "acb[compression]" |
Serialization | JSON, YAML, TOML, MsgPack | pdm add "acb[serialization]" |
Multiple Features | Combined dependencies | pdm add "acb[redis,sql,nosql]" |
Web Application | Typical web app stack | pdm add "acb[redis,sql,storage]" |
Complete | All dependencies | pdm add "acb[all]" |
ACB follows a component-based architecture with automatic discovery and registration of modules:
acb/
├── actions/ # Reusable utility functions (compress, encode, hash)
├── adapters/ # Integration modules for external systems
│ ├── cache/ # Memory and Redis caching
│ ├── dns/ # DNS management
│ ├── logger/ # Logging adapters (Loguru, structlog)
│ ├── sql/ # Database adapters (MySQL, PostgreSQL)
│ ├── nosql/ # NoSQL adapters (MongoDB, Firestore, Redis)
│ ├── storage/ # Storage adapters (S3, GCS, Azure, local)
│ └── ... # Additional adapter categories
├── config.py # Configuration system using Pydantic
├── depends.py # Dependency injection framework
└── ...
ACB is structured around these fundamental building blocks:
Actions are modular, self-contained utility functions that perform specific tasks. They are automatically discovered and registered, making them immediately available throughout your application.
Action Category | Description | Implementations |
---|---|---|
Compress/Decompress | Efficient data compression | gzip, brotli |
Encode/Decode | Data serialization | JSON, YAML, TOML, MsgPack |
Hash | Secure hashing functions | blake3, crc32c, md5 |
Example: Using the compress action
from acb.actions.compress import compress, decompress
# Compress some text data using brotli
compressed = compress.brotli("Hello, ACB!", level=3)
# Decompress back to the original text
original = decompress.brotli(compressed)
For more detailed documentation on actions, see the Actions README.
Adapters provide standardized interfaces to external systems and services. Each adapter category includes a base class that defines the interface and multiple implementations.
Adapter Category | Description | Implementations |
---|---|---|
Cache | Data caching | Memory, Redis |
DNS | Domain name management | Cloud DNS |
FTP/SFTP | File transfer protocols | FTP, SFTP |
Logger | Structured logging | Loguru, structlog |
Database | Data persistence | SQL (MySQL, PostgreSQL), NoSQL (MongoDB, Firestore, Redis) |
Storage | File storage | File, Memory, S3, Cloud Storage, Azure |
Secret | Secret management | Infisical, Secret Manager |
Example: Using the cache adapter
from acb.depends import depends
from acb.adapters import import_adapter
# Import the cache adapter (automatically uses the one enabled in config)
Cache = import_adapter("cache")
# Get the cache instance via dependency injection
cache = depends.get(Cache)
# Use the cache with a consistent API regardless of implementation
await cache.set("my_key", "my_value", ttl=60)
value = await cache.get("my_key")
For more detailed documentation on adapters, see the Adapters README.
ACB's configuration system is built on Pydantic and supports multiple configuration sources:
- YAML Files: Environment-specific settings in YAML format
- Secret Files: Secure storage of sensitive information
- Secret Managers: Integration with external secret management services
from acb.config import Settings
from pydantic import SecretStr
class MyServiceSettings(Settings):
api_url: str = "https://api.example.com"
api_version: str = "v1"
timeout: int = 30
max_retries: int = 3
api_key: SecretStr = SecretStr("default-key") # Automatically handled securely
ACB features a simple yet powerful dependency injection system that makes component wiring automatic and testable.
Example: Using dependency injection
from acb.depends import depends
from acb.config import Config
# Register your custom component
depends.set(MyService)
# Inject dependencies into functions
@depends.inject
def process_data(data, config: Config = depends(), logger = depends("logger")):
logger.info(f"Processing data with app: {config.app.name}")
# Process data...
return result
- Create a new project with PDM:
mkdir myapp
cd myapp
pdm init
pdm add acb
- Create a basic application structure:
myapp/
├── myapp/
│ ├── __init__.py
│ ├── actions/
│ │ └── __init__.py
│ ├── adapters/
│ │ └── __init__.py
│ └── main.py
└── settings/
├── app.yml
├── debug.yml
└── adapters.yml
- Initialize ACB in your main.py:
from acb import register_pkg
from acb.depends import depends
from acb.config import Config
# Register your package with ACB
register_pkg()
# Access configuration
config = depends.get(Config)
# Import adapters
Logger = depends.get("logger")
async def main():
Logger.info(f"Starting {config.app.name} application")
# Your application logic here
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Create your initial configuration files:
app:
name: "MyApp"
title: "My ACB Application"
domain: "myapp.example.com"
version: "0.1.0"
# Choose your adapter implementations
cache: memory
logger: loguru
storage: file
ACB provides comprehensive debugging tools to help troubleshoot your application:
- Granular debug levels: Control debug output for specific components
- Pretty-printed output: Human-readable debug information
- Performance timers: Measure and optimize execution time
The framework includes a robust logging system with structured logging and multiple output formats:
- Asynchronous logging: Non-blocking log operations
- Structured data: JSON-formatted logs for better analysis
- Multiple adapters: Choose between different logging implementations
Create your own reusable actions by adding Python modules to the actions directory:
# myapp/actions/validate.py
from pydantic import BaseModel
class Validate(BaseModel):
@staticmethod
def email(email: str) -> bool:
"""Validate an email address"""
import re
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
return bool(re.match(pattern, email))
validate = Validate()
Extend ACB with your own adapters to integrate with additional services:
- Create the adapter directory structure:
myapp/adapters/payment/
├── __init__.py
├── _base.py
├── stripe.py
└── paypal.py
- Define the base interface in
_base.py
:
from acb.config import AdapterBase, Settings
class PaymentBaseSettings(Settings):
currency: str = "USD"
default_timeout: int = 30
class PaymentBase(AdapterBase):
async def charge(self, amount: float, description: str) -> str:
"""Charge a payment and return a transaction ID"""
raise NotImplementedError()
async def refund(self, transaction_id: str) -> bool:
"""Refund a previous transaction"""
raise NotImplementedError()
- Implement specific providers like
stripe.py
:
from ._base import PaymentBase, PaymentBaseSettings
from pydantic import SecretStr
import stripe
class StripeSettings(PaymentBaseSettings):
api_key: SecretStr = SecretStr("sk_test_default")
class Stripe(PaymentBase):
settings: StripeSettings = None
async def init(self) -> None:
stripe.api_key = self.settings.api_key.get_secret_value()
async def charge(self, amount: float, description: str) -> str:
response = await stripe.PaymentIntent.create(
amount=int(amount * 100), # Convert to cents
currency=self.settings.currency,
description=description
)
return response.id
For more detailed documentation about ACB components:
- Core Systems: Configuration, dependency injection, debugging, and logging
- Actions: Detailed guide to built-in actions and creating custom ones
- Adapters: Comprehensive documentation on adapter system and implementations
- Cache Adapter: Memory and Redis caching
- SQL Adapter: SQL database connections
- Storage Adapter: File and object storage
ACB "blocks" logo used by permission from Andy Coe Band.
Special thanks to the following open-source projects for powering ACB:
This project is licensed under the terms of the BSD 3-Clause license.
Here are some notable projects built with ACB:
- FastBlocks: A rapid development framework that leverages ACB's asynchronous components to build scalable web applications.
Contributions to ACB are welcome! We follow a workflow inspired by the Crackerjack development guidelines. To ensure consistency and high quality, please adhere to the following steps when contributing:
-
Fork and Clone Fork the repository and clone it locally.
-
Set Up Your Development Environment Use PDM for dependency and virtual environment management. Add ACB as a development dependency:
pdm add -G dev acb
-
Run Pre-commit Hooks & Tests Before submitting a pull request, ensure your changes pass all quality checks and tests. We recommend running the following command (which is inspired by Crackerjack's automated workflow):
python -m crackerjack -x -t -p <version> -c
Alternatively, you can use:
python -m crackerjack -a <version>
This command cleans your code, runs linting, tests, bumps the version (micro, minor, or major), and commits changes.
-
Submit a Pull Request If everything passes, submit a pull request describing your changes. Include details about your contribution and reference any related issues.
-
Feedback and Review Our maintainers will review your changes. Please be responsive to feedback and be prepared to update your pull request as needed.
For more detailed development guidelines and the Crackerjack philosophy that influences our workflow, please refer to our internal development documentation.
Happy contributing!