yakenator/todos2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 풀스택 할일관리 앱 구현 (통합 모달 + 간트차트)

Browse Source 
- Backend: FastAPI + MongoDB + Redis (카테고리, 할일 CRUD, 파일 첨부, 검색, 대시보드)
- Frontend: Next.js 15 + Tailwind + React Query + Zustand
- 통합 TodoModal: 생성/수정 모달 통합, 탭 구조 (기본/태그와 첨부)
- 간트차트: 카테고리별 할일 타임라인 시각화
- TodoCard: 제목/카테고리/우선순위/태그/첨부 한줄 표시
- Docker Compose 배포 (Frontend:3010, Backend:8010, MongoDB:27021, Redis:6391)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jungwoo choi
2026-02-12 15:45:03 +09:00
parent b54811ad8d
commit 074b5133bf
81 changed files with 17027 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
backend/app/config.py Normal file
View File

@ -0,0 +1,44 @@
from pydantic_settings import BaseSettings
from pydantic import Field
from functools import lru_cache
class Settings(BaseSettings):
mongodb_url: str = Field(
default="mongodb://mongodb:27017",
description="MongoDB 연결 URL",
)
mongodb_database: str = Field(
default="todos2",
alias="DB_NAME",
description="MongoDB 데이터베이스 이름",
)
redis_url: str = Field(
default="redis://redis:6379",
description="Redis 연결 URL",
)
frontend_url: str = Field(
default="http://localhost:3000",
description="프론트엔드 URL (CORS 허용)",
)
upload_dir: str = Field(
default="/app/uploads",
description="파일 업로드 디렉토리",
)
max_file_size: int = Field(
default=10 * 1024 * 1024,
description="최대 파일 크기 (bytes, 기본 10MB)",
)
max_files_per_todo: int = Field(
default=5,
description="Todo당 최대 첨부 파일 수",
)
class Config:
env_file = ".env"
populate_by_name = True # alias와 필드명 모두 허용
@lru_cache
def get_settings() -> Settings:
return Settings()

98
backend/app/database.py Normal file
View File

@ -0,0 +1,98 @@
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
import redis.asyncio as aioredis
from app.config import get_settings
class Database:
client: AsyncIOMotorClient | None = None
db: AsyncIOMotorDatabase | None = None
redis: aioredis.Redis | None = None
db = Database()
def get_database() -> AsyncIOMotorDatabase:
"""MongoDB 데이터베이스 인스턴스를 반환한다.
주의: `if not db.db:` 사용 금지 (pymongo 4.9.x NotImplementedError)
"""
if db.db is None:
raise RuntimeError("Database not initialized")
return db.db
def get_redis() -> aioredis.Redis:
"""Redis 클라이언트 인스턴스를 반환한다."""
if db.redis is None:
raise RuntimeError("Redis not initialized")
return db.redis
async def connect_db():
"""MongoDB + Redis 연결 초기화"""
settings = get_settings()
# MongoDB
db.client = AsyncIOMotorClient(settings.mongodb_url)
db.db = db.client[settings.mongodb_database]
# 인덱스 생성
await create_indexes(db.db)
# Redis
db.redis = aioredis.from_url(
settings.redis_url,
encoding="utf-8",
decode_responses=True,
)
async def disconnect_db():
"""연결 종료"""
if db.client is not None:
db.client.close()
if db.redis is not None:
await db.redis.close()
async def create_indexes(database: AsyncIOMotorDatabase):
"""컬렉션 인덱스 생성"""
todos = database["todos"]
categories = database["categories"]
# todos 텍스트 검색 인덱스
await todos.create_index(
[("title", "text"), ("content", "text"), ("tags", "text")],
name="text_search_index",
weights={"title": 10, "content": 5, "tags": 3},
)
# todos 복합 인덱스
await todos.create_index(
[("category_id", 1), ("created_at", -1)],
name="category_created",
)
await todos.create_index(
[("completed", 1), ("created_at", -1)],
name="completed_created",
)
await todos.create_index(
[("priority", 1), ("created_at", -1)],
name="priority_created",
)
await todos.create_index(
[("tags", 1)],
name="tags",
)
await todos.create_index(
[("due_date", 1)],
name="due_date",
)
await todos.create_index(
[("completed", 1), ("due_date", 1)],
name="completed_due_date",
)
# categories 인덱스
await categories.create_index("name", unique=True, name="category_name_unique")
await categories.create_index("order", name="category_order")

64
backend/app/main.py
View File

@ -1,17 +1,55 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from datetime import datetime
app = FastAPI(title="API", version="1.0.0")
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
from app.config import get_settings
from app.database import connect_db, disconnect_db
@app.get("/health")
async def health_check():
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
@asynccontextmanager
async def lifespan(app: FastAPI):
"""서버 시작/종료 이벤트: MongoDB + Redis 연결 관리"""
await connect_db()
yield
await disconnect_db()
def create_app() -> FastAPI:
settings = get_settings()
app = FastAPI(
title="todos2 API",
description="확장형 할일 관리 애플리케이션 API",
version="1.0.0",
lifespan=lifespan,
)
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 라우터 등록
from app.routers import todos, categories, tags, search, dashboard, uploads
app.include_router(todos.router)
app.include_router(categories.router)
app.include_router(tags.router)
app.include_router(search.router)
app.include_router(dashboard.router)
app.include_router(uploads.router)
# 헬스 체크
@app.get("/health", tags=["health"])
async def health_check():
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
return app
app = create_app()

47
backend/app/models/__init__.py Normal file
View File

@ -0,0 +1,47 @@
from app.models.common import PyObjectId, ErrorResponse, PaginatedResponse
from app.models.todo import (
Priority,
TodoCreate,
TodoUpdate,
TodoResponse,
TodoListResponse,
BatchRequest,
BatchResponse,
ToggleResponse,
TagInfo,
SearchResponse,
DashboardStats,
OverviewStats,
CategoryStats,
PriorityStats,
UpcomingDeadline,
)
from app.models.category import (
CategoryCreate,
CategoryUpdate,
CategoryResponse,
)
__all__ = [
"PyObjectId",
"ErrorResponse",
"PaginatedResponse",
"Priority",
"TodoCreate",
"TodoUpdate",
"TodoResponse",
"TodoListResponse",
"BatchRequest",
"BatchResponse",
"ToggleResponse",
"TagInfo",
"SearchResponse",
"DashboardStats",
"OverviewStats",
"CategoryStats",
"PriorityStats",
"UpcomingDeadline",
"CategoryCreate",
"CategoryUpdate",
"CategoryResponse",
]

49
backend/app/models/category.py Normal file
View File

@ -0,0 +1,49 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, field_validator
# === Request 스키마 ===
class CategoryCreate(BaseModel):
"""카테고리 생성 요청 (F-007)"""
name: str = Field(..., min_length=1, max_length=50)
color: str = Field("#6B7280", pattern=r"^#[0-9A-Fa-f]{6}$")
@field_validator("name")
@classmethod
def name_not_blank(cls, v: str) -> str:
v = v.strip()
if not v:
raise ValueError("카테고리 이름은 공백만으로 구성할 수 없습니다")
return v
class CategoryUpdate(BaseModel):
"""카테고리 수정 요청 (F-009) - Partial Update"""
name: Optional[str] = Field(None, min_length=1, max_length=50)
color: Optional[str] = Field(None, pattern=r"^#[0-9A-Fa-f]{6}$")
order: Optional[int] = Field(None, ge=0)
@field_validator("name")
@classmethod
def name_not_blank(cls, v: Optional[str]) -> Optional[str]:
if v is not None:
v = v.strip()
if not v:
raise ValueError("카테고리 이름은 공백만으로 구성할 수 없습니다")
return v
# === Response 스키마 ===
class CategoryResponse(BaseModel):
"""카테고리 응답 (F-008)"""
id: str
name: str
color: str
order: int
todo_count: int = 0
created_at: datetime

29
backend/app/models/common.py Normal file
View File

@ -0,0 +1,29 @@
from typing import Annotated, Any
from bson import ObjectId
from pydantic import BaseModel, BeforeValidator
def validate_object_id(v: Any) -> str:
"""ObjectId <-> str 변환을 위한 유효성 검증기"""
if isinstance(v, ObjectId):
return str(v)
if isinstance(v, str) and ObjectId.is_valid(v):
return v
raise ValueError(f"Invalid ObjectId: {v}")
PyObjectId = Annotated[str, BeforeValidator(validate_object_id)]
class ErrorResponse(BaseModel):
"""표준 에러 응답"""
detail: str
class PaginatedResponse(BaseModel):
"""페이지네이션 응답 래퍼"""
items: list[Any]
total: int
page: int
limit: int
total_pages: int

221
backend/app/models/todo.py Normal file
View File

@ -0,0 +1,221 @@
from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, field_validator
class Priority(str, Enum):
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
# === Request 스키마 ===
class TodoCreate(BaseModel):
"""할일 생성 요청 (F-001)"""
title: str = Field(..., min_length=1, max_length=200)
content: Optional[str] = Field(None, max_length=2000)
category_id: Optional[str] = None
tags: list[str] = Field(default_factory=list, max_length=10)
priority: Priority = Priority.MEDIUM
start_date: Optional[datetime] = None
due_date: Optional[datetime] = None
@field_validator("title")
@classmethod
def title_not_blank(cls, v: str) -> str:
v = v.strip()
if not v:
raise ValueError("제목은 공백만으로 구성할 수 없습니다")
return v
@field_validator("tags")
@classmethod
def normalize_tags(cls, v: list[str]) -> list[str]:
"""소문자 정규화, 중복 제거, 공백 trim"""
seen: set[str] = set()
result: list[str] = []
for tag in v:
tag = tag.strip().lower()
if tag and len(tag) <= 30 and tag not in seen:
seen.add(tag)
result.append(tag)
return result
@field_validator("category_id")
@classmethod
def validate_category_id(cls, v: Optional[str]) -> Optional[str]:
from bson import ObjectId
if v is not None and not ObjectId.is_valid(v):
raise ValueError("유효하지 않은 카테고리 ID 형식입니다")
return v
class TodoUpdate(BaseModel):
"""할일 수정 요청 (F-004) - Partial Update"""
title: Optional[str] = Field(None, min_length=1, max_length=200)
content: Optional[str] = Field(None, max_length=2000)
completed: Optional[bool] = None
category_id: Optional[str] = None
tags: Optional[list[str]] = None
priority: Optional[Priority] = None
start_date: Optional[datetime] = None
due_date: Optional[datetime] = None
@field_validator("title")
@classmethod
def title_not_blank(cls, v: Optional[str]) -> Optional[str]:
if v is not None:
v = v.strip()
if not v:
raise ValueError("제목은 공백만으로 구성할 수 없습니다")
return v
@field_validator("tags")
@classmethod
def normalize_tags(cls, v: Optional[list[str]]) -> Optional[list[str]]:
if v is None:
return None
seen: set[str] = set()
result: list[str] = []
for tag in v:
tag = tag.strip().lower()
if tag and len(tag) <= 30 and tag not in seen:
seen.add(tag)
result.append(tag)
return result
@field_validator("category_id")
@classmethod
def validate_category_id(cls, v: Optional[str]) -> Optional[str]:
from bson import ObjectId
if v is not None and not ObjectId.is_valid(v):
raise ValueError("유효하지 않은 카테고리 ID 형식입니다")
return v
class BatchRequest(BaseModel):
"""일괄 작업 요청 (F-019, F-020, F-021)"""
action: str = Field(..., pattern="^(complete|delete|move_category)$")
ids: list[str] = Field(..., min_length=1)
category_id: Optional[str] = None # move_category 시에만 사용
@field_validator("ids")
@classmethod
def validate_ids(cls, v: list[str]) -> list[str]:
from bson import ObjectId
for id_str in v:
if not ObjectId.is_valid(id_str):
raise ValueError(f"유효하지 않은 ID 형식: {id_str}")
return v
# === Attachment 스키마 ===
class Attachment(BaseModel):
"""첨부파일 메타데이터"""
id: str
filename: str
stored_filename: str
content_type: str
size: int
uploaded_at: datetime
# === Response 스키마 ===
class TodoResponse(BaseModel):
"""할일 응답 (카테고리 정보 포함 가능)"""
id: str
title: str
content: Optional[str] = None
completed: bool
priority: Priority
category_id: Optional[str] = None
category_name: Optional[str] = None
category_color: Optional[str] = None
tags: list[str] = []
start_date: Optional[datetime] = None
due_date: Optional[datetime] = None
attachments: list[Attachment] = []
created_at: datetime
updated_at: datetime
class TodoListResponse(BaseModel):
"""할일 목록 페이지네이션 응답 (F-002)"""
items: list[TodoResponse]
total: int
page: int
limit: int
total_pages: int
class ToggleResponse(BaseModel):
"""완료 토글 응답 (F-006)"""
id: str
completed: bool
class BatchResponse(BaseModel):
"""일괄 작업 응답 (F-019, F-020, F-021)"""
action: str
processed: int
failed: int
class TagInfo(BaseModel):
"""태그 정보 (F-013)"""
name: str
count: int
class SearchResponse(BaseModel):
"""검색 응답 (F-017)"""
items: list[TodoResponse]
total: int
query: str
page: int
limit: int
# === 대시보드 통계 모델 ===
class OverviewStats(BaseModel):
total: int
completed: int
incomplete: int
completion_rate: float
class CategoryStats(BaseModel):
category_id: Optional[str] = None
name: str
color: str
count: int
class PriorityStats(BaseModel):
high: int
medium: int
low: int
class UpcomingDeadline(BaseModel):
id: str
title: str
due_date: datetime
priority: Priority
class DashboardStats(BaseModel):
"""대시보드 통계 응답 (F-018)"""
overview: OverviewStats
by_category: list[CategoryStats]
by_priority: PriorityStats
upcoming_deadlines: list[UpcomingDeadline]

9
backend/app/routers/__init__.py Normal file
View File

@ -0,0 +1,9 @@
from app.routers import todos, categories, tags, search, dashboard
__all__ = [
"todos",
"categories",
"tags",
"search",
"dashboard",
]

55
backend/app/routers/categories.py Normal file
View File

@ -0,0 +1,55 @@
from fastapi import APIRouter, Depends, Response, status
from app.database import get_database, get_redis
from app.models.category import (
CategoryCreate,
CategoryUpdate,
CategoryResponse,
)
from app.services.category_service import CategoryService
router = APIRouter(prefix="/api/categories", tags=["categories"])
def _get_service(
db=Depends(get_database),
redis=Depends(get_redis),
) -> CategoryService:
return CategoryService(db, redis)
@router.get("", response_model=list[CategoryResponse])
async def list_categories(
service: CategoryService = Depends(_get_service),
):
"""F-008: 카테고리 목록 조회 (order 오름차순, todo_count 포함)"""
return await service.list_categories()
@router.post("", response_model=CategoryResponse, status_code=status.HTTP_201_CREATED)
async def create_category(
data: CategoryCreate,
service: CategoryService = Depends(_get_service),
):
"""F-007: 카테고리 생성"""
return await service.create_category(data)
@router.put("/{category_id}", response_model=CategoryResponse)
async def update_category(
category_id: str,
data: CategoryUpdate,
service: CategoryService = Depends(_get_service),
):
"""F-009: 카테고리 수정"""
return await service.update_category(category_id, data)
@router.delete("/{category_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_category(
category_id: str,
service: CategoryService = Depends(_get_service),
):
"""F-010: 카테고리 삭제 (해당 카테고리의 할일 category_id null 처리)"""
await service.delete_category(category_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)

17
backend/app/routers/dashboard.py Normal file
View File

@ -0,0 +1,17 @@
from fastapi import APIRouter, Depends
from app.database import get_database, get_redis
from app.models.todo import DashboardStats
from app.services.dashboard_service import DashboardService
router = APIRouter(prefix="/api/dashboard", tags=["dashboard"])
@router.get("/stats", response_model=DashboardStats)
async def get_dashboard_stats(
db=Depends(get_database),
redis=Depends(get_redis),
):
"""F-018: 대시보드 통계 (Redis 캐싱, TTL 60초)"""
service = DashboardService(db, redis)
return await service.get_stats()

19
backend/app/routers/search.py Normal file
View File

@ -0,0 +1,19 @@
from fastapi import APIRouter, Depends, Query
from app.database import get_database
from app.models.todo import SearchResponse
from app.services.search_service import SearchService
router = APIRouter(prefix="/api/search", tags=["search"])
@router.get("", response_model=SearchResponse)
async def search_todos(
q: str = Query(..., min_length=1, max_length=200, description="검색 키워드"),
page: int = Query(1, ge=1, description="페이지 번호"),
limit: int = Query(20, ge=1, le=100, description="페이지당 항목 수"),
db=Depends(get_database),
):
"""F-017: 전문 검색 (제목/내용/태그, text score 기준 정렬)"""
service = SearchService(db)
return await service.search(query=q, page=page, limit=limit)

16
backend/app/routers/tags.py Normal file
View File

@ -0,0 +1,16 @@
from fastapi import APIRouter, Depends
from app.database import get_database
from app.models.todo import TagInfo
from app.services.todo_service import TodoService
router = APIRouter(prefix="/api/tags", tags=["tags"])
@router.get("", response_model=list[TagInfo])
async def list_tags(
db=Depends(get_database),
):
"""F-013: 사용 중인 태그 목록 조회 (사용 횟수 포함, 내림차순)"""
service = TodoService(db)
return await service.get_tags()

106
backend/app/routers/todos.py Normal file
View File

@ -0,0 +1,106 @@
from typing import Optional
from fastapi import APIRouter, Depends, Query, Response, status
from app.database import get_database, get_redis
from app.models.todo import (
TodoCreate,
TodoUpdate,
TodoResponse,
TodoListResponse,
BatchRequest,
BatchResponse,
ToggleResponse,
)
from app.services.todo_service import TodoService
router = APIRouter(prefix="/api/todos", tags=["todos"])
def _get_service(
db=Depends(get_database),
redis=Depends(get_redis),
) -> TodoService:
return TodoService(db, redis)
# 중요: /batch를 /{id} 보다 위에 등록
@router.post("/batch", response_model=BatchResponse)
async def batch_action(
data: BatchRequest,
service: TodoService = Depends(_get_service),
):
"""F-019~F-021: 일괄 작업 (완료/삭제/카테고리 변경)"""
return await service.batch_action(data)
@router.get("", response_model=TodoListResponse)
async def list_todos(
page: int = Query(1, ge=1, description="페이지 번호"),
limit: int = Query(20, ge=1, le=100, description="페이지당 항목 수"),
completed: Optional[bool] = Query(None, description="완료 상태 필터"),
category_id: Optional[str] = Query(None, description="카테고리 필터"),
priority: Optional[str] = Query(None, description="우선순위 필터 (high/medium/low)"),
tag: Optional[str] = Query(None, description="태그 필터"),
sort: str = Query("created_at", description="정렬 기준 (created_at/due_date/priority)"),
order: str = Query("desc", description="정렬 방향 (asc/desc)"),
service: TodoService = Depends(_get_service),
):
"""F-002: 할일 목록 조회 (필터/정렬/페이지네이션)"""
return await service.list_todos(
page=page,
limit=limit,
completed=completed,
category_id=category_id,
priority=priority,
tag=tag,
sort=sort,
order=order,
)
@router.post("", response_model=TodoResponse, status_code=status.HTTP_201_CREATED)
async def create_todo(
data: TodoCreate,
service: TodoService = Depends(_get_service),
):
"""F-001: 할일 생성"""
return await service.create_todo(data)
@router.get("/{todo_id}", response_model=TodoResponse)
async def get_todo(
todo_id: str,
service: TodoService = Depends(_get_service),
):
"""F-003: 할일 상세 조회"""
return await service.get_todo(todo_id)
@router.put("/{todo_id}", response_model=TodoResponse)
async def update_todo(
todo_id: str,
data: TodoUpdate,
service: TodoService = Depends(_get_service),
):
"""F-004: 할일 수정"""
return await service.update_todo(todo_id, data)
@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(
todo_id: str,
service: TodoService = Depends(_get_service),
):
"""F-005: 할일 삭제"""
await service.delete_todo(todo_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.patch("/{todo_id}/toggle", response_model=ToggleResponse)
async def toggle_todo(
todo_id: str,
service: TodoService = Depends(_get_service),
):
"""F-006: 할일 완료 토글"""
return await service.toggle_todo(todo_id)

56
backend/app/routers/uploads.py Normal file
View File

@ -0,0 +1,56 @@
from typing import List
from fastapi import APIRouter, Depends, File, UploadFile, status
from fastapi.responses import FileResponse
from app.database import get_database
from app.models.todo import Attachment
from app.services.file_service import FileService
router = APIRouter(prefix="/api/todos", tags=["attachments"])
def _get_service(db=Depends(get_database)) -> FileService:
return FileService(db)
@router.post(
"/{todo_id}/attachments",
response_model=list[Attachment],
status_code=status.HTTP_201_CREATED,
)
async def upload_attachments(
todo_id: str,
files: List[UploadFile] = File(...),
service: FileService = Depends(_get_service),
):
"""파일 업로드 (최대 5개, 파일당 10MB)"""
return await service.upload_files(todo_id, files)
@router.get("/{todo_id}/attachments/{attachment_id}/download")
async def download_attachment(
todo_id: str,
attachment_id: str,
service: FileService = Depends(_get_service),
):
"""파일 다운로드"""
file_path, filename = await service.get_file_info(todo_id, attachment_id)
return FileResponse(
path=str(file_path),
filename=filename,
media_type="application/octet-stream",
)
@router.delete(
"/{todo_id}/attachments/{attachment_id}",
status_code=status.HTTP_204_NO_CONTENT,
)
async def delete_attachment(
todo_id: str,
attachment_id: str,
service: FileService = Depends(_get_service),
):
"""첨부파일 삭제"""
await service.delete_attachment(todo_id, attachment_id)

11
backend/app/services/__init__.py Normal file
View File

@ -0,0 +1,11 @@
from app.services.todo_service import TodoService
from app.services.category_service import CategoryService
from app.services.search_service import SearchService
from app.services.dashboard_service import DashboardService
__all__ = [
"TodoService",
"CategoryService",
"SearchService",
"DashboardService",
]

161
backend/app/services/category_service.py Normal file
View File

@ -0,0 +1,161 @@
from datetime import datetime, timezone
from typing import Optional
from bson import ObjectId
from fastapi import HTTPException
from motor.motor_asyncio import AsyncIOMotorDatabase
import redis.asyncio as aioredis
from app.models.category import (
CategoryCreate,
CategoryUpdate,
CategoryResponse,
)
DASHBOARD_CACHE_KEY = "dashboard:stats"
class CategoryService:
"""카테고리 CRUD 비즈니스 로직"""
def __init__(self, db: AsyncIOMotorDatabase, redis_client: Optional[aioredis.Redis] = None):
self.collection = db["categories"]
self.todos = db["todos"]
self.redis = redis_client
async def _invalidate_cache(self):
"""대시보드 캐시 무효화"""
if self.redis is not None:
try:
await self.redis.delete(DASHBOARD_CACHE_KEY)
except Exception:
pass
async def list_categories(self) -> list[CategoryResponse]:
"""F-008: 카테고리 목록 조회 (order 오름차순, todo_count 포함)"""
categories = await self.collection.find().sort("order", 1).to_list(None)
result = []
for cat in categories:
cat_id_str = str(cat["_id"])
# 해당 카테고리의 할일 수 조회
todo_count = await self.todos.count_documents({"category_id": cat_id_str})
result.append(CategoryResponse(
id=cat_id_str,
name=cat["name"],
color=cat.get("color", "#6B7280"),
order=cat.get("order", 0),
todo_count=todo_count,
created_at=cat["created_at"],
))
return result
async def create_category(self, data: CategoryCreate) -> CategoryResponse:
"""F-007: 카테고리 생성"""
# 이름 중복 검사
existing = await self.collection.find_one({"name": data.name})
if existing:
raise HTTPException(status_code=409, detail="이미 존재하는 카테고리 이름입니다")
# order 자동 설정 (현재 최대값 + 1)
max_order_doc = await self.collection.find_one(
{},
sort=[("order", -1)],
)
next_order = (max_order_doc["order"] + 1) if max_order_doc and "order" in max_order_doc else 0
now = datetime.now(timezone.utc)
doc = {
"name": data.name,
"color": data.color,
"order": next_order,
"created_at": now,
}
result = await self.collection.insert_one(doc)
doc["_id"] = result.inserted_id
await self._invalidate_cache()
return CategoryResponse(
id=str(doc["_id"]),
name=doc["name"],
color=doc["color"],
order=doc["order"],
todo_count=0,
created_at=doc["created_at"],
)
async def update_category(self, category_id: str, data: CategoryUpdate) -> CategoryResponse:
"""F-009: 카테고리 수정 (Partial Update)"""
if not ObjectId.is_valid(category_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
# 존재 확인
existing = await self.collection.find_one({"_id": ObjectId(category_id)})
if not existing:
raise HTTPException(status_code=404, detail="카테고리를 찾을 수 없습니다")
# 변경 필드 추출
update_data = data.model_dump(exclude_unset=True)
if not update_data:
raise HTTPException(status_code=400, detail="수정할 필드가 없습니다")
# name 변경 시 중복 검사
if "name" in update_data:
dup = await self.collection.find_one({
"name": update_data["name"],
"_id": {"$ne": ObjectId(category_id)},
})
if dup:
raise HTTPException(status_code=409, detail="이미 존재하는 카테고리 이름입니다")
result = await self.collection.find_one_and_update(
{"_id": ObjectId(category_id)},
{"$set": update_data},
return_document=True,
)
if not result:
raise HTTPException(status_code=404, detail="카테고리를 찾을 수 없습니다")
# todo_count 조회
cat_id_str = str(result["_id"])
todo_count = await self.todos.count_documents({"category_id": cat_id_str})
await self._invalidate_cache()
return CategoryResponse(
id=cat_id_str,
name=result["name"],
color=result.get("color", "#6B7280"),
order=result.get("order", 0),
todo_count=todo_count,
created_at=result["created_at"],
)
async def delete_category(self, category_id: str) -> None:
"""F-010: 카테고리 삭제 + 해당 카테고리의 할일 category_id null 처리"""
if not ObjectId.is_valid(category_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
# 존재 확인
existing = await self.collection.find_one({"_id": ObjectId(category_id)})
if not existing:
raise HTTPException(status_code=404, detail="카테고리를 찾을 수 없습니다")
# 해당 카테고리의 모든 todo.category_id -> null 갱신
await self.todos.update_many(
{"category_id": category_id},
{"$set": {
"category_id": None,
"updated_at": datetime.now(timezone.utc),
}},
)
# 카테고리 삭제
await self.collection.delete_one({"_id": ObjectId(category_id)})
await self._invalidate_cache()

182
backend/app/services/dashboard_service.py Normal file
View File

@ -0,0 +1,182 @@
import json
from datetime import datetime, timezone
from bson import ObjectId
from motor.motor_asyncio import AsyncIOMotorDatabase
import redis.asyncio as aioredis
from app.models.todo import (
DashboardStats,
OverviewStats,
CategoryStats,
PriorityStats,
UpcomingDeadline,
)
DASHBOARD_CACHE_KEY = "dashboard:stats"
DASHBOARD_CACHE_TTL = 60 # 60초
class DashboardService:
"""대시보드 통계 + Redis 캐싱"""
def __init__(self, db: AsyncIOMotorDatabase, redis_client: aioredis.Redis):
self.todos = db["todos"]
self.categories = db["categories"]
self.redis = redis_client
async def get_stats(self) -> DashboardStats:
"""F-018: 대시보드 통계
1. Redis 캐시 확인
2. 캐시 히트 -> JSON 파싱 후 반환
3. 캐시 미스 -> MongoDB 집계 -> Redis 저장 (TTL 60s) -> 반환
"""
# Redis 캐시 확인
try:
cached = await self.redis.get(DASHBOARD_CACHE_KEY)
if cached:
data = json.loads(cached)
return DashboardStats(**data)
except Exception:
pass # Redis 에러 시 DB에서 직접 조회
# 캐시 미스: MongoDB 집계
stats = await self._compute_stats()
# Redis에 캐싱
try:
stats_json = stats.model_dump_json()
await self.redis.setex(DASHBOARD_CACHE_KEY, DASHBOARD_CACHE_TTL, stats_json)
except Exception:
pass # Redis 저장 실패는 무시
return stats
async def _compute_stats(self) -> DashboardStats:
"""MongoDB 집계 파이프라인으로 통계 계산"""
# 1. Overview (전체/완료/미완료/완료율)
total = await self.todos.count_documents({})
completed = await self.todos.count_documents({"completed": True})
incomplete = total - completed
completion_rate = round((completed / total * 100), 1) if total > 0 else 0.0
overview = OverviewStats(
total=total,
completed=completed,
incomplete=incomplete,
completion_rate=completion_rate,
)
# 2. By Category (카테고리별 할일 분포)
by_category = await self._compute_by_category()
# 3. By Priority (우선순위별 현황)
by_priority = await self._compute_by_priority()
# 4. Upcoming Deadlines (마감 임박 할일 상위 5개)
upcoming_deadlines = await self._compute_upcoming_deadlines()
return DashboardStats(
overview=overview,
by_category=by_category,
by_priority=by_priority,
upcoming_deadlines=upcoming_deadlines,
)
async def _compute_by_category(self) -> list[CategoryStats]:
"""카테고리별 할일 분포 집계"""
pipeline = [
{"$group": {
"_id": "$category_id",
"count": {"$sum": 1},
}},
{"$sort": {"count": -1}},
]
results = await self.todos.aggregate(pipeline).to_list(None)
# 카테고리 정보 조회
cat_ids = set()
for r in results:
if r["_id"]:
try:
cat_ids.add(ObjectId(r["_id"]))
except Exception:
pass
cat_map: dict[str, dict] = {}
if cat_ids:
async for cat in self.categories.find({"_id": {"$in": list(cat_ids)}}):
cat_map[str(cat["_id"])] = cat
category_stats = []
for r in results:
cat_id = r["_id"]
if cat_id and cat_id in cat_map:
cat = cat_map[cat_id]
category_stats.append(CategoryStats(
category_id=cat_id,
name=cat["name"],
color=cat.get("color", "#6B7280"),
count=r["count"],
))
else:
category_stats.append(CategoryStats(
category_id=None,
name="미분류",
color="#9CA3AF",
count=r["count"],
))
return category_stats
async def _compute_by_priority(self) -> PriorityStats:
"""우선순위별 현황 집계"""
pipeline = [
{"$group": {
"_id": "$priority",
"count": {"$sum": 1},
}},
]
results = await self.todos.aggregate(pipeline).to_list(None)
priority_map = {"high": 0, "medium": 0, "low": 0}
for r in results:
if r["_id"] in priority_map:
priority_map[r["_id"]] = r["count"]
return PriorityStats(
high=priority_map["high"],
medium=priority_map["medium"],
low=priority_map["low"],
)
async def _compute_upcoming_deadlines(self) -> list[UpcomingDeadline]:
"""마감 임박 할일 상위 5개 (미완료, 마감일 있는 것, 마감일 오름차순)"""
now = datetime.now(timezone.utc)
cursor = self.todos.find({
"completed": False,
"due_date": {"$ne": None, "$gte": now},
}).sort("due_date", 1).limit(5)
docs = await cursor.to_list(5)
deadlines = []
for doc in docs:
deadlines.append(UpcomingDeadline(
id=str(doc["_id"]),
title=doc["title"],
due_date=doc["due_date"],
priority=doc["priority"],
))
return deadlines
async def invalidate_cache(self) -> None:
"""캐시 무효화: 할일 CUD 작업 후 호출"""
try:
await self.redis.delete(DASHBOARD_CACHE_KEY)
except Exception:
pass

154
backend/app/services/file_service.py Normal file
View File

@ -0,0 +1,154 @@
import shutil
import uuid
from datetime import datetime, timezone
from pathlib import Path
from bson import ObjectId
from fastapi import HTTPException, UploadFile
from motor.motor_asyncio import AsyncIOMotorDatabase
from app.config import get_settings
from app.models.todo import Attachment
class FileService:
"""파일 업로드/다운로드/삭제 서비스"""
def __init__(self, db: AsyncIOMotorDatabase):
self.collection = db["todos"]
self.settings = get_settings()
self.upload_dir = Path(self.settings.upload_dir)
def _ensure_todo_dir(self, todo_id: str) -> Path:
todo_dir = self.upload_dir / todo_id
todo_dir.mkdir(parents=True, exist_ok=True)
return todo_dir
async def upload_files(
self, todo_id: str, files: list[UploadFile]
) -> list[Attachment]:
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
todo = await self.collection.find_one({"_id": ObjectId(todo_id)})
if not todo:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
existing_count = len(todo.get("attachments", []))
if existing_count + len(files) > self.settings.max_files_per_todo:
raise HTTPException(
status_code=400,
detail=f"최대 {self.settings.max_files_per_todo}개의 파일만 첨부할 수 있습니다",
)
todo_dir = self._ensure_todo_dir(todo_id)
attachments: list[Attachment] = []
saved_paths: list[Path] = []
try:
for file in files:
content = await file.read()
if len(content) > self.settings.max_file_size:
raise HTTPException(
status_code=400,
detail=f"파일 '{file.filename}'이 최대 크기(10MB)를 초과합니다",
)
file_id = str(uuid.uuid4())
ext = Path(file.filename or "").suffix
stored_filename = f"{file_id}{ext}"
file_path = todo_dir / stored_filename
with open(file_path, "wb") as f:
f.write(content)
saved_paths.append(file_path)
attachments.append(
Attachment(
id=file_id,
filename=file.filename or "unknown",
stored_filename=stored_filename,
content_type=file.content_type or "application/octet-stream",
size=len(content),
uploaded_at=datetime.now(timezone.utc),
)
)
await self.collection.update_one(
{"_id": ObjectId(todo_id)},
{
"$push": {
"attachments": {
"$each": [a.model_dump() for a in attachments]
}
},
"$set": {"updated_at": datetime.now(timezone.utc)},
},
)
except HTTPException:
for p in saved_paths:
p.unlink(missing_ok=True)
raise
except Exception:
for p in saved_paths:
p.unlink(missing_ok=True)
raise HTTPException(status_code=500, detail="파일 업로드에 실패했습니다")
return attachments
async def delete_attachment(self, todo_id: str, attachment_id: str) -> None:
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
todo = await self.collection.find_one({"_id": ObjectId(todo_id)})
if not todo:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
attachment = None
for att in todo.get("attachments", []):
if att["id"] == attachment_id:
attachment = att
break
if not attachment:
raise HTTPException(status_code=404, detail="첨부파일을 찾을 수 없습니다")
file_path = self.upload_dir / todo_id / attachment["stored_filename"]
if file_path.exists():
file_path.unlink()
await self.collection.update_one(
{"_id": ObjectId(todo_id)},
{
"$pull": {"attachments": {"id": attachment_id}},
"$set": {"updated_at": datetime.now(timezone.utc)},
},
)
async def get_file_info(
self, todo_id: str, attachment_id: str
) -> tuple[Path, str]:
"""파일 경로와 원본 파일명을 반환"""
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
todo = await self.collection.find_one({"_id": ObjectId(todo_id)})
if not todo:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
for att in todo.get("attachments", []):
if att["id"] == attachment_id:
file_path = self.upload_dir / todo_id / att["stored_filename"]
if not file_path.exists():
raise HTTPException(
status_code=404, detail="파일이 디스크에서 찾을 수 없습니다"
)
return file_path, att["filename"]
raise HTTPException(status_code=404, detail="첨부파일을 찾을 수 없습니다")
async def delete_all_todo_files(self, todo_id: str) -> None:
"""Todo 삭제 시 해당 Todo의 모든 파일 삭제"""
todo_dir = self.upload_dir / todo_id
if todo_dir.exists():
shutil.rmtree(todo_dir)

92
backend/app/services/search_service.py Normal file
View File

@ -0,0 +1,92 @@
import math
from bson import ObjectId
from fastapi import HTTPException
from motor.motor_asyncio import AsyncIOMotorDatabase
from app.models.todo import TodoResponse, SearchResponse
class SearchService:
"""MongoDB text index 기반 검색 로직"""
def __init__(self, db: AsyncIOMotorDatabase):
self.collection = db["todos"]
self.categories = db["categories"]
async def _populate_categories_bulk(self, todo_docs: list[dict]) -> list[TodoResponse]:
"""여러 Todo 문서의 카테고리 정보를 일괄 조회하여 TodoResponse 리스트를 반환한다."""
# 카테고리 ID 수집
cat_ids = set()
for doc in todo_docs:
cat_id = doc.get("category_id")
if cat_id:
try:
cat_ids.add(ObjectId(cat_id))
except Exception:
pass
# 카테고리 일괄 조회
cat_map: dict[str, dict] = {}
if cat_ids:
async for cat in self.categories.find({"_id": {"$in": list(cat_ids)}}):
cat_map[str(cat["_id"])] = cat
# TodoResponse 변환
result = []
for doc in todo_docs:
cat_id = str(doc["category_id"]) if doc.get("category_id") else None
cat = cat_map.get(cat_id) if cat_id else None
result.append(TodoResponse(
id=str(doc["_id"]),
title=doc["title"],
content=doc.get("content"),
completed=doc["completed"],
priority=doc["priority"],
category_id=cat_id,
category_name=cat["name"] if cat else None,
category_color=cat["color"] if cat else None,
tags=doc.get("tags", []),
due_date=doc.get("due_date"),
created_at=doc["created_at"],
updated_at=doc["updated_at"],
))
return result
async def search(self, query: str, page: int = 1, limit: int = 20) -> SearchResponse:
"""F-017: 전문 검색
MongoDB text index를 활용한 전문 검색, text score 기준 정렬, 페이지네이션
"""
if not query or not query.strip():
raise HTTPException(status_code=422, detail="검색어를 입력해주세요")
query = query.strip()
# MongoDB text search
search_filter = {"$text": {"$search": query}}
# 총 개수 조회
total = await self.collection.count_documents(search_filter)
# text score 기준 정렬 + 페이지네이션
skip = (page - 1) * limit
cursor = self.collection.find(
search_filter,
{"score": {"$meta": "textScore"}},
).sort(
[("score", {"$meta": "textScore"})]
).skip(skip).limit(limit)
docs = await cursor.to_list(limit)
items = await self._populate_categories_bulk(docs)
return SearchResponse(
items=items,
total=total,
query=query,
page=page,
limit=limit,
)

396
backend/app/services/todo_service.py Normal file
View File

@ -0,0 +1,396 @@
import math
from datetime import datetime, timezone
from typing import Optional
from bson import ObjectId
from fastapi import HTTPException
from motor.motor_asyncio import AsyncIOMotorDatabase
import redis.asyncio as aioredis
from app.models.todo import (
Priority,
TodoCreate,
TodoUpdate,
TodoResponse,
TodoListResponse,
BatchRequest,
BatchResponse,
ToggleResponse,
TagInfo,
)
# 우선순위 정렬용 매핑 (높음이 먼저 오도록)
PRIORITY_ORDER = {"high": 0, "medium": 1, "low": 2}
DASHBOARD_CACHE_KEY = "dashboard:stats"
class TodoService:
"""할일 CRUD + 일괄 작업 + 태그 집계 비즈니스 로직"""
def __init__(self, db: AsyncIOMotorDatabase, redis_client: Optional[aioredis.Redis] = None):
self.collection = db["todos"]
self.categories = db["categories"]
self.redis = redis_client
async def _invalidate_cache(self):
"""대시보드 캐시 무효화"""
if self.redis is not None:
try:
await self.redis.delete(DASHBOARD_CACHE_KEY)
except Exception:
pass # Redis 오류 시 무시 (캐시 무효화 실패는 치명적이지 않음)
async def _populate_category(self, todo_doc: dict) -> TodoResponse:
"""Todo 문서에 카테고리 정보를 추가하여 TodoResponse를 반환한다."""
category_name = None
category_color = None
cat_id = todo_doc.get("category_id")
if cat_id:
category = await self.categories.find_one({"_id": ObjectId(cat_id)})
if category:
category_name = category["name"]
category_color = category["color"]
return TodoResponse(
id=str(todo_doc["_id"]),
title=todo_doc["title"],
content=todo_doc.get("content"),
completed=todo_doc["completed"],
priority=todo_doc["priority"],
category_id=str(todo_doc["category_id"]) if todo_doc.get("category_id") else None,
category_name=category_name,
category_color=category_color,
tags=todo_doc.get("tags", []),
start_date=todo_doc.get("start_date"),
due_date=todo_doc.get("due_date"),
attachments=todo_doc.get("attachments", []),
created_at=todo_doc["created_at"],
updated_at=todo_doc["updated_at"],
)
async def _populate_categories_bulk(self, todo_docs: list[dict]) -> list[TodoResponse]:
"""여러 Todo 문서의 카테고리 정보를 일괄 조회하여 TodoResponse 리스트를 반환한다."""
# 카테고리 ID 수집
cat_ids = set()
for doc in todo_docs:
cat_id = doc.get("category_id")
if cat_id:
try:
cat_ids.add(ObjectId(cat_id))
except Exception:
pass
# 카테고리 일괄 조회
cat_map: dict[str, dict] = {}
if cat_ids:
async for cat in self.categories.find({"_id": {"$in": list(cat_ids)}}):
cat_map[str(cat["_id"])] = cat
# TodoResponse 변환
result = []
for doc in todo_docs:
cat_id = str(doc["category_id"]) if doc.get("category_id") else None
cat = cat_map.get(cat_id) if cat_id else None
result.append(TodoResponse(
id=str(doc["_id"]),
title=doc["title"],
content=doc.get("content"),
completed=doc["completed"],
priority=doc["priority"],
category_id=cat_id,
category_name=cat["name"] if cat else None,
category_color=cat["color"] if cat else None,
tags=doc.get("tags", []),
start_date=doc.get("start_date"),
due_date=doc.get("due_date"),
attachments=doc.get("attachments", []),
created_at=doc["created_at"],
updated_at=doc["updated_at"],
))
return result
# --- CRUD ---
async def create_todo(self, data: TodoCreate) -> TodoResponse:
"""F-001: 할일 생성"""
# category_id 존재 검증
if data.category_id:
cat = await self.categories.find_one({"_id": ObjectId(data.category_id)})
if not cat:
raise HTTPException(status_code=404, detail="카테고리를 찾을 수 없습니다")
now = datetime.now(timezone.utc)
doc = {
"title": data.title,
"content": data.content,
"completed": False,
"priority": data.priority.value,
"category_id": data.category_id,
"tags": data.tags,
"start_date": data.start_date,
"due_date": data.due_date,
"created_at": now,
"updated_at": now,
}
result = await self.collection.insert_one(doc)
doc["_id"] = result.inserted_id
await self._invalidate_cache()
return await self._populate_category(doc)
async def list_todos(
self,
page: int = 1,
limit: int = 20,
completed: Optional[bool] = None,
category_id: Optional[str] = None,
priority: Optional[str] = None,
tag: Optional[str] = None,
sort: str = "created_at",
order: str = "desc",
) -> TodoListResponse:
"""F-002: 할일 목록 조회 (필터, 정렬, 페이지네이션)"""
# 필터 조건 구성 (AND)
query: dict = {}
if completed is not None:
query["completed"] = completed
if category_id:
query["category_id"] = category_id
if priority:
query["priority"] = priority
if tag:
query["tags"] = tag
# 정렬 조건 구성
sort_direction = 1 if order == "asc" else -1
if sort == "priority":
# priority 정렬 시 매핑 사용 (MongoDB에선 문자열 정렬이므로 보조 필드 필요)
# 대안: aggregation pipeline 사용
pipeline = []
if query:
pipeline.append({"$match": query})
pipeline.append({
"$addFields": {
"priority_order": {
"$switch": {
"branches": [
{"case": {"$eq": ["$priority", "high"]}, "then": 0},
{"case": {"$eq": ["$priority", "medium"]}, "then": 1},
{"case": {"$eq": ["$priority", "low"]}, "then": 2},
],
"default": 1,
}
}
}
})
pipeline.append({"$sort": {"priority_order": sort_direction, "created_at": -1}})
# 총 개수 조회
count_pipeline = list(pipeline) + [{"$count": "total"}]
count_result = await self.collection.aggregate(count_pipeline).to_list(1)
total = count_result[0]["total"] if count_result else 0
# 페이지네이션
skip = (page - 1) * limit
pipeline.append({"$skip": skip})
pipeline.append({"$limit": limit})
pipeline.append({"$project": {"priority_order": 0}})
docs = await self.collection.aggregate(pipeline).to_list(limit)
else:
# 일반 정렬
sort_field = sort if sort in ("created_at", "due_date") else "created_at"
sort_spec = [(sort_field, sort_direction)]
total = await self.collection.count_documents(query)
skip = (page - 1) * limit
cursor = self.collection.find(query).sort(sort_spec).skip(skip).limit(limit)
docs = await cursor.to_list(limit)
total_pages = math.ceil(total / limit) if limit > 0 else 0
items = await self._populate_categories_bulk(docs)
return TodoListResponse(
items=items,
total=total,
page=page,
limit=limit,
total_pages=total_pages,
)
async def get_todo(self, todo_id: str) -> TodoResponse:
"""F-003: 할일 상세 조회"""
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
doc = await self.collection.find_one({"_id": ObjectId(todo_id)})
if not doc:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
return await self._populate_category(doc)
async def update_todo(self, todo_id: str, data: TodoUpdate) -> TodoResponse:
"""F-004: 할일 수정 (Partial Update)"""
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
# 변경 필드만 추출
update_data = data.model_dump(exclude_unset=True)
if not update_data:
raise HTTPException(status_code=400, detail="수정할 필드가 없습니다")
# category_id 존재 검증
if "category_id" in update_data and update_data["category_id"] is not None:
cat = await self.categories.find_one({"_id": ObjectId(update_data["category_id"])})
if not cat:
raise HTTPException(status_code=404, detail="카테고리를 찾을 수 없습니다")
# priority enum -> string 변환
if "priority" in update_data and update_data["priority"] is not None:
update_data["priority"] = update_data["priority"].value if hasattr(update_data["priority"], "value") else update_data["priority"]
update_data["updated_at"] = datetime.now(timezone.utc)
result = await self.collection.find_one_and_update(
{"_id": ObjectId(todo_id)},
{"$set": update_data},
return_document=True,
)
if not result:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
await self._invalidate_cache()
return await self._populate_category(result)
async def delete_todo(self, todo_id: str) -> None:
"""F-005: 할일 삭제"""
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
result = await self.collection.delete_one({"_id": ObjectId(todo_id)})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
# 첨부파일 삭제
from app.services.file_service import FileService
file_service = FileService(self.collection.database)
await file_service.delete_all_todo_files(todo_id)
await self._invalidate_cache()
async def toggle_todo(self, todo_id: str) -> ToggleResponse:
"""F-006: 완료 토글"""
if not ObjectId.is_valid(todo_id):
raise HTTPException(status_code=422, detail="유효하지 않은 ID 형식입니다")
doc = await self.collection.find_one({"_id": ObjectId(todo_id)})
if not doc:
raise HTTPException(status_code=404, detail="할일을 찾을 수 없습니다")
new_completed = not doc["completed"]
await self.collection.update_one(
{"_id": ObjectId(todo_id)},
{"$set": {
"completed": new_completed,
"updated_at": datetime.now(timezone.utc),
}},
)
await self._invalidate_cache()
return ToggleResponse(id=todo_id, completed=new_completed)
# --- 일괄 작업 ---
async def batch_action(self, data: BatchRequest) -> BatchResponse:
"""F-019~F-021: 일괄 작업 분기"""
if data.action == "complete":
return await self._batch_complete(data.ids)
elif data.action == "delete":
return await self._batch_delete(data.ids)
elif data.action == "move_category":
return await self._batch_move_category(data.ids, data.category_id)
else:
raise HTTPException(status_code=400, detail="지원하지 않는 작업입니다")
async def _batch_complete(self, ids: list[str]) -> BatchResponse:
"""F-019: 일괄 완료"""
object_ids = [ObjectId(id_str) for id_str in ids]
result = await self.collection.update_many(
{"_id": {"$in": object_ids}},
{"$set": {
"completed": True,
"updated_at": datetime.now(timezone.utc),
}},
)
await self._invalidate_cache()
return BatchResponse(
action="complete",
processed=result.modified_count,
failed=len(ids) - result.modified_count,
)
async def _batch_delete(self, ids: list[str]) -> BatchResponse:
"""F-020: 일괄 삭제"""
object_ids = [ObjectId(id_str) for id_str in ids]
result = await self.collection.delete_many({"_id": {"$in": object_ids}})
await self._invalidate_cache()
return BatchResponse(
action="delete",
processed=result.deleted_count,
failed=len(ids) - result.deleted_count,
)
async def _batch_move_category(self, ids: list[str], category_id: Optional[str]) -> BatchResponse:
"""F-021: 일괄 카테고리 변경"""
# category_id 검증 (null이 아닌 경우)
if category_id is not None:
if not ObjectId.is_valid(category_id):
raise HTTPException(status_code=422, detail="유효하지 않은 카테고리 ID 형식입니다")
cat = await self.categories.find_one({"_id": ObjectId(category_id)})
if not cat:
raise HTTPException(status_code=404, detail="카테고리를 찾을 수 없습니다")
object_ids = [ObjectId(id_str) for id_str in ids]
result = await self.collection.update_many(
{"_id": {"$in": object_ids}},
{"$set": {
"category_id": category_id,
"updated_at": datetime.now(timezone.utc),
}},
)
await self._invalidate_cache()
return BatchResponse(
action="move_category",
processed=result.modified_count,
failed=len(ids) - result.modified_count,
)
# --- 태그 ---
async def get_tags(self) -> list[TagInfo]:
"""F-013: 태그 목록 조회 (사용 횟수 포함, 사용 횟수 내림차순)"""
pipeline = [
{"$unwind": "$tags"},
{"$group": {"_id": "$tags", "count": {"$sum": 1}}},
{"$sort": {"count": -1}},
{"$project": {"name": "$_id", "count": 1, "_id": 0}},
]
results = await self.collection.aggregate(pipeline).to_list(None)
return [TagInfo(name=r["name"], count=r["count"]) for r in results]