template for CRUD REST with fastapi

fastapi-template.py

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="Backend",
        description="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)