first commit

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

@@ -11,7 +11,8 @@ services:
       - db
     networks:
       - web_network
-
+    env_file:
+      - env.example
   db:
     image: postgres:13
     volumes:

src/general/views.py

@@ -1,17 +1,30 @@
 """
 Provide status views.
 """
+from typing import List
 from fastapi import (
     APIRouter,
+    Depends,
+    HTTPException,
     status,
 )
 
 from src.common.schemas.common import DetailsResponse
 from src.general.routes import GeneralRoutesPrefixes
+from src.users.dependencies import get_current_user
 
+from src.users.models.pydantic import (
+    UserModel,
+    UserAddressShort,
+    UserAddressDetail,
+)
+from src.users.services import (
+    UserAddressService,
+    get_user_address_service,
+)
 
-router = APIRouter()
 
+router = APIRouter()
 
 @router.get(
     GeneralRoutesPrefixes.health_check,
@@ -27,3 +40,23 @@ def health_check() -> DetailsResponse:
         Response showing whether server is alive.
     """
     return DetailsResponse(details='UP')
+
+@router.get("/", response_model=List[UserAddressShort])
+async def list_addresses(
+    current_user: UserModel = Depends(get_current_user),
+    service: UserAddressService = Depends(get_user_address_service),
+):
+    addresses = await service.get_user_addresses(user_id=current_user.id)
+    return [UserAddressShort.model_validate(addr) for addr in addresses]
+
+
+@router.get("/{address_id}", response_model=UserAddressDetail)
+async def address_detail(
+    address_id: int,
+    current_user: UserModel = Depends(get_current_user),
+    service: UserAddressService = Depends(get_user_address_service),
+):
+    address = await service.get_user_address_detail(user_id=current_user.id, address_id=address_id)
+    if not address:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not allowed to access this address")
+    return UserAddressDetail.model_validate(address)

src/users/dependencies.py

@@ -0,0 +1,11 @@
+from src.users.models.pydantic import UserModel
+
+
+async def get_current_user() -> UserModel:
+    return UserModel(
+        id=1,
+        first_name="Test",
+        last_name="User",
+        email="test@example.com",
+        phone_number="+380001112233",
+    )

src/users/models/pydantic.py

@@ -1,3 +1,6 @@
+from typing import Optional, Union
+from pydantic import BaseModel, ConfigDict, EmailStr
+
 from typing import Union
 
 from pydantic import (
@@ -19,3 +22,36 @@ class UserModel(BaseModel):
 
 class UserWithPassword(UserModel):
     hashed_password: str
+
+class UserAddressBase(BaseModel):
+    title: Optional[str] = None
+    city: str
+    street: str
+    house: str
+    apartment: Optional[str] = None
+    post_code: Optional[str] = None
+    floor: Optional[str] = None
+    additional_info: Optional[str] = None
+
+
+class UserAddressCreate(UserAddressBase):
+    pass
+
+
+class UserAddressModel(UserAddressBase):
+    id: int
+    user_id: int
+
+    model_config = ConfigDict(from_attributes=True)
+
+class UserAddressShort(BaseModel):
+    id: int
+    title: Optional[str] = None
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class UserAddressDetail(UserAddressBase):
+    id: int
+
+    model_config = ConfigDict(from_attributes=True)

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
@@ -9,8 +9,9 @@
 from src.users.models.pydantic import (
     UserModel,
     UserWithPassword,
+    UserAddressModel,
 )
-from src.users.models.sqlalchemy import User
+from src.users.models.sqlalchemy import User, UserAddress
 
 
 class UserRepository(BaseSQLAlchemyRepository[User, UserModel]):
@@ -32,6 +33,23 @@ async def get_by_email(self, email: str) -> Optional[UserWithPassword]:
 
         return UserWithPassword.model_validate(user)
 
-
 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 get_by_user_id(self, user_id: int) -> List[UserAddressModel]:
+        return await self.filter(user_id=user_id)
+
+    async def get_one_by_user(self, user_id: int, address_id: int) -> UserAddressModel | None:
+        stmt = select(self.model).where(self.model.id == address_id, self.model.user_id == user_id)
+        result = await self.session.execute(stmt)
+        instance = result.scalar_one_or_none()
+        if not instance:
+            return None
+        return self.pydantic_model.model_validate(instance)
+
+def get_user_address_repository(session: AsyncSession = Depends(get_session)) -> UserAddressRepository:
+    return UserAddressRepository(session=session)

src/users/services.py

@@ -1,15 +1,18 @@
-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,
+    UserAddressModel,
 )
 from src.users.repository import (
     UserRepository,
     get_user_repository,
+    UserAddressRepository,
+    get_user_address_repository,
 )
 
 
@@ -31,3 +34,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[UserAddressModel]):
+    def __init__(self, repository: UserAddressRepository):
+        super().__init__(repository)
+
+    async def get_user_addresses(self, user_id: int) -> List[UserAddressModel]:
+        return await self.repository.get_by_user_id(user_id=user_id)
+
+    async def get_user_address_detail(self, user_id: int, address_id: int) -> Optional[UserAddressModel]:
+        return await self.repository.get_one_by_user(user_id=user_id, address_id=address_id)
+
+
+def get_user_address_service(
+    repo: UserAddressRepository = Depends(get_user_address_repository),
+) -> UserAddressService:
+    return UserAddressService(repository=repo)