$ 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
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!"}
$ poetry add sqlalchemy pydantic-settings
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)
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!"}
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
password: str
class UserLogin(BaseModel):
username: str
password: str
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()
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"}
import hashlib
def hash_password(password: str) -> str:
return hashlib.sha256(password.encode()).hexdigest()
아래 부분을 수정
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"}
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
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"}
$ poetry new snowdeer-login-sample
Created package snowdeer_login_sample in snowdeer-login-sample
아래와 같은 폴더 구조가 생성됨
.
├── pyproject.toml
├── README.md
├── snowdeer_login_sample
│ └── __init__.py
└── tests
└── __init__.py
$ cd snowdeer-login-sample
$ poetry add "fastapi[standard]" uvicorn sqlalchemy
snowdeer-login-sample/snowdeer_login_sample/main.py에 아래 내용 생성
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello")
async def hello():
return {"message": "hello, snowdeer !!"}
# 아래 명령어로 실행
$ poetry run uvicorn snowdeer_login_sample.main:app --reload
# 다른 터미널에서 확인
$ curl localhost:8000/hello
{"message":"hello, snowdeer !!"}
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)
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)
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from .database import SessionLocal, User
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/hello")
async def hello():
return {"message": "hello, snowdeer !!"}
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy.orm import Session
from .database import SessionLocal, User
app = FastAPI()
class UserCreate(BaseModel):
username: str
password: str
class UserLogin(BaseModel):
username: str
password: str
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/hello")
async def hello():
return {"message": "Hello FastAPI with DB!"}
@app.post("/register")
async def register(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(username=user.username, password=user.password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return {"msg": "User created", "user_id": db_user.id}
@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(status_code=400, detail="Invalid credentials")
return {"msg": "Login successful", "user_id": db_user.id}
아래 명령어로 동작 확인
# 아직 회원 가입을 안했기 때문에 오류가 생겨야 정상
$ 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","user_id":1}
# 잘못된 패스워드로 테스트
$ curl -X POST http://localhost:8000/login \
-H "Content-Type: application/json" \
-d '{"username": "snowdeer", "password": "wrong-password"}'
{"detail":"Invalid credentials"}
import hashlib
def hash_password(password: str) -> str:
return hashlib.sha256(password.encode()).hexdigest()
아래 부분을 수정
from .utils import hash_password
# ...
@app.post("/register")
async def register(user: UserCreate, db: Session = Depends(get_db)):
hashed_pw = hash_password(user.password)
db_user = User(username=user.username, password=hashed_pw)
db.add(db_user)
db.commit()
db.refresh(db_user)
return {"msg": "User created"}
@app.post("/login")
async def login(user: UserLogin, db: Session = Depends(get_db)):
hashed_pw = hash_password(user.password)
db_user = db.query(User).filter(
User.username == user.username, User.password == hashed_pw
).first()
if not db_user:
raise HTTPException(status_code=400, detail="Invalid credentials")
return {"msg": "Login successful"}