FASTAPI 로 구현해보는 Vertical Slice Architecture - 4 (user management with prisma)

user 가입 처리를 위해서 features 폴더에 user 폴더를 추가하자.

그리고 endpoints.py 를 추가 하자

 

prisma 를 각 handler (add, update, delete) 에서 사용하기 위해

infrastructure 아래 prisma.py 를 추가 하자

prisma.py

from prisma import Prisma

prisma = Prisma()

 

이제 add.py, update.py, delete.py 를 추가 하자

그리고 prisma 를 import 하여 사용하자

add.py

from typing import Annotated, Optional
from fastapi import Body, Depends
from mediatr import Mediator
from src.infrastructure.prisma import prisma
from pydantic import BaseModel


class Command(BaseModel):
    email: str
    name: Optional[str] = None
    pass


class Response(BaseModel):
    result: bool
    pass


@Mediator.handler
async def handle(request: Command):
    await prisma.user.create(data={
        "name": f"{request.name}",
        "email": f"{request.email}",
    })
    return Response(result=True)

update.py

from typing import Optional
from mediatr import Mediator
from src.infrastructure.prisma import prisma
from pydantic import BaseModel


class Command(BaseModel):
    email: str
    name: Optional[str] = None
    pass


class Response(BaseModel):
    result: bool
    pass


@Mediator.handler
async def handle(request: Command):
    await prisma.user.update(
        where={
            "email": f"{request.email}"
        },
        data={
            "name": f"{request.name}",
        }
    )
    return Response(result=True)

delete.py

from mediatr import Mediator
from pydantic import BaseModel
from src.infrastructure.prisma import prisma


class Command(BaseModel):
    email: str
    pass


class Response(BaseModel):
    result: bool
    pass


@Mediator.handler
async def handle(request: Command):
    await prisma.user.delete(
        where={
            "email": f"{request.email}"
        }
    )
    return Response(result=True)

endpoints.py

from fastapi import APIRouter
from mediatr import Mediator

from src.shared.api_logger.apiroute import LoggingAPIRoute

from .add import Command as add_command
from .update import Command as update_command
from .delete import Command as delete_command

import logging

logger = logging.getLogger()

router = APIRouter(
    prefix="/user",
    tags=["user"],
    responses={404: {"description": "Not found"}},
    route_class=LoggingAPIRoute)


@router.post("/")
async def add(request: add_command):
    mediator = Mediator()
    return await mediator.send_async(request)


@router.put("/")
async def update(request: update_command):
    mediator = Mediator()
    return await mediator.send_async(request)


@router.delete("/")
async def delete(request: delete_command):
    mediator = Mediator()
    return await mediator.send_async(request)

main.py 에서  unhandled exception 을 catch 해서 처리 하자

main.py

import logging
from fastapi import Depends, FastAPI
from fastapi.responses import JSONResponse

from .features.hello import endpoints as hello_endpoints
from .features.user import endpoints as user_endpoints
from src.infrastructure.prisma import prisma

app = FastAPI(title="FastAPI, vertical slice architecture")

logging.config.fileConfig("log.ini")
logger = logging.getLogger()

app.include_router(hello_endpoints.router)
app.include_router(user_endpoints.router)


@app.on_event("startup")
async def startup_event():
    await prisma.connect()
    print("Startup event")


@app.on_event("shutdown")
async def shutdown_event():
    await prisma.disconnect()
    print("Shutdown event")

# global error handler


@app.exception_handler(Exception)
async def validation_exception_handler(request, err):
    base_error_message = f"Failed to execute: {request.method}: {request.url}"
    # Change here to LOGGER
    logger.error(msg={"message": f"{base_error_message}. Detail: {err}"})
    return JSONResponse(status_code=400, content={"message": f"{base_error_message}. Detail: {err}"})

위와 같은 형태로 add , update, delete 를 처리 하는걸 확인 가능 하다.

 

prisma 는 orm 으로 생각보다 편리한 도구이다. 

python 에서 공식 지원이 아니라 아쉬운 면이 있지만

앞으로 어떤 형태가 될지 확인해 보자

 

 

관련영상

https://youtu.be/L3Knp5nnX14