PROD.2026/RekomenciBackend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(): add validation and documentation to input schemas

Browse Source
This commit is contained in:
gitgernit
2025-11-21 13:45:52 +03:00
parent 3eb4666eff
commit 483c7f90ea
7 changed files with 248 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -18,6 +18,7 @@ dependencies = [
"aioboto3==15.5.0",
"prometheus-fastapi-instrumentator>=7.1.0",
"python-multipart>=0.0.20",
"pydantic[email]>=2.12.4",
]
[dependency-groups]
src/template_project/web_api/routes/auth.py
+67 -15
View File
@@ -1,7 +1,7 @@
from dishka import FromDishka
from dishka.integrations.fastapi import DishkaRoute
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, SecretStr
from pydantic import BaseModel, EmailStr, Field, SecretStr
from template_project.application.auth_identity.errors import (
AuthError,
@@ -18,15 +18,29 @@ router = APIRouter(route_class=DishkaRoute, tags=["Auth"])
class EmailSignUpRequest(BaseModel):
email: str
password: SecretStr
email: EmailStr = Field(..., description="User email address", examples=["user@example.com"])
password: SecretStr = Field(
..., min_length=8, description="User password (minimum 8 characters)", examples=["password123"]
)
model_config = {"json_schema_extra": {"example": {"email": "user@example.com", "password": "password123"}}}
class EmailSignUpResponse(BaseModel):
access_token: str
access_token: str = Field(..., description="JWT access token for authentication")
model_config = {"json_schema_extra": {"example": {"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."}}}
@router.post("/auth/sign_up/email")
@router.post(
"/auth/sign_up/email",
summary="Sign up with email",
description="Register a new user with email and password",
responses={
200: {"description": "User successfully registered", "model": EmailSignUpResponse},
409: {"description": "User with this email already exists"},
},
)
async def sign_up_email(
request: EmailSignUpRequest,
interactor: FromDishka[SignUpInteractor],
@@ -44,14 +58,26 @@ async def sign_up_email(
class YandexSignUpRequest(BaseModel):
code: str
code: str = Field(..., min_length=1, description="Yandex OAuth authorization code", examples=["abc123def456"])
model_config = {"json_schema_extra": {"example": {"code": "abc123def456"}}}
class YandexSignUpResponse(BaseModel):
access_token: str
access_token: str = Field(..., description="JWT access token for authentication")
model_config = {"json_schema_extra": {"example": {"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."}}}
@router.post("/auth/sign_up/yandex")
@router.post(
"/auth/sign_up/yandex",
summary="Sign up with Yandex",
description="Register a new user using Yandex OAuth",
responses={
200: {"description": "User successfully registered", "model": YandexSignUpResponse},
401: {"description": "Authentication failed - invalid code or OAuth error"},
},
)
async def sign_up_yandex(
request: YandexSignUpRequest,
interactor: FromDishka[SignUpInteractor],
@@ -66,15 +92,28 @@ async def sign_up_yandex(
class EmailSignInRequest(BaseModel):
email: str
password: SecretStr
email: EmailStr = Field(..., description="User email address", examples=["user@example.com"])
password: SecretStr = Field(..., min_length=1, description="User password", examples=["password123"])
model_config = {"json_schema_extra": {"example": {"email": "user@example.com", "password": "password123"}}}
class EmailSignInResponse(BaseModel):
access_token: str
access_token: str = Field(..., description="JWT access token for authentication")
model_config = {"json_schema_extra": {"example": {"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."}}}
@router.post("/auth/sign_in/email")
@router.post(
"/auth/sign_in/email",
summary="Sign in with email",
description="Authenticate user with email and password",
responses={
200: {"description": "User successfully authenticated", "model": EmailSignInResponse},
401: {"description": "Invalid credentials"},
404: {"description": "User not found"},
},
)
async def sign_in_email(
request: EmailSignInRequest,
interactor: FromDishka[SignInInteractor],
@@ -94,14 +133,27 @@ async def sign_in_email(
class YandexSignInRequest(BaseModel):
code: str
code: str = Field(..., min_length=1, description="Yandex OAuth authorization code", examples=["abc123def456"])
model_config = {"json_schema_extra": {"example": {"code": "abc123def456"}}}
class YandexSignInResponse(BaseModel):
access_token: str
access_token: str = Field(..., description="JWT access token for authentication")
model_config = {"json_schema_extra": {"example": {"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."}}}
@router.post("/auth/sign_in/yandex")
@router.post(
"/auth/sign_in/yandex",
summary="Sign in with Yandex",
description="Authenticate user using Yandex OAuth",
responses={
200: {"description": "User successfully authenticated", "model": YandexSignInResponse},
401: {"description": "Invalid authorization code or authentication failed"},
404: {"description": "User not found"},
},
)
async def sign_in_yandex(
request: YandexSignInRequest,
interactor: FromDishka[SignInInteractor],
src/template_project/web_api/routes/healthcheck.py
+12 -3
View File
@@ -1,14 +1,23 @@
from dishka.integrations.fastapi import DishkaRoute
from fastapi import APIRouter
from pydantic import BaseModel
from pydantic import BaseModel, Field
router = APIRouter(route_class=DishkaRoute, tags=["Health"])
class HealthcheckResponse(BaseModel):
ok: bool
ok: bool = Field(..., description="Service health status")
model_config = {"json_schema_extra": {"example": {"ok": True}}}
@router.get("/healthcheck")
@router.get(
"/healthcheck",
summary="Health check",
description="Check if the service is running and healthy",
responses={
200: {"description": "Service is healthy", "model": HealthcheckResponse},
},
)
async def healthcheck() -> HealthcheckResponse:
return HealthcheckResponse(ok=True)
src/template_project/web_api/routes/notification.py
+35 -6
View File
@@ -2,7 +2,7 @@ from dishka import FromDishka
from dishka.integrations.fastapi import DishkaRoute
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from pydantic import BaseModel, Field
from template_project.application.notification_device.errors import NotificationDeviceNotFoundError
from template_project.application.notification_device.interactors.register_device import (
@@ -19,15 +19,36 @@ router = APIRouter(route_class=DishkaRoute, tags=["Notifications"], dependencies
class SendNotificationRequestModel(BaseModel):
title: str
body: str
title: str = Field(..., min_length=1, max_length=100, description="Notification title", examples=["New Message"])
body: str = Field(
..., min_length=1, max_length=500, description="Notification body text", examples=["You have a new message"]
)
model_config = {"json_schema_extra": {"example": {"title": "New Message", "body": "You have a new message"}}}
class RegisterNotificationDeviceRequestModel(BaseModel):
device_id: str
device_id: str = Field(
...,
min_length=1,
max_length=200,
description="Device ID for push notifications (FCM token)",
examples=["fcm_token_123456789"],
)
model_config = {"json_schema_extra": {"example": {"device_id": "fcm_token_123456789"}}}
@router.post("/notifications/send")
@router.post(
"/notifications/send",
summary="Send notification",
description="Send a push notification to the current user's registered device",
responses={
200: {"description": "Notification sent successfully"},
401: {"description": "Unauthorized - invalid or missing token"},
404: {"description": "Notification device not found - user needs to register device first"},
},
)
async def send_notification(
request: SendNotificationRequestModel,
interactor: FromDishka[NotificationInteractor],
@@ -39,7 +60,15 @@ async def send_notification(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Notification device not found") from error
@router.post("/notifications/register_device")
@router.post(
"/notifications/register_device",
summary="Register notification device",
description="Register a device ID (FCM token) for push notifications for the current user",
responses={
200: {"description": "Device registered successfully"},
401: {"description": "Unauthorized - invalid or missing token"},
},
)
async def register_notification_device(
request: RegisterNotificationDeviceRequestModel,
interactor: FromDishka[RegisterNotificationDeviceInteractor],
src/template_project/web_api/routes/profile.py
+84 -24
View File
@@ -2,7 +2,7 @@ from dishka import FromDishka
from dishka.integrations.fastapi import DishkaRoute
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from pydantic import BaseModel, EmailStr, Field, HttpUrl
from template_project.application.user.entity import UserId
from template_project.application.user.profile.interactors.get_profile import GetProfileInteractor
@@ -16,16 +16,39 @@ router = APIRouter(route_class=DishkaRoute, tags=["Profile"], dependencies=[Depe
class GetProfileResponse(BaseModel):
id: UserId
email: str | None
display_name: str | None
first_name: str | None
last_name: str | None
avatar_url: str | None
phone: str | None
id: UserId = Field(..., description="User ID")
email: str | None = Field(None, description="User email address")
display_name: str | None = Field(None, description="User display name")
first_name: str | None = Field(None, description="User first name")
last_name: str | None = Field(None, description="User last name")
avatar_url: str | None = Field(None, description="URL to user avatar")
phone: str | None = Field(None, description="User phone number")
model_config = {
"json_schema_extra": {
"example": {
"id": "01234567-89ab-cdef-0123-456789abcdef",
"email": "user@example.com",
"display_name": "John Doe",
"first_name": "John",
"last_name": "Doe",
"avatar_url": "https://example.com/avatar.jpg",
"phone": "+1234567890",
}
}
}
@router.get("/profile")
@router.get(
"/profile",
summary="Get user profile",
description="Retrieve the current authenticated user's profile information",
responses={
200: {"description": "Profile retrieved successfully", "model": GetProfileResponse},
401: {"description": "Unauthorized - invalid or missing token"},
404: {"description": "Profile not found"},
},
)
async def get_profile(
interactor: FromDishka[GetProfileInteractor],
) -> GetProfileResponse:
@@ -44,25 +67,62 @@ async def get_profile(
class PatchProfileRequestModel(BaseModel):
email: str | None = None
display_name: str | None = None
first_name: str | None = None
last_name: str | None = None
avatar_url: str | None = None
phone: str | None = None
email: EmailStr | None = Field(None, description="User email address", examples=["user@example.com"])
display_name: str | None = Field(None, max_length=100, description="User display name", examples=["John Doe"])
first_name: str | None = Field(None, max_length=50, description="User first name", examples=["John"])
last_name: str | None = Field(None, max_length=50, description="User last name", examples=["Doe"])
avatar_url: HttpUrl | None = Field(
None, description="URL to user avatar", examples=["https://example.com/avatar.jpg"]
)
phone: str | None = Field(None, max_length=20, description="User phone number", examples=["+1234567890"])
model_config = {
"json_schema_extra": {
"example": {
"email": "user@example.com",
"display_name": "John Doe",
"first_name": "John",
"last_name": "Doe",
"avatar_url": "https://example.com/avatar.jpg",
"phone": "+1234567890",
}
}
}
class PatchProfileResponse(BaseModel):
id: UserId
email: str | None
display_name: str | None
first_name: str | None
last_name: str | None
avatar_url: str | None
phone: str | None
id: UserId = Field(..., description="User ID")
email: str | None = Field(None, description="User email address")
display_name: str | None = Field(None, description="User display name")
first_name: str | None = Field(None, description="User first name")
last_name: str | None = Field(None, description="User last name")
avatar_url: str | None = Field(None, description="URL to user avatar")
phone: str | None = Field(None, description="User phone number")
model_config = {
"json_schema_extra": {
"example": {
"id": "01234567-89ab-cdef-0123-456789abcdef",
"email": "user@example.com",
"display_name": "John Doe",
"first_name": "John",
"last_name": "Doe",
"avatar_url": "https://example.com/avatar.jpg",
"phone": "+1234567890",
}
}
}
@router.patch("/profile")
@router.patch(
"/profile",
summary="Update user profile",
description="Update the current authenticated user's profile information. All fields are optional.",
responses={
200: {"description": "Profile updated successfully", "model": PatchProfileResponse},
401: {"description": "Unauthorized - invalid or missing token"},
},
)
async def patch_profile(
request: PatchProfileRequestModel,
interactor: FromDishka[PatchProfileInteractor],
@@ -72,7 +132,7 @@ async def patch_profile(
display_name=request.display_name,
first_name=request.first_name,
last_name=request.last_name,
avatar_url=request.avatar_url,
avatar_url=str(request.avatar_url) if request.avatar_url else None,
phone=request.phone,
)
response = await interactor.execute(patch_request)
src/template_project/web_api/routes/storage.py
+20 -8
View File
@@ -2,9 +2,9 @@ from io import BytesIO
from dishka import FromDishka
from dishka.integrations.fastapi import DishkaRoute
from fastapi import APIRouter, Depends, UploadFile, status
from fastapi.responses import JSONResponse
from fastapi import APIRouter, Depends, UploadFile
from fastapi.security import HTTPBearer
from pydantic import BaseModel, Field
from uuid_utils.compat import uuid7
from template_project.application.common.file_storage import FileStorage
@@ -13,16 +13,28 @@ security = HTTPBearer()
router = APIRouter(route_class=DishkaRoute, tags=["Storage"], dependencies=[Depends(security)])
@router.post("/storage/upload_file")
class UploadFileResponse(BaseModel):
path: str = Field(..., description="Generated file path (UUID7)")
model_config = {"json_schema_extra": {"example": {"path": "01234567-89ab-cdef-0123-456789abcdef"}}}
@router.post(
"/storage/upload_file",
summary="Upload file",
description="Upload a file to storage. The file path is automatically generated using UUID7.",
responses={
200: {"description": "File uploaded successfully", "model": UploadFileResponse},
401: {"description": "Unauthorized - invalid or missing token"},
413: {"description": "File too large"},
},
)
async def upload_file(
file: UploadFile,
file_storage: FromDishka[FileStorage],
) -> JSONResponse:
) -> UploadFileResponse:
path = str(uuid7())
file_content = await file.read()
file_io = BytesIO(file_content)
await file_storage.upload(path=path, image=file_io)
return JSONResponse(
status_code=status.HTTP_200_OK,
content={"path": path},
)
return UploadFileResponse(path=path)
uv.lock
Generated
+29
View File
@@ -623,6 +623,28 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b7/b9/89381173b4f336e986d72471198614806cd313e0f85c143ccb677c310223/dishka-1.7.2-py3-none-any.whl", hash = "sha256:f6faa6ab321903926b825b3337d77172ee693450279b314434864978d01fbad3", size = 94774, upload-time = "2025-09-24T21:23:03.246Z" },
]
[[package]]
name = "dnspython"
version = "2.8.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/8c/8b/57666417c0f90f08bcafa776861060426765fdb422eb10212086fb811d26/dnspython-2.8.0.tar.gz", hash = "sha256:181d3c6996452cb1189c4046c61599b84a5a86e099562ffde77d26984ff26d0f", size = 368251, upload-time = "2025-09-07T18:58:00.022Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ba/5a/18ad964b0086c6e62e2e7500f7edc89e3faa45033c71c1893d34eed2b2de/dnspython-2.8.0-py3-none-any.whl", hash = "sha256:01d9bbc4a2d76bf0db7c1f729812ded6d912bd318d3b1cf81d30c0f845dbf3af", size = 331094, upload-time = "2025-09-07T18:57:58.071Z" },
]
[[package]]
name = "email-validator"
version = "2.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "dnspython" },
{ name = "idna" },
]
sdist = { url = "https://files.pythonhosted.org/packages/f5/22/900cb125c76b7aaa450ce02fd727f452243f2e91a61af068b40adba60ea9/email_validator-2.3.0.tar.gz", hash = "sha256:9fc05c37f2f6cf439ff414f8fc46d917929974a82244c20eb10231ba60c54426", size = 51238, upload-time = "2025-08-26T13:09:06.831Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/de/15/545e2b6cf2e3be84bc1ed85613edd75b8aea69807a71c26f4ca6a9258e82/email_validator-2.3.0-py3-none-any.whl", hash = "sha256:80f13f623413e6b197ae73bb10bf4eb0908faf509ad8362c5edeb0be7fd450b4", size = 35604, upload-time = "2025-08-26T13:09:05.858Z" },
]
[[package]]
name = "fastapi"
version = "0.119.0"
@@ -1600,6 +1622,11 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/82/2f/e68750da9b04856e2a7ec56fc6f034a5a79775e9b9a81882252789873798/pydantic-2.12.4-py3-none-any.whl", hash = "sha256:92d3d202a745d46f9be6df459ac5a064fdaa3c1c4cd8adcfa332ccf3c05f871e", size = 463400, upload-time = "2025-11-05T10:50:06.732Z" },
]
[package.optional-dependencies]
email = [
{ name = "email-validator" },
]
[[package]]
name = "pydantic-core"
version = "2.41.5"
@@ -1952,6 +1979,7 @@ dependencies = [
{ name = "httpx" },
{ name = "prometheus-fastapi-instrumentator" },
{ name = "psycopg", extra = ["binary"] },
{ name = "pydantic", extra = ["email"] },
{ name = "python-multipart" },
{ name = "sqlalchemy" },
{ name = "uuid-utils" },
@@ -2001,6 +2029,7 @@ requires-dist = [
{ name = "httpx", specifier = "==0.28.1" },
{ name = "prometheus-fastapi-instrumentator", specifier = ">=7.1.0" },
{ name = "psycopg", extras = ["binary"], specifier = ">=3.2.12" },
{ name = "pydantic", extras = ["email"], specifier = ">=2.12.4" },
{ name = "python-multipart", specifier = ">=0.0.20" },
{ name = "sqlalchemy", specifier = "==2.0.44" },
{ name = "uuid-utils", specifier = "==0.11.1" },