my_memo_app
- 인증(로그인) → 해싱 기법으로 비밀번호 저장
- 사용자(user)에 따라 메모 테이블 저장
- CRUD 적용
🐇 코드 Review (API)
1. 라이브러리 임포트
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel
from typing import Optional
from passlib.context import CryptContext
from starlette.middleware.sessions import SessionMiddleware
코드 설명
- Request: HTTP 요청을 나타내는 객체로, 요청의 메타데이터와 본문에 접근 가능
- Depends: 의존성 주입을 위한 기능으로, 특정 함수나 클래스의 인스턴스를 자동으로 주입 가능
- HTTPException: HTTP 오류를 발생시키기 위한 예외 클래스
- Jinja2Templates: Jinja2 템플릿 엔진을 사용하여 HTML 파일을 렌더링하는 데 사용
- Session: SQLAlchemy의 세션 객체로, 데이터베이스와의 상호작용을 관리
- create_engine: SQLAlchemy의 데이터베이스 엔진을 생성하는 함수
- Column, Integer, String, ForeignKey: SQLAlchemy에서 데이터베이스 테이블의 열을 정의하는 데 사용되는 클래스
- declarative_base: SQLAlchemy ORM에서 사용할 기본 클래스를 생성
- BaseModel: Pydantic의 기본 모델 클래스로, 데이터 검증 및 직렬화를 지원
- Optional: 타입 힌팅에서 선택적 필드를 정의할 때 사용
- CryptContext: 비밀번호 해싱 및 검증을 위한 설정을 관리
- SessionMiddleware: 세션 관리를 위한 미들웨어로, 사용자 세션을 유지하는 데 사용
2. 비밀번호 해싱 설정
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)
- CryptContext: 비밀번호 해싱을 위한 다양한 알고리즘을 설정 여기서는 bcrypt를 사용함
- get_password_hash: 사용자가 입력한 평문 비밀번호를 해시하여 안전하게 저장할 수 있는 형태로 변환 해시된 비밀번호는 데이터베이스에 저장
- **verify_password**: 사용자가 로그인할 때 입력한 평문 비밀번호와 데이터베이스에 저장된 해시 비밀번호를 비교하여 일치 여부를 확인
3. FastAPI 애플리케이션 생성
app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-secret-key")
templates = Jinja2Templates(directory="templates")
- SessionMiddleware: 세션 관리를 위한 미들웨어를 추가 secret_key는 세션 데이터의 암호화를 위한 비밀 키입니다.
- Jinja2Templates: HTML 템플릿을 렌더링하기 위한 설정 directory는 템플릿 파일이 위치한 디렉토리
4. 데이터베이스 설정
DATABASE_URL = "mysql+pymysql://~~~~~~~@localhost/my_memo_app"
engine = create_engine(DATABASE_URL)
Base = declarative_base()- DATABASE_URL: 데이터베이스에 연결하기 위한 URL 여기서는 MySQL 데이터베이스를 사용하고 있음. yun은 사용자 이름, 0000은 비밀번호, localhost는 데이터베이스 서버의 주소, my_memo_app은 데이터베이스 이름
- create_engine: 데이터베이스와의 연결을 관리하는 엔진을 생성 이 엔진을 통해 SQLAlchemy가 데이터베이스와 상호작용함 → sqlalchemy는 파이썬에서 sql을 사용하게 할 수 있는 ORM 라이브러리임
- declarative_base: SQLAlchemy ORM에서 사용할 기본 클래스를 생성 이 클래스를 상속받아 데이터베이스 모델을 정의
5. 데이터베이스 모델 정의
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))
- User: 사용자 정보를 저장하기 위한 데이터베이스 모델 → Base를 상속받아 SQLAlchemy ORM의 기능을 사용할 수 있음
- tablename: 데이터베이스에서 사용할 테이블 이름을 정의
- Column: 테이블의 각 열을 정의
6. 데이터 검증 모델 정의
class UserCreate(BaseModel):
username: str
email: str
password: str
class UserLogin(BaseModel):
username: str
password: str
- UserCreate: 회원가입 시 입력받는 데이터 모델 → Pydantic의 BaseModel을 상속받아 데이터 검증을 수행 → 사용자가 입력한 데이터가 올바른 형식인지 확인
- UserLogin: 로그인 시 입력받는 데이터 모델 마찬가지로 Pydantic의 BaseModel을 상속받음
7. 데이터베이스 세션 관리
def get_db():
db = Session(bind=engine)
try:
yield db
finally:
db.close()
- get_db: 데이터베이스 세션을 생성하고, 요청이 끝난 후 세션을 닫음
- 이 함수는 FastAPI의 의존성 주입 시스템을 통해 사용
- Session(bind=engine): 데이터베이스와 연결된 세션을 생성
- yield db: 생성된 세션을 반환 이 시점에서 클라이언트의 요청을 처리 가능
- finally: 요청이 끝난 후 세션을 닫음 이는 데이터베이스 연결을 안전하게 종료하는 데 중요함
8. 테이블 생성
Base.metadata.create_all(bind=engine)
create_all : 정의된 모든 모델에 대해 데이터베이스에 테이블을 생성 User모델에 해당하는 users 테이블이 데이터베이스에 생성됨 만약 테이블이 이미 존재한다면, 아무런 변화 X
9. API 엔드포인트 정의
- 회원가입: /signup 엔드포인트에서 사용자 정보를 받아 회원가입을 처리함
@app.post("/signup")
async def signup(signup_data: UserCreate, db: Session = Depends(get_db)):
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": "회원가입이 성공했습니다."}
- 사용자가 입력한 사용자 이름이 이미 존재하는지 확인합니다. 존재한다면 400 오류를 발생시킵니다.
- 비밀번호를 해시하여 새로운 사용자 객체를 생성하고, 데이터베이스에 추가합니다.
- 데이터베이스에 커밋하여 변경 사항을 저장합니다. 오류가 발생하면 500 오류를 발생시킵니다.
- 로그인: /login 엔드포인트에서 사용자 인증을 처리
@app.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="로그인이 실패했습니다.")
- 사용자가 입력한 사용자 이름으로 데이터베이스에서 사용자를 조회합니다.
- 비밀번호가 일치하면 세션에 사용자 이름을 저장하고 성공 메시지를 반환합니다. 일치하지 않으면 401 오류를 발생시킵니다.
- 로그아웃: /logout 엔드포인트에서 세션을 종료
@app.post("/logout")
async def logout(request: Request):
request.session.pop("username", None)
return {"message": "로그아웃이 성공했습니다."}
- 메모 생성: /memos/ 엔드포인트에서 메모 생성
@app.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
- 세션에서 사용자 이름을 가져와 사용자가 로그인했는지 확인 로그인하지 않았다면 401 오류를 발생
- 사용자가 존재하는지 확인한 후, 새로운 메모를 생성하고 데이터베이스에 추가
- 메모 조회: /memos/ 엔드포인트에서 사용자의 메모를 조회
@app.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})
- 로그인된 사용자의 메모를 조회하여 HTML 템플릿으로 렌더링함
- 메모 수정: /memos/{memo_id} 엔드포인트에서 특정 메모를 수정
@app.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
사용자가 로그인했는지 확인한 후, 수정할 메모를 데이터베이스에서 조회 메모가 존재하면 제목과 내용을 수정하고, 변경 사항을 커밋
- 메모 삭제: /memos/{memo_id} 엔드포인트에서 특정 메모를 삭제
@app.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"})
사용자가 로그인했는지 확인한 후, 삭제할 메모를 데이터베이스에서 조회 메모가 존재하면 삭제하고, 변경 사항을 커밋
루트 및 소개 페이지
@app.get('/')
async def read_root(request: Request):
return templates.TemplateResponse('home.html', {"request": request})
@app.get("/about")
async def about():
return {"message": "이것은 마이 메모 앱의 소개 페이지입니다."}
- 루트 엔드포인트: 기본 페이지를 렌더링함 home.html 템플릿을 사용하여 사용자에게 보여줄 내용을 생성 {"request": request}는 템플릿에서 요청 정보를 사용할 수 있도록 전달
- 소개 페이지: 애플리케이션에 대한 간단한 소개 메시지를 반환 JSON 형식으로 메시지를 반환
🐇 전체 코드
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel
from typing import Optional
from passlib.context import CryptContext
from starlette.middleware.sessions import SessionMiddleware
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)
app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-secret-key")
templates = Jinja2Templates(directory="templates")
DATABASE_URL = "mysql+pymysql://~~~~~~@localhost/my_memo_app"
engine = create_engine(DATABASE_URL)
Base = declarative_base()
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 UserCreate(BaseModel):
username: str
email: str
password: str # 해시전 패스워드를 받습니다.
# 회원로그인시 데이터 검증
class UserLogin(BaseModel):
username: str
password: str # 해시전 패스워드를 받습니다.
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))
class MemoCreate(BaseModel):
title: str
content: str
class MemoUpdate(BaseModel):
title: Optional[str] = None
content: Optional[str] = None
def get_db():
db = Session(bind=engine)
try:
yield db
finally:
db.close()
Base.metadata.create_all(bind=engine)
# 회원 가입
@app.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": "회원가입이 성공했습니다."}
# 로그인
@app.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="로그인이 실패했습니다.")
# 로그아웃
@app.post("/logout")
async def logout(request: Request):
request.session.pop("username", None)
return {"message": "로그아웃이 성공했습니다."}
# 메모 생성
@app.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
# 메모 조회
@app.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})
# 메모 수정
@app.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
# 메모 삭제
@app.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"})
# 기존 라우트
@app.get('/')
async def read_root(request: Request):
return templates.TemplateResponse('home.html', {"request": request})
@app.get("/about")
async def about():
return {"message": "이것은 마이 메모 앱의 소개 페이지입니다."}
function submitLoginForm(event) {
event.preventDefault();
const formData = new FormData(event.target);
const data = {
username: formData.get('username'),
password: formData.get('password')
};
fetch('/login', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify(data)
})
.then(response => response.json().then(body => ({ status: response.status, body: body })))
.then(result => {
if (result.status === 200) {
alert(result.body.message); // 성공 메시지 표시
window.location.href = '/memos'; // 성공 시 리다이렉트
} else {
throw new Error(result.body.detail || '로그인을 실패했습니다.'); // 서버가 제공하는 에러 메시지 또는 기본 메시지
}
})
.catch((error) => {
console.error('Error:', error);
alert(error.message); // 에러 메시지 표시
});
}
function submitSignupForm(event) {
event.preventDefault();
const formData = new FormData(event.target);
const data = {
username: formData.get('username'),
email: formData.get('email'),
password: formData.get('password')
};
fetch('/signup', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify(data)
})
.then(response => response.json().then(body => ({ status: response.status, body: body })))
.then(result => {
if (result.status === 200) {
alert(result.body.message); // 회원가입 성공 메시지 표시
window.location.href = '/'; // 성공 시 리다이렉트
} else {
throw new Error(result.body.detail || '회원가입을 실패했습니다.'); // 서버가 제공하는 에러 메시지 또는 기본 메시지
}
})
.catch((error) => {
console.error('Error:', error);
alert(error.message); // 에러 메시지 표시
});
}
function createMemo() {
var title = document.getElementById('new-title').value;
var content = document.getElementById('new-content').value;
fetch('/memos/', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ title: title, content: content })
})
.then(response => response.json())
.then(data => {
console.log(data);
window.location.reload(); // 페이지 새로고침
})
.catch((error) => {
console.error('Error:', error);
});
}
function toggleEdit(id) {
var titleEl = document.getElementById('title-' + id);
var contentEl = document.getElementById('content-' + id);
var isReadOnly = titleEl.readOnly;
titleEl.readOnly = !isReadOnly;
contentEl.readOnly = !isReadOnly;
if (!isReadOnly) {
updateMemo(id);
}
}
function updateMemo(id) {
var title = document.getElementById('title-' + id).value;
var content = document.getElementById('content-' + id).value;
fetch('/memos/' + id, {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ title: title, content: content })
})
.then(response => response.json())
.then(data => {
console.log(data);
alert('메모가 업데이트되었습니다.');
})
.catch((error) => {
console.error('Error:', error);
});
}
function deleteMemo(id) {
if (!confirm('메모를 삭제하시겠습니까?')) return;
fetch('/memos/' + id, {
method: 'DELETE',
})
.then(response => response.json())
.then(data => {
console.log(data);
window.location.reload(); // 페이지 새로고침
})
.catch((error) => {
console.error('Error:', error);
});
}
function logout() {
fetch('/logout', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
}
})
.then(response => response.json())
.then(data => {
console.log(data);
window.location.href = '/'; // 로그아웃 후 홈페이지로 리다이렉트
})
.catch((error) => {
console.error('Error:', error);
});
}
'프로젝트' 카테고리의 다른 글
| 이미지 크롤링(구글,네이버) (1) | 2025.04.30 |
|---|---|
| 크몽 작업물 - 인스타그램 팔로워 크롤링(22.09) (0) | 2025.04.28 |
| 크몽 작업물 - 유튜브 크롤링(22.11) (1) | 2024.11.23 |