Minor fixes

Author: szymon6927

README.md

@@ -15,7 +15,7 @@ This is for educational purposes only and the code is not production-ready.
 
 ## Stack
 
-- Python 3.9
+- Python 3.10
 - FastAPI
 - MongoDB
 

example.env

@@ -1,8 +1,9 @@
+ENVIRONMENT=dev
 MONGO_CONNECTION_URL=mongodb://localhost:27017/
 MONGO_DATABASE_NAME=hexagonal_architecture_python
 S3_CLIENTS_BUCKET=client-reports
 DROPBOX_ACCESS_TOKEN=xxxx
 BUCKET_REGION=eu-west-1
 AWS_DEFAULT_REGION=eu-west-1
 AWS_ACCESS_KEY_ID=test
-AWS_SECRET_ACCESS_KEY=test
+AWS_SECRET_ACCESS_KEY=test

pyproject.toml

@@ -6,7 +6,7 @@ requires = ["poetry-core"]
 
 [tool.black]
 line-length = 120
-target-version = ["py39"]
+target-version = ["py310"]
 
 [tool.coverage.report]
 exclude_lines = [
@@ -49,7 +49,7 @@ warn_unused_configs = true
 authors = ["Szymon Miks <miks.szymon@gmail.com>"]
 description = ""
 name = "hexagonal-architecture-python"
-packages = [{include = "hexagonal_architecture_python"}]
+packages = [{include = "src"}]
 readme = "README.md"
 version = "0.1.0"
 
@@ -65,8 +65,8 @@ mypy-boto3-sns = "^1.24.68"
 pymongo = "^4.2.0"
 python = "^3.10"
 python-dateutil = "^2.8.2"
-requests = "^2.28.1"
 python-dotenv = "^0.21.0"
+requests = "^2.28.1"
 
 [tool.poetry.group.dev.dependencies]
 black = "^22.10.0"
@@ -85,15 +85,19 @@ toml-sort = "^0.20.1"
 good-names = "id,i,j,k"
 
 [tool.pylint.DESIGN]
-max-args = 5
+max-args = 7
 max-attributes = 8
 min-public-methods = 1
 
 [tool.pylint.FORMAT]
 max-line-length = 120
 
+[tool.pylint.MASTER]
+extension-pkg-whitelist = "pydantic"
+
 [tool.pylint."MESSAGES CONTROL"]
 disable = "missing-docstring, line-too-long, logging-fstring-interpolation, duplicate-code"
+extension-pkg-whitelist = "pydantic"
 
 [tool.pylint.MISCELLANEOUS]
 notes = "XXX"

src/__init__.py

@@ -1,3 +1,3 @@
 from dotenv import load_dotenv
 
-load_dotenv()
+load_dotenv()

src/app.py

@@ -1,17 +1,64 @@
+from typing import Any
 
 import uvicorn
 from fastapi import FastAPI
+from fastapi.openapi.utils import get_openapi
+from fastapi.requests import Request
+from fastapi.responses import JSONResponse
 
+from src.building_blocks.errors import APIErrorMessage, DomainError, RepositoryError, ResourceNotFound
 from src.clients.controllers import router as clients_router
 from src.gym_classes.controllers import router as gym_classes_router
 from src.gym_passes.controllers import router as gym_passes_router
 
-
 app = FastAPI()
 app.include_router(gym_classes_router)
 app.include_router(clients_router)
 app.include_router(gym_passes_router)
 
 
+@app.exception_handler(DomainError)
+async def domain_error_handler(request: Request, exc: DomainError) -> JSONResponse:
+ error_msg = APIErrorMessage(type=exc.__class__.__name__, message=f"Oops! {exc}")
+ return JSONResponse(
+ status_code=400,
+ content=error_msg.dict(),
+ )
+
+
+@app.exception_handler(ResourceNotFound)
+async def resource_not_found_handler(request: Request, exc: ResourceNotFound) -> JSONResponse:
+ error_msg = APIErrorMessage(type=exc.__class__.__name__, message=str(exc))
+ return JSONResponse(status_code=404, content=error_msg.dict())
+
+
+@app.exception_handler(RepositoryError)
+async def repository_error_handler(request: Request, exc: RepositoryError) -> JSONResponse:
+ error_msg = APIErrorMessage(
+ type=exc.__class__.__name__, message="Oops! Something went wrong, please try again later..."
+ )
+ return JSONResponse(
+ status_code=500,
+ content=error_msg.dict(),
+ )
+
+
+def custom_openapi() -> dict[str, Any]:
+ if app.openapi_schema:
+ return app.openapi_schema # type: ignore
+
+ openapi_schema = get_openapi(
+ title="hexagonal-architecture-python",
+ version="1.0.0",
+ description="Hexagonal architecture in Python build on top of FastAPI",
+ routes=app.routes,
+ )
+ app.openapi_schema = openapi_schema
+
+ return app.openapi_schema # type: ignore
+
+
+app.openapi = custom_openapi # type: ignore
+
 if __name__ == "__main__":
  uvicorn.run(app, host="localhost", port=8000)

src/building_blocks/errors.py

@@ -1,7 +1,14 @@
+from pydantic import BaseModel
+
+
 class DomainError(Exception):
  pass
 
 
+class ResourceNotFound(DomainError):
+ pass
+
+
 class RepositoryError(DomainError):
  @classmethod
  def save_operation_failed(cls) -> "RepositoryError":
@@ -10,3 +17,8 @@ def save_operation_failed(cls) -> "RepositoryError":
  @classmethod
  def get_operation_failed(cls) -> "RepositoryError":
  return cls("An error occurred while retrieving the data from the database!")
+
+
+class APIErrorMessage(BaseModel):
+ type: str
+ message: str

src/clients/application/client_service.py

@@ -34,6 +34,7 @@ def __init__(
 
  def create(self, input_dto: CreateClientDTO) -> ClientDTO:
  client = Client.create(input_dto.first_name, input_dto.last_name, input_dto.email)
+
  self._client_repo.save(client)
 
  snapshot = client.to_snapshot()
@@ -54,7 +55,9 @@ def change_personal_data(self, input_dto: ChangeClientPersonalDataDTO) -> Client
  def archive(self, input_dto: ArchiveClientDTO) -> None:
  client = self._client_repo.get(ClientId.of(input_dto.client_id))
  client.archive()
+
  self._client_repo.save(client)
+
  self._gym_pass_facade.disable_for(str(client.id.value))
 
  def export(self, input_dto: ExportClientsDTO) -> None:

src/clients/application/exporter_factory.py

@@ -1,4 +1,5 @@
 import os
+from enum import Enum, unique
 
 import boto3
 from dropbox import Dropbox
@@ -8,15 +9,22 @@
 from src.clients.infrastructure.s3_clients_exporter import S3ClientsExporter
 
 
+@unique
+class Destination(str, Enum):
+ S3 = "s3"
+ DROPBOX = "dropbox"
+
+
 class ExporterFactory:
  @staticmethod
- def build(destination: str) -> IClientsExporter:
- if destination == "s3":
- s3_sdk = boto3.client("s3", endpoint_url="http://localhost:4566")
+ def build(destination: Destination) -> IClientsExporter:
+ if destination == Destination.S3:
+ endpoint_url = "http://localhost:4566" if os.getenv("ENVIRONMENT") == "dev" else None
+ s3_sdk = boto3.client("s3", endpoint_url=endpoint_url)
  bucket_name = os.environ["S3_CLIENTS_BUCKET"]
  return S3ClientsExporter(s3_sdk, bucket_name)
 
- if destination == "dropbox":
+ if destination == Destination.DROPBOX:
  dropbox_access_token = os.environ["DROPBOX_ACCESS_TOKEN"]
  dropbox_client = Dropbox(dropbox_access_token)
  return DropboxClientsExporter(dropbox_client)

src/clients/bootstrap.py

@@ -2,7 +2,7 @@
 
 from src.building_blocks.db import get_mongo_database
 from src.clients.application.client_service import ClientService
-from src.clients.application.exporter_factory import ExporterFactory
+from src.clients.application.exporter_factory import Destination, ExporterFactory
 from src.clients.domain.client_repository import IClientRepository
 from src.clients.domain.clients_exporter import IClientsExporter
 from src.clients.infrastructure.mongo_client_repository import MongoDBClientRepository
@@ -11,7 +11,7 @@
 
 def bootstrap_di() -> None:
  repository = MongoDBClientRepository(get_mongo_database())
- clients_exporter = ExporterFactory.build("s3")
+ clients_exporter = ExporterFactory.build(Destination.S3)
 
  di[IClientRepository] = repository
  di[IClientsExporter] = clients_exporter

src/clients/controllers.py

@@ -2,6 +2,7 @@
 from fastapi.responses import JSONResponse, Response
 from kink import di
 
+from src.building_blocks.errors import APIErrorMessage
 from src.clients.application.client_service import ClientService
 from src.clients.application.dto import (
  ArchiveClientDTO,
@@ -14,15 +15,25 @@
 router = APIRouter()
 
 
-@router.post("/clients", response_model=ClientDTO, tags=["clients"])
+@router.post(
+ "/clients",
+ response_model=ClientDTO,
+ responses={400: {"model": APIErrorMessage}, 500: {"model": APIErrorMessage}},
+ tags=["clients"],
+)
 async def create_client(
  request: CreateClientDTO, service: ClientService = Depends(lambda: di[ClientService])
 ) -> JSONResponse:
  result = service.create(request)
  return JSONResponse(content=result.dict(), status_code=status.HTTP_201_CREATED)
 
 
-@router.put("/clients/{client_id}", response_model=ClientDTO, tags=["clients"])
+@router.put(
+ "/clients/{client_id}",
+ response_model=ClientDTO,
+ responses={400: {"model": APIErrorMessage}, 404: {"model": APIErrorMessage}, 500: {"model": APIErrorMessage}},
+ tags=["clients"],
+)
 async def change_personal_data(
  client_id: str, request: ChangeClientPersonalDataDTO, service: ClientService = Depends(lambda: di[ClientService])
 ) -> JSONResponse:
@@ -31,15 +42,23 @@ async def change_personal_data(
  return JSONResponse(content=result.dict(), status_code=status.HTTP_200_OK)
 
 
-@router.delete("/clients/{client_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["clients"])
+@router.delete(
+ "/clients/{client_id}",
+ status_code=status.HTTP_204_NO_CONTENT,
+ responses={400: {"model": APIErrorMessage}, 404: {"model": APIErrorMessage}, 500: {"model": APIErrorMessage}},
+ tags=["clients"],
+)
 async def archive(client_id: str, service: ClientService = Depends(lambda: di[ClientService])) -> Response:
  service.archive(ArchiveClientDTO(client_id=client_id))
  return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
-@router.post("/clients/exports", status_code=status.HTTP_204_NO_CONTENT, tags=["clients"])
-async def export(
- request: ExportClientsDTO, service: ClientService = Depends(lambda: di[ClientService])
-) -> Response:
+@router.post(
+ "/clients/exports",
+ status_code=status.HTTP_204_NO_CONTENT,
+ responses={400: {"model": APIErrorMessage}, 500: {"model": APIErrorMessage}},
+ tags=["clients"],
+)
+async def export(request: ExportClientsDTO, service: ClientService = Depends(lambda: di[ClientService])) -> Response:
  service.export(request)
  return Response(status_code=status.HTTP_204_NO_CONTENT)