Python Login 백엔드 샘플 2(Poetry, FastAPI, PyTest)

|

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"}

Python Login 백엔드 샘플(Poetry, FastAPI, SQLite)

|

Sample Login 백엔드 서비스 코드

샘플 프로젝트 생성

$ 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 !!"}

SQLAlchemy 활용 Users 테이블 생성

snowdeer-login-sample/snowdeer_login_sample/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)
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)

snowdeer-login-sample/snowdeer_login_sample/main.py

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 !!"}

회원가입(Register) 및 로그인(Login) 엔드포인트 추가

snowdeer-login-sample/snowdeer_login_sample/main.py

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"}

SHA256 암호화 적용

snowdeer-login-sample/snowdeer_login_sample/utils.py

import hashlib

def hash_password(password: str) -> str:
    return hashlib.sha256(password.encode()).hexdigest()

snowdeer-login-sample/snowdeer_login_sample/main.py

아래 부분을 수정

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"}

주요 HTTP Status

|

주요 HTTP Status Code

200 OK

정상 응답 상태 코드

301 Moved Permanently

HTTP 요청을 보낸 주소의 URL이 변경되었다는 상태 코드.
보통 Location 헤더가 포함되는 것이 일반적이며,
Location 헤더에 새로운 주소가 포함되어 있음

HTTP/1.1 301 Moved Permanently
Location: http://www.snowdeer.com/index.html

400 Bad Request

잘못된 요청일 때 보내는 응답 코드.
주로 Request에 잘못된 값들이 보내졌을 때 사용함

401 Unauthorized

사용자가 로그인이 필요한 경우 401을 리턴

403 Forbidden

로그인은 되어 있으나, 해당 리소스에 대한 접근 권한이 없는 경우 리턴

404 Not Found

HTTP 요청을 보내는 URL 주소가 존재하지 않을 경우 보내는 응답 코드

500 Internal Server Error

서버 내부 오류가 발생했을 때 리턴하는 응답 코드

Poetry 샘플 코드

|

Poetry 활용 샘플 코드 작성

Poetry Init

$ mkdir snowdeer-poerty
$ cd snowdeer-poerty
$ poetry init -n

Poetry Add [Packages]

$ poetry add "fastapi[standard]" uvicorn

Creating virtualenv snowdeer-poetry in /Users/snowdeer/Workspace/snowdeer/snowdeer-poetry/.venv
Using version ^0.125.0 for fastapi
Using version ^0.38.0 for uvicorn

Updating dependencies
Resolving dependencies... Downloading https://files.pythonhosted.org/packages/fa/a8/5b41e0da817d64113292ab1f8247140aac61cbf6cfd085d6a0fa77f4984f/websockets-15.0.
Resolving dependencies... Downloading https://files.pythonhosted.org/packages/c3/e2/51d9ee443aabcd5aa581d45b18b6198ced364b5cd97e5504c5d782ceb82c/fastar-0.8.0-cp3
Resolving dependencies... Downloading https://files.pythonhosted.org/packages/c3/e2/51d9ee443aabcd5aa581d45b18b6198ced364b5cd97e5504c5d782ceb82c/fastar-0.8.0-cp3
Resolving dependencies... Downloading https://files.pythonhosted.org/packages/86/7a/b970cd0138b0ece72eb28f086e933f9ed75b795716ad3de5ab22994b3b54/rignore-0.7.6-cp
Resolving dependencies... (6.2s)

Package operations: 42 installs, 0 updates, 0 removals

  • Installing typing-extensions (4.15.0)
  • Installing exceptiongroup (1.3.1)
  • Installing idna (3.11)
  • Installing mdurl (0.1.2)
  • Installing anyio (4.12.0)
  • Installing certifi (2025.11.12)
  • Installing dnspython (2.8.0)
  • Installing h11 (0.16.0)
  • Installing markdown-it-py (4.0.0)
  • Installing pygments (2.19.2)
  • Installing annotated-types (0.7.0)
  • Installing click (8.3.1)
  • Installing email-validator (2.3.0)
  • Installing httpcore (1.0.9)
  • Installing httptools (0.7.1)
  • Installing pydantic-core (2.41.5)
  • Installing python-dotenv (1.2.1)
  • Installing pyyaml (6.0.3)
  • Installing rich (14.2.0)
  • Installing shellingham (1.5.4)
  • Installing typing-inspection (0.4.2)
  • Installing urllib3 (2.6.2)
  • Installing uvloop (0.22.1)
  • Installing watchfiles (1.1.1)
  • Installing websockets (15.0.1)
  • Installing fastar (0.8.0)
  • Installing httpx (0.28.1)
  • Installing pydantic (2.12.5)
  • Installing rich-toolkit (0.17.1)
  • Installing rignore (0.7.6)
  • Installing sentry-sdk (2.48.0)
  • Installing typer (0.20.1)
  • Installing uvicorn (0.38.0)
  • Installing fastapi-cloud-cli (0.7.0)
  • Installing markupsafe (3.0.3)
  • Installing tomli (2.3.0)
  • Installing annotated-doc (0.0.4)
  • Installing fastapi-cli (0.0.16)
  • Installing jinja2 (3.1.6)
  • Installing python-multipart (0.0.21)
  • Installing starlette (0.50.0)
  • Installing fastapi (0.125.0)

Writing lock file

디렉토리 구조

아래와 같은 구조의 디렉토리 생성

.
├── poetry.lock
├── pyproject.toml
├── README.md
└── snowdeer_poetry
    ├── __init__.py
    └── main.py

snowdeer_poetry/main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
async def hello_endpoint():
    return ["hello", "snowdeer"]

실행

그 이후 루트 디렉토리에서 아래 명령어 실행

$ poetry run uvicorn snowdeer_poetry.main:app --host 0.0.0.0 --port 8314 --reload

INFO:     Will watch for changes in these directories: ['/Users/snowdeer/Workspace/snowdeer/snowdeer-poetry']
INFO:     Uvicorn running on http://0.0.0.0:8314 (Press CTRL+C to quit)
INFO:     Started reloader process [60728] using WatchFiles
INFO:     Started server process [60770]
INFO:     Waiting for application startup.

코드에서 uvicorn 실행하는 법

만약 코드에서 uvicorn을 실행하려면 아래와 같이 main.py 수정

수정된 main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
async def hello_endpoint():
    return ["hello", "snowdeer"]

import uvicorn

def dev_server():
    uvicorn.run("snowdeer_poetry.main:app", host="0.0.0.0", port=8314, reload=True)

pyproject.toml

# 파일 맽 끝에 아래 라인 추가

[tool.poetry.scripts]
dev = "snowdeer_poetry.main:dev_server"

실행

$ poetry run dev

INFO:     Will watch for changes in these directories: ['/Users/snowdeer/Workspace/snowdeer/snowdeer-poetry']
INFO:     Uvicorn running on http://0.0.0.0:8314 (Press CTRL+C to quit)
INFO:     Started reloader process [64017] using WatchFiles
INFO:     Started server process [64022]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

영상 캡셔닝 AI 모델(ViT) 테스트

|

실행 환경

맥북(M1 Pro) 기반으로 실습 테스트

패키지 설치

# 가상환경 생성
python -m venv venv
source venv/bin/activate

# 패키지 설치
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/nightly/cpu
pip install transformers pillow opencv-python accelerate

사용 가능 디바이스 확인

Apple Silicon 환경에서는 mps 디바이스 사용 가능

ViT 기반 모델

영상에서 8개의 keyframe을 추출해서 캡션을 생성하는 코드

main.py

import torch
import cv2
from PIL import Image
from transformers import VisionEncoderDecoderModel, ViTImageProcessor, GPT2Tokenizer
import argparse
import sys

def setup_device():
    """M1 Pro MPS 또는 CPU 설정"""
    if torch.backends.mps.is_available():
        return "mps"
    return "cpu"

def load_model(device):
    """모델과 토크나이저 로드"""
    model_name = "nlpconnect/vit-gpt2-image-captioning"
    model = VisionEncoderDecoderModel.from_pretrained(model_name)
    feature_extractor = ViTImageProcessor.from_pretrained(model_name)
    tokenizer = GPT2Tokenizer.from_pretrained(model_name)
    
    # 패딩 토큰 설정
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
    
    model.to(device)
    model.eval()
    return model, feature_extractor, tokenizer

def extract_key_frames(video_path, max_frames=8):
    """주요 프레임 추출"""
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    
    frames = []
    step = max(1, total_frames // max_frames)
    
    for i in range(0, total_frames, step):
        cap.set(cv2.CAP_PROP_POS_FRAMES, i)
        ret, frame = cap.read()
        if ret:
            rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            frames.append(Image.fromarray(rgb_frame))
    
    cap.release()
    return frames

def generate_captions(model, feature_extractor, tokenizer, frames, device):
    """각 프레임 캡션 생성"""
    captions = []
    
    for frame in frames:
        # 이미지 전처리
        pixel_values = feature_extractor(images=frame, return_tensors="pt").pixel_values.to(device)
        
        # 캡션 생성
        with torch.no_grad():
            generated_ids = model.generate(
                pixel_values, 
                max_length=30,
                num_beams=5,
                early_stopping=True,
                pad_token_id=tokenizer.eos_token_id
            )
        
        # 토크나이저로 디코딩
        caption = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
        captions.append(caption.strip())
    
    return captions

def summarize_video(captions):
    """캡션들을 하나의 영어 문장으로 요약"""
    if not captions:
        return "No content detected in the video."
    
    # 중복 제거 후 가장 대표적인 캡션 선택
    unique_captions = list(set(captions))
    if len(unique_captions) <= 2:
        return " and ".join(unique_captions)
    
    # 주요 장면들 결합
    main_scenes = unique_captions[:3]
    return f"A video featuring {', '.join(main_scenes[:-1])} and {main_scenes[-1]}."

def main(video_path):
    print(f"Processing video: {video_path}")
    
    # 디바이스 및 모델 설정
    device = setup_device()
    print(f"Using device: {device}")
    
    model, feature_extractor, tokenizer = load_model(device)
    
    # 프레임 추출
    frames = extract_key_frames(video_path, max_frames=8)
    print(f"Extracted {len(frames)} key frames")
    
    # 캡션 생성
    captions = generate_captions(model, feature_extractor, tokenizer, frames, device)
    print("Generated captions:", captions)
    
    # 최종 요약
    summary = summarize_video(captions)
    print("\n" + "="*50)
    print("VIDEO SUMMARY (영어 한 문장):")
    print(summary)
    print("="*50)
    
    return summary

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python video_caption.py ")
        sys.exit(1)
    
    video_path = sys.argv[1]
    main(video_path)
</pre>

### 실행 결과

$ python video.py example6.mp4

Processing video: example6.mp4
Using device: mps
Extracted 9 key frames
The attention mask is not set and cannot be inferred from input because pad token is same as eos token. As a consequence, you may observe unexpected behavior. Please pass your input's `attention_mask` to obtain reliable results.
We strongly recommend passing in an `attention_mask` since your input_ids may be padded. See https://huggingface.co/docs/transformers/troubleshooting#incorrect-output-when-padding-tokens-arent-masked.
You may ignore this warning if your `pad_token_id` (50256) is identical to the `bos_token_id` (50256), `eos_token_id` (50256), or the `sep_token_id` (None), and your input is not padded.
Generated captions: ['a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a man riding a wave on top of a surfboard', 'a surfer riding a wave in the ocean']

==================================================
VIDEO SUMMARY (영어 한 문장):
a surfer riding a wave in the ocean and a man riding a wave on top of a surfboard
==================================================