Fastapi CRUD

# Project Structure (unchanged)
crud-app/
│
├── api/
│   └── v1/
│       └── endpoints/
│           └── items.py
│
├── core/
│   ├── config.py
│   └── logger.py
│
├── crud/
│   └── crud_item.py
│
├── db/
│   └── session.py
│
├── exceptions/
│   └── http_exceptions.py
│
├── models/
│   └── item.py
│
├── schemas/
│   └── item.py
│
├── tests/
│   ├── conftest.py
│   ├── test_api.py
│   └── test_crud.py
│
├── main.py
├── Dockerfile
├── Makefile
├── dev.env
├── prod.env
└── requirements.txt

# 수정된 파일: api/v1/endpoints/items.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from crud import crud_item
from models.item import Item
from schemas.item import ItemCreate, ItemUpdate, ItemInDB
from db.session import get_db
from core.logger import logger
from exceptions.http_exceptions import ItemNotFoundException, ItemAlreadyExistsException

router = APIRouter()

@router.get("/items/", response_model=List[ItemInDB], summary="Get all items")
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """
    Retrieve all items.

    - **skip**: Number of items to skip (pagination)
    - **limit**: Maximum number of items to return
    """
    items = crud_item.get_items(db, skip=skip, limit=limit)
    logger.info(f"Retrieved {len(items)} items")
    return items

@router.post("/items/", response_model=ItemInDB, summary="Create a new item")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    """
    Create a new item.

    - **name**: Required. The name of the item
    - **description**: Optional. A description of the item
    """
    db_item = crud_item.get_item_by_name(db, name=item.name)
    if db_item:
        logger.warning(f"Attempt to create duplicate item: {item.name}")
        raise ItemAlreadyExistsException(name=item.name)
    return crud_item.create_item(db=db, item=item)

@router.get("/items/{item_id}", response_model=ItemInDB, summary="Get a specific item")
def read_item(item_id: int, db: Session = Depends(get_db)):
    """
    Get a specific item by ID.

    - **item_id**: Required. The ID of the item to retrieve
    """
    db_item = crud_item.get_item(db, item_id=item_id)
    if db_item is None:
        logger.warning(f"Item not found: {item_id}")
        raise ItemNotFoundException(item_id=item_id)
    return db_item

@router.put("/items/{item_id}", response_model=ItemInDB, summary="Update an item")
def update_item(item_id: int, item: ItemUpdate, db: Session = Depends(get_db)):
    """
    Update an existing item.

    - **item_id**: Required. The ID of the item to update
    - **name**: Optional. The new name of the item
    - **description**: Optional. The new description of the item
    """
    db_item = crud_item.get_item(db, item_id=item_id)
    if db_item is None:
        logger.warning(f"Attempt to update non-existent item: {item_id}")
        raise ItemNotFoundException(item_id=item_id)
    return crud_item.update_item(db=db, item_id=item_id, item=item)

@router.delete("/items/{item_id}", response_model=ItemInDB, summary="Delete an item")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    """
    Delete an existing item.

    - **item_id**: Required. The ID of the item to delete
    """
    db_item = crud_item.get_item(db, item_id=item_id)
    if db_item is None:
        logger.warning(f"Attempt to delete non-existent item: {item_id}")
        raise ItemNotFoundException(item_id=item_id)
    return crud_item.delete_item(db=db, item_id=item_id)

# 나머지 파일들은 변경 없음

# main.py
from fastapi import FastAPI
from api.v1.endpoints import items
from core.config import settings
from db.session import engine
from models import item as item_model

app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

# Create database tables
item_model.Base.metadata.create_all(bind=engine)

# Include routers
app.include_router(items.router, prefix=settings.API_V1_STR)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host=settings.HOST,
        port=settings.PORT,
        reload=settings.DEBUG
    )

# core/config.py
from pydantic import BaseSettings
import os

class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "CRUD App"
    DEBUG: bool = os.getenv("DEBUG", "False") == "True"
    HOST: str = os.getenv("HOST", "0.0.0.0")
    PORT: int = int(os.getenv("PORT", 8000))
    DATABASE_URL: str = os.getenv("DATABASE_URL", "sqlite:///./sql_app.db")
    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")

    class Config:
        env_file = "dev.env" if os.getenv("ENV") == "dev" else "prod.env"

settings = Settings()

# core/logger.py
import logging
from core.config import settings

# Configure logging
logging.basicConfig(
    level=settings.LOG_LEVEL,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)

logger = logging.getLogger(__name__)

# crud/crud_item.py
from sqlalchemy.orm import Session
from models.item import Item
from schemas.item import ItemCreate, ItemUpdate

def get_item(db: Session, item_id: int):
    return db.query(Item).filter(Item.id == item_id).first()

def get_item_by_name(db: Session, name: str):
    return db.query(Item).filter(Item.name == name).first()

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Item).offset(skip).limit(limit).all()

def create_item(db: Session, item: ItemCreate):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def update_item(db: Session, item_id: int, item: ItemUpdate):
    db_item = get_item(db, item_id)
    update_data = item.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_item, key, value)
    db.commit()
    db.refresh(db_item)
    return db_item

def delete_item(db: Session, item_id: int):
    db_item = get_item(db, item_id)
    db.delete(db_item)
    db.commit()
    return db_item

# db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from core.config import settings

engine = create_engine(settings.DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# exceptions/http_exceptions.py
from fastapi import HTTPException

class ItemNotFoundException(HTTPException):
    def __init__(self, item_id: int):
        super().__init__(status_code=404, detail=f"Item with id {item_id} not found")

class ItemAlreadyExistsException(HTTPException):
    def __init__(self, name: str):
        super().__init__(status_code=400, detail=f"Item with name {name} already exists")

# models/item.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)
    description = Column(String, index=True)

# schemas/item.py
from pydantic import BaseModel, Field

class ItemBase(BaseModel):
    name: str = Field(..., example="Laptop")
    description: str = Field(None, example="A high-performance laptop")

class ItemCreate(ItemBase):
    pass

class ItemUpdate(ItemBase):
    name: str = Field(None, example="Updated Laptop")
    description: str = Field(None, example="An updated high-performance laptop")

class ItemInDB(ItemBase):
    id: int

    class Config:
        orm_mode = True

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from main import app
from db.session import get_db
from models.item import Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="function")
def db():
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def client(db):
    def override_get_db():
        try:
            yield db
        finally:
            db.close()
    
    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    del app.dependency_overrides[get_db]

# tests/test_api.py
from fastapi.testclient import TestClient

def test_create_item(client):
    response = client.post(
        "/api/v1/items/",
        json={"name": "Test Item", "description": "This is a test item"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Test Item"
    assert data["description"] == "This is a test item"
    assert "id" in data

def test_read_item(client):
    # First, create an item
    response = client.post(
        "/api/v1/items/",
        json={"name": "Test Item", "description": "This is a test item"},
    )
    created_item = response.json()

    # Then, read the item
    response = client.get(f"/api/v1/items/{created_item['id']}")
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Test Item"
    assert data["description"] == "This is a test item"
    assert data["id"] == created_item["id"]

# tests/test_crud.py
from crud import crud_item
from schemas.item import ItemCreate

def test_create_item(db):
    item = ItemCreate(name="Test Item", description="This is a test item")
    db_item = crud_item.create_item(db, item)
    assert db_item.name == "Test Item"
    assert db_item.description == "This is a test item"

def test_get_item(db):
    item = ItemCreate(name="Test Item", description="This is a test item")
    db_item = crud_item.create_item(db, item)
    
    retrieved_item = crud_item.get_item(db, db_item.id)
    assert retrieved_item.name == "Test Item"
    assert retrieved_item.description == "This is a test item"

# Dockerfile
FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# Makefile
.PHONY: install run-dev run-prod test

install:
	pip install -r requirements.txt

run-dev:
	ENV=dev uvicorn main:app --reload

run-prod:
	ENV=prod gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

test:
	pytest

# dev.env
DEBUG=True
HOST=localhost
PORT=8000
DATABASE_URL=sqlite:///./sql_app_dev.db
LOG_LEVEL=DEBUG

# prod.env
DEBUG=False
HOST=0.0.0.0
PORT=8000
DATABASE_URL=sqlite:///./sql_app_prod.db
LOG_LEVEL=INFO

# requirements.txt
fastapi==0.68.0
uvicorn==0.15.0
sqlalchemy==1.4.23
pydantic==1.8.2
python-dotenv==0.19.0
pytest==6.2.5
gunicorn==20.1.0

windows

cd crud-app
set ENV=dev
uvicorn main:app --reload

cd crud-app
set ENV=prod
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

linux

cd crud-app
ENV=dev uvicorn main:app --reload

cd crud-app
ENV=prod gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker