2024-08-21 WED
0. 준비
ch15_memo_app_mvc_final
ㄴstatic
ㄴtemplates
ㄴhome.html
ㄴmemos.html
ㄴmain.py
ㄴcontrollers.py
ㄴdatabase.py
ㄴdependencies.py
ㄴschemas.py
create database memo_db;
파일은 기능에 따라 나눔. 아래는 각 파일별 기능 간략 설명
1. main.py
from fastapi import FastAPI, Request
from starlette.middleware.sessions import SessionMiddleware
from fastapi.templating import Jinja2Templates
from database import Base, engine
from controllers import router
app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-secret-key")
Base.metadata.create_all(bind=engine)
app.include_router(router)
templates = Jinja2Templates(directory="templates")
@app.get('/')
async def read_root(request: Request):
return templates.TemplateResponse('home.html', {"request": request})

2. controllers.py
from fastapi import APIRouter, Request, Depends, HTTPException
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from models import User, Memo
from dependencies import get_db, get_password_hash, verify_password
from schemas import UserCreate, UserLogin, MemoCreate, MemoUpdate
router = APIRouter()
templates = Jinja2Templates(directory="templates")
# 회원 가입
@router.post("/signup")
async def signup(signup_data: UserCreate, db: Session = Depends(get_db)):
# 먼저 username이 이미 존재하는지 확인
existing_user = db.query(User).filter(User.username == signup_data.username).first()
if existing_user:
raise HTTPException(status_code=400, detail="이미 동일 사용자 이름이 가입되어 있습니다.")
hashed_password = get_password_hash(signup_data.password)
new_user = User(username=signup_data.username, email=signup_data.email, hashed_password=hashed_password)
db.add(new_user)
try:
db.commit()
except Exception as e:
print (e)
raise HTTPException(status_code=500, detail="회원가입이 실패했습니다. 기입한 내용을 확인해보세요.")
db.refresh(new_user)
return {"message": "회원가입이 성공했습니다."}
db.refresh(new_user) : 새로 생성된 사용자 정보를 데이터베이스에서 다시 읽어와 new_user 객체에 넣어줌.# 로그인
@router.post("/login")
async def login(request: Request, signin_data: UserLogin, db: Session = Depends(get_db)):
user = db.query(User).filter(User.username == signin_data.username).first()
if user and verify_password(signin_data.password, user.hashed_password):
request.session["username"] = user.username
return {"message":"로그인이 성공했습니다."}
else:
raise HTTPException(status_code=401, detail="로그인이 실패했습니다.")
# 로그아웃
@router.post("/logout")
async def logout(request: Request):
request.session.pop("username", None)
return {"message": "로그아웃이 성공했습니다."}
@router.post("/memos/")
async def create_memo(request: Request, memo: MemoCreate, db: Session = Depends(get_db)):
username = request.session.get("username")
if username is None:
raise HTTPException(status_code=401, detail="Not authorized")
user = db.query(User).filter(User.username == username).first()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
new_memo = Memo(user_id=user.id, title=memo.title, content=memo.content)
db.add(new_memo)
db.commit()
db.refresh(new_memo)
return new_memo
@router.get("/memos/")
async def list_memos(request: Request, db: Session = Depends(get_db)):
username = request.session.get("username")
if username is None:
raise HTTPException(status_code=401, detail="Not authorized")
user = db.query(User).filter(User.username == username).first()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
memos = db.query(Memo).filter(Memo.user_id == user.id).all()
return templates.TemplateResponse("memos.html", {"request": request, "memos": memos, "username": username})
@router.put("/memos/{memo_id}")
async def update_memo(request: Request, memo_id: int, memo: MemoUpdate, db: Session = Depends(get_db)):
username = request.session.get("username")
if username is None:
raise HTTPException(status_code=401, detail="Not authorized")
user = db.query(User).filter(User.username == username).first()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
db_memo = db.query(Memo).filter(Memo.user_id == user.id, Memo.id == memo_id).first()
if db_memo is None:
return ({"error": "Memo not found"})
if memo.title is not None:
db_memo.title = memo.title
if memo.content is not None:
db_memo.content = memo.content
db.commit()
db.refresh(db_memo)
return db_memo
@router.delete("/memos/{memo_id}")
async def delete_memo(request: Request, memo_id: int, db: Session = Depends(get_db)):
username = request.session.get("username")
if username is None:
raise HTTPException(status_code=401, detail="Not authorized")
user = db.query(User).filter(User.username == username).first()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
db_memo = db.query(Memo).filter(Memo.user_id == user.id, Memo.id == memo_id).first()
if db_memo is None:
return ({"error": "Memo not found"})
db.delete(db_memo)
db.commit()
return ({"message": "Memo deleted"})
3. database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
DATABASE_URL = "mysql+pymysql://root:1234@localhost/memo_db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Base = declarative_base() : sqlalchemy에서 사용하는 데이터베이스 모델 정의하기 위한 기본 클래스주의
Base vs BaseModel
| Base | BaseModel | |
|---|---|---|
| 사용목적 | SQLAlchemy | Pydantic |
| 기능 | db 모델 구조 정의 | 데이터 유효성검사, Json |
| 사용방법 | ORM 사용 | Pyndatic 클래스 상속받아 모델 정의 |
4. dependencies.py
from passlib.context import CryptContext
from database import SessionLocal
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password):
return pwd_context.hash(password)
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
pip install passlib : 설치해야함.
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") : 비밀번호 암호화 해서 받음.
get_password_hash : 비밀번호 해시화 하는 함수
verify_password : 평문 비밀번호와 암호화된 비밀번호 서로 맞는지 비교하는 함수. verify() 함수에 들어온 두 비밀번호가 서로 일치하면 true
get_db : 세션을 생성하고 관리하는 함수.
SessionLocal()로 db 세션 생성.
yield로 함수 제너레이터로 만듦. 왜?? 비동기 처리 하려고
db.close() : db세션은 작업 완료 후 반드시 종료 시켜야 함.
5. models.py
from database import Base
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(100), unique=True, index=True)
email = Column(String(200))
hashed_password = Column(String(512))
class Memo(Base):
__tablename__ = 'memo'
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey('users.id'))
title = Column(String(100))
content = Column(String(1000))
User 클래스로 "users"라는 테이블 생성
User, Memo 클래스는 아까 "models.py" 에서 정의 했던 Base 상속 받아서 사용
Memo 클래스로 "memo"라는 테이블 생성

6. schemas.py
from pydantic import BaseModel
from typing import Optional
# 회원가입시 데이터 검증
class UserCreate(BaseModel):
username: str
email: str
password: str # 해시전 패스워드를 받습니다.
# 회원로그인시 데이터 검증
class UserLogin(BaseModel):
username: str
password: str # 해시전 패스워드를 받습니다.
class MemoCreate(BaseModel):
title: str
content: str
class MemoUpdate(BaseModel):
title: Optional[str] = None
content: Optional[str] = None
Optional[str] = None : 들어오면 해당 값을 입력하고 값이 들어오지 않으면 None을 출력한다는 뜻. 즉, 값이 없어도 상관 없고, 둘 중 하나만 들어와도 상관이 없다.7. home.html
8. memos.html