Last active
January 18, 2024 06:38
-
-
Save kubosuke/7796d8fd78107699fa34cb2a673acf8f to your computer and use it in GitHub Desktop.
Revisions
-
kubosuke revised this gist
Jan 18, 2024 . 1 changed file with 2 additions and 2 deletions.There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -144,8 +144,8 @@ def init_routers(app: FastAPI) -> None: def create_app() -> FastAPI: app = FastAPI( title="Backend", description="Backend API", version="1.0.0", docs_url="/docs", redoc_url="/redoc", -
Jan 18, 2024 .There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -0,0 +1,163 @@ from typing import List from pydantic.dataclasses import dataclass from fastapi import APIRouter, FastAPI from fastapi.middleware.cors import CORSMiddleware import uvicorn from pydantic import BaseModel # Domain Layer # Domain Object (Entities & Value Objects) @dataclass class User: user_id: str name: str = "John Doe" # Repository # I/O from external services/datasource class UserRepository: def list(self) -> List[User]: return [] def get(self, user_id: str) -> User: return User(user_id=user_id) def update(self, user_id: str, name: str) -> None: raise NotImplementedError def delete(self, user_id: str) -> None: raise NotImplementedError # Controller # I/O of domain objects' behaviours class UserController: user_repository: UserRepository def __init__(self, user_repository: UserRepository) -> None: self.user_repository = user_repository def list(self) -> List[User]: return self.user_repository.list() def get(self, user_id: str) -> List[User]: return self.user_repository.get(user_id=user_id) def update(self, user_id: str, name=str) -> List[User]: return self.user_repository.update(user_id=user_id, name=name) def delete(self, user_id: str) -> List[User]: return self.user_repository.delete(user_id=user_id) # View # Interface adopters class UserResponse(BaseModel): content: User class UsersResponse(BaseModel): content: List[User] class SuccessResponse(BaseModel): message: str = "Success" router = APIRouter() # able to do DI this part user_repository = UserRepository() user_controller = UserController(user_repository) @router.get("/healthz", summary="healthcheck") async def healthcheck(): return {"status": "OK"} @router.get( "/api/v1/users", response_model=UsersResponse, summary="List Users", ) async def list_user(): users = user_controller.list() return UsersResponse(content=users) @router.get( "/api/v1/users/{user_id}", response_model=UserResponse, summary="Get Users", ) async def get_user(user_id: str): user = user_controller.get(user_id=user_id) return UserResponse(content=user) @router.put( "/api/v1/users/{user_id}", response_model=SuccessResponse, summary="Update User", ) async def update_user(user_id: str, name: str): user_controller.update(user_id=user_id, name=name) return SuccessResponse() @router.delete( "/api/v1/users/{id}", response_model=SuccessResponse, summary="Delete User", ) async def delete_user(user_id: str): user_controller.delete(user_id=user_id) return SuccessResponse() # Main def init_cors(app: FastAPI) -> None: app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def init_routers(app: FastAPI) -> None: app.include_router(router) def create_app() -> FastAPI: app = FastAPI( title="GOD Backend", description="GOD Backend API", version="1.0.0", docs_url="/docs", redoc_url="/redoc", ) init_routers(app=app) init_cors(app=app) return app app = create_app() # uvicorn main:app --reload if __name__ == "__main__": uvicorn.run(app="main:app", host="0.0.0.0", port=8080, reload=True)