address and address detail

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
 

src/authentication/routes.py

@@ -1,4 +1,5 @@
 AUTH_PREFIX = '/auth'
+#AUTH_PREFIX = ''
 
 
 class AuthRoutesPrefixes:

src/common/repository/sqlalchemy.py

@@ -63,3 +63,4 @@ async def filter(self, **kwargs) -> List[PType]:
         result = await self.session.execute(stmt)
         instances = result.scalars().all()
         return [self.pydantic_model.model_validate(instance) for instance in instances]
+

src/main.py

@@ -9,6 +9,7 @@
 from src.general.views import router as status_router
 from src.routes import BaseRoutesPrefixes
 from src.users.views import user_router
+from src.users.views import address_router
 
 def include_routes(application: FastAPI) -> None:
     application.include_router(
@@ -29,6 +30,11 @@ def include_routes(application: FastAPI) -> None:
             prefix=BaseRoutesPrefixes.account,
             tags=['Account'],
         )
+    application.include_router(
+            router=address_router,
+            prefix=BaseRoutesPrefixes.account,
+            tags=['Account'],
+        )
 
 def get_application() -> FastAPI:
     application = FastAPI(

src/routes.py

@@ -4,5 +4,7 @@ class BaseRoutesPrefixes:
     openapi: str = '/openapi.json'
 
     catalogue: str = '/catalogue'
-    authentication: str = '/auth'
+    authentication: str = ''
+#    authentication: str = '/auth'
     account: str = '/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,24 @@ class UserModel(BaseModel):
 
 class UserWithPassword(UserModel):
     hashed_password: str
+
+
+class UserAddressModel(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: Union[int, None] = None
+    title: str
+
+
+class UserAddressModelDetail(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: Union[int, None] = None
+    title: str
+    city: str
+    street: str
+    house: str
+    apartment: Optional[str] = None
+    post_code: Optional[str] = None
+    floor: Optional[str] = None
+    additional_info: Optional[str] = None

src/users/repository.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, List
 
 from fastapi import Depends
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -8,9 +8,9 @@
 from src.common.repository.sqlalchemy import BaseSQLAlchemyRepository
 from src.users.models.pydantic import (
     UserModel,
-    UserWithPassword,
+    UserWithPassword, UserAddressModel, UserAddressModelDetail,
 )
-from src.users.models.sqlalchemy import User
+from src.users.models.sqlalchemy import User, UserAddress
 
 
 class UserRepository(BaseSQLAlchemyRepository[User, UserModel]):
@@ -35,3 +35,33 @@ 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, UserAddressModel]):
+    def __init__(self, session: AsyncSession):
+        super().__init__(model=UserAddress, pydantic_model=UserAddressModel, session=session)
+
+    async def create(self, *args, **kwargs):
+        raise NotImplementedError
+
+    async def delete(self, *args, **kwargs):
+        raise NotImplementedError
+
+    async def get_by_user_id(self, user_id: int) -> List[UserAddressModel]:
+        stmt = select(self.model).where(self.model.user_id == user_id)
+        result = await self.session.execute(stmt)
+        addresses = result.scalars().all()
+
+        return [UserAddressModel.model_validate(address) for address in addresses]
+
+    async def get_details(self, pk: int) -> UserAddress:
+        stmt = select(self.model).where(self.model.id == pk)
+        result = await self.session.execute(stmt)
+        addresses_detail = result.scalar_one_or_none()
+
+        return addresses_detail
+
+def get_user_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):
-    ...
+    address: str = '/address'
+    detail: str = '/detail'

src/users/services.py

@@ -1,15 +1,15 @@
-from typing import Optional
+from typing import Optional, List
 
 from fastapi import Depends
 
 from src.authentication.security import verify_password
 from src.common.service import BaseService
 from src.users.models.pydantic import (
-    UserModel,
+    UserModel, UserAddressModel, UserAddressModelDetail,
 )
 from src.users.repository import (
     UserRepository,
-    get_user_repository,
+    get_user_repository, UserAddressRepository, get_user_address_repository,
 )
 
 
@@ -31,3 +31,20 @@ 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[UserAddressModel]):
+    def __init__(self, repository: UserAddressRepository):
+        super().__init__(repository)
+
+    async def get_by_user_id(self, user_id: int) -> List[UserAddressModel]:
+        return await self.repository.get_by_user_id(user_id=user_id)
+
+    async def get_details(self, pk: int) -> UserAddressModelDetail:
+        return await self.repository.get_details(pk)
+
+
+def get_user_address_service(repo: UserAddressRepository = Depends(get_user_address_repository)) -> UserAddressService:
+    return UserAddressService(repository=repo)

src/users/views/__init__.py

@@ -1 +1,2 @@
 from .user import router as user_router
+from .address import router as address_router

src/users/views/address.py

@@ -0,0 +1,69 @@
+from typing import (
+    Annotated,
+    Union, List,
+)
+
+from fastapi import (
+    APIRouter,
+    Depends,
+    Response,
+    status, HTTPException,
+)
+
+from src.authentication.utils import get_current_user
+from src.common.exceptions.base import ObjectDoesNotExistException
+
+from src.common.schemas.common import ErrorResponse
+from src.users.models.pydantic import UserAddressModel, UserModel, UserAddressModelDetail
+
+from src.users.routes import UserRoutesPrefixes
+from src.users.services import UserAddressService, get_user_address_service
+
+
+
+
+router = APIRouter(prefix=UserRoutesPrefixes.address)
+
+
+@router.get(
+    UserRoutesPrefixes.root,
+    responses={
+        status.HTTP_200_OK: {'model': UserAddressModel},
+        status.HTTP_404_NOT_FOUND: {'model': ErrorResponse},
+    },
+    status_code=status.HTTP_200_OK,
+    response_model=List[UserAddressModel],
+)
+async def user_address(
+    current_user: Annotated[UserModel, Depends(get_current_user)],
+    addresses_service: Annotated[UserAddressService, Depends(get_user_address_service)]
+) -> List[UserAddressModel]:
+
+    return await addresses_service.get_by_user_id(current_user.id)
+
+
+
+@router.get(
+    UserRoutesPrefixes.detail,
+    responses={
+        status.HTTP_200_OK: {'model': UserAddressModelDetail},
+        status.HTTP_404_NOT_FOUND: {'model': ErrorResponse},
+    },
+    status_code=status.HTTP_200_OK,
+    response_model=Union[UserAddressModelDetail, ErrorResponse],
+)
+async def address_detail(
+    pk: int,
+    current_user: Annotated[UserModel, Depends(get_current_user)],
+    service: Annotated[get_user_address_service, Depends()],
+) -> Union[UserAddressModelDetail, ErrorResponse]:
+
+    try:
+        address = await service.get_details(pk)
+    except ObjectDoesNotExistException as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+    if address.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
+    return UserAddressModelDetail.model_validate(address)