Розробка API для Управління Адресами Користувача

Dockerfile

@@ -6,7 +6,7 @@ COPY pyproject.toml pyproject.toml
 
 RUN pip install poetry
 RUN poetry config virtualenvs.create false
-RUN poetry install --no-dev
+RUN poetry install --no-root
 
 COPY . /app
 

docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3.8'
 
 services:
   web:

src/catalogue/routes.py

@@ -7,3 +7,6 @@ class CatalogueRoutesPrefixes:
 
 class ProductRoutesPrefixes(BaseCrudPrefixes):
     ...
+
+class AddressRoutesPrefixes(BaseCrudPrefixes):
+    ...

src/routes.py

@@ -1,8 +1,8 @@
 class BaseRoutesPrefixes:
-    swagger: str = '/docs'
-    redoc: str = '/redoc'
-    openapi: str = '/openapi.json'
+    swagger = '/docs'
+    redoc = '/redoc'
+    openapi = '/openapi.json'
 
-    catalogue: str = '/catalogue'
-    authentication: str = '/auth'
-    account: str = '/account'
+    catalogue = '/catalogue'
+    authentication = '/auth'
+    account = '/account'

src/users/models/pydantic.py

@@ -1,4 +1,4 @@
-from typing import Union
+from typing import Union, Optional
 
 from pydantic import (
     BaseModel,
@@ -19,3 +19,18 @@ class UserModel(BaseModel):
 
 class UserWithPassword(UserModel):
     hashed_password: str
+
+class UserAddressListResponse(BaseModel):
+    id: int
+    title: str
+
+class UserAddressDetailResponse(BaseModel):
+    id: int
+    title: Optional[str]
+    city: str
+    street: str
+    house: str
+    apartment: Optional[str]
+    post_code: Optional[str]
+    floor: Optional[str]
+    additional_info: Optional[str]

src/users/repository.py

@@ -8,10 +8,10 @@
 from src.common.repository.sqlalchemy import BaseSQLAlchemyRepository
 from src.users.models.pydantic import (
     UserModel,
-    UserWithPassword,
+    UserWithPassword, UserAddressDetailResponse, UserAddressListResponse
 )
-from src.users.models.sqlalchemy import User
-
+from src.users.models.sqlalchemy import User, UserAddress
+from src.common.exceptions.base import ObjectDoesNotExistException
 
 class UserRepository(BaseSQLAlchemyRepository[User, UserModel]):
     def __init__(self, session: AsyncSession):
@@ -35,3 +35,30 @@ async def get_by_email(self, email: str) -> Optional[UserWithPassword]:
 
 def get_user_repository(session: AsyncSession = Depends(get_session)) -> UserRepository:
     return UserRepository(session=session)
+
+class UserAddressRepository(BaseSQLAlchemyRepository[UserAddress, UserAddressDetailResponse]):
+    def __init__(self, session: AsyncSession):
+        super().__init__(model=UserAddress, pydantic_model=UserAddressDetailResponse, session=session)
+
+    async def create(self, *args, **kwargs):
+        raise NotImplementedError
+
+    async def delete(self, *args, **kwargs):
+        raise NotImplementedError
+
+    async def get_by_user(self, user_id : int) -> list[UserAddressListResponse]:
+        stmt = select(self.model).where(self.model.user_id == user_id)
+        result = await self.session.execute(stmt)
+        addresses = result.scalars().all()
+        return [UserAddressListResponse.model_validate(a) for a in addresses]
+
+    async def get_detail(self, address_id: int) -> UserAddressDetailResponse:
+        stmt = select(self.model).where(self.model.id == address_id)
+        result = await self.session.execute(stmt)
+        address = result.scalar_one_or_none()
+        if not address:
+            raise ObjectDoesNotExistException(f"Address with id {address_id} not found")
+        return UserAddressDetailResponse.model_validate(address)
+
+def get_address_repository(session: AsyncSession = Depends(get_session)) -> UserAddressRepository:
+    return UserAddressRepository(session=session)

src/users/routes.py

@@ -6,4 +6,5 @@ class UserManagementRoutesPrefixes:
 
 
 class UserRoutesPrefixes(BaseCrudPrefixes):
-    ...
+    root = "/addresses"
+    detail = "/addresses/{address_id}"

src/users/services.py

@@ -5,11 +5,12 @@
 from src.authentication.security import verify_password
 from src.common.service import BaseService
 from src.users.models.pydantic import (
-    UserModel,
+    UserModel, UserAddressDetailResponse, UserAddressListResponse
 )
+from src.users.models.sqlalchemy import UserAddress
 from src.users.repository import (
-    UserRepository,
-    get_user_repository,
+    UserRepository, UserAddressRepository,
+    get_user_repository, get_address_repository
 )
 
 
@@ -31,3 +32,19 @@ async def authenticate(self, email: str, password: str) -> Optional[UserModel]:
 
 def get_user_service(repo: UserRepository = Depends(get_user_repository)) -> UserService:
     return UserService(repository=repo)
+
+class UserAddressService(BaseService[UserAddress]):
+    def __init__(self, repository: UserAddressRepository):
+        super().__init__(repository)
+
+    async def get_by_user_list(self, user_id: int):
+        return await self.repository.get_by_user(user_id=user_id)
+
+    async def get_detail_for_user(self, address_id: int, user_id: int):
+        address = await self.repository.get_detail(address_id=address_id)
+        if address.user_id != user_id:
+            raise PermissionError("Forbidden")
+        return UserAddressDetailResponse.model_validate(address)
+
+def get_address_service(repo: UserAddressRepository = Depends(get_address_repository)) -> UserAddressService:
+    return UserAddressService(repository=repo)

src/users/views/user_address.py

@@ -11,14 +11,17 @@
 )
 
 from src.catalogue.models.pydantic import ProductModel
+from src.users.models.sqlalchemy import UserAddress
+from src.users.models.pydantic import UserAddressListResponse, UserAddressDetailResponse
 from src.catalogue.routes import (
     CatalogueRoutesPrefixes,
-    ProductRoutesPrefixes,
+    ProductRoutesPrefixes, AddressRoutesPrefixes,
 )
 from src.catalogue.services import get_product_service
+from src.users.services import get_address_service, UserAddressService
 from src.common.exceptions.base import ObjectDoesNotExistException
 from src.common.schemas.common import ErrorResponse
-
+from src.authentication.utils import get_current_user
 
 router = APIRouter(prefix=CatalogueRoutesPrefixes.product)
 
@@ -65,3 +68,56 @@ async def product_detail(
         return ErrorResponse(message=exc.message)
 
     return response
+
+
+
+
+
+
+@router.get(
+    AddressRoutesPrefixes.root,
+    status_code=status.HTTP_200_OK,
+    response_model=list[UserAddressListResponse],
+)
+async def address_list(
+    address_service: Annotated[UserAddressService, Depends(get_address_service)],
+    current_user = Depends(get_current_user),
+) -> list[UserAddressListResponse]:
+    """
+    Get list of addresses for the current user.
+    """
+    addresses = await address_service.get_by_user_list(user_id=current_user.id)
+    return addresses
+
+
+@router.get(
+    AddressRoutesPrefixes.detail,
+    responses={
+        status.HTTP_200_OK: {'model': UserAddress},
+        status.HTTP_404_NOT_FOUND: {'model': ErrorResponse},
+    },
+    status_code=status.HTTP_200_OK,
+    response_model=Union[UserAddressDetailResponse, ErrorResponse],
+)
+async def address_detail(
+    address_id: int,
+    response: Response,
+    address_service: Annotated[UserAddressService, Depends(get_address_service)],
+    current_user = Depends(get_current_user),
+) -> Union[UserAddressDetailResponse, ErrorResponse]:
+    """
+    Retrieve address details for the current user.
+    """
+    try:
+        address = await address_service.get_detail_for_user(
+            address_id=address_id,
+            user_id=current_user.id
+        )
+    except ObjectDoesNotExistException as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ErrorResponse(message=exc.message)
+    except PermissionError:
+        response.status_code = status.HTTP_403_FORBIDDEN
+        return ErrorResponse(message="Forbidden")
+
+    return address