Quickstart
Define dependencies
To showcase the basics of Wirio, we will create a container able to resolve the following:
EmailService: A simple service with no dependencies.UserService: A service depending onEmailService.
class EmailService:
pass
class UserService:
def __init__(self, email_service: EmailService) -> None:
self.email_service = email_service
Register services
The next step is to create a container and register the dependencies we just defined.
from wirio.service_collection import ServiceCollection
services = ServiceCollection()
services.add_transient(EmailService) # (1)!
services.add_transient(UserService)
- Both services are registered as transient, meaning a new instance will be created each time they're requested
We'll use .add_X depending on the desired lifetime. For example: .add_transient for transient services, .add_singleton for singleton services and .add_scoped for scoped services.
Use
Finally, we convert the service collection into a service provider, which will validate and build the dependency graph, and we'll be able to request instances from it.
To resolve dependencies from the service provider, we annotate the parameter with the type we want to resolve.
from wirio.annotations import FromServices
@app.post("/users")
async def create_user(
user_service: Annotated[UserService, FromServices()],
) -> None:
pass
services.configure_fastapi(app) # (1)!
- This will configure FastAPI to use Wirio's dependency injection
To resolve dependencies from the service provider, we call get_required_service with the type we want to resolve.
Full code
from typing import Annotated
from fastapi import FastAPI
from wirio.annotations import FromServices
from wirio.service_collection import ServiceCollection
class EmailService:
pass
class UserService:
def __init__(self, email_service: EmailService) -> None:
self.email_service = email_service
async def create_user(self) -> None:
pass
app = FastAPI()
@app.post("/users")
async def create_user(
user_service: Annotated[UserService, FromServices()],
) -> None:
pass
services = ServiceCollection()
services.add_transient(EmailService)
services.add_transient(UserService)
services.configure_fastapi(app)
import asyncio
from wirio.service_collection import ServiceCollection
class EmailService:
pass
class UserService:
def __init__(self, email_service: EmailService) -> None:
self.email_service = email_service
async def create_user(self) -> None:
pass
async def main() -> None:
services = ServiceCollection()
services.add_transient(EmailService)
services.add_transient(UserService)
async with services.build_service_provider() as service_provider:
user_service = await service_provider.get_required_service(UserService)
if __name__ == "__main__":
asyncio.run(main())
Test
We can substitute dependencies on the fly meanwhile the context manager is active.
with service_provider.override_service(EmailService, email_service_mock):
user_service = await service_provider.get_required_service(UserService)
For more information, check the testing documentation.