Factory Pattern¶
Factories handle complex object construction, especially when objects need configuration or produce different variants.
When to Use Factories¶
Use factories when:
- Object construction requires configuration
- The same class has multiple valid configurations
- Construction logic is complex
- You need to delay or control instantiation
JWTAuthFactory Example¶
The most common factory in this project is JWTAuthFactory:
# src/fastdjango/core/authentication/delivery/fastapi/auth.py
from dataclasses import dataclass
@dataclass(kw_only=True)
class JWTAuthFactory(BaseFactory):
"""Factory for creating JWT authentication dependencies."""
_jwt_service: JWTService
_user_use_case: UserUseCase
def __call__(
self,
*,
require_staff: bool = False,
require_superuser: bool = False,
) -> JWTAuth:
"""Create a JWT auth dependency.
Args:
require_staff: If True, user must have is_staff=True
require_superuser: If True, user must have is_superuser=True
Returns:
JWTAuth or JWTAuthWithPermissions instance
"""
if require_staff or require_superuser:
return JWTAuthWithPermissions(
jwt_service=self._jwt_service,
user_use_case=self._user_use_case,
require_staff=require_staff,
require_superuser=require_superuser,
)
return JWTAuth(jwt_service=self._jwt_service, user_use_case=self._user_use_case)
Usage in Controllers¶
@dataclass(kw_only=True)
class UserController(BaseAsyncController):
_jwt_auth_factory: JWTAuthFactory
_user_use_case: UserUseCase
def __post_init__(self) -> None:
# Create different auth configurations
self._jwt_auth = self._jwt_auth_factory()
self._staff_jwt_auth = self._jwt_auth_factory(require_staff=True)
super().__post_init__()
def register(self, registry: APIRouter) -> None:
# Public endpoint (no auth)
registry.add_api_route("/v1/users/", self.create_user, methods=["POST"])
# Authenticated endpoint
registry.add_api_route(
"/v1/users/me",
self.get_current_user,
methods=["GET"],
dependencies=[Depends(self._jwt_auth)],
)
# Staff-only endpoint
registry.add_api_route(
"/v1/users/{user_id}",
self.get_user_by_id,
methods=["GET"],
dependencies=[Depends(self._staff_jwt_auth)],
)
Auth dependencies are created in __post_init__ and stored as instance attributes (no field(init=False) needed).
FastAPIFactory Example¶
The FastAPIFactory creates the entire FastAPI application:
# src/fastdjango/entrypoints/fastapi/factories.py
@dataclass(kw_only=True)
class FastAPIFactory(BaseFactory):
_application_settings: ApplicationSettings
_fastapi_settings: FastAPISettings
_cors_settings: CORSSettings
_lifespan: Lifespan
_telemetry_instrumentor: OpenTelemetryInstrumentor
_django_wsgi_factory: DjangoWSGIFactory
# Controllers are injected as fields
_health_controller: HealthController
_authentication_token_controller: AuthenticationTokenController
_user_controller: UserController
def __call__(
self,
*,
include_django: bool = True,
add_trusted_hosts_middleware: bool = True,
add_cors_middleware: bool = True,
) -> FastAPI:
docs_url = (
"/docs" if self._application_settings.environment != Environment.PRODUCTION else None
)
app = FastAPI(
title="API",
lifespan=self._lifespan,
docs_url=docs_url,
redoc_url=None,
)
self._telemetry_instrumentor.instrument_fastapi(app=app)
self._add_middlewares(app=app, ...)
self._register_controllers(app=app)
if include_django:
django_wsgi = self._django_wsgi_factory()
app.mount("/django", WSGIMiddleware(django_wsgi))
return app
def _register_controllers(self, app: FastAPI) -> None:
health_router = APIRouter(tags=["health"])
self._health_controller.register(health_router)
app.include_router(health_router)
# ... more controllers
ContainerBasedFactory¶
Test factories extend ContainerBasedFactory to access the IoC container:
# tests/foundation/factories.py
class ContainerBasedFactory(BaseTestFactory, ABC):
def __init__(self, container: Container) -> None:
self._container = container
TestClientFactory¶
# tests/integration/factories.py
class TestClientFactory(ContainerBasedFactory):
"""Factory for creating test HTTP clients."""
def __call__(
self,
auth_for_user: User | None = None,
headers: dict[str, str] | None = None,
**kwargs: Any,
) -> TestClient:
"""Create a test client.
Args:
auth_for_user: User to authenticate as (auto-generates token)
headers: Additional headers to include
Returns:
Configured TestClient
"""
headers = headers or {}
if auth_for_user is not None:
jwt_service = self._container.resolve(JWTService)
token = jwt_service.issue_access_token(user_id=auth_for_user.id)
headers["Authorization"] = f"Bearer {token}"
api_factory = self._container.resolve(FastAPIFactory)
app = api_factory(
include_django=False,
add_trusted_hosts_middleware=False,
add_cors_middleware=False,
)
return TestClient(app=app, headers=headers, base_url="http://testserver", **kwargs)
Factory vs Direct Instantiation¶
When to Use Direct Instantiation¶
Simple objects without configuration:
When to Use Factory¶
Objects with configuration options:
# Factory allows different configurations
auth = jwt_auth_factory() # Basic auth
staff_auth = jwt_auth_factory(require_staff=True) # Staff auth
Factory Registration¶
Factories are usually auto-wired by type, so no manual registration is required:
When the dependency should be the value built by a callable factory class, use
add_factory_class so diwire injects the factory constructor and resolves the
value returned by __call__:
CeleryAppFactory Example¶
# src/fastdjango/entrypoints/celery/factories.py
@dataclass(kw_only=True)
class CeleryAppFactory(BaseFactory):
"""Factory for creating Celery applications."""
_application_settings: ApplicationSettings
_broker_settings: CeleryBrokerSettings
_celery_settings: CelerySettings
_instance: Celery | None = field(default=None, init=False)
def __call__(self) -> Celery:
"""Create or return the Celery application.
Reuses the same Celery app for this factory instance.
"""
if self._instance is not None:
return self._instance
celery_app = Celery(
"main",
broker=self._broker_settings.url.get_secret_value(),
backend=self._broker_settings.url.get_secret_value(),
)
celery_app.conf.update(
timezone=self._application_settings.time_zone,
enable_utc=True,
**self._celery_settings.model_dump(),
)
celery_app.conf.beat_schedule = {
"ping-every-minute": {
"task": TaskName.PING,
"schedule": 60.0,
},
}
self._instance = celery_app
return self._instance
The factory caches the app instance so task registration and worker setup share the same Celery application object.
Benefits¶
1. Flexible Configuration¶
# Same factory, different configurations
basic_auth = factory()
staff_auth = factory(require_staff=True)
superuser_auth = factory(require_superuser=True)
2. Encapsulated Complexity¶
# Complex construction hidden behind simple interface
app = fast_api_factory() # All setup handled internally
3. Testability¶
# Factories can be mocked or configured for tests
def test_with_custom_factory():
factory = JWTAuthFactory(_jwt_service=mock_jwt, _user_use_case=mock_user)
auth = factory()
# Test with controlled dependencies
4. Lazy Initialization¶
# Create factory early, instantiate later
def __post_init__(self) -> None:
# Only creates auth when needed
self._jwt_auth = self._jwt_auth_factory()
Summary¶
Factories:
- Encapsulate complex construction logic
- Enable configurable object creation
- Support multiple variants from single factory
- Integrate with IoC container for dependencies
- Improve testability through controlled construction