■ FastAPI 클래스를 사용해 MongoDB 데이터베이스에서 단순 CRUD 애플리케이션을 만드는 방법을 보여준다.
▶ model/__init__.py
|
1 2 3 |
▶ model/event.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from typing import Optional, List from beanie import Document from pydantic import BaseModel class Event(Document): title : str image : str description : str tagList : List[str] location : str class Settings: name = "event" class EventUpdate(BaseModel): title : Optional[str] image : Optional[str] description : Optional[str] tagList : Optional[List[str]] location : Optional[str] |
▶ model/user.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
from typing import Optional, List from beanie import Document from pydantic import BaseModel, EmailStr from model.event import Event class User(Document): email : EmailStr password : str eventList : Optional[List[Event]] class Settings: name = "user" class UserSignIn(BaseModel): email : EmailStr password : str |
▶ database/__init__.py
|
1 2 3 |
▶ database/connection.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
from typing import Any, List, Optional from motor.motor_asyncio import AsyncIOMotorClient from beanie import init_beanie, PydanticObjectId from pydantic import BaseSettings, BaseModel from model.event import Event from model.user import User class MongoDBSettings(BaseSettings): DATABASE_URL : Optional[str] = None async def initializeDatabase(self): client = AsyncIOMotorClient(self.DATABASE_URL) await init_beanie(database = client.get_default_database(), document_models = [Event, User]) class Config: env_file = ".env" class Database: def __init__(self, modelMetaClass): self.modelMetaClass = modelMetaClass async def save(self, document) -> str: document = await document.create() return document.id async def get(self, id : PydanticObjectId) -> Any: document = await self.modelMetaClass.get(id) if document: return document return False async def getList(self) -> List[Any]: documentList = await self.modelMetaClass.find_all().to_list() return documentList async def update(self, id : PydanticObjectId, baseModel : BaseModel) -> Any: documentID = id baseModelDictionary = baseModel.dict() baseModelDictionary = {k: v for k, v in baseModelDictionary.items() if v is not None} updateQuery = {"$set": {field: value for field, value in baseModelDictionary.items()}} document = await self.get(documentID) if not document: return False await document.update(updateQuery) return document async def delete(self, id : PydanticObjectId) -> bool: document = await self.get(id) if not document: return False await document.delete() return True |
▶ route/__init__.py
|
1 2 3 |
▶ route/event.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
from typing import List from beanie import PydanticObjectId from fastapi import APIRouter, HTTPException, status from database.connection import Database from model.event import Event, EventUpdate eventDatabase = Database(Event) eventAPIRouter = APIRouter(tags = ["Event"]) @eventAPIRouter.get("/", response_model = List[Event]) async def getEventList() -> List[Event]: eventList = await eventDatabase.getList() return eventList @eventAPIRouter.get("/{id}", response_model = Event) async def getEvent(id : PydanticObjectId) -> Event: event = await eventDatabase.get(id) if not event: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist") return event @eventAPIRouter.post("/new") async def createEvent(event : Event) -> dict: await eventDatabase.save(event) return {"message" : "Event created successfully"} @eventAPIRouter.put("/{id}", response_model = Event) async def updateEvent(id : PydanticObjectId, body : EventUpdate) -> Event: eventUpdated = await eventDatabase.update(id, body) if not eventUpdated: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist") return eventUpdated @eventAPIRouter.delete("/{id}") async def deleteEvent(id : PydanticObjectId) -> dict: event = await eventDatabase.delete(id) if not event: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist") return {"message" : "Event deleted successfully."} |
▶ route/user.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
from fastapi import APIRouter, HTTPException, status from database.connection import Database from model.user import User, UserSignIn userAPIRouter = APIRouter(tags = ["User"]) userDatabase = Database(User) @userAPIRouter.post("/signup") async def signUp(user : User) -> dict: existingUser = await User.find_one(User.email == user.email) if existingUser: raise HTTPException(status_code = status.HTTP_409_CONFLICT, detail = "User with email provided exists already.") await userDatabase.save(user) return {"message" : "User created successfully"} @userAPIRouter.post("/signin") async def signIn(userSignIn : UserSignIn) -> dict: existingUser = await User.find_one(User.email == userSignIn.email) if not existingUser: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "User with email does not exist.") if existingUser.password == userSignIn.password: return {"message" : "User signed in successfully."} raise HTTPException(status_code = status.HTTP_401_UNAUTHORIZED, detail = "Invalid details passed.") |
▶ main.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
from fastapi import FastAPI from fastapi.responses import RedirectResponse import uvicorn from database.connection import MongoDBSettings from route.event import eventAPIRouter from route.user import userAPIRouter fastAPI = FastAPI() mongoDBSettings = MongoDBSettings() fastAPI.include_router(userAPIRouter , prefix = "/user" ) fastAPI.include_router(eventAPIRouter, prefix = "/event") @fastAPI.on_event("startup") async def startUp(): await mongoDBSettings.initializeDatabase() @fastAPI.get("/") async def home(): return RedirectResponse(url = "/event/") if __name__ == "__main__": uvicorn.run("main:fastAPI", host = "127.0.0.0", port = 8000, reload = True) |
▶ requirements.txt
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
anyio==3.5.0 asgiref==3.5.0 bcrypt==3.2.0 beanie==1.10.4 cffi==1.15.0 click==8.0.4 dnspython==2.2.0 email-validator==1.1.3 fastapi==0.74.1 h11==0.13.0 idna==3.3 Jinja2==3.0.3 MarkupSafe==2.1.0 motor==2.5.1 multidict==6.0.2 passlib==1.7.4 pycparser==2.21 pydantic==1.9.0 PyJWT==2.3.0 pymongo==3.12.3 python-dotenv==0.20.0 python-multipart==0.0.5 six==1.16.0 sniffio==1.2.0 SQLAlchemy==1.4.32 sqlalchemy2-stubs==0.0.2a20 sqlmodel==0.0.6 starlette==0.17.1 toml==0.10.2 typing_extensions==4.1.1 uvicorn==0.17.5 yarl==1.7.2 |
▶ 터미널 테스트 실행 명령
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
curl -X "POST" "http://127.0.0.1:8000/event/new" -H "accept: application/json" -H "Content-Type: application/json" \ -d '{ "title" : "FastAPI Book Launch", "image" : "fastapi-book.jpeg", "description" : "test", "tagList" : ["python", "fastapi", "book", "launch"], "location" : "Google Meet" }' curl -X "GET" "http://127.0.0.1:8000/event/" -H "accept: application/json" curl -X "GET" "http://127.0.0.1:8000/event/6651f3041a7d774524511b11" -H "accept: application/json" curl -X "PUT" "http://127.0.0.1:8000/event/6651f3041a7d774524511b11" -H "accept: application/json" -H "Content-Type: application/json" -d '{"location" : "Hybrid"}' curl -X "POST" "http://127.0.0.1:8000/user/signup" -H "accept: application/json" -H "Content-Type: application/json" \ -d '{ "email" : "user@example.com", "password" : "string", "eventList" : [] }' curl -X "POST" "http://127.0.0.1:8000/user/signin" -H "accept: application/json" -H "Content-Type: application/json" \ -d '{ "email" : "user@example.com", "password" : "string" }' |