# O2Sound Backend 개발자 가이드 ## 프로젝트 구조 ``` backend/ ├── app/ │ ├── core/ # 핵심 설정 및 유틸리티 │ │ ├── celery_app.py # Celery 설정 │ │ ├── database.py # 데이터베이스 연결 │ │ ├── env_setting.py # 환경 변수 설정 │ │ └── redis/ # Redis 관련 설정 │ ├── domain/ # 도메인 모델 │ │ └── models/ # SQLAlchemy 모델 │ ├── infra/ # 인프라 레이어 │ │ └── google/ # Google OAuth 구현 │ ├── presentation/ # 프레젠테이션 레이어 │ │ ├── api/ # API 라우터 │ │ │ └── v1/ # v1 API 엔드포인트 │ │ └── schemas/ # Pydantic 스키마 │ ├── services/ # 비즈니스 로직 │ ├── shared/ # 공통 유틸리티 │ │ ├── decorator/ # 데코레이터 │ │ ├── logger.py # 로깅 설정 │ │ └── progress.py # 진행률 추적 │ └── workers/ # Celery 작업 │ └── tasks.py # 비동기 작업 정의 ├── docs/ # API 문서 ├── uploads/ # 업로드된 파일 ├── main.py # FastAPI 앱 엔트리포인트 ├── dependencies.py # 의존성 주입 ├── pyproject.toml # Poetry 설정 └── Dockerfile # Docker 설정 ``` ## 핵심 개념 ### 1. DDD (Domain-Driven Design) 구조 프로젝트는 DDD 원칙을 따라 구성되어 있습니다: - **Domain Layer**: 비즈니스 엔티티와 도메인 로직 - **Application Layer**: 서비스와 비즈니스 유스케이스 - **Infrastructure Layer**: 외부 시스템과의 통합 - **Presentation Layer**: API 엔드포인트와 스키마 ### 2. 의존성 주입 FastAPI의 의존성 주입 시스템을 활용: ```python from app.dependencies import get_user_service @router.post("/items") async def get_items( request: GetItemsRequest, user_service: UserService = Depends(get_user_service) ): return user_service.get_items(request) ``` ### 3. Response Wrapper 모든 API 응답은 일관된 형식으로 래핑됩니다: ```python from app.shared.decorator.response_wrapper import response_wrapper @router.post("/login") @response_wrapper async def login(request: LoginRequest): # 자동으로 성공/실패 형식으로 래핑됨 return auth_service.login(request) ``` ## 주요 기능 구현 ### 1. 인증 시스템 #### 세션 기반 인증 ```python # main.py app.add_middleware(SessionMiddleware, secret_key=settings.SESSION_SECRET_KEY) ``` #### OAuth 2.0 (Google) ```python # infra/google/service.py class GoogleService: async def get_login_url(self, return_url: Optional[str], request: Request): # OAuth 플로우 시작 async def handle_callback(self, request: Request): # 콜백 처리 및 토큰 교환 ``` ### 2. 비동기 작업 처리 (Celery) #### 작업 정의 ```python # workers/tasks.py @celery_app.task(bind=True) def task1_crawl(self, url: str, root_task_id: str): # 웹 크롤링 작업 process = Process(redis_manager) process.update_progress(root_task_id, "metadata", True) ``` #### 워크플로우 체인 ```python # presentation/api/v1/moviemakers.py workflow = chain( crawl, chord( [lyrics_to_music, image_to_video], task5_merge_results.s(root_task_id) ) ) ``` ### 3. 진행률 추적 ```python # shared/progress.py class Process: def __init__(self, redis_manager: RedisManager): self.redis = redis_manager async def init_task_status(self, task_id: str): # 초기 상태 설정 async def update_progress(self, task_id: str, step: str, completed: bool): # 진행률 업데이트 ``` ## 개발 환경 설정 ### 1. 초기 설정 ```bash # Poetry 설치 pip install poetry # 의존성 설치 poetry install # 환경 변수 설정 cp .env.local .env ``` ### 2. 개발 서버 실행 ```bash # Redis 실행 (Docker) docker run -d -p 6379:6379 redis # PostgreSQL 실행 (Docker) docker run -d -p 5432:5432 \ -e POSTGRES_USER=o2sound \ -e POSTGRES_PASSWORD=password \ -e POSTGRES_DB=o2sound \ postgres:14 # Celery Worker 실행 poetry run celery -A app.core.celery_app worker --loglevel=info # FastAPI 개발 서버 poetry run uvicorn main:app --reload --port 8000 ``` ### 3. 환경 변수 설정 (.env) ```env # 기본 설정 PROJECT_NAME=O2Sound API_V1_STR=/api/v1 DEBUG=True # 데이터베이스 DATABASE_URL=postgresql://o2sound:password@localhost:5432/o2sound # Redis REDIS_URL=redis://localhost:6379 # 세션 SESSION_SECRET_KEY=your-secret-key-here # Google OAuth GOOGLE_CLIENT_ID=your-client-id GOOGLE_CLIENT_SECRET=your-client-secret GOOGLE_REDIRECT_URI=http://localhost:8000/social/google/callback # Celery CELERY_BROKER_URL=redis://localhost:6379/0 CELERY_RESULT_BACKEND=redis://localhost:6379/0 ``` ## 새로운 기능 추가하기 ### 1. 새 엔드포인트 추가 #### Step 1: 스키마 정의 ```python # presentation/schemas/new_feature_schema.py from pydantic import BaseModel from uuid import UUID class NewFeatureRequest(BaseModel): user_id: UUID feature_data: str class NewFeatureResponse(BaseModel): success: bool result: str ``` #### Step 2: 서비스 구현 ```python # services/new_feature_service.py class NewFeatureService: def __init__(self, db_session): self.db = db_session def process_feature(self, request: NewFeatureRequest): # 비즈니스 로직 구현 return NewFeatureResponse(success=True, result="processed") ``` #### Step 3: 라우터 추가 ```python # presentation/api/v1/new_feature.py from fastapi import APIRouter, Depends router = APIRouter(prefix="/new-feature", tags=["new-feature"]) @router.post("/process") @response_wrapper async def process_feature( request: NewFeatureRequest, service: NewFeatureService = Depends(get_new_feature_service) ): return service.process_feature(request) ``` #### Step 4: 메인 앱에 등록 ```python # main.py from app.presentation.api.v1.new_feature import router as new_feature_router app.include_router(new_feature_router) ``` ### 2. 새 Celery 작업 추가 ```python # workers/tasks.py @celery_app.task(bind=True) def new_async_task(self, data: dict, task_id: str): try: # 작업 시작 update_progress(task_id, "started", 0) # 처리 로직 result = process_data(data) # 완료 update_progress(task_id, "completed", 100) return result except Exception as e: update_progress(task_id, "failed", -1) raise ``` ## 테스트 ### 1. 단위 테스트 ```python # tests/test_auth.py import pytest from app.services.auth_service import AuthService def test_user_registration(): service = AuthService(mock_db_session) result = service.join(JoinRequest( user_id="test_user", name="Test User", password="password123" )) assert result.name == "Test User" ``` ### 2. API 테스트 ```python # tests/test_api.py from fastapi.testclient import TestClient from main import app client = TestClient(app) def test_health_check(): response = client.get("/health") assert response.status_code == 200 assert response.json()["status"] == "healthy" ``` ### 3. 통합 테스트 ```bash # Postman 컬렉션 실행 newman run tests/postman/O2Sound.postman_collection.json ``` ## 배포 ### 1. Docker 빌드 ```bash docker build -t o2sound-backend . ``` ### 2. Docker Compose ```yaml # docker-compose.yml version: '3.8' services: backend: image: o2sound-backend ports: - "8000:8000" environment: - DATABASE_URL=postgresql://user:pass@db:5432/o2sound - REDIS_URL=redis://redis:6379 depends_on: - db - redis worker: image: o2sound-backend command: celery -A app.core.celery_app worker environment: - CELERY_BROKER_URL=redis://redis:6379/0 depends_on: - redis db: image: postgres:14 environment: - POSTGRES_USER=user - POSTGRES_PASSWORD=pass - POSTGRES_DB=o2sound redis: image: redis:7 ``` ### 3. 프로덕션 체크리스트 - [ ] 환경 변수 확인 - [ ] 데이터베이스 마이그레이션 - [ ] Redis 연결 확인 - [ ] Celery Worker 실행 - [ ] 로그 설정 - [ ] 모니터링 설정 - [ ] 백업 전략 수립 ## 모니터링 ### 1. 로그 확인 ```bash # FastAPI 로그 tail -f logs/app.log # Celery 로그 tail -f logs/celery.log ``` ### 2. Celery Flower (웹 UI) ```bash celery -A app.core.celery_app flower --port=5555 ``` ### 3. 성능 모니터링 - APM 도구 (예: New Relic, DataDog) - Prometheus + Grafana - ELK Stack (Elasticsearch, Logstash, Kibana) ## 문제 해결 ### 일반적인 문제 1. **Redis 연결 실패** ``` ❌ Redis 연결 실패: Connection refused ``` 해결: Redis 서버가 실행 중인지 확인 2. **데이터베이스 연결 오류** ``` sqlalchemy.exc.OperationalError ``` 해결: DATABASE_URL 및 PostgreSQL 서버 상태 확인 3. **Celery 작업 실패** ``` Task task1_crawl[...] raised unexpected ``` 해결: Celery Worker 로그 확인 및 재시작 ### 디버깅 팁 1. **FastAPI 자동 문서** - Swagger UI: `http://localhost:8000/docs` - ReDoc: `http://localhost:8000/redoc` 2. **대화형 디버깅** ```python import pdb; pdb.set_trace() ``` 3. **로그 레벨 조정** ```python # 개발 환경에서 상세 로그 logger.setLevel(logging.DEBUG) ``` ## 추가 리소스 - [FastAPI 공식 문서](https://fastapi.tiangolo.com/) - [Celery 공식 문서](https://docs.celeryproject.org/) - [SQLAlchemy 공식 문서](https://www.sqlalchemy.org/) - [Redis 공식 문서](https://redis.io/documentation)