snowdeer-user-api
샘플 프로젝트 생성
$ poetry new snowdeer-user-api
Created package snowdeer_user_api in snowdeer-user-api
아래와 같은 폴더 구조가 생성됨
.
├── pyproject.toml
├── README.md
├── snowdeer_user_api
│ └── __init__.py
└── tests
└── __init__.py
패키지 설치
$ cd snowdeer-user-api
$ poetry add "fastapi[standard]" uvicorn
$ poetry add --group dev ruff pytest
샘플 코드 작성
snowdeer_user_api/main.py
from fastapi import FastAPI
app = FastAPI(title="snowdeer's User API")
@app.get("/hello")
async def hello():
return {"message": "Hello, snowdeer!"}
# 아래 명령어로 실행
$ poetry run uvicorn snowdeer_user_api.main:app --reload
# 다른 터미널에서 확인
$ curl localhost:8000/hello
{"message":"Hello Snowdeer Users API!"}
SQLAlchemy 활용 Users 테이블 생성
$ poetry add sqlalchemy pydantic-settings
snowdeer_user_api/database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./users.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
password = Column(String)
snowdeer_user_api/main.py
from fastapi import FastAPI
from .database import engine, Base, SessionLocal
Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
app = FastAPI(title="snowdeer's User API")
@app.get("/hello")
async def hello():
return {"message": "Hello, snowdeer!"}
회원가입(Register) 및 로그인(Login) 엔드포인트 추가
snowdeer_user_api/schemas.py
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
password: str
class UserLogin(BaseModel):
username: str
password: str
snowdeer_user_api/database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./users.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
password = Column(String)
Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
snowdeer_user_api/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .schemas import UserCreate, UserLogin
from .database import get_db, User
app = FastAPI(title="snowdeer's User API")
@app.get("/hello")
async def hello():
return {"message": "Hello, snowdeer!"}
@app.post("/register")
async def register(user: UserCreate, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.username == user.username).first()
if db_user:
raise HTTPException(400, "Username already exists")
new_user = User(username=user.username, password=user.password)
db.add(new_user)
db.commit()
return {"msg": "User created"}
@app.post("/login")
async def login(user: UserLogin, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.username == user.username).first()
if not db_user or db_user.password != user.password:
raise HTTPException(401, "Invalid credentials")
return {"msg": "Login successful"}
아래 명령어로 동작 확인
# 아직 회원 가입을 안했기 때문에 오류가 생겨야 정상
$ curl -X POST http://localhost:8000/login \
-H "Content-Type: application/json" \
-d '{"username": "snowdeer", "password": "snowdeer"}'
{"detail":"Invalid credentials"}
# 회원 가입
curl -X POST http://localhost:8000/register \
-H "Content-Type: application/json" \
-d '{"username": "snowdeer", "password": "snowdeer"}'
{"msg":"User created","user_id":1}
# 로그인 테스트
$ curl -X POST http://localhost:8000/login \
-H "Content-Type: application/json" \
-d '{"username": "snowdeer", "password": "snowdeer"}'
{"msg":"Login successful"}
# 잘못된 패스워드로 테스트
$ curl -X POST http://localhost:8000/login \
-H "Content-Type: application/json" \
-d '{"username": "snowdeer", "password": "wrong-password"}'
{"detail":"Invalid credentials"}
SHA256 암호화 적용
snowdeer_user_api/utils.py
import hashlib
def hash_password(password: str) -> str:
return hashlib.sha256(password.encode()).hexdigest()
snowdeer_user_api/main.py
아래 부분을 수정
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .schemas import UserCreate, UserLogin
from .database import get_db, User
from .utils import hash_password
app = FastAPI(title="snowdeer's User API")
@app.get("/hello")
async def hello():
return {"message": "Hello, snowdeer!"}
@app.post("/register")
async def register(user: UserCreate, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.username == user.username).first()
if db_user:
raise HTTPException(400, "Username already exists")
new_user = User(username=user.username, password=hash_password(user.password))
db.add(new_user)
db.commit()
return {"msg": "User created"}
@app.post("/login")
async def login(user: UserLogin, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.username == user.username).first()
if not db_user or db_user.password != hash_password(user.password):
raise HTTPException(401, "Invalid credentials")
return {"msg": "Login successful"}
암호화에 Salt 적용
snowdeer_user_api/utils.py
import hashlib
import secrets
def generate_salt() -> str:
return secrets.token_hex(16)
def hash_password(password: str, salt: str = None) -> tuple[str, str]:
if not salt:
salt = generate_salt()
salted_password = f"{password}{salt}"
hashed = hashlib.sha256(salted_password.encode()).hexdigest()
return hashed, salt
snowdeer_user_api/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .schemas import UserCreate, UserLogin
from .database import get_db, User
from .utils import hash_password
app = FastAPI(title="snowdeer's User API")
@app.get("/hello")
async def hello():
return {"message": "Hello, snowdeer!"}
@app.post("/register")
async def register(user: UserCreate, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.username == user.username).first()
if db_user:
raise HTTPException(400, "Username already exists")
hashed, salt = hash_password(user.password)
new_user = User(username=user.username, password=f"{hashed}:{salt}")
db.add(new_user)
db.commit()
return {"msg": "User created"}
@app.post("/login")
async def login(user: UserLogin, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.username == user.username).first()
if not db_user:
raise HTTPException(401, "Invalid credentials")
_, salt = db_user.password.split(":")
hashed_input, _ = hash_password(user.password, salt)
if hashed_input != db_user.password.split(":")[0]:
raise HTTPException(401, "Invalid credentials")
return {"msg": "Login successful"}
