■ FastAPI 클래스를 사용해 JWT 인증 애플리케이션을 만드는 방법을 보여준다. (MongoDB 연동)
▶ .env
|
1 2 3 4 |
DATABASE_URL=mongodb://localhost:27017/testdb SECRET_KEY=pass1234567 |
※ testdb : MongoDB 데이터베이스명
※ pass1234567 : JWT 비밀키
▶ application_setting.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
from typing import Optional from pydantic import BaseSettings from model.event import Event from model.user import User from database.db_helper import DBHelper class ApplicationSetting(BaseSettings): DATABASE_URL : Optional[str] = None SECRET_KEY : Optional[str] = None async def initializeDatabase(self): dbHelper = DBHelper() await dbHelper.initialize(self.DATABASE_URL, [Event, User]) class Config: env_file = ".env" |
※ .env : 애플리케이션 설정 파일명
▶ auth/hash_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 |
from passlib.context import CryptContext cryptContext = CryptContext(schemes = ["bcrypt"], deprecated = "auto") class HashHelper: def createHash(self, password : str): return cryptContext.hash(password) def verifyHash(self, plain_password : str, hashed_password : str): return cryptContext.verify(plain_password, hashed_password) |
▶ auth/jwt_helper.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 |
import time from datetime import datetime from jose import jwt, JWTError from fastapi import HTTPException, status from application_setting import ApplicationSetting applicationSetting = ApplicationSetting() def ceateAccessToken(user : str): payload = { "user" : user, "expires" : time.time() + 3600 } token = jwt.encode(payload, applicationSetting.SECRET_KEY, algorithm = "HS256") return token def verifyAccessToken(token : str): try: decodedTokenDictionary = jwt.decode(token, applicationSetting.SECRET_KEY, algorithms = ["HS256"]) expiresTimeStamp = decodedTokenDictionary.get("expires") if expiresTimeStamp is None: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "No access token supplied") if datetime.utcnow() > datetime.utcfromtimestamp(expiresTimeStamp): raise HTTPException(status_code = status.HTTP_403_FORBIDDEN, detail = "Token expired!") return decodedTokenDictionary except JWTError: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "Invalid token") |
▶ auth/auth_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
from auth.jwt_helper import verifyAccessToken from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer oauth2PasswordBearer = OAuth2PasswordBearer(tokenUrl = "/user/signin") async def authenticate(token : str = Depends(oauth2PasswordBearer)) -> str: if not token: raise HTTPException(status_code = status.HTTP_403_FORBIDDEN, detail = "Sign in for access") decodedTokenDictionary = verifyAccessToken(token) return decodedTokenDictionary["user"] |
▶ model/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 |
from pydantic import BaseModel, EmailStr from beanie import Document class User(Document): email : EmailStr password : str class Collection: name = "user" class Config: json_schema_extra = { "example" : { "email" : "fastapi@packt.com", "password" : "strong!!!" } } class TokenResponse(BaseModel): access_token : str token_type : str |
▶ 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 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
from typing import Optional, List from pydantic import BaseModel from beanie import Document class Event(Document): creator : Optional[str] title : str image : str description : str tagList : List[str] location : str class Collection: name = "event" class Config: json_schema_extra = { "example" : { "title" : "FastAPI BookLaunch", "image" : "https://linktomyimage.com/image.png", "description" : "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!", "tagList" : ["python", "fastapi", "book", "launch"], "location" : "Google Meet" } } class EventUpdate(BaseModel): title : Optional[str] image : Optional[str] description : Optional[str] tagList : Optional[List[str]] location : Optional[str] class Config: json_schema_extra = { "example" : { "title" : "FastAPI BookLaunch", "image" : "https://linktomyimage.com/image.png", "description" : "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!", "tagList" : ["python", "fastapi", "book", "launch"], "location" : "Google Meet" } } |
▶ database/db_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
from motor.motor_asyncio import AsyncIOMotorClient from beanie import init_beanie class DBHelper: def __init__(self): pass async def initialize(self, databaseURL, documentModelList): self.databaseURL = databaseURL self.documentModelList = documentModelList client = AsyncIOMotorClient(self.databaseURL) await init_beanie(database = client.get_default_database(), document_models = self.documentModelList) |
▶ database/document_helper.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 |
from pydantic import BaseModel from beanie import PydanticObjectId class DocumentHelper: def __init__(self, modelMetaClass): self.modelMetaClass = modelMetaClass async def create(self, document): await document.create() return async def get(self, id : PydanticObjectId): document = await self.modelMetaClass.get(id) if document: return document return False async def getList(self): documentList = await self.modelMetaClass.find_all().to_list() return documentList async def update(self, id : PydanticObjectId, updateModel : BaseModel): documentID = id updateModelDictionary = updateModel.dict() updateModelDictionary = {k : v for k, v in updateModelDictionary.items() if v is not None} updateQuery = {"$set" : {field : value for field, value in updateModelDictionary.items()}} document = await self.get(documentID) if not document: return False await document.update(updateQuery) return document async def delete(self, id : PydanticObjectId): document = await self.get(id) if not document: return False await document.delete() return True |
▶ 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 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm from model.user import User, TokenResponse from auth.hash_helper import HashHelper from auth.jwt_helper import ceateAccessToken from database.document_helper import DocumentHelper userAPIRouter = APIRouter(tags = ["User"]) userDocumentHelper = DocumentHelper(User) hashHelper = HashHelper() @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.") hashedPassword = hashHelper.createHash(user.password) user.password = hashedPassword await userDocumentHelper.create(user) return {"message" : "User created successfully"} @userAPIRouter.post("/signin", response_model = TokenResponse) async def signIn(oauth2PasswordRequestForm : OAuth2PasswordRequestForm = Depends()) -> dict: existingUser = await User.find_one(User.email == oauth2PasswordRequestForm.username) if not existingUser: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "User with email does not exist.") if hashHelper.verifyHash(oauth2PasswordRequestForm.password, existingUser.password): accessToken = ceateAccessToken(existingUser.email) return { "access_token" : accessToken, "token_type" : "Bearer" } raise HTTPException(status_code = status.HTTP_401_UNAUTHORIZED, detail = "Invalid details passed.") |
▶ 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 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
from typing import List from beanie import PydanticObjectId from fastapi import APIRouter, Depends, HTTPException, status from model.event import Event, EventUpdate from auth.auth_helper import authenticate from database.document_helper import DocumentHelper eventAPIRouter = APIRouter(tags = ["Event"]) eventDocumentHelper = DocumentHelper(Event) @eventAPIRouter.get("/", response_model = List[Event]) async def getEventList() -> List[Event]: eventList = await eventDocumentHelper.getList() return eventList @eventAPIRouter.get("/{id}", response_model = Event) async def getEvent(id : PydanticObjectId) -> Event: event = await eventDocumentHelper.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, user : str = Depends(authenticate)) -> dict: event.creator = user await eventDocumentHelper.create(event) return {"message" : "Event created successfully"} @eventAPIRouter.put("/{id}", response_model = Event) async def updateEvent(id : PydanticObjectId, eventUpdate : EventUpdate, user : str = Depends(authenticate)) -> Event: event = await eventDocumentHelper.get(id) if event.creator != user: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "Operation not allowed") updatedEvent = await eventDocumentHelper.update(id, eventUpdate) if not updatedEvent: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist") return updatedEvent @eventAPIRouter.delete("/{id}") async def deleteEvent(id : PydanticObjectId, user : str = Depends(authenticate)) -> dict: event = await eventDocumentHelper.get(id) if not event: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event not found") if event.creator != user: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "Operation not allowed") event = await eventDocumentHelper.delete(id) return {"message" : "Event deleted successfully."} |
▶ 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 29 30 31 32 33 34 35 36 37 |
import uvicorn from fastapi import FastAPI from fastapi.responses import RedirectResponse from fastapi.middleware.cors import CORSMiddleware from application_setting import ApplicationSetting from route.event import eventAPIRouter from route.user import userAPIRouter applicationSetting = ApplicationSetting() fastAPI = FastAPI() fastAPI.add_middleware( CORSMiddleware, allow_origins = ["*"], allow_credentials = True, allow_methods = ["*"], allow_headers = ["*"], ) fastAPI.include_router(userAPIRouter , prefix = "/user" ) fastAPI.include_router(eventAPIRouter, prefix = "/event") @fastAPI.on_event("startup") async def startUp(): await applicationSetting.initializeDatabase() @fastAPI.get("/") async def home(): return RedirectResponse(url="/event/") if __name__ == "__main__": uvicorn.run("main:fastAPI", host = "0.0.0.0", port = 8000, reload = True) |
▶ requirements.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 |
anyio==3.5.0 asgiref==3.5.0 bcrypt==3.2.0 beanie==1.10.4 cffi==1.15.0 click==8.0.4 cryptography==36.0.2 dnspython==2.2.0 ecdsa==0.17.0 email-validator==1.1.3 fastapi==0.74.1 h11==0.13.0 idna==3.3 MarkupSafe==2.1.0 motor==2.5.1 multidict==6.0.2 passlib==1.7.4 pyasn1==0.4.8 pycparser==2.21 pydantic==1.9.0 pymongo==3.12.3 python-dotenv==0.20.0 python-jose==3.3.0 python-multipart==0.0.5 rsa==4.8 six==1.16.0 sniffio==1.2.0 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 30 31 32 33 34 35 36 37 38 39 40 41 |
# 신규 사용자 추가 curl -X "POST" "http://127.0.0.1:8000/user/signup" -H "accept: application/json" -H "Content-Type: application/json" \ -d '{ "email" : "reader@packt.com", "password" : "test1234" }' # 사용자 로그인 curl -X "POST" "http://127.0.0.1:8000/user/signin" -H "accept: application/json" -H "Content-Type: application/x-www-form-urlencoded" \ -d 'grant_type=&username=reader%40packt.com&password=test1234&scope=&client_id&client_secret=' {"access_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE3MTY4MzMyMDAuNTMxOTAxMX0.iLw7W8LMn5q2DUwPbOS0DTs6eDR969sIfyAFx1wa3yM","token_type":"Bearer"} # 신규 이벤트 추가 curl -X "POST" "http://127.0.0.1:8000/event/new" \ -H "accept: application/json" \ -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE3MTY4MzMyMDAuNTMxOTAxMX0.iLw7W8LMn5q2DUwPbOS0DTs6eDR969sIfyAFx1wa3yM" \ -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 "PUT" "http://127.0.0.1:8000/event/6654bb467ac4278dc60627f0" \ -H "accept: application/json" \ -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE3MTY4MzMyMDAuNTMxOTAxMX0.iLw7W8LMn5q2DUwPbOS0DTs6eDR969sIfyAFx1wa3yM" \ -H "Content-Type: application/json" \ -d '{ "title" : "FastAPI Book Launch 2" }' # 이벤트 삭제 curl -X "DELETE" "http://127.0.0.1:8000/event/6654bb467ac4278dc60627f0" \ -H "accept: application/json" \ -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE3MTY4MzMyMDAuNTMxOTAxMX0.iLw7W8LMn5q2DUwPbOS0DTs6eDR969sIfyAFx1wa3yM" |
※ eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE3MTY4MzMyMDAuNTMxOTAxMX0.iLw7W8LMn5q2DUwPbOS0DTs6eDR969sIfyAFx1wa3yM : JWT 토큰
※ 6654bb467ac4278dc60627f0 : MongoDB 문서 ID