DEVOCEAN

# 추상화된 인터페이스 정의
class UserRepositoryInterface:
    def get_user(self, user_id: int):
        pass

# MySQL 저장소 구현
class MySQLUserRepository(UserRepositoryInterface):
    def get_user(self, user_id: int):
        # MySQL 데이터베이스에서 사용자 조회
        return {"id": user_id, "name": "MySQL User"}

from fastapi import Depends

def get_user_service(repo: UserRepositoryInterface = Depends(MySQLUserRepository)):
    def get_user(user_id: int):
        return repo.get_user(user_id)
    return get_user

from fastapi import FastAPI

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: int, service=Depends(get_user_service)):
    return service(user_id)

# MongoDB 저장소로 교체
def get_user_service(repo: UserRepositoryInterface = Depends(MongoDBUserRepository)):
    # 이제 MongoDB를 사용하여 사용자 정보를 조회합니다.
    ...

graph TD
    A[FastAPI Application] -->|의존| B[UserRepositoryInterface]
    
    B -.->|구현| C[MySQLUserRepository]
    B -.->|구현| D[MongoDBUserRepository]
    
    E[get_user_service] -->|의존| B
    F["/user/{user_id} Endpoint"] -->|호출| E
    
    G[Depends] -->|런타임 의존성 주입| C
    G -->|런타임 의존성 주입| D
    
    H[Runtime Injection] -.-> G

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

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

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

Base = declarative_base()

def create_tables():
    # Import models here to ensure they're registered with Base
    from app.models import User  # Make sure this import exists
    Base.metadata.create_all(bind=engine)

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

from sqlalchemy import Column, Integer, String
from app.database import Base
from pydantic import BaseModel

# SQLAlchemy Model
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)
    hashed_password = Column(String)

# Pydantic Models
class UserBase(BaseModel):
    username: str
    email: str
    full_name: str

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int

    class Config:
        from_attributes = True

from sqlalchemy.orm import Session
from typing import List, Optional
from app.models import User, UserCreate

# Repository Interface
class UserRepositoryInterface:
    def get_user_by_username(self, db: Session, username: str) -> Optional[User]:
        pass

    def get_user_by_email(self, db: Session, email: str) -> Optional[User]:
        pass

    def get_users(self, db: Session, skip: int = 0, limit: int = 10) -> List[User]:
        pass

    def create_user(self, db: Session, user: UserCreate) -> User:
        pass

# UserRepository 구현체
class UserRepository(UserRepositoryInterface):
    def get_user_by_username(self, db: Session, username: str) -> Optional[User]:
        return db.query(User).filter(User.username == username).first()

    def get_user_by_email(self, db: Session, email: str) -> Optional[User]:
        return db.query(User).filter(User.email == email).first()

    def get_users(self, db: Session, skip: int = 0, limit: int = 10) -> List[User]:
        return db.query(User).offset(skip).limit(limit).all()

    def create_user(self, db: Session, user: UserCreate) -> User:
        db_user = User(
            username=user.username,
            email=user.email,
            full_name=user.full_name
        )
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

from sqlalchemy.orm import Session
from app.user_repository import UserRepositoryInterface
from app.models import UserCreate, User
from fastapi import HTTPException
from typing import List

class UserService:
    def __init__(self, repository: UserRepositoryInterface):
        self.repository = repository

    def create_user(self, db: Session, user: UserCreate) -> User:
        db_user = self.repository.get_user_by_email(db, user.email)
        if db_user:
            raise HTTPException(status_code=400, detail="Email already registered")
        return self.repository.create_user(db, user)

    def get_users(self, db: Session, skip: int = 0, limit: int = 10) -> List[User]:
        return self.repository.get_users(db, skip, limit)

    def get_user_by_username(self, db: Session, username: str) -> User:
        user = self.repository.get_user_by_username(db, username)
        if user is None:
            raise HTTPException(status_code=404, detail="User not found")
        return user

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from app.database import get_db, create_tables
from app.models import UserCreate, UserResponse
from app.user_service import UserService
from app.user_repository import UserRepository
from typing import List

# Create tables before the app starts
create_tables()

app = FastAPI()

# Repository와 Service를 의존성 주입으로 설정
def get_user_service(repo=Depends(UserRepository)):
    return UserService(repo)

# Create User API Endpoint
@app.post("/users/", response_model=UserResponse)
def create_user_endpoint(
    user: UserCreate,
    db: Session = Depends(get_db),
    service: UserService = Depends(get_user_service)
):
    return service.create_user(db, user)

# Get Users API Endpoint
@app.get("/users/", response_model=List[UserResponse])
def read_users(
    skip: int = 0,
    limit: int = 10,
    db: Session = Depends(get_db),
    service: UserService = Depends(get_user_service)
):
    return service.get_users(db, skip, limit)

# Get User by Username API Endpoint
@app.get("/users/{username}", response_model=UserResponse)
def read_user(
    username: str,
    db: Session = Depends(get_db),
    service: UserService = Depends(get_user_service)
):
    return service.get_user_by_username(db, username)