FastAPI Todo 라우팅(main.py 수정) 확장성 확인 - 65

Authentication Routing /  Todo Routing /  Prefix Routing /  External Routing /  Dependencies Routing

 

1. 라우팅 역할

• 애플리케이션의 구조를 위한 유연한 도구
• 확장 가능 한 아키텍처를 지원
• 파일 구조화를 도움

 

2. 구성할 프로젝트 구조

TodoApp
main.py database.py models.py	TodoApp/routers		TodoApp/company
	auth.py	todos.py	companyapis.py	dependencies.py

---

인증 라우팅

routers 디렉터리 내에 todos.py 생성

현재 디렉토리 구조

 

todos.py 파일 수정(main.py를 복사해서 붙여 넣기)

import sys
sys.path.append("..")
from typing import Optional
#FastAPI DELETE, APITRouter 생성
#from fastapi import FastAPI, Depends, HTTPException
from fastapi import Depends, HTTPException, APIRouter
import models
from database import engine, SessionLocal
from sqlalchemy.orm import Session
from pydantic import BaseModel, Field

#TodoApp 삭제, routers 삭제
#from TodoApp.routers.auth import get_current_user, get_user_exception
#from routers.auth import get_current_user, get_user_exception
from .auth import get_current_user, get_user_exception


#app >> router
#app = FastAPI()
router = APIRouter()

models.Base.metadata.create_all(bind=engine)


def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

class Todo(BaseModel):
    title: str
    description: Optional[str]
    priority: int = Field(gt=0, lt=6, description="The priority must be between 1-5")
    complete: bool

#app >> router
#@app.get("/")
@router.get("/")
async def read_all(db: Session = Depends(get_db)):
    return db.query(models.Todos).all()

#app >> router
#@app.get("/todos/user")
@router.get("/todos/user")
async def read_all_by_user(user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
    if user is None:
        raise get_user_exception()
    return db.query(models.Todos).filter(models.Todos.owner_id == user.get("id")).all()

#app >> router
#@app.post("/")
@router.post("/")
async def create_todo(todo: Todo, user: dict = Depends(get_current_user), db: Session = Depends(get_db)):

    if user is None:
        raise get_user_exception()
    todo_model = models.Todos()
    todo_model.title = todo.title
    todo_model.description = todo.description
    todo_model.priority = todo.priority
    todo_model.complete = todo.complete
    todo_model.owner_id = user.get("id")

    db.add(todo_model)
    # db.flush()
    db.commit()

    return successful_response(200)

#app >> router
#app.put("/{todo_id}")
@router.put("/{todo_id}")
async def update_todo(todo_id: int,
                      todo: Todo,
                      user: dict = Depends(get_current_user),
                      db: Session = Depends(get_db)):

    if user is None:
        raise get_user_exception()

    todo_model = db.query(models.Todos)\
        .filter(models.Todos.id == todo_id)\
        .filter(models.Todos.owner_id == user.get("id")).first()

    if todo_model is None:
        raise http_exception()

    todo_model.title = todo.title
    todo_model.description = todo.description
    todo_model.priority = todo.priority
    todo_model.complete = todo.complete

    db.add(todo_model)
    # db.flush()
    db.commit()
    return successful_response(200)

#app >> router
#@app.get("/todo/{todo_id}")
@router.get("/todo/{todo_id}")
async def read_todo(todo_id: int, user:dict = Depends(get_current_user), db: Session = Depends(get_db)):
    if user is None:
        raise get_user_exception()
    todo_model = db.query(models.Todos).filter(models.Todos.id == todo_id).filter(models.Todos.owner_id == user.get("id")).first()
    if todo_model is not None:
        return todo_model
    raise http_exception()

#app >> router
#@app.delete("/{todo_id}")
@router.delete("/{todo_id}")
async def delete_todo(todo_id: int, user: dict = Depends(get_current_user), db: Session = Depends(get_db)):

    if user is None:
        raise get_user_exception()

    todo_model = db.query(models.Todos)\
        .filter(models.Todos.id == todo_id)\
        .filter(models.Todos.owner_id == user.get("id")).first()

    if todo_model is None:
        raise http_exception()

    db.query(models.Todos).filter(models.Todos.id == todo_id).delete()

    # db.flush()
    db.commit()

    return successful_response(200)

def successful_response(status_code:int):
    return {
        'status': status_code,
        'transaction': 'Successful'
    }

def http_exception():
    return HTTPException(status_code=404, detail="Todo not found")

 

main.py 수정

from fastapi import FastAPI
import models
from database import engine
from routers import auth, todos

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

app.include_router(auth.router)
app.include_router(todos.router)

app.include_router()은 main.py와 router를 연결해준다.

 

 

3. Swagger 확인

uvicorn main:app --reload

routing을 통해서 auth.py와 todos.py의 함수들을 잘 가져오고 있는 걸 볼 수 있다.