Use Annotated for DI in FastAPI examples

This commit is contained in:
ZipFile 2025-02-02 17:36:48 +00:00
parent 2330122de6
commit 63813d67ea
5 changed files with 41 additions and 24 deletions

View File

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

View File

@ -1,18 +1,22 @@
"""Application module."""
from dependency_injector.wiring import inject, Provide
from fastapi import FastAPI, Depends
from typing import Annotated
from fastapi import Depends, FastAPI
from dependency_injector.wiring import Provide, inject
from .containers import Container
from .services import Service
app = FastAPI()
@app.api_route("/")
@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()
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.wiring import Provide, inject
@ -18,7 +21,9 @@ app = FastAPI()
@app.api_route("/")
@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()
return {"result": result}

View File

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

View File

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