사용자 라우트는 로그인, 로그아웃, 등록으로 구성된다. 인증을 완료한 사용자는 이벤트를 생성, 변경, 삭제할 수 있으며, 인증을 거치지 않은 사용자는 생성된 이벤트를 확인하는 것만 가능하다.
routes
폴더의 users.py
에 사용자 라우트를 정의해보자
먼저 등록 라우트이다. 등록 라우트는 애플리케이션에 내장된 데이터베이스를 사용하며, 사용자를 등록하기 전 데이터베이스에 비슷한 이메일이 존재하는지 확인한다.
from fastapi import APIRouter, HTTPException, status
from models.users import User, UserSignIn
user_router = APIRouter(
tags = ["User"]
)
users = {}
@user_router.post("/signup")
async def sign_new_user(data: User) -> dict:
if data.email in users:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="User wuth supplied username exists"
)
users[data.email] = data
return {
"message":"User successfully registered"
}
다음은 로그인(/signin
) 라우트이다.
이 라우트는 로그인하려는 사용자가 데이터베이스에 존재하는지를 먼저 확인하고, 없으면 예외를 발생시킨다.
사용자가 존재하면 패스워드가 일치하는지 확인해서 성공 또는 실패 메시지를 반환한다.
@user_router.post("/signin")
async def sign_user_in(user: UserSignIn) -> dict:
if user.email not in users:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User does not exist"
)
if users[user.email].password != user.passward:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Wrong credentials passed"
)
return {
"message":"User signed in successfully"
}
이제 사용자 처리용 라우트를 정의했으니, main.py
에 라우트를 등록하고 애플리케이션 실행을 해보자.
from fastapi import FastAPI
from routes.users import user_router
import uvicorn
# FastAPI() 인스턴스를 만들고 앞서 정의한 라우트 등록
app = FastAPI()
# 라우트 등록
app.include_router(user_router, prefix="/user")
if __name__=="__main__":
# uvicorn.run() 메서드를 사용해 8000번 포트에서 애플리케이션을 실행하도록 설정
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
이제 실행해보자.
python3 main.py
애플리케이션이 실행됐으니 사용자 라우트를 테스트해보자
curl -X 'POST' \
'http://127.0.0.1:8000/user/signup' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"email":"fastapi@packt.com",
"password":"Stro0ng!",
"username":"FastPackt"
}'
{"message":"User successfully registered"}%
curl -X 'POST' \
'http://127.0.0.1:8000/user/signin' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"email":"fastapi@packt.com",
"password":"Stro0ng!"
}'
{"message":"User signed in successfully"}%
curl -X 'POST' \
'http://127.0.0.1:8000/user/signin' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"email":"fastapi@packt.com",
"password":"password!"
}'
{"detail":"Wrong credentials passed"}%
사용자 라우트가 준비됐으니 이제 이벤트 라우트를 구현해야한다.
routes 폴더의 events.py를 열어서 다음과 같이 의존 라이브러리를 임포트하고 이벤트 라우트를 정의한다.
from fastapi import APIRouter, Body, HTTPException, status
from models.events import Event
from typing import List
event_router = APIRouter(
tags = ["Events"]
)
events = []
모든 이벤트를 추출하거나 특정 ID의 이벤트만 추출하는 라우트를 정의한다.
@event_router.get("/", response_model=Event)
async def retrieve_all_events() -> List[Event]:
if len(events)==0:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="No events found"
)
return events
@event_router.get("/{id}", response_model=Event)
async def retrieve_event(id: int) -> Event:
for event in events:
if event.id == id:
return event
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Event with supplied ID does not exist"
)
이제 이벤트 생성 및 삭제 라우트를 정의하자.
첫 번째 라우트는 이벤트 생성,
두 번째는 데이터베이스에 있는 단일 이벤트 삭제,
세 번째는 전체 이벤트 삭제이다.
@event_router.post("/new")
async def create_event(body: Event = Body(...)) -> dict:
events.append(body)
return {
"message":"Event created successfully"
}
@event_router.delete("/{id}")
async def delete_event(id: int) -> dict:
for event in events:
if event.id == id:
events.remove(event)
return {
"message" : "Event deleted successfully"
}
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Event with supplied ID does not exist"
)
@event_router.delete("/")
async def delete_all_events() -> dict:
events.clear()
return {
"message":"Events deleted successfully."
}
이제 main.py
의 라우트 설정을 변경해서 이벤트 라우트를 추가하자.
from fastapi import FastAPI
from routes.users import user_router
from routes.events import event_router # event 라우트
import uvicorn
# FastAPI() 인스턴스를 만들고 앞서 정의한 라우트 등록
app = FastAPI()
# 라우트 등록
app.include_router(user_router, prefix="/user")
app.include_router(event_router, prefix="/event") # event 라우트 등록
if __name__=="__main__":
# uvicorn.run() 메서드를 사용해 8000번 포트에서 애플리케이션을 실행하도록 설정
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
이제 이벤트 라우트도 테스트해보자.
curl -X 'GET' \
'http://127.0.0.1:8000/event/' \
-H 'accept: application/json'
이벤트가 없기에, 500 에러가 발생하고 아래와 같은 출력을 확인할 수 있다.
{"detail":"No events found"}%
curl -X 'POST' \
'http://127.0.0.1:8000/event/new' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"id":1,
"title":"FastAPI Book Launch",
"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!",
"tags":["python", "fastapi", "book", "launch"],
"location": "Google Meet"
}'
{"message":"Event created successfully"}%
curl -X 'GET' \
'http://127.0.0.1:8000/event/1' \
-H 'accept: application/json'
앞서 추가한 이벤트가 그대로 표시된다.
{
"id":1,
"title":"FastAPI Book Launch",
"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!",
"tags":["python","fastapi","book","launch"],
"location":"Google Meet"
}%
curl -X 'DELETE' \
'http://127.0.0.1:8000/event/1' \
-H 'accept: application/json'
{"message":"Event deleted successfully"}%
동일한 삭제 명령을 다시 실행해보자.
{"detail":"Event with supplied ID does not exist"}%
해당 ID의 이벤트가 존재하지 않기 때문문에 detail
이 출력된다.
여기까지 이벤트 플래너 애플리케이션의 라우트와 모델을 성공적으로 구현했으며, 테스트를 통해 제대로 실행되는지도 확인했다.