Use Annotated for DI in FastAPI examples (#853)

This commit is contained in:
ZipFile 2025-02-23 18:21:31 +02:00 committed by GitHub
parent 23acf01c15
commit 8b625d81ad
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 41 additions and 24 deletions

View File

@ -64,7 +64,7 @@ FastAPI example:
@app.api_route("/") @app.api_route("/")
@inject @inject
async def index(service: Service = Depends(Provide[Container.service])): async def index(service: Annotated[Service, Depends(Provide[Container.service])]):
value = await service.process() value = await service.process()
return {"result": value} return {"result": value}

View File

@ -1,18 +1,22 @@
"""Application module.""" """Application module."""
from dependency_injector.wiring import inject, Provide from typing import Annotated
from fastapi import FastAPI, Depends
from fastapi import Depends, FastAPI
from dependency_injector.wiring import Provide, inject
from .containers import Container from .containers import Container
from .services import Service from .services import Service
app = FastAPI() app = FastAPI()
@app.api_route("/") @app.api_route("/")
@inject @inject
async def index(service: Service = Depends(Provide[Container.service])): async def index(
service: Annotated[Service, Depends(Provide[Container.service])]
) -> dict[str, str]:
value = await service.process() value = await service.process()
return {"result": value} return {"result": value}

View File

@ -1,4 +1,7 @@
from fastapi import FastAPI, Depends from typing import Annotated
from fastapi import Depends, FastAPI
from dependency_injector import containers, providers from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject from dependency_injector.wiring import Provide, inject
@ -18,7 +21,9 @@ app = FastAPI()
@app.api_route("/") @app.api_route("/")
@inject @inject
async def index(service: Service = Depends(Provide[Container.service])): async def index(
service: Annotated[Service, Depends(Provide[Container.service])]
) -> dict[str, str]:
result = await service.process() result = await service.process()
return {"result": result} return {"result": result}

View File

@ -1,11 +1,14 @@
"""Endpoints module.""" """Endpoints module."""
from typing import Annotated
from fastapi import APIRouter, Depends, Response, status from fastapi import APIRouter, Depends, Response, status
from dependency_injector.wiring import inject, Provide
from dependency_injector.wiring import Provide, inject
from .containers import Container from .containers import Container
from .services import UserService
from .repositories import NotFoundError from .repositories import NotFoundError
from .services import UserService
router = APIRouter() router = APIRouter()
@ -13,7 +16,7 @@ router = APIRouter()
@router.get("/users") @router.get("/users")
@inject @inject
def get_list( def get_list(
user_service: UserService = Depends(Provide[Container.user_service]), user_service: Annotated[UserService, Depends(Provide[Container.user_service])],
): ):
return user_service.get_users() return user_service.get_users()
@ -21,8 +24,8 @@ def get_list(
@router.get("/users/{user_id}") @router.get("/users/{user_id}")
@inject @inject
def get_by_id( def get_by_id(
user_id: int, user_id: int,
user_service: UserService = Depends(Provide[Container.user_service]), user_service: Annotated[UserService, Depends(Provide[Container.user_service])],
): ):
try: try:
return user_service.get_user_by_id(user_id) return user_service.get_user_by_id(user_id)
@ -33,7 +36,7 @@ def get_by_id(
@router.post("/users", status_code=status.HTTP_201_CREATED) @router.post("/users", status_code=status.HTTP_201_CREATED)
@inject @inject
def add( def add(
user_service: UserService = Depends(Provide[Container.user_service]), user_service: Annotated[UserService, Depends(Provide[Container.user_service])],
): ):
return user_service.create_user() return user_service.create_user()
@ -41,9 +44,9 @@ def add(
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) @router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
@inject @inject
def remove( def remove(
user_id: int, user_id: int,
user_service: UserService = Depends(Provide[Container.user_service]), user_service: Annotated[UserService, Depends(Provide[Container.user_service])],
): ) -> Response:
try: try:
user_service.delete_user_by_id(user_id) user_service.delete_user_by_id(user_id)
except NotFoundError: except NotFoundError:

View File

@ -1,13 +1,14 @@
"""Endpoints module.""" """Endpoints module."""
from typing import Optional, List from typing import Annotated, List
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from pydantic import BaseModel from pydantic import BaseModel
from dependency_injector.wiring import inject, Provide
from .services import SearchService from dependency_injector.wiring import Provide, inject
from .containers import Container from .containers import Container
from .services import SearchService
class Gif(BaseModel): class Gif(BaseModel):
@ -26,11 +27,15 @@ router = APIRouter()
@router.get("/", response_model=Response) @router.get("/", response_model=Response)
@inject @inject
async def index( async def index(
query: Optional[str] = None, default_query: Annotated[str, Depends(Provide[Container.config.default.query])],
limit: Optional[str] = None, default_limit: Annotated[
default_query: str = Depends(Provide[Container.config.default.query]), int, Depends(Provide[Container.config.default.limit.as_int()])
default_limit: int = Depends(Provide[Container.config.default.limit.as_int()]), ],
search_service: SearchService = Depends(Provide[Container.search_service]), search_service: Annotated[
SearchService, Depends(Provide[Container.search_service])
],
query: str | None = None,
limit: int | None = None,
): ):
query = query or default_query query = query or default_query
limit = limit or default_limit limit = limit or default_limit