Chapter 25: DI Container#
1. Stop Constructing Dependencies Inside Your Code#
@post("/api/orders")async def create_order(request, response): db = Database.get_connection() # tight coupling mailer = Messenger() # constructed inline payments = PaymentGateway( # hardcoded config api_key=os.environ["PAYMENT_KEY"] ) ...Every route constructs its own dependencies. Testing requires monkey-patching or mocking at the module level. Swapping the payment gateway means touching every route that uses it.
A DI container moves construction out of the route. You register services once, at startup. Routes ask the container for what they need by name.
2. The Container Class#
from tina4_python.container import Containercontainer = Container()The Container is a lightweight registry. It does not scan files, read XML, or require annotations on your classes. You register explicitly. You retrieve explicitly.
3. register() - Transient Services#
A transient service creates a new instance every time get() is called:
from tina4_python.container import Containercontainer = Container()# Register a factory functioncontainer.register("logger", lambda: Logger(level="INFO"))# Each call creates a fresh instancelogger1 = container.get("logger")logger2 = container.get("logger")assert logger1 is not logger2 # Different objectsUse transient registration for services that must not share state: request-scoped objects, disposable HTTP clients, or mock implementations in tests.
4. singleton() - Cached Services#
A singleton creates the instance once and returns the same object on every subsequent call:
import osfrom tina4_python.container import Containerfrom tina4_python.api import Apicontainer = Container()# Register a singleton factorycontainer.singleton("payment_gateway", lambda: Api( bearer_token=os.environ["PAYMENT_API_KEY"], timeout=15))# Both calls return the exact same objectgw1 = container.get("payment_gateway")gw2 = container.get("payment_gateway")assert gw1 is gw2 # Same objectUse singleton registration for: database connections, HTTP clients, configuration objects, external service adapters.
5. get() and has()#
# Retrieve a serviceservice = container.get("payment_gateway")# Check before retrievingif container.has("email_service"): mailer = container.get("email_service") mailer.send(...)get() raises KeyError if the service is not registered. has() lets you check first, or use it to provide a default:
mailer = container.get("email_service") if container.has("email_service") else None6. reset()#
Clear the singleton cache without removing registrations. Useful in tests to force fresh construction:
container.singleton("db", lambda: Database.connect())# In tests: reset between test cases so each gets a clean DBcontainer.reset()db = container.get("db") # New connection createdreset() clears all cached singletons. Transient registrations are unaffected (they never cache). Call reset() in test teardown to prevent state from leaking between tests.
7. Building a Service Container for Your App#
Define your container in one place:
# src/services.pyimport osfrom tina4_python.container import Containerfrom tina4_python.api import Apicontainer = Container()# Database connection (singleton -- one connection reused)container.singleton("db", lambda: Database.get_connection())# Payment gateway (singleton -- one client, shared config)container.singleton("payments", lambda: Api( bearer_token=os.environ["PAYMENT_API_KEY"], timeout=15))# Email service (singleton)container.singleton("mailer", lambda: Api( bearer_token=os.environ["SENDGRID_API_KEY"]))# Request logger (transient -- new instance per use)container.register("request_logger", lambda: RequestLogger( level=os.environ.get("LOG_LEVEL", "INFO")))Import container wherever you need it:
# src/routes/orders.pyfrom src.services import container@post("/api/orders")async def create_order(request, response): payments = container.get("payments") mailer = container.get("mailer") ...8. Testing with the Container#
Swap implementations for tests without changing any route code:
# tests/test_orders.pyimport pytestfrom src.services import containerclass MockPaymentGateway: def post(self, url, body): return { "http_code": 200, "body": {"charge_id": "ch_test_001", "status": "succeeded"}, "headers": {}, "error": None }class MockMailer: def __init__(self): self.sent = [] def post(self, url, body): self.sent.append(body) return {"http_code": 200, "body": {"id": "msg_001"}, "headers": {}, "error": None}@pytest.fixture(autouse=True)def reset_container(): # Clear singleton cache before each test container.reset() yield container.reset()def test_create_order_success(client): mock_mailer = MockMailer() # Override with test doubles container.singleton("payments", lambda: MockPaymentGateway()) container.singleton("mailer", lambda: mock_mailer) resp = client.post("/api/orders", json={ "customer_id": 1, "items": [{"product_id": "KB-001", "quantity": 1, "price": 79.99}], "email": "alice@example.com" }) assert resp.status_code == 201 assert resp.json()["order_id"] is not None assert len(mock_mailer.sent) == 1 # Confirmation email was sentThe route code never changes. The container is the only seam.
9. Full Example: Order Route with DI#
# src/routes/orders_di.pyfrom tina4_python.core.router import postfrom tina4_python.debug import Logfrom src.services import container@post("/api/di/orders")async def create_order_di(request, response): body = request.body if not body.get("customer_id") or not body.get("items"): return response({"error": "customer_id and items are required"}, 400) payments = container.get("payments") mailer = container.get("mailer") total = round(sum(item.get("price", 0) * item.get("quantity", 1) for item in body["items"]), 2) # Charge payment charge_result = payments.post("https://api.payment-provider.com/v1/charges", { "amount": int(total * 100), "currency": "usd" }) if charge_result["error"] or charge_result["http_code"] not in (200, 201): Log.error("Payment failed", customer_id=body["customer_id"], total=total) return response({"error": "Payment failed"}, 402) charge_id = charge_result["body"]["charge_id"] order_id = f"ORD-{body['customer_id']}-{charge_id}" # Send confirmation mailer.post("https://api.sendgrid.com/v3/mail/send", { "to": body.get("email"), "subject": f"Order Confirmation {order_id}", "body": f"Your order of ${total} has been confirmed." }) Log.info("Order created", order_id=order_id, total=total) return response({ "order_id": order_id, "total": total, "charge_id": charge_id }, 201)10. Exercise: Configurable Notification Service#
Build a notification service that can switch between email and SMS providers via the container.
Requirements#
- Define two notifier classes:
EmailNotifierandSmsNotifier, each with asend(to, message)method - Register the active notifier as
"notifier"in the container (controlled by aNOTIFIERenv var) - Create
POST /api/notifythat retrieves"notifier"from the container and sends a message - Write a test that registers a mock notifier and verifies messages are sent
Test with:#
# Set NOTIFIER=email or NOTIFIER=sms in .envcurl -X POST http://localhost:7146/api/notify \ -H "Content-Type: application/json" \ -d '{"to": "alice@example.com", "message": "Your order has shipped!"}'11. Solution#
# src/notifiers.pyimport osclass EmailNotifier: def send(self, to, message): print(f"[EMAIL] To: {to} | Message: {message}") return {"channel": "email", "to": to}class SmsNotifier: def send(self, to, message): print(f"[SMS] To: {to} | Message: {message}") return {"channel": "sms", "to": to}def make_notifier(): channel = os.environ.get("NOTIFIER", "email") if channel == "sms": return SmsNotifier() return EmailNotifier()# src/services.py (additions)from src.notifiers import make_notifiercontainer.singleton("notifier", make_notifier)# src/routes/notify.pyfrom tina4_python.core.router import postfrom src.services import container@post("/api/notify")async def notify(request, response): body = request.body if not body.get("to") or not body.get("message"): return response({"error": "to and message are required"}, 400) notifier = container.get("notifier") result = notifier.send(body["to"], body["message"]) return response({"sent": True, **result})# tests/test_notify.pyfrom src.services import containerclass MockNotifier: def __init__(self): self.messages = [] def send(self, to, message): self.messages.append({"to": to, "message": message}) return {"channel": "mock", "to": to}def test_notify_sends_message(client): mock = MockNotifier() container.reset() container.singleton("notifier", lambda: mock) resp = client.post("/api/notify", json={"to": "bob@example.com", "message": "Hello!"}) assert resp.status_code == 200 assert mock.messages[0]["to"] == "bob@example.com" container.reset()12. Gotchas#
1. Mutable singleton shared across requests#
Problem: A singleton holds per-request state and leaks it between concurrent requests.
Fix: Use register() (transient) for anything that holds request-specific state. Singletons should be stateless or thread-safe.
2. reset() removes singleton cache but not registrations#
Problem: After container.reset(), calling container.has("payments") still returns True but the next get() creates a new instance.
Fix: This is correct behaviour. reset() only clears the cache. The factory function is still registered and will run again on the next get(). Use reset() deliberately in test teardown.
3. Circular dependencies#
Problem: Service A's factory calls container.get("B") and B's factory calls container.get("A"). Both hang or raise a recursion error.
Fix: Restructure so that dependencies are one-directional. Extract shared logic into a third service that neither A nor B depends back upon.
4. Importing container before .env is loaded#
Problem: container.singleton("payments", lambda: Api(bearer_token=os.environ["PAYMENT_KEY"])) raises KeyError because .env has not been loaded yet.
Fix: Tina4 loads .env before importing route files. If you initialise the container in a module that is imported before startup, use a lambda (lazy factory) so os.environ is not accessed until get() is called.