Python Dependency Injection Practice Problems & Exercises
Practice: Dependency Injection
← Back to lessonEasy
Refactor a tightly-coupled OrderService to use constructor injection.
# Before: tightly coupled (bad)
class TightOrderService:
def __init__(self):
self._stock = StockService() # creates its own dependencies
self._email = EmailService()
# After: constructor injection (good)
class StockService:
def update(self, item: str, qty: int) -> None:
print(f"Inventory updated for {item}")
class EmailService:
def send(self, to: str, message: str) -> None:
print(f"Email sent to {to}: {message}")
class OrderService:
def __init__(self, stock: StockService, email: EmailService):
self._stock = stock
self._email = email
self._next_id = 1
def place(self, customer: str, item: str, qty: int, unit_price: int) -> int:
order_id = self._next_id
self._next_id += 1
total = qty * unit_price
print(f"Order {order_id}: {item} x {qty} = {total} cents")
self._stock.update(item, qty)
self._email.send(customer, f"Order {order_id} placed")
return order_id
# Composition root
stock = StockService()
email = EmailService()
svc = OrderService(stock, email)
svc.place("[email protected]", "Widget", 2, 999)Solution
The solution above demonstrates the core principle: dependencies are pushed in, not pulled out. OrderService.__init__ declares what it needs; the caller decides what to provide.
Three injection styles:
# 1. Constructor injection (preferred)
class Service:
def __init__(self, dep: IDep): self._dep = dep
# 2. Method injection (for optional/per-call deps)
class Service:
def process(self, dep: IDep): dep.do_work()
# 3. Property injection (rare, for optional deps)
class Service:
@property
def dep(self): return self._dep
@dep.setter
def dep(self, v): self._dep = v
Constructor injection is preferred because dependencies are explicit, required at construction time, and the object is always in a valid state.
Expected Output
Order 1: Widget x 2 = 1998 cents\nInventory updated for Widget\nEmail sent to [email protected]: Order 1 placedHints
Hint 1: Constructor injection means passing dependencies as arguments to __init__, not creating them inside the class.
Hint 2: The class should not call StockService() or EmailService() internally — it receives already-constructed instances.
Write fake implementations of StockService and EmailService, inject them into OrderService, and assert on the recorded calls.
from dataclasses import dataclass, field
class StockService:
def update(self, item: str, qty: int) -> None:
print(f"Stock updated: {item} -{qty}")
class EmailService:
def send(self, to: str, message: str) -> None:
print(f"Email to {to}: {message}")
class OrderService:
def __init__(self, stock: StockService, email: EmailService):
self._stock = stock
self._email = email
self._next_id = 1
def place(self, customer: str, item: str, qty: int, price: int) -> int:
order_id = self._next_id
self._next_id += 1
self._stock.update(item, qty)
self._email.send(customer, f"Order {order_id} placed")
return order_id
# Fakes
@dataclass
class FakeStockService:
calls: list = field(default_factory=list)
def update(self, item: str, qty: int) -> None:
self.calls.append((item, qty))
@dataclass
class FakeEmailService:
sent: list = field(default_factory=list)
def send(self, to: str, message: str) -> None:
self.sent.append((to, message))
# Tests
fake_stock = FakeStockService()
fake_email = FakeEmailService()
svc = OrderService(fake_stock, fake_email)
order_id = svc.place("[email protected]", "TestItem", 3, 100)
assert order_id == 1, "Expected order ID 1"
print("Test 1 passed: order created with ID 1")
assert fake_stock.calls == [("TestItem", 3)]
print("Test 2 passed: stock updated for TestItem")
assert fake_email.sent[0][0] == "[email protected]"
print("Test 3 passed: email sent to [email protected]")Solution
The solution is above. This demonstrates the primary testing benefit of DI:
Without DI: Testing OrderService would call real SMTP servers and real database writes. Tests are slow, fragile, and require infrastructure setup.
With DI: You inject FakeEmailService — no SMTP, no database. Tests run in microseconds and are completely isolated.
Fake vs Mock distinction:
- Fake: A real implementation that works in-memory (like
FakeEmailServicethat stores sent emails in a list). - Mock: A generated test double that records calls and can verify them (like
unittest.mock.MagicMock).
Fakes are preferred for repository/service dependencies because they model the real behavior rather than just recording calls.
Expected Output
Test 1 passed: order created with ID 1\nTest 2 passed: stock updated for TestItem\nTest 3 passed: email sent to [email protected]Hints
Hint 1: Create fake implementations of the dependencies that record calls instead of actually doing the work.
Hint 2: Inject the fakes into OrderService — the class should work identically regardless of which implementations it receives.
Identify all Dependency Injection anti-patterns in the code below and explain how to fix each one.
import json
# Global registry (service locator)
_REGISTRY = {}
def register(name, instance): _REGISTRY[name] = instance
def get_service(name): return _REGISTRY[name]
class ReportService:
def __init__(self):
# Anti-pattern 1: creates its own database connection
self._db = DatabaseConnection("postgresql://localhost/prod")
# Anti-pattern 2: reads config from a hard-coded file
with open("/etc/app/config.json") as f:
self._config = json.load(f)
# Anti-pattern 3: looks up a dependency from a global registry
self._mailer = get_service("mailer")
def generate(self, report_id: int):
# Anti-pattern 4: creates a collaborator mid-method
formatter = PdfFormatter()
data = self._db.query(f"SELECT * FROM reports WHERE id = {report_id}")
return formatter.format(data)
Solution
Anti-pattern 1: DatabaseConnection(...) inside __init__
Problem: You can never test ReportService without a real PostgreSQL server. You cannot swap to SQLite for tests.
Fix:
class ReportService:
def __init__(self, db: IDatabasePort): # injected
self._db = db
Anti-pattern 2: Hard-coded file path
Problem: Tests break unless /etc/app/config.json exists on every machine. Config is hidden inside the class.
Fix:
class ReportService:
def __init__(self, db: IDatabasePort, config: dict):
self._config = config
# Caller loads config and passes it in
Anti-pattern 3: Service locator get_service("mailer")
Problem: The dependency is hidden — you cannot tell from the class signature that ReportService needs a mailer. Tests must pre-populate the global registry. This is called the "service locator anti-pattern."
Fix:
class ReportService:
def __init__(self, db: IDatabasePort, mailer: IMailerPort):
self._mailer = mailer
Anti-pattern 4: PdfFormatter() created mid-method
Problem: Cannot switch to HtmlFormatter or CsvFormatter without editing the class. Cannot test with a fake formatter.
Fix:
class ReportService:
def __init__(self, db: IDatabasePort, formatter: IFormatterPort):
self._formatter = formatter
Corrected version:
class ReportService:
def __init__(self, db: IDatabasePort, config: dict, mailer: IMailerPort, formatter: IFormatterPort):
self._db = db
self._config = config
self._mailer = mailer
self._formatter = formatter
All four dependencies are now explicit, injectable, and testable.
Expected Output
See solution for analysisHints
Hint 1: Look for: new() calls inside constructors, global state access, service locator pattern, and hidden dependencies.
Hint 2: Each anti-pattern makes the class harder to test. Explain what makes each one a problem.
Medium
Build a minimal DI container that supports singleton-scoped services and factory functions.
from typing import Callable, Any
class Container:
def __init__(self):
self._factories: dict[str, Callable] = {}
self._singletons: dict[str, Any] = {}
def register(self, name: str, factory: Callable[["Container"], Any]) -> None:
self._factories[name] = factory
def resolve(self, name: str) -> Any:
if name not in self._singletons:
if name not in self._factories:
raise KeyError(f"No registration for: {name}")
self._singletons[name] = self._factories[name](self)
return self._singletons[name]
# Services
class StockService:
def update(self, item: str, qty: int) -> None:
pass # real impl would write to DB
class EmailService:
def send(self, to: str, msg: str) -> None:
print(f"Email to {to}: {msg}")
class OrderService:
def __init__(self, stock: StockService, email: EmailService):
self._stock = stock
self._email = email
self._next_id = 1
def place(self, customer: str, item: str, qty: int) -> int:
order_id = self._next_id
self._next_id += 1
self._stock.update(item, qty)
self._email.send(customer, f"Order {order_id} placed")
return order_id
# Register
container = Container()
container.register("stock", lambda c: StockService())
container.register("email", lambda c: EmailService())
container.register("order_svc", lambda c: OrderService(c.resolve("stock"), c.resolve("email")))
# Resolve
svc = container.resolve("order_svc")
print("OrderService resolved")
print(f"StockService is singleton: {container.resolve('stock') is container.resolve('stock')}")
print(f"EmailService is singleton: {container.resolve('email') is container.resolve('email')}")
svc.place("[email protected]", "Gadget", 1)Solution
The solution is above. The container has three responsibilities:
- Store factory functions keyed by name.
- Resolve dependencies by calling the factory on first access.
- Cache resolved instances (singleton scope).
Adding transient scope:
def register_transient(self, name: str, factory: Callable) -> None:
self._transient_factories[name] = factory
def resolve_transient(self, name: str) -> Any:
return self._transient_factories[name](self)
Production DI containers for Python: dependency-injector, lagom, punq. They add type-based resolution, scope management, and lazy initialization.
from typing import Callable, TypeVar, Type
T = TypeVar("T")
class Container:
"""A minimal dependency injection container.
Supports:
- register(name, factory_fn)
- resolve(name) -> instance (singleton by default)
"""
passExpected Output
OrderService resolved\nStockService is singleton: True\nEmailService is singleton: True\nOrder placed for [email protected]Hints
Hint 1: Store factories in a dict. On first resolve(), call the factory and cache the result. Subsequent calls return the cached instance (singleton scope).
Hint 2: The factory function takes the container as its argument so it can resolve its own dependencies: lambda c: OrderService(c.resolve("stock"), c.resolve("email")).
Implement a ScopedContainer with singleton, transient, and scoped (per-request) dependency lifetimes.
from dataclasses import dataclass, field
from typing import Callable, Any
import contextlib
class ScopedContainer:
def __init__(self):
self._singleton_factories: dict[str, Callable] = {}
self._transient_factories: dict[str, Callable] = {}
self._scoped_factories: dict[str, Callable] = {}
self._singletons: dict[str, Any] = {}
self._scope_cache: dict[str, Any] = {}
def singleton(self, name: str, factory: Callable) -> None:
self._singleton_factories[name] = factory
def transient(self, name: str, factory: Callable) -> None:
self._transient_factories[name] = factory
def scoped(self, name: str, factory: Callable) -> None:
self._scoped_factories[name] = factory
def resolve(self, name: str) -> Any:
if name in self._singleton_factories:
if name not in self._singletons:
self._singletons[name] = self._singleton_factories[name]()
return self._singletons[name]
if name in self._transient_factories:
return self._transient_factories[name]()
if name in self._scoped_factories:
if name not in self._scope_cache:
self._scope_cache[name] = self._scoped_factories[name]()
return self._scope_cache[name]
raise KeyError(f"Unknown: {name}")
@contextlib.contextmanager
def create_scope(self):
old_cache = self._scope_cache
self._scope_cache = {}
try:
yield self
finally:
self._scope_cache = old_cache
class Config:
def __init__(self): self.id = id(self)
class DbSession:
def __init__(self): self.id = id(self)
c = ScopedContainer()
c.singleton("config", lambda: Config())
c.transient("session", lambda: DbSession())
c.scoped("request_ctx", lambda: DbSession())
print(f"Singleton same instance: {c.resolve('config') is c.resolve('config')}")
print(f"Transient same instance: {c.resolve('session') is c.resolve('session')}")
with c.create_scope() as s1:
ctx_a = s1.resolve("request_ctx")
ctx_b = s1.resolve("request_ctx")
print(f"Scoped same in scope: {ctx_a is ctx_b}")
with c.create_scope() as s2:
ctx_c = s2.resolve("request_ctx")
print(f"Scoped different across scopes: {ctx_a is not ctx_c}")Solution
The solution is above. Understanding scopes is critical in web frameworks:
Web framework analogy:
- Singleton: Database connection pool, configuration object. Created once at startup.
- Transient: Stateless helper, calculator. Fresh instance every time — safe to share state temporarily.
- Scoped: Database session, request context. One per HTTP request — created at request start, disposed at request end.
FastAPI's equivalent:
# Singleton
db_pool = create_pool()
# Transient
async def get_formatter():
return JsonFormatter() # new each time
# Scoped (per-request)
async def get_db():
session = SessionLocal()
try:
yield session
finally:
session.close()
from dataclasses import dataclass, field
from typing import Callable, Any
import uuid
class ScopedContainer:
"""Container with three scopes:
- singleton: one instance for the app lifetime
- transient: new instance every resolve
- scoped: one instance per 'scope' context manager
"""
passExpected Output
Singleton same instance: True\nTransient same instance: False\nScoped same in scope: True\nScoped different across scopes: TrueHints
Hint 1: For singleton scope, cache in a class-level dict. For transient, always call the factory. For scoped, use a context manager that creates a local cache dict and clears it on exit.
Hint 2: Use a threading.local() or a simple dict reset in __exit__ to manage scoped state.
Implement an @inject decorator that automatically fills unset function parameters from a global service container.
from typing import Callable, Any
import inspect
from functools import wraps
_CONTAINER: dict[str, Any] = {}
def provide(name: str, instance: Any) -> None:
_CONTAINER[name] = instance
def inject(func: Callable) -> Callable:
params = list(inspect.signature(func).parameters.keys())
@wraps(func)
def wrapper(*args, **kwargs):
# Auto-fill missing keyword args from container
for param in params[len(args):]:
if param not in kwargs and param in _CONTAINER:
kwargs[param] = _CONTAINER[param]
return func(*args, **kwargs)
return wrapper
# Services
class InventoryService:
def check(self, item_id: int) -> bool:
print("Inventory checked")
return True
class NotificationService:
def send(self, email: str) -> None:
print(f"Notification sent to {email}")
# Register
provide("inventory", InventoryService())
provide("notification", NotificationService())
# Usage with @inject
@inject
def process_order(order_id: int, customer_email: str, inventory: InventoryService = None, notification: NotificationService = None):
print(f"Processing order for {customer_email}")
inventory.check(order_id)
notification.send(customer_email)
process_order(1, "[email protected]") # inventory and notification auto-injectedSolution
The solution is above. This mirrors how frameworks like injector and some DI libraries work under the hood.
Limitations of decorator-based DI:
- Uses a global container (the service locator anti-pattern at the module level).
- Type annotations are not used for resolution — only parameter names.
- Hard to test without modifying global state.
Better production approach — use type annotations:
def inject(func: Callable) -> Callable:
hints = get_type_hints(func)
@wraps(func)
def wrapper(*args, **kwargs):
sig = inspect.signature(func)
params = list(sig.parameters.keys())
for param in params[len(args):]:
if param not in kwargs and param in hints:
cls = hints[param]
if cls in _TYPE_CONTAINER:
kwargs[param] = _TYPE_CONTAINER[cls]
return func(*args, **kwargs)
return wrapper
from typing import Callable, Any
import inspect
# Implement an @inject decorator that automatically resolves
# function parameters from a global container by parameter name.
_CONTAINER: dict[str, Any] = {}
def provide(name: str, instance: Any) -> None:
_CONTAINER[name] = instance
def inject(func: Callable) -> Callable:
"""Decorator that injects arguments from _CONTAINER by parameter name."""
passExpected Output
Processing order for [email protected]\nInventory checked\nNotification sent to [email protected]Hints
Hint 1: Use inspect.signature(func).parameters to get the parameter names. For each parameter that is not provided by the caller, look it up in _CONTAINER by name.
Hint 2: The decorated function should accept positional args from the caller and fill in unset keyword args from the container.
Implement a minimal Depends system that mirrors FastAPI's dependency injection mechanism, including nested dependencies and test overrides.
from typing import Callable, Any
import inspect
from functools import wraps
class Depends:
def __init__(self, dependency: Callable):
self.dependency = dependency
def resolve(handler: Callable, **overrides) -> Any:
sig = inspect.signature(handler)
kwargs = {}
for name, param in sig.parameters.items():
if name in overrides:
kwargs[name] = overrides[name]
elif isinstance(param.default, Depends):
# Recursively resolve nested dependency
kwargs[name] = resolve(param.default.dependency, **overrides)
# Parameters with no default and not in overrides: skip (caller provides them)
return handler(**kwargs)
# Dependency functions
def get_db():
return "RealDB"
def get_cache():
return "RealCache"
def get_current_user(db=Depends(get_db)):
return f"user_from_{db}"
# Handler
def handle_request(
user=Depends(get_current_user),
db=Depends(get_db),
cache=Depends(get_cache),
):
return f"user={user}, db={db}, cache={cache}"
result = resolve(handle_request)
print(f"Request handled: {result}")
# Override for testing
result_test = resolve(handle_request, db="FakeDB", cache="FakeCache", user="testuser")
print(f"With overrides: {result_test}")Solution
The solution is above. This is exactly how FastAPI's Depends() works internally, minus the async/await handling.
Real FastAPI usage:
async def get_db() -> AsyncSession:
async with SessionLocal() as session:
yield session
async def get_current_user(
token: str = Header(...),
db: AsyncSession = Depends(get_db)
) -> User:
return await verify_token(token, db)
@router.get("/profile")
async def get_profile(user: User = Depends(get_current_user)):
return {"email": user.email}
Test overrides in FastAPI:
def get_fake_db():
return FakeSession()
app.dependency_overrides[get_db] = get_fake_db
# Now all endpoints that Depends(get_db) receive FakeSession instead
from typing import Callable, Any
from dataclasses import dataclass
# Implement a minimal Depends system (like FastAPI's):
# class Depends:
# def __init__(self, dependency: Callable)
#
# def resolve(handler: Callable, **overrides) -> Any:
# Inspects handler's parameters, resolves Depends() annotations,
# and calls the handler with all resolved deps.
# overrides let tests provide fake deps.Expected Output
Request handled: user=alice, db=FakeDB, cache=FakeCache\nWith overrides: user=testuser, db=FakeDB, cache=FakeCacheHints
Hint 1: Inspect the handler signature. For each parameter with a Depends() default, call the wrapped dependency function recursively (dependencies can have their own Depends).
Hint 2: The overrides dict maps parameter names to replacement values — this is how you inject fakes in tests without modifying the handler.
Hard
Build an auto-wiring container that resolves class dependencies by inspecting __init__ type annotations recursively.
from typing import get_type_hints, Type, TypeVar, Any
import inspect
T = TypeVar("T")
class AutoWireContainer:
def __init__(self):
self._registered: set[type] = set()
self._singletons: dict[type, Any] = {}
self._overrides: dict[type, Any] = {}
def register(self, cls: type) -> None:
self._registered.add(cls)
def override(self, cls: type, instance: Any) -> None:
self._overrides[cls] = instance
def resolve(self, cls: type) -> Any:
if cls in self._overrides:
return self._overrides[cls]
if cls in self._singletons:
return self._singletons[cls]
# Get __init__ parameter types (skip 'self' and 'return')
try:
hints = get_type_hints(cls.__init__)
except Exception:
hints = {}
hints.pop("return", None)
kwargs = {}
for param_name, param_type in hints.items():
kwargs[param_name] = self.resolve(param_type)
instance = cls(**kwargs)
self._singletons[cls] = instance
return instance
# Services
class StockService:
def update(self, item: str, qty: int) -> None:
pass
class EmailService:
def send(self, to: str, msg: str) -> None:
pass
class OrderService:
def __init__(self, stock: StockService, email: EmailService):
self._stock = stock
self._email = email
def place(self, customer: str, item: str, qty: int) -> str:
self._stock.update(item, qty)
self._email.send(customer, f"Order placed for {item}")
return "order placed"
# Auto-wire
container = AutoWireContainer()
container.register(StockService)
container.register(EmailService)
container.register(OrderService)
svc = container.resolve(OrderService)
print("OrderService resolved")
print(f"OrderService.stock is StockService: {isinstance(svc._stock, StockService)}")
print(f"OrderService.email is EmailService: {isinstance(svc._email, EmailService)}")
print(svc.place("[email protected]", "Widget", 1))Solution
The solution is above. Auto-wiring reads type annotations and recursively resolves them — no manual factory registration needed.
Limitations:
- Only works if
__init__parameters are type-annotated. - Cannot resolve
Protocoltypes without explicit overrides. - Does not handle
Optional,Union, or primitive types automatically.
Production libraries handle these edge cases: dependency-injector (most popular), lagom (type-based), punq (lightweight).
Override for testing:
fake_email = FakeEmailService()
container.override(EmailService, fake_email)
svc = container.resolve(OrderService)
# svc._email is now fake_email
from typing import get_type_hints, Type, TypeVar, Any
import inspect
T = TypeVar("T")
class AutoWireContainer:
"""Container that resolves dependencies by inspecting type annotations.
register(cls) - register a class for autowiring
resolve(cls) - resolve an instance, creating dependencies recursively
"""
passExpected Output
OrderService resolved\nOrderService.stock is StockService: True\nOrderService.email is EmailService: True\norder placedHints
Hint 1: Use inspect.signature(cls.__init__) to get the constructor parameter types. For each annotated parameter, call resolve() recursively on that type.
Hint 2: Store resolved instances in a dict keyed by class to provide singleton behavior.
Build a cycle detector for dependency graphs using DFS. Use it to validate a DI container before startup.
def find_cycles(graph: dict[str, list[str]]) -> list[list[str]]:
visited = set()
in_path = set()
cycles = []
def dfs(node: str, path: list[str]) -> None:
if node in in_path:
cycle_start = path.index(node)
cycles.append(path[cycle_start:] + [node])
return
if node in visited:
return
visited.add(node)
in_path.add(node)
path.append(node)
for dep in graph.get(node, []):
dfs(dep, path)
path.pop()
in_path.discard(node)
for node in graph:
if node not in visited:
dfs(node, [])
return cycles
# Clean graph
clean = {
"OrderService": ["StockService", "EmailService"],
"StockService": ["DatabaseConnection"],
"EmailService": [],
"DatabaseConnection": [],
}
cycles = find_cycles(clean)
print("Clean graph: no cycles" if not cycles else f"Cycles: {cycles}")
# Cyclic graph
cyclic = {
"A": ["B"],
"B": ["C"],
"C": ["A"], # cycle!
"D": ["A"],
}
cycles = find_cycles(cyclic)
print(f"Cyclic graph cycles: {cycles}")Solution
The solution is above. Run this check at container startup:
class AutoWireContainer:
def validate(self) -> None:
graph = {}
for cls in self._registered:
hints = get_type_hints(cls.__init__)
hints.pop("return", None)
graph[cls.__name__] = [t.__name__ for t in hints.values()]
cycles = find_cycles(graph)
if cycles:
raise RuntimeError(f"Circular dependencies detected: {cycles}")
Real circular dependency example:
class AuthService:
def __init__(self, user_svc: "UserService"): ...
class UserService:
def __init__(self, auth_svc: AuthService): ... # circular!
Fix: Introduce an interface, use lazy resolution, or extract the shared logic into a third service that neither depends on the other.
from typing import Callable
# Given a dependency graph as a dict of name -> list of dependency names,
# write a function that detects circular dependencies and returns
# the cycles found (if any).
def find_cycles(graph: dict[str, list[str]]) -> list[list[str]]:
"""Return all cycles found in the dependency graph."""
passExpected Output
Clean graph: no cycles\nCyclic graph cycles: [['A', 'B', 'C', 'A']]Hints
Hint 1: Use DFS with a "currently visiting" set. When you encounter a node already in the current path, you have found a cycle.
Hint 2: Track the path as a list so you can reconstruct the cycle from the node where the cycle starts to where it was re-encountered.
Implement request-scoped DI using contextvars.ContextVar so each simulated request gets its own scoped service instances.
from contextvars import ContextVar
from typing import Callable, Any
import asyncio
_scope: ContextVar[dict] = ContextVar("_scope", default={})
class Config:
"""Singleton - shared across all requests."""
value = "production"
class DbSession:
"""Scoped - one per request."""
_counter = 0
def __init__(self):
DbSession._counter += 1
self.session_id = DbSession._counter
_singletons: dict[type, Any] = {}
_scoped_factories: dict[type, Callable] = {}
def register_singleton(cls: type) -> None:
_singletons[cls] = cls()
def register_scoped(cls: type, factory: Callable = None) -> None:
_scoped_factories[cls] = factory or cls
def resolve(cls: type) -> Any:
if cls in _singletons:
return _singletons[cls]
if cls in _scoped_factories:
scope_cache = _scope.get()
if cls not in scope_cache:
scope_cache[cls] = _scoped_factories[cls]()
return scope_cache[cls]
raise KeyError(f"Not registered: {cls}")
async def handle_request(request_name: str) -> dict:
# Each request sets its own scope cache
_scope.set({})
session_a = resolve(DbSession)
session_b = resolve(DbSession) # should be same instance within request
config = resolve(Config)
assert session_a is session_b, "Scoped: should be same within request"
return {"request": request_name, "session_id": session_a.session_id, "config": config.value}
register_singleton(Config)
register_scoped(DbSession)
async def main():
r1, r2 = await asyncio.gather(
handle_request("Request 1"),
handle_request("Request 2"),
)
print(f"Request 1: session_id differs from Request 2: {r1['session_id'] != r2['session_id']}")
print(f"Request 1: config is same as Request 2: {r1['config'] == r2['config']}")
asyncio.run(main())Solution
The solution is above. ContextVar is the asyncio-safe replacement for threading.local(). Each coroutine (request) gets its own copy of the scope cache — no shared state, no race conditions.
FastAPI's actual implementation (simplified):
from fastapi import Request
async def get_db(request: Request) -> AsyncSession:
if "db" not in request.state.__dict__:
request.state.db = SessionLocal()
try:
yield request.state.db
finally:
await request.state.db.close()
FastAPI stores scoped state in request.state, which is per-request by design.
from contextvars import ContextVar
from typing import Callable, Any
from dataclasses import dataclass, field
# Implement a request-scoped DI system using contextvars:
# - Each "request" runs in its own context
# - Scoped services are created once per context
# - Singleton services are shared across all contexts
# - Use threading or asyncio to simulate concurrent requestsExpected Output
Request 1: session_id differs from Request 2: True\nRequest 1: config is same as Request 2: TrueHints
Hint 1: Use ContextVar to store a dict of scoped instances. Each asyncio task gets its own copy of the ContextVar value.
Hint 2: Set the ContextVar at the start of each request with contextvars.copy_context().run() or by setting a new dict at request entry.
Build a decorator-driven DI container with singleton/transient scopes, auto-wiring, test overrides, and dependency validation.
from typing import get_type_hints, Type, TypeVar, Any
import inspect
T = TypeVar("T")
class RegistrationError(Exception):
pass
class DIContainer:
def __init__(self):
self._singletons_reg: set[type] = set()
self._transients_reg: set[type] = set()
self._singleton_cache: dict[type, Any] = {}
self._overrides: dict[type, Any] = {}
def singleton(self, cls: type) -> type:
self._singletons_reg.add(cls)
return cls
def transient(self, cls: type) -> type:
self._transients_reg.add(cls)
return cls
def override(self, cls: type, instance: Any) -> None:
self._overrides[cls] = instance
def clear_overrides(self) -> None:
self._overrides.clear()
def _build(self, cls: type) -> Any:
try:
hints = get_type_hints(cls.__init__)
except Exception:
hints = {}
hints.pop("return", None)
kwargs = {}
for param_name, param_type in hints.items():
kwargs[param_name] = self.resolve(param_type)
return cls(**kwargs)
def resolve(self, cls: type) -> Any:
if cls in self._overrides:
return self._overrides[cls]
if cls in self._singletons_reg:
if cls not in self._singleton_cache:
self._singleton_cache[cls] = self._build(cls)
return self._singleton_cache[cls]
if cls in self._transients_reg:
return self._build(cls)
raise RegistrationError(f"Type not registered: {cls.__name__}")
container = DIContainer()
# Register services
@container.singleton
class Config:
db_url: str = "postgresql://localhost/app"
@container.singleton
class NotificationService:
def send(self, msg: str) -> None:
print(f"[Notification] {msg}")
@container.singleton
class OrderService:
def __init__(self, config: Config, notifier: NotificationService):
self._config = config
self._notifier = notifier
def place(self, item: str) -> None:
self._notifier.send(f"Order placed: {item}")
# Test it
svc = container.resolve(OrderService)
print("OrderService wired successfully")
svc.place("Widget")
# Override for testing
class FakeNotifier:
def __init__(self): self.last_msg = None
def send(self, msg: str) -> None:
self.last_msg = msg
print(f"[FakeNotifier] {msg}")
fake = FakeNotifier()
container.override(NotificationService, fake)
container._singleton_cache.clear() # reset to re-wire
svc2 = container.resolve(OrderService)
svc2.place("Gadget")
print(f"With override: notification was {type(svc2._notifier).__name__}")
container.clear_overrides()
# Validation
class Unregistered:
pass
@container.transient
class BrokenService:
def __init__(self, dep: Unregistered):
pass
try:
container.resolve(BrokenService)
except RegistrationError as e:
print(f"Validation: RegistrationError caught for missing dep")Solution
The solution is above. This container is production-usable for medium-sized applications. It demonstrates all the DI patterns in one coherent implementation:
- Decorator registration (
@container.singleton,@container.transient) — clean, visible at the class definition. - Auto-wiring — reads
__init__type hints, no manual factory functions. - Singleton cache — one instance per type across the app lifetime.
- Test overrides — inject fakes without modifying any production code.
- Error on missing registration — fail fast with a clear message at resolve time, not at call time.
Production improvement: Add @container.interface(INotificationPort, NotificationService) to support resolving by protocol rather than concrete class.
# Build a production-quality DI container that supports:
# @container.singleton - decorator to register a class as singleton
# @container.transient - decorator to register a class as transient
# container.resolve(cls) - auto-wire from type annotations
# container.override(cls, instance) - test override
# Validate: detect unregistered dependencies at resolve timeExpected Output
OrderService wired successfully\nOrder placed\nWith override: notification was FakeNotifier\nValidation: RegistrationError caught for missing depHints
Hint 1: Use class decorators to add classes to the container registry. The decorator returns the class unchanged but registers it internally.
Hint 2: At resolve() time, walk the __init__ type hints, resolve each recursively, check for overrides, and raise a clear error for unregistered types.