Dependency Injection for FP
Project description
partial-injector
Dependency injection for functional Python. Wire plain functions together with functools.partial — no decorators, no metaclasses, no framework lock-in.
from partial_injector.partial_container import Container
def get_db_url() -> str:
return "postgresql://localhost/mydb"
def find_user(db_url: str, user_id: int) -> dict:
... # db_url pre-filled by the container
container = Container()
container.register_singleton(get_db_url)
container.register_singleton(find_user)
container.build()
find = container.resolve(find_user)
user = find(user_id=42) # db_url already injected
Why partial-injector?
Most Python DI frameworks are built around classes — you annotate __init__ parameters, inherit from a base, or stack decorators. That works fine for OOP, but Python's best code is often just functions.
partial-injector is built differently:
- Functions are first-class citizens. Register a plain function; the container inspects its type annotations, finds matching registrations, and returns a
functools.partialwith those arguments pre-filled. You call it as normal. - No decorators required. Your business logic has zero knowledge of the container. Swap the container out in tests without touching production code.
- Proper transient lifecycle. Transient dependencies registered as functions or instances are cloned fresh on every resolution — no captive-dependency anti-pattern where a singleton ends up holding a permanently frozen copy of what should be a live object.
- Automatic build order.
build()runs a topological sort of all registered keys and detects cycles with a descriptive path — you see exactly which keys form the loop. - Async-native. Everything works transparently with
async deffunctions: injected wrappers, factory results, and return-injection.
Installation
pip install partial-injector
Requires Python 3.12+
Table of Contents
- Core Concepts
- Registration Types
- Registration Keys
- Dependency Matching
- FromContainer
- Multiple Registrations for the Same Key
- Conditional Registrations
- inject_returns — wiring factory outputs
- inject_items — per-element injection in lists
- Async Support
- Transient Lifecycle in Depth
- Error Handling
- API Reference
Core Concepts
The three-step lifecycle
container = Container()
# 1. Register — declare what exists and how to build it
container.register_singleton(my_service, key=MyService)
# 2. Build — resolve dependency order, wire everything up
container.build()
# 3. Resolve — retrieve the wired object
svc = container.resolve(MyService)
No registrations are allowed after build(). No resolution is allowed before it.
Registration Types
| Method | When it builds | What resolve() returns |
|---|---|---|
register_singleton |
Once, at build() time |
The same wired object every time |
register_transient |
Deferred until resolve() |
A fresh copy on every call |
register_singleton_factory |
Calls the factory once at build() |
Cached result of that factory call |
register_transient_factory |
Calls the factory on every resolve() |
New result each time |
container = Container()
# singleton — one shared instance
container.register_singleton(DatabasePool(max_connections=10), key=DatabasePool)
# transient — fresh copy per resolve
container.register_transient(RequestContext(), key=RequestContext)
# singleton factory — builds once from a callable
container.register_singleton_factory(lambda: connect("postgresql://localhost/db"), key="conn")
# transient factory — calls the factory fresh each time
container.register_transient_factory(lambda: uuid.uuid4(), key="request_id")
container.build()
Registration Keys
Every registration needs a key for later lookup. The default key is the registered object itself — convenient for functions and types:
container.register_singleton(my_function) # key = my_function
container.register_singleton(MyService()) # key = MyService() (the instance — rarely useful)
container.register_singleton(MyService(), key=MyService) # key = MyService (the class — good)
Keys can be any hashable value: a type, a string, a TypeAlias, or a function:
# string key
container.register_singleton("postgresql://localhost/db", key="db_url")
# type key
container.register_singleton(42, key=int)
# TypeAlias key (PEP 695)
type Greeter = Callable[[str], str]
container.register_singleton(make_greeter, key=Greeter)
Dependency Matching
When the container wires a function, it inspects every parameter and tries to find a match in the registry. The lookup order is:
- Parameter name — is there a registration whose key equals the parameter name?
- Type annotation — is there a registration whose key equals the annotation?
- Group key — is there a group of registrations under
list[annotation]?
Parameters with no match are left unbound — the caller supplies them.
type Multiplier = Callable[[], int]
def get_multiplier() -> int:
return 6
def compute(get_multiplier: Multiplier, x: int) -> int:
# get_multiplier is resolved from the container
# x is left for the caller to supply
return get_multiplier() * x
container = Container()
container.register_singleton(get_multiplier, key=Multiplier)
container.register_singleton(compute)
container.build()
fn = container.resolve(compute)
print(fn(x=7)) # 42
Tip: Using
TypeAliaskeys for callable types (type Greeter = Callable[[str], str]) gives your dependency graph a vocabulary beyond raw types and avoids accidental collisions with built-in types likestrorint.
FromContainer
FromContainer lets one registration derive its value from another, with an optional selector lambda. This is especially useful for slicing a configuration object into individual settings:
from partial_injector.partial_container import Container, FromContainer
from dataclasses import dataclass
@dataclass
class Config:
db_url: str
secret_key: str
debug: bool
container = Container()
container.register_singleton(
Config(db_url="postgresql://localhost/db", secret_key="s3cr3t", debug=False),
key=Config,
)
container.register_singleton(FromContainer(Config, lambda c: c.db_url), key="db_url")
container.register_singleton(FromContainer(Config, lambda c: c.secret_key), key="secret_key")
container.register_singleton(FromContainer(Config, lambda c: c.debug), key="debug")
container.build()
db_url = container.resolve("db_url") # "postgresql://localhost/db"
FromContainer can also appear inside factory_args and factory_kwargs, where it is resolved just before the factory is called:
def connect(url: str, debug: bool) -> Connection: ...
container.register_singleton_factory(
connect,
key=Connection,
factory_args=[FromContainer(Config, lambda c: c.db_url)],
factory_kwargs={"debug": FromContainer(Config, lambda c: c.debug)},
)
Multiple Registrations for the Same Key
Register more than one object under the same key and the container automatically groups them. Resolve as list[Key] to get all, or as the bare Key if only one survives its condition:
type Middleware = Callable[[Request], Request]
def auth_middleware(request: Request) -> Request: ...
def logging_middleware(request: Request) -> Request: ...
def rate_limit_middleware(request: Request) -> Request: ...
container = Container()
container.register_singleton(auth_middleware, key=Middleware)
container.register_singleton(logging_middleware, key=Middleware)
container.register_singleton(rate_limit_middleware, key=Middleware)
container.build()
middlewares = container.resolve(list[Middleware]) # [auth, logging, rate_limit]
Injecting a group into a function — annotate the parameter as list[Key]:
type Validator = Callable[[str], bool]
def validate_all(validators: list[Validator], value: str) -> bool:
return all(v(value) for v in validators)
container.register_singleton(validate_all)
container.build()
fn = container.resolve(validate_all)
fn(value="hello") # all validators pre-filled, only value is open
Conditional Registrations
Every register_* method accepts a condition callable. The result determines whether the registration is included at all.
import os
container = Container()
container.register_singleton(
ProductionDatabase(),
key="db",
condition=lambda: os.getenv("ENV") == "production",
)
container.register_singleton(
InMemoryDatabase(),
key="db",
condition=lambda: os.getenv("ENV") != "production",
)
container.build()
db = container.resolve("db") # whichever condition passed
Condition evaluation timing:
| Registration type | When condition runs |
|---|---|
SINGLETON / SINGLETON_FACTORY |
At build() time |
TRANSIENT / TRANSIENT_FACTORY |
Lazily, at each resolve() call |
This means transient conditions can respond to runtime state — a feature flag read from a database, a per-request context, etc.
Conditions with injected arguments — pass condition_args or condition_kwargs to inject FromContainer values into the condition callable:
def is_feature_enabled(config: Config) -> bool:
return config.feature_flags.get("new_ui", False)
container.register_singleton(
NewUIRenderer(),
key=Renderer,
condition=is_feature_enabled,
condition_args=[FromContainer(Config)],
)
Strict mode — set throw_if_condition_not_satisfied=True to raise PartialContainerError when no registration for a key passes its condition, instead of silently omitting it.
inject_returns — wiring factory outputs
When a registered function returns another function, set inject_returns=True to wire that returned function's parameters through the container too. This supports factory patterns without any boilerplate:
type Handler = Callable[[str], None]
type HandlerFactory = Callable[[], Handler]
def make_handler(logger: Callable[[str], None]) -> Handler:
def handle(message: str) -> None:
logger(f"[handler] {message}")
return handle
container = Container()
container.register_singleton(print, key=Callable[[str], None])
container.register_singleton(make_handler, key=HandlerFactory, inject_returns=True)
container.build()
factory = container.resolve(HandlerFactory)
handler = factory() # returns `handle` with `logger` already injected
handler("ping") # prints "[handler] ping"
Return injection is recursive — the returned function can itself return a function that also gets wired — and works transparently with async def.
inject_items — per-element injection in lists
Register a list and set inject_items=True to have each element wired individually rather than treating the list as a single opaque object:
type Processor = Callable[[int], int]
def scale(factor: int, value: int) -> int:
return value * factor
def shift(offset: int, value: int) -> int:
return value + offset
container = Container()
container.register_singleton(3, key=int)
container.register_singleton([scale, shift], key=list[Processor], inject_items=True)
container.build()
processors = container.resolve(list[Processor])
# scale has `factor=3` pre-filled; shift has `offset=3` pre-filled
results = [p(10) for p in processors] # [30, 13]
Async Support
There is nothing special to configure. The container handles async def functions at every level:
async def fetch_user(db: Database, user_id: int) -> User: ...
async def handle_request(fetch_user: Callable, request: Request) -> Response:
user = await fetch_user(user_id=request.user_id)
...
container = Container()
container.register_singleton(db_instance, key=Database)
container.register_singleton(fetch_user)
container.register_singleton(handle_request)
container.build()
handler = container.resolve(handle_request)
response = await handler(request=req) # db pre-filled; request supplied by caller
Async functions also work with inject_returns — the wrapper awaits the outer function and wires the returned callable before handing it back.
Transient Lifecycle in Depth
partial-injector enforces strict transient semantics with no captive-dependency anti-pattern.
What is captive dependency? It occurs when a singleton captures a transient at build time and holds onto that single frozen instance permanently — defeating the purpose of registering it as transient.
How partial-injector prevents it: When a singleton function depends on a transient, the container does not resolve the transient eagerly. Instead it builds a closure that re-resolves the transient on every invocation:
from dataclasses import dataclass, field
import uuid
@dataclass
class RequestContext:
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
def process(ctx: RequestContext, payload: str) -> str:
return f"[{ctx.request_id}] {payload}"
container = Container()
container.register_transient(RequestContext(), key=RequestContext)
container.register_singleton(process)
container.build()
fn = container.resolve(process)
print(fn(payload="a")) # [3f2a...] a
print(fn(payload="b")) # [9c1e...] b ← fresh RequestContext every time
Transient copy semantics:
| Registered object | What resolve() returns |
|---|---|
| Plain instance | copy.deepcopy of the original |
| Function | New FunctionType sharing bytecode but with independent __dict__ |
List with inject_items=True |
Each element deep-copied and re-wired independently |
Error Handling
All errors raise PartialContainerError from partial_injector.error_handling:
from partial_injector.error_handling import PartialContainerError
try:
container.resolve(MyService)
except PartialContainerError as e:
print(e.message)
| Situation | Error message |
|---|---|
Register after build() |
"Container already built" |
resolve() before build() |
"Container not built" |
| Key not registered | "Object with key <key> not registered" |
| Key registered but not built | "Object with key <key> not built" |
Condition not met, throw=True |
"No object with key <key> was built because the built condition has not been met." |
| Circular dependency | "Circular dependency detected: A → B → C → A" |
API Reference
Container
from partial_injector.partial_container import Container
Registration
container.register_singleton(
instance,
key=None, # lookup key; defaults to instance itself
inject_returns=False, # wire the callable returned by this function
inject_items=False, # wire each element of a registered list individually
condition=None, # () -> bool; False skips this registration
condition_args=None, # positional args passed to condition
condition_kwargs=None, # keyword args passed to condition
throw_if_condition_not_satisfied=False,
)
container.register_transient(instance, key=None, ...) # same signature
container.register_singleton_factory(
factory,
key=None,
factory_args=None, # positional args for factory (FromContainer allowed)
factory_kwargs=None, # keyword args for factory (FromContainer allowed)
inject_returns=False,
condition=None, ...,
)
container.register_transient_factory(factory, key=None, ...) # same signature
Lifecycle
container.build() # topological sort + wire; must be called before resolve
container.resolve(key) # return the wired object for key
FromContainer
from partial_injector.partial_container import FromContainer
FromContainer(source_key) # resolve source_key as-is
FromContainer(source_key, lambda val: val.attr) # apply selector to resolved value
Used as a registration value or as an element inside factory_args / factory_kwargs.
License
MIT
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 partial_injector-4.0.1.tar.gz.
File metadata
- Download URL: partial_injector-4.0.1.tar.gz
- Upload date:
- Size: 23.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: uv/0.11.13 {"installer":{"name":"uv","version":"0.11.13","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":null,"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
6d09c16ca43d4b27376c3ca4cd331d6dc08a5296930ad30071f1a14557a0b085
|
|
| MD5 |
0e027e324ec2764de7c40d817bf10c20
|
|
| BLAKE2b-256 |
6ccbf9e82ccaf383ad759f8fa30a73cf34c33cdf2b6273cb88970bd254e5657f
|
File details
Details for the file partial_injector-4.0.1-py3-none-any.whl.
File metadata
- Download URL: partial_injector-4.0.1-py3-none-any.whl
- Upload date:
- Size: 22.5 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: uv/0.11.13 {"installer":{"name":"uv","version":"0.11.13","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":null,"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
a2d560b2d185209104216a98566c073a9b65b41350c2e12725595830732df767
|
|
| MD5 |
a1af9354d1f7c76bef09c5008a35762b
|
|
| BLAKE2b-256 |
feb176bc4975862d29c7c367285c53879e7a5cd2f8aa33a0325f549ce0c2d7e8
|