A Python dependency injection container inspired by Jakarta CDI and Spring. Supports sync and async resolution, multiple scopes, lifecycle hooks, and configuration modules.
Project description
providify
A Python dependency injection container inspired by Jakarta CDI and Spring. Supports sync and async resolution, multiple scopes, lifecycle hooks, and configuration modules.
Installation
poetry install
Requires Python 3.12+.
Quick start
from providify import DIContainer, Component, Singleton
class Notifier:
def send(self, msg: str) -> None: ...
@Component
class EmailNotifier(Notifier):
def send(self, msg: str) -> None:
print(f"email: {msg}")
@Singleton
class AlertService:
def __init__(self, notifier: Notifier) -> None:
self._notifier = notifier # injected automatically
def alert(self, msg: str) -> None:
self._notifier.send(msg)
container = DIContainer()
container.bind(Notifier, EmailNotifier)
container.register(AlertService)
svc = container.get(AlertService)
svc.alert("hello") # -> email: hello
Core concepts
The container operates in two phases:
- Registration — declare bindings via
bind(),register(),provide(),scan(), orinstall() - Resolution — the first
get()/aget()call validates all bindings, then resolves them
Constructor parameters are injected automatically when a matching binding exists.
No annotation is needed for plain type hints — the container inspects __init__ at resolution time.
Scope decorators
Mark a class so the container knows how to manage its lifetime.
from providify import Component, Singleton, RequestScoped, SessionScoped
@Component # new instance on every resolution (default)
class EmailSender: ...
@Singleton # one instance for the lifetime of the container
class Database: ...
@RequestScoped # one instance per active request context
class RequestLogger: ...
@SessionScoped # one instance per active session context
class UserSession: ...
All scope decorators accept optional keyword arguments:
@Singleton(qualifier="primary", priority=1, inherited=True)
class PrimaryDB(Database): ...
| Argument | Type | Meaning |
|---|---|---|
qualifier |
str |
Named qualifier — used to distinguish multiple bindings of the same type |
priority |
int |
Lower number wins when multiple candidates match (default 0) |
inherited |
bool |
Subclasses inherit this metadata via MRO walk (default False) |
@Provider
Register a factory function instead of a class. The return type determines the resolved interface.
from providify import Provider
@Provider
def make_sender() -> EmailSender:
return EmailSender(host="smtp.example.com")
# singleton=True caches the result — provider called only once
@Provider(singleton=True)
def make_db() -> Database:
return Database(url=os.environ["DB_URL"])
# async providers are supported — resolve with aget()
@Provider(singleton=True)
async def make_pool() -> ConnectionPool:
pool = ConnectionPool()
await pool.connect()
return pool
Providers also accept qualifier= and priority=.
Container API
from providify import DIContainer
container = DIContainer()
# ── Registration ──────────────────────────────────────────────────
container.bind(Interface, ConcreteClass) # bind interface -> implementation
container.register(ConcreteClass) # self-bind: interface == implementation
container.provide(factory_fn) # register a @Provider function
container.scan("myapp.services") # auto-discover decorated classes in a module
container.install(MyModule) # install a @Configuration module (see below)
# ── Sync resolution ───────────────────────────────────────────────
svc = container.get(Service)
svc = container.get(Service, qualifier="primary")
svc = container.get(Service, priority=1)
svcs = container.get_all(Service) # all matching bindings, sorted by priority
# ── Async resolution ──────────────────────────────────────────────
svc = await container.aget(Service)
svc = await container.aget(Service, qualifier="primary")
svcs = await container.aget_all(Service)
# ── Global singleton ──────────────────────────────────────────────
container = DIContainer.current() # sync — thread-safe
container = await DIContainer.acurrent() # async — never blocks the event loop
DIContainer.reset() # wipe global (useful in tests)
# ── Scoped global — swap in a fresh container for one block ───────
with DIContainer.scoped() as c:
c.bind(...)
c.get(Service)
# original global is restored on exit, even if an exception is raised
async with DIContainer.scoped() as c:
await c.aget(Service)
# ── Instance lifecycle ────────────────────────────────────────────
with container: # calls shutdown() on __exit__
...
async with container: # calls ashutdown() on __aexit__
...
Injection types
Plain type annotation
The simplest case — annotate the parameter with the type to inject.
@Component
class OrderService:
def __init__(self, db: Database) -> None:
self.db = db
Inject[T] — with options
Use Inject[T] when you need a qualifier, exact priority, or optional injection.
from providify import Inject
@Component
class ReportService:
def __init__(
self,
db: Inject[Database],
cache: Inject(Cache, qualifier="redis"),
metrics: Inject(Metrics, optional=True), # None if nothing is bound
audit: Inject(AuditLog, priority=1),
) -> None: ...
InjectInstances[T] — all bindings as a list
Inject every registered implementation of an interface, sorted by priority.
from providify import InjectInstances
@Component
class NotificationFanout:
def __init__(self, senders: InjectInstances[Sender]) -> None:
self.senders = senders # list[Sender]
def notify(self, msg: str) -> None:
for sender in self.senders:
sender.send(msg)
Lazy[T] — deferred injection
Wraps the dependency in a LazyProxy. The real instance is not resolved until
.get() (or .aget()) is called for the first time. Useful for two things:
- Breaking circular dependencies —
Acan holdLazy[B]whileBholdsAdirectly - Scope-safe singletons — a
@SingletonholdingLazy[T]for a@RequestScopeddep will re-resolve on every.get()call instead of caching a stale request instance
from providify import Lazy
@Singleton
class ReportService:
def __init__(self, repo: Lazy[ReportRepository]) -> None:
self._repo = repo # proxy — ReportRepository not resolved yet
def run(self) -> Report:
return self._repo.get().fetch_all() # resolved here on first call
# Async resolution
async def run_async(self) -> Report:
repo = await self._repo.aget()
return await repo.fetch_all_async()
Lazy also accepts qualifier= and priority=:
Lazy(Cache, qualifier="redis", priority=1)
Scope contexts
@RequestScoped and @SessionScoped bindings require an active scope context.
# Sync request scope
with container.scope_context.request():
svc = container.get(RequestLogger) # same instance within this block
# Async request scope
async with container.scope_context.arequest():
svc = await container.aget(RequestLogger)
# Session scope — provide a stable ID to share state across multiple requests
with container.scope_context.session("user-abc") as sid:
profile = container.get(UserProfile)
# Resume the same session later
with container.scope_context.session("user-abc"):
profile = container.get(UserProfile) # same cached instance
# Destroy a session on logout
container.scope_context.invalidate_session("user-abc")
Resolving a
@RequestScopedor@SessionScopedbinding outside an active context raisesRuntimeErrorimmediately.
Lifecycle hooks
@PostConstruct
Called by the container immediately after the instance is constructed and all dependencies are injected. Both sync and async forms are supported.
from providify import PostConstruct
@Singleton
class SearchIndex:
@PostConstruct
def build(self) -> None:
self._load_from_disk()
# Async — must resolve with aget()
@PostConstruct
async def async_build(self) -> None:
await self._fetch_from_s3()
@PreDestroy
Called during shutdown() / ashutdown() for every cached singleton instance.
DEPENDENT instances are not owned by the container and are never destroyed this way.
from providify import PreDestroy
@Singleton
class ConnectionPool:
@PreDestroy
def close(self) -> None:
self._pool.close()
# Async — use ashutdown() to invoke
@PreDestroy
async def async_close(self) -> None:
await self._pool.aclose()
Shutdown
container.shutdown() # calls @PreDestroy on all cached singletons, clears caches
await container.ashutdown() # async — awaits async @PreDestroy hooks
Calling shutdown() when any cached singleton has an async @PreDestroy raises RuntimeError —
use ashutdown() in that case.
@Configuration modules
Group related @Provider methods in a single class.
Spring-style: the module's own __init__ parameters are injected by the container at install() time,
so providers can share config or other injected collaborators via self.
from providify import Configuration
from providify.decorator.scope import Provider, Singleton
@Singleton
class AppConfig:
db_url = "postgresql://localhost/mydb"
pool_size = 10
@Configuration
class DatabaseModule:
def __init__(self, config: AppConfig) -> None:
self._config = config # injected at install() time
@Provider(singleton=True)
def connection_pool(self) -> ConnectionPool:
return ConnectionPool(self._config.db_url, size=self._config.pool_size)
@Provider
def user_repo(self) -> UserRepository:
return UserRepository(self._connection_pool())
container.register(AppConfig)
container.install(DatabaseModule) # sync
await container.ainstall(DatabaseModule) # async — use when module deps need aget()
All @Provider options (qualifier=, priority=, singleton=) work normally inside modules.
Named qualifiers and priority
@Named and @Priority decorators
Qualifiers and priorities can be applied inline via the scope decorator or as
separate @Named / @Priority modifiers on top of any scope decorator.
from providify import Named, Priority
# Inline form — shorter, good for simple cases
@Singleton(qualifier="primary", priority=1)
class PrimaryDB(Database): ...
# Modifier form — useful when the qualifier or priority is a separate concern
@Singleton
@Named(name="replica")
@Priority(priority=2)
class ReplicaDB(Database): ...
@Named requires keyword argument name= — bare @Named raises TypeError immediately.
Both modifiers work on @Provider functions too:
@Provider(singleton=True)
@Named(name="readonly")
@Priority(priority=5)
def make_replica() -> Database:
return ReplicaDB(url=os.environ["REPLICA_URL"])
Warm-up — eager singleton instantiation
By default singletons are created lazily on the first get() call.
Call warm_up() to pre-create them at startup so the first real request
doesn't pay the construction cost.
# Sync — raises RuntimeError if any singleton has an async provider
container.warm_up()
container.warm_up(qualifier="db") # only bindings with qualifier="db"
container.warm_up(priority=0) # only bindings with priority=0
# Async — handles both sync and async singleton providers
await container.awarm_up()
await container.awarm_up(qualifier="db")
warm_up() is all-or-nothing: if any matching singleton is backed by an async
provider it raises before touching the cache, so the cache is never left
partially warmed. Use awarm_up() when you have async providers.
Named qualifiers and priority (resolution)
@Singleton(qualifier="primary")
class PrimaryDB(Database): ...
@Singleton(qualifier="replica", priority=1)
class ReplicaDB(Database): ...
# Resolve by name
db = container.get(Database, qualifier="primary")
# Resolve all, sorted by priority (lowest number first)
all_dbs = container.get_all(Database)
Circular dependency detection
The container detects circular dependencies at resolution time and raises
CircularDependencyError with a readable chain:
CircularDependencyError: Circular dependency detected: OrderService -> PaymentService -> OrderService
To break a cycle intentionally, use Lazy[T]:
@Component
class A:
def __init__(self, b: Lazy[B]) -> None:
self._b = b # proxy — B is not resolved during A's construction
@Component
class B:
def __init__(self, a: A) -> None:
self.a = a # A is fully constructed here — no cycle
Running tests
cd tests
poetry install
poetry run pytest
Tests are organised by feature — one file per subsystem:
| File | Covers |
|---|---|
test_binding.py |
ClassBinding, ProviderBinding construction and errors |
test_container.py |
bind, register, provide, get, get_all, current, scoped |
test_scopes.py |
SINGLETON, DEPENDENT, REQUEST, SESSION, scope violation detection |
test_inject.py |
Inject[T], InjectInstances[T], optional=True/False |
test_lazy.py |
LazyProxy unit tests, Lazy[T] injection, circular-via-lazy |
test_lifecycle.py |
@PostConstruct, @PreDestroy, shutdown, ashutdown |
test_async.py |
aget, aget_all, async providers, async context manager |
test_configuration.py |
@Configuration, install(), ainstall(), Spring-style injection |
test_circular.py |
CircularDependencyError, diamond dependency, Lazy cycle-break |
test_warmup.py |
warm_up(), awarm_up(), all-or-nothing guard, qualifier/priority filter |
test_decorators.py |
@Named, @Priority, @Inheritable, stacking, error paths |
Scope reference
| Decorator | Lifetime |
|---|---|
@Component |
New instance on every get() |
@Singleton |
One instance per container — shared for the container's lifetime |
@RequestScoped |
One instance per scope_context.request() block |
@SessionScoped |
One instance per scope_context.session(id) — survives across requests |
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file providify-0.1.0.tar.gz.
File metadata
- Download URL: providify-0.1.0.tar.gz
- Upload date:
- Size: 50.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/2.2.1 CPython/3.11.11 Linux/6.6.87.2-microsoft-standard-WSL2
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
8ceb28f64c08728d9663af880d12aa565ba1619dc7f0fcd4b2a7536428eb3a30
|
|
| MD5 |
6f342d9123a6ddf1bde07b8c91a8d212
|
|
| BLAKE2b-256 |
4878b2bbcf9e92667756295a39daee2b5dc0d9d9cb7f5abc501d1f0e631d6759
|
File details
Details for the file providify-0.1.0-py3-none-any.whl.
File metadata
- Download URL: providify-0.1.0-py3-none-any.whl
- Upload date:
- Size: 54.5 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/2.2.1 CPython/3.11.11 Linux/6.6.87.2-microsoft-standard-WSL2
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
e4b78172387c048084e63417a6d8ef356ac5a9cf5f16148868d117bd5679a40a
|
|
| MD5 |
0de17159f5a2b4f38866830eba3154d3
|
|
| BLAKE2b-256 |
92c5e28eb8c3519034da8cb80cc8c1b062f57dd4cb3bef779298f25ec92b26ec
|