Function Injection
The @inject_from_container decorator injects dependencies directly into function parameters. Use this when building
your own integration or using Wireup in a framework without built-in support.
Basic Usage¶
Decorate any function and annotate the parameters you want injected:
from typing import Annotated
from wireup import Inject, Injected, inject_from_container
@inject_from_container(container)
def process_order(
order_service: Injected[OrderService],
db_url: Annotated[str, Inject(config="database_url")],
) -> None:
order_service.process()
The decorator:
- Creates a new scope before the function runs
- Injects all annotated parameters from that scope
- Closes the scope when the function returns (triggering cleanup)
Async Functions¶
The decorator works with async functions. The container must be created using wireup.create_async_container.
@inject_from_container(container)
async def process_data(service: Injected[DataService]):
await service.process()
Annotations Required
Only parameters annotated with Injected[T] or Annotated[T, Inject(...)] are injected. Unannotated parameters are
left alone for the caller to provide.
Advanced Usage¶
Using an Existing Scope¶
If a scope already exists (e.g., created by middleware), pass a callable that returns it as the second argument. The decorator will use that scope instead of creating a new one.
from contextvars import ContextVar
from wireup import ScopedSyncContainer, inject_from_container
scoped_container: ContextVar[ScopedSyncContainer] = ContextVar(
"scoped_container"
)
@inject_from_container(container, scoped_container.get)
def handle_request(service: Injected[RequestService]) -> None: ...
Creating a Decorator Alias¶
For cleaner code, create an alias:
inject = inject_from_container(container, scoped_container.get)
@inject
def handle_request(service: Injected[RequestService]) -> None: ...
Dynamic Function Injection¶
Sometimes the function you want to inject is only known at runtime, for example when scheduling background work on the event loop. In these cases, inject the root container into an infrastructure service and use it to wrap the callable before scheduling it.
Example: Scheduling Background Tasks
import asyncio
from functools import lru_cache
from wireup import AsyncContainer, Injected, inject_from_container, injectable
from wireup.ioc.types import AnyCallable
@injectable
class BackgroundTasks:
def __init__(self, container: AsyncContainer) -> None:
# Build wrappers for callables using this container instance.
self._inject = inject_from_container(container)
# Keep strong references to in-flight tasks until they complete.
self._tasks: set[asyncio.Task] = set()
@lru_cache(maxsize=128)
def _wrap(self, fn: AnyCallable) -> AnyCallable:
# Reuse wrappers for frequently scheduled functions.
return self._inject(fn)
def schedule(self, fn, /, *args, **kwargs) -> asyncio.Task:
task = asyncio.create_task(self._wrap(fn)(*args, **kwargs))
self._tasks.add(task)
task.add_done_callback(self._tasks.discard)
return task
async def send_email(
user_id: str,
email_service: Injected[EmailService],
) -> None:
await email_service.send_welcome_email(user_id)
async def handle_signup(
user_id: str,
background_tasks: Injected[BackgroundTasks],
) -> None:
background_tasks.schedule(send_email, user_id)
This avoids storing a global container variable while still letting you inject dependencies into functions that are created or selected dynamically. See Container: Injecting The Container.
Good to know
In the above example, each scheduled task runs in its own Wireup scope, so scoped dependencies are isolated from the caller and from other scheduled tasks.
API Reference¶
wireup.inject_from_container¶
Inject dependencies into the decorated function based on annotations. Wireup containers will
attempt to provide only parameters annotated with Inject.
See the documentation for more details: https://maldoinc.github.io/wireup/latest/function_injection/
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
container
|
SyncContainer | AsyncContainer
|
The root container created via |
required |
scoped_container_supplier
|
Callable[[], ScopedSyncContainer | ScopedAsyncContainer] | None
|
An optional callable that returns the current scoped container instance. If provided, it will be used to create scoped dependencies. If not provided, the container will automatically enter a scope. Provide a scoped_container_supplier if you need to manage the container's scope manually. |
None
|
hide_annotated_names
|
bool
|
If True, the parameters annotated with Wireup annotations will be removed from the signature of the decorated function. |
False
|