Daily Work MCP Server

from __future__ import annotations import csv import io from datetime import date from typing import List from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from app.core.database import get_session from app.schemas.entries import ( CreateEntryRequest, CreateEntryResponse, CreateTagRequest, EntryRange, EntryResponse, ExportEntriesResponse, ExportFormat, ReminderStatusResponse, SearchEntriesResponse, SummaryItem, SummaryResponse, TagListResponse, TagResponse, ) from app.services.entry_service import EntryService, TagService router = APIRouter() def _entry_to_response(entry, service: EntryService) -> EntryResponse: return EntryResponse( id=entry.id, description=entry.description, createdAt=entry.created_at, tags=[tag.name for tag in entry.tags], summary=service.format_summary(entry), ) @router.post("/entries", response_model=CreateEntryResponse) def create_entry( payload: CreateEntryRequest, session: Session = Depends(get_session), ) -> CreateEntryResponse: service = EntryService(session) try: entry = service.create_entry(description=payload.description, tag_names=payload.tags) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return CreateEntryResponse( id=entry.id, createdAt=entry.created_at, summary=service.format_summary(entry), ) @router.get("/entries", response_model=SearchEntriesResponse) def list_entries( range: EntryRange = EntryRange.daily, target_date: date | None = Query(default=None, alias="date"), tags: List[str] | None = Query(default=None), session: Session = Depends(get_session), ) -> SearchEntriesResponse: service = EntryService(session) entries = service.list_entries( range_type=range.value, target_date=target_date, tag_names=tags, ) return SearchEntriesResponse(items=[_entry_to_response(entry, service) for entry in entries]) @router.get("/entries/search", response_model=SearchEntriesResponse) def search_entries( q: str = Query(..., min_length=1), start_date: date | None = Query(default=None), end_date: date | None = Query(default=None), session: Session = Depends(get_session), ) -> SearchEntriesResponse: service = EntryService(session) entries = service.search_entries(query=q, start_date=start_date, end_date=end_date) return SearchEntriesResponse(items=[_entry_to_response(entry, service) for entry in entries]) @router.get("/entries/export") def export_entries( format: ExportFormat = ExportFormat.json, start_date: date | None = Query(default=None), end_date: date | None = Query(default=None), session: Session = Depends(get_session), ): service = EntryService(session) entries = service.export_entries(start_date=start_date, end_date=end_date) responses = [_entry_to_response(entry, service) for entry in entries] if format == ExportFormat.json: return ExportEntriesResponse(items=responses) buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow(["id", "description", "createdAt", "tags", "summary"]) for item in responses: writer.writerow( [ item.id, item.description, item.createdAt.isoformat(), "|".join(item.tags), item.summary, ] ) buffer.seek(0) return StreamingResponse( buffer, media_type="text/csv", headers={"Content-Disposition": "attachment; filename=entries.csv"}, ) @router.get("/entries/summary", response_model=SummaryResponse) def summarize_entries( days: int = Query(7, ge=1, le=31), session: Session = Depends(get_session), ) -> SummaryResponse: service = EntryService(session) try: summary = service.summarize(days=days) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc items = [SummaryItem(tag=item["tag"], count=item["count"]) for item in summary] return SummaryResponse(items=items) @router.get("/entries/reminder", response_model=ReminderStatusResponse) def reminder_status(session: Session = Depends(get_session)) -> ReminderStatusResponse: service = EntryService(session) return ReminderStatusResponse(hasEntryToday=service.has_entry_today()) @router.get("/tags", response_model=TagListResponse) def list_tags(session: Session = Depends(get_session)) -> TagListResponse: service = TagService(session) tags = service.list_tags() return TagListResponse(items=[TagResponse.model_validate(tag) for tag in tags]) @router.post("/tags", response_model=TagResponse) def create_tag( payload: CreateTagRequest, session: Session = Depends(get_session), ) -> TagResponse: service = TagService(session) try: tag = service.create_tag(name=payload.name, description=payload.description) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return TagResponse.model_validate(tag)