lucid-container 0.2.0

Dependency injection container with autowiring for Python.

Lucid Container

Dependency injection that wires itself. Bind abstracts to concretes, swap implementations by config, and let the container figure out the rest.

Stop manually constructing dependency trees. Stop passing 12 arguments through 4 layers of functions. Stop refactoring every call site when you swap Redis for Memcached. Lucid Container gives Python a proper IoC container with autowiring — the kind of DI that gets out of your way.

---

Before & After

Without Lucid Container:

config = load_config()
logger = FileLogger(config["log_path"])
redis = RedisClient(config["redis_host"], config["redis_port"])
cache = RedisCache(redis, prefix=config["cache_prefix"])
mailer = SMTPMailer(config["smtp_host"], config["smtp_port"], logger)
user_repo = UserRepository(cache, logger)
auth_service = AuthService(user_repo, mailer, cache, logger)
payment_service = PaymentService(auth_service, logger, cache)
order_service = OrderService(payment_service, user_repo, mailer, cache, logger)

# And now do this again in tests with different implementations...

With Lucid Container:

from lucid_container import Container

app = Container()
app.singleton(CacheContract, RedisCache)
app.singleton(MailerContract, SMTPMailer)
app.singleton(LoggerContract, FileLogger)

# Container reads type hints, builds the entire tree automatically
order_service = app.make(OrderService)

# Swap to in-memory for tests — one line, zero refactoring
app.singleton(CacheContract, InMemoryCache)

---

Installation

pip install lucid-container

Requires Python 3.10 or higher. No dependencies.

---

Quick Start

Basic binding and resolution

from lucid_container import Container

app = Container()

# Bind an abstract (interface) to a concrete (implementation)
app.bind(CacheContract, RedisCache)

# Resolve — container inspects RedisCache.__init__ type hints,
# recursively resolves dependencies, and returns a fully constructed instance
cache = app.make(CacheContract)

Singletons

# Same instance every time
app.singleton(LoggerContract, FileLogger)

logger1 = app.make(LoggerContract)
logger2 = app.make(LoggerContract)
assert logger1 is logger2  # True

Autowiring (zero config)

class OrderService:
    def __init__(self, payments: PaymentContract, mailer: MailerContract):
        self.payments = payments
        self.mailer = mailer

# If PaymentContract and MailerContract are bound, this just works.
# The container reads __init__ type hints and resolves each one.
order_service = app.make(OrderService)

---

Full API Reference

Container()

Creates a new container instance.

from lucid_container import Container

app = Container()

---

.bind(abstract, concrete)

Registers a binding. Every call to make(abstract) creates a new instance.

Parameter	Type	Description
abstract	type | str	The interface, base class, or string key.
concrete	type | Callable[[Container], Any]	A class to autowire, or a factory callable that receives the container.

# Class binding — container autowires the constructor
app.bind(CacheContract, RedisCache)

# Factory binding — you control construction
app.bind(CacheContract, lambda c: RedisCache(
    host=c.make("config")["redis_host"],
    port=c.make("config")["redis_port"],
))

---

.singleton(abstract, concrete)

Same as .bind(), but the instance is created once and cached. Every subsequent make() returns the same object.

app.singleton(DatabaseContract, PostgresDatabase)

db1 = app.make(DatabaseContract)
db2 = app.make(DatabaseContract)
assert db1 is db2

---

.instance(abstract, obj)

Registers an existing object as a singleton. Useful for config objects, pre-built clients, or test doubles.

config = {"debug": True, "db_host": "localhost"}
app.instance("config", config)
app.instance(LoggerContract, my_custom_logger)

---

.alias(name, abstract)

Creates a shorthand name for an abstract.

app.singleton(CacheContract, RedisCache)
app.alias("cache", CacheContract)

# Both return the same thing
app.make(CacheContract)
app.make("cache")

---

.make(abstract)

Resolves an abstract to a fully constructed instance. This is the core method.

Parameter	Type	Description
abstract	type | str	What to resolve.

Resolution order:

1. Check aliases — if abstract is a string alias, resolve to the canonical abstract.
2. Check singleton cache — if already resolved as a singleton, return cached instance.
3. Check bindings — if a binding exists, use it.
4. Autowire — if abstract is a concrete class (not an ABC), attempt to autowire it directly.
5. Raise ResolutionError if nothing works.

cache = app.make(CacheContract)
config = app.make("config")

---

.has(abstract)

Returns True if the abstract is bound, aliased, or has a cached singleton.

app.bind(CacheContract, RedisCache)
app.has(CacheContract)  # True
app.has(MailerContract)  # False

---

.flush()

Clears all bindings, singletons, and aliases. Useful in tests.

app.flush()

---

.bound_to(abstract)

Returns the concrete class or factory bound to an abstract, or None.

app.bind(CacheContract, RedisCache)
app.bound_to(CacheContract)  # RedisCache

---

Autowiring

The container's most powerful feature. When resolving a class, the container:

1. Inspects __init__ using inspect.signature() and typing.get_type_hints().
2. For each parameter with a type annotation that's bound in the container, recursively calls make().
3. For parameters with default values and no binding, uses the default.
4. For parameters without a type hint or binding, raises ResolutionError.

class NotificationService:
    def __init__(
        self,
        mailer: MailerContract,        # resolved from container
        cache: CacheContract,          # resolved from container
        retries: int = 3,              # uses default (not in container)
    ):
        self.mailer = mailer
        self.cache = cache
        self.retries = retries

# Works automatically — mailer and cache are resolved, retries uses default
service = app.make(NotificationService)

Autowiring also works for concrete classes that aren't explicitly bound:

# UserService is never registered — but its dependencies are.
# The container can still build it.
class UserService:
    def __init__(self, repo: UserRepository, logger: LoggerContract):
        ...

service = app.make(UserService)  # Just works

---

Contextual Binding

Sometimes the same abstract should resolve to different concretes depending on who's asking.

app.when(PhotoController).needs(StorageContract).give(S3Storage)
app.when(ReportController).needs(StorageContract).give(LocalStorage)

# When building PhotoController, StorageContract → S3Storage
# When building ReportController, StorageContract → LocalStorage
photo = app.make(PhotoController)
report = app.make(ReportController)

---

Tagging

Group related bindings under a tag, then resolve them all at once.

app.bind("reports.daily", DailySalesReport)
app.bind("reports.weekly", WeeklyInventoryReport)
app.bind("reports.monthly", MonthlyRevenueReport)

app.tag(["reports.daily", "reports.weekly", "reports.monthly"], "reports")

# Resolve all tagged bindings
all_reports = app.tagged("reports")
# Returns: [DailySalesReport(...), WeeklyInventoryReport(...), MonthlyRevenueReport(...)]

---

Container Events

Hook into resolution for logging, decoration, or post-construction setup.

.resolving(abstract, callback)

Called every time abstract is resolved, before returning. Receives the resolved instance and the container. Return value replaces the instance.

def add_logging(instance, container):
    instance.logger = container.make(LoggerContract)
    return instance

app.resolving(PaymentService, add_logging)

.after_resolving(abstract, callback)

Called after the instance is returned. Useful for non-mutating side effects like event emission or metrics.

app.after_resolving(PaymentService, lambda inst, c: metrics.increment("payment_service.resolved"))

Global resolving (no abstract filter)

# Called for EVERY resolution
app.resolving_any(lambda instance, container: print(f"Resolved: {type(instance).__name__}"))

Callbacks can be async def — see Async Support below.

---

Method Injection

Resolve dependencies for a function call, not just constructor injection.

def send_welcome(user_id: int, mailer: MailerContract, cache: CacheContract):
    user = cache.get(f"user:{user_id}")
    mailer.send(user.email, "Welcome!")

# Container injects mailer and cache, you provide user_id
app.call(send_welcome, {"user_id": 42})

---

Async Support

All core operations have async counterparts. Sync and async APIs coexist on the same container — you can mix them freely.

async_make

Resolves an abstract asynchronously. Supports async def factory functions.

async def make_cache(container: Container) -> RedisCache:
    await asyncio.sleep(0)  # any async setup
    return RedisCache()

app.bind(CacheContract, make_cache)

cache = await app.async_make(CacheContract)

Autowiring works the same way as make() — each dependency is resolved via async_make, so async factory chains resolve correctly.

async_call

Injects dependencies and calls an async (or sync) function.

async def handle_request(request_id: str, cache: CacheContract, mailer: MailerContract):
    data = await cache.get(request_id)
    await mailer.send_async(data)

await app.async_call(handle_request, {"request_id": "abc123"})

async_tagged

Resolves all abstracts under a tag using async_make.

handlers = await app.async_tagged("event_handlers")

Async Service Providers

Override register() or boot() as async def. Use async_register_provider and async_boot to drive them.

class CacheServiceProvider(ServiceProvider):

    async def register(self) -> None:
        self.app.singleton(CacheContract, make_cache)  # async factory

    async def boot(self) -> None:
        cache = await self.app.async_make(CacheContract)
        await cache.ping()


app = Container()
await app.async_register_provider(CacheServiceProvider)
await app.async_boot()

Sync providers work unchanged with async_boot — the container detects whether each boot() is a coroutine and handles it accordingly. You can mix sync and async providers in the same app.

Async Callbacks

resolving, after_resolving, and resolving_any callbacks can be async def when used in the async path.

async def stamp(instance, container):
    instance.resolved_at = await get_current_time()
    return instance

app.resolving(PaymentService, stamp)

service = await app.async_make(PaymentService)  # stamp is awaited

Async DriverManager

Override create_{name}_driver as async def and call async_driver() instead of driver().

class CacheManager(DriverManager):

    async def create_redis_driver(self, config: dict) -> RedisCache:
        client = await aioredis.from_url(config["url"])
        return RedisCache(client)

    def create_memory_driver(self, config: dict) -> MemoryCache:
        return MemoryCache()

    def get_default_driver(self) -> str:
        return self.config.get("default", "memory")


manager = CacheManager(app, {"default": "redis", "drivers": {"redis": {"url": "redis://localhost"}}})
cache = await manager.async_driver("redis")

# Custom async creator via extend():
async def make_dynamo(config: dict):
    return await DynamoDB.connect(**config)

manager.extend("dynamo", make_dynamo)
driver = await manager.async_driver("dynamo")

Backward compatibility

All sync APIs (make, call, tagged, boot, register_provider, driver) are completely unchanged. Upgrading to 0.2.0 does not require any code changes for existing users.

---

Service Providers

Service providers are the recommended way to organize bindings. Each provider is responsible for one subsystem — cache, mail, queue, logging, etc.

Base class

from lucid_container import ServiceProvider

class CacheServiceProvider(ServiceProvider):

    def register(self):
        """Bind into the container. Called first on ALL providers."""
        self.app.singleton(CacheContract, lambda c: self._create_cache(c))

    def boot(self):
        """Called after ALL providers have registered. Safe to resolve."""
        cache = self.app.make(CacheContract)
        cache.set("app.booted", True)

    def _create_cache(self, container):
        config = container.make("config").get("cache", {})
        driver = config.get("driver", "memory")

        if driver == "redis":
            return RedisCache(
                host=config.get("host", "localhost"),
                port=config.get("port", 6379),
            )
        elif driver == "file":
            return FileCache(path=config.get("path", "/tmp/cache"))
        else:
            return InMemoryCache()

Registering providers

app = Container()
app.instance("config", load_config())

app.register_provider(CacheServiceProvider)
app.register_provider(MailServiceProvider)
app.register_provider(QueueServiceProvider)

# register() is called on each immediately.
# boot() is called on all providers after all register() calls complete.
app.boot()

Provider lifecycle

1. app.register_provider(P) → instantiates P(app) and calls P.register().
2. app.boot() → calls boot() on every registered provider, in order.
3. After boot, the container is fully ready.

This two-phase lifecycle is critical: during register(), providers can only bind. During boot(), providers can safely resolve from the container because all bindings exist.

---

Driver Manager

For subsystems with swappable backends (cache, queue, mail, storage, etc.), use the DriverManager base class.

from lucid_container import DriverManager

class CacheManager(DriverManager):
    """Manages cache driver instantiation and switching."""

    def create_redis_driver(self, config: dict):
        return RedisCache(host=config["host"], port=config["port"])

    def create_memory_driver(self, config: dict):
        return InMemoryCache()

    def create_file_driver(self, config: dict):
        return FileCache(path=config["path"])

    def get_default_driver(self) -> str:
        return self.config.get("default", "memory")

How DriverManager works

• Subclass names driver creators as create_{name}_driver(self, config).
• get_default_driver() returns the name of the default driver from config.
• .driver(name=None) returns a cached driver instance. None uses the default.
• .async_driver(name=None) is the async counterpart — supports async def creator methods.
• .extend(name, creator) registers custom drivers at runtime (sync or async creators).

# In a service provider:
def register(self):
    self.app.singleton(CacheManager, lambda c: CacheManager(
        container=c,
        config=c.make("config").get("cache", {}),
    ))
    self.app.bind(CacheContract, lambda c: c.make(CacheManager).driver())
    self.app.alias("cache", CacheContract)

# Usage — swap drivers by config:
config = {"cache": {"default": "redis", "drivers": {"redis": {"host": "localhost", "port": 6379}}}}

# Or on the fly:
file_cache = app.make(CacheManager).driver("file")

# Third-party packages can extend:
app.make(CacheManager).extend("dynamodb", lambda config: DynamoDBCache(**config))

---

Real-World Examples

Web application bootstrap

from lucid_container import Container

app = Container()

# Config
app.instance("config", {
    "cache": {"default": "redis", "drivers": {"redis": {"host": "localhost"}}},
    "mail": {"default": "smtp", "drivers": {"smtp": {"host": "mail.example.com"}}},
    "database": {"host": "localhost", "name": "myapp"},
})

# Register providers
app.register_provider(LoggingServiceProvider)
app.register_provider(DatabaseServiceProvider)
app.register_provider(CacheServiceProvider)
app.register_provider(MailServiceProvider)

app.boot()

# In your request handler — the entire tree resolves automatically
controller = app.make(UserController)
response = controller.handle(request)

Testing with swapped implementations

def test_order_processing():
    app = Container()

    # Bind test doubles
    app.instance(CacheContract, InMemoryCache())
    app.instance(MailerContract, FakeMailer())
    app.instance(PaymentContract, FakePaymentGateway(always_succeeds=True))

    # Everything downstream uses the fakes automatically
    service = app.make(OrderService)
    result = service.process(order)

    assert result.status == "completed"
    assert app.make(MailerContract).last_sent.subject == "Order Confirmed"

Contextual binding for multi-tenant

app.when(TenantAController).needs(DatabaseContract).give(
    lambda c: PostgresDatabase(host="tenant-a.db.example.com")
)
app.when(TenantBController).needs(DatabaseContract).give(
    lambda c: PostgresDatabase(host="tenant-b.db.example.com")
)

---

Architecture

Project Structure

lucid-container/
├── src/
│   └── lucid_container/
│       ├── __init__.py             # Public API exports
│       ├── container.py            # Container class
│       ├── service_provider.py     # ServiceProvider base class
│       ├── driver_manager.py       # DriverManager base class
│       ├── contextual_binding.py   # ContextualBindingBuilder
│       ├── autowire.py             # Autowiring logic (inspect + type hints)
│       └── exceptions.py           # ResolutionError, BindingError, CircularDependencyError
├── tests/
│   ├── __init__.py
│   ├── test_binding.py             # bind, singleton, instance
│   ├── test_autowire.py            # Autowiring from type hints
│   ├── test_aliases.py             # Alias resolution
│   ├── test_contextual.py          # Contextual bindings
│   ├── test_tagging.py             # tag and tagged()
│   ├── test_events.py              # resolving / after_resolving hooks
│   ├── test_method_injection.py    # container.call()
│   ├── test_service_provider.py    # Provider lifecycle
│   ├── test_driver_manager.py      # DriverManager pattern
│   ├── test_flush.py               # flush() and re-binding
│   ├── test_edge_cases.py          # Circular deps, missing bindings, scoped containers
│   └── test_async.py               # async_make, async_call, async providers, async drivers
├── pyproject.toml
├── readme.md
├── CHANGELOG.md
└── LICENSE

Implementation Notes

Container internals:

The container maintains the following internal data structures:

self._bindings: dict[Any, Callable]        # abstract → factory or class
self._singletons: dict[Any, Any]           # abstract → cached instance
self._aliases: dict[str, Any]              # string alias → abstract
self._tags: dict[str, list[Any]]           # tag name → list of abstracts
self._contextual: dict[type, dict[type, Any]]  # consumer → {abstract → concrete}
self._lock: threading.RLock                # guards singleton creation (re-entrant)
self._local: threading.local               # per-thread build stack storage

Autowiring algorithm:

def _autowire(self, concrete: type) -> Any:
    hints = typing.get_type_hints(concrete.__init__)
    sig = inspect.signature(concrete.__init__)
    kwargs = {}

    for name, param in sig.parameters.items():
        if name == "self":
            continue

        hint = hints.get(name)

        # 1. Check contextual bindings for the current build stack
        # 2. If hint is bound in container → recursive make()
        # 3. If param has a default → skip (use default)
        # 4. Otherwise → raise ResolutionError

    return concrete(**kwargs)

The container maintains a per-thread build stack (threading.local) of types currently being constructed to detect circular dependencies. If type A requires B and B requires A, the container raises CircularDependencyError with the full cycle path instead of infinite recursion. Because the stack is thread-local, concurrent resolutions in different threads are fully isolated from one another.

Async detection:

The async path uses inspect.iscoroutinefunction to detect whether a factory, callback, or provider method is a coroutine function. This means you don't need to subclass or mark anything — just define async def and the container handles it.

Contextual binding resolution:

During autowiring, the container checks the build stack. If the class currently being built has a contextual override for the dependency being resolved, that override takes priority over the global binding.

Factory vs class binding:

• If concrete is a callable but not a class → treated as a factory. Called with concrete(container) (or await concrete(container) in the async path).
• If concrete is a class → autowired. The container inspects __init__ and builds it.
• Determined by inspect.isclass(concrete).

Thread safety:

make() is safe to call from multiple threads. The implementation uses:

• Double-checked locking (threading.RLock) on the singleton creation path — the cached fast path is entirely lock-free; only the one-time build is serialised. RLock (re-entrant) is used instead of Lock so a singleton factory that calls make() for its own dependencies does not deadlock.
• Per-thread build stack (threading.local) for circular-dependency detection — each thread owns its own stack, so concurrent resolutions never corrupt each other.
• Callback list snapshots — callback lists are copied before iteration, guarding against concurrent registration.

Bootstrap the container from a single thread before spawning workers. Calling bind(), singleton(), or flush() concurrently with make() is not protected and should be avoided. See the Thread Safety guide for the full model.

Scoped containers (child containers):

request_container = app.scoped()
request_container.instance("request", current_request)

# Resolves "request" from child, everything else falls through to parent
controller = request_container.make(UserController)

A scoped container inherits all bindings from the parent but has its own singleton cache. Bindings added to the child don't affect the parent.

Public API

from lucid_container import (
    Container,
    ServiceProvider,
    DriverManager,
    ResolutionError,
    BindingError,
    CircularDependencyError,
)

Exceptions

Exception	When
ResolutionError	Cannot resolve an abstract — no binding, no autowiring possible.
BindingError	Invalid binding (e.g., binding to a non-callable).
CircularDependencyError	A depends on B depends on A. Includes the cycle path.

---

Part of the Lucid Ecosystem

Lucid Container is the backbone that every future Lucid package builds on.

Released:

• lucid-pipeline — Clean, expressive pipelines for multi-step data processing.

Coming soon:

• lucid-cache — Multi-driver cache with a unified API, wired through the container.
• lucid-config — Cascading configuration management.
• lucid-events — Event dispatcher with listeners and subscribers.

---

License

MIT License. See LICENSE for details.