MCP Weather & Accruals Server

from fastapi import FastAPI, HTTPException from fastapi.encoders import jsonable_encoder from decimal import Decimal from .db import database, metadata, engine from .models import users, accruals from .schemas import UserCreate, User, AccrualCreate, Accrual, AccrualStats, AccrualUpdate from .crud import ( create_user, get_users, create_accrual, get_accruals, get_accrual_stats, update_accrual, delete_accrual, ) from typing import Optional app = FastAPI(title="Accruals API", version="0.2.0") @app.on_event("startup") async def startup(): # создаём таблицы если их нет (sync через SQLAlchemy engine) metadata.create_all(bind=engine) await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() @app.get("/") async def root(): return {"message": "Accruals API server is running"} @app.get("/health") async def health(): """Health endpoint для готовности/жизни сервиса""" try: await database.fetch_one("SELECT 1") except Exception: raise HTTPException(status_code=503, detail="database unavailable") return {"status": "ok"} # ===== Users endpoints ===== @app.post("/users/", response_model=User) async def api_create_user(user: UserCreate): existing = await database.fetch_one(users.select().where(users.c.email == user.email)) if existing: raise HTTPException(status_code=400, detail="Email already registered") created = await create_user(user) return created @app.get("/users/") async def api_get_users(): results = await get_users() return results # ===== Accruals endpoints ===== @app.post("/accruals/") async def api_create_accrual(accrual: AccrualCreate): """Create a new accrual record""" existing = await database.fetch_one( accruals.select().where(accruals.c.id_accrual == accrual.id_accrual) ) if existing: raise HTTPException(status_code=400, detail="Accrual with this id_accrual already exists") created = await create_accrual(accrual) return created @app.put("/accruals/{id_accrual}") async def api_update_accrual(id_accrual: str, accrual: AccrualUpdate): """Update an existing accrual by id_accrual""" existing = await database.fetch_one( accruals.select().where(accruals.c.id_accrual == id_accrual) ) if not existing: raise HTTPException(status_code=404, detail="Accrual not found") updates = accrual.dict(exclude_unset=True) if not updates: raise HTTPException(status_code=400, detail="No fields to update") updated = await update_accrual(id_accrual, updates) if not updated: raise HTTPException(status_code=500, detail="Failed to update accrual") return updated @app.delete("/accruals/{id_accrual}") async def api_delete_accrual(id_accrual: str): """Delete accrual by id_accrual""" existing = await database.fetch_one( accruals.select().where(accruals.c.id_accrual == id_accrual) ) if not existing: raise HTTPException(status_code=404, detail="Accrual not found") await delete_accrual(id_accrual) return {"status": "deleted", "id_accrual": id_accrual} @app.get("/accruals/") async def api_get_accruals( skip: int = 0, limit: int = 100, service_group: Optional[str] = None, sales_platform: Optional[str] = None, accrual_type: Optional[str] = None, ): """Get accruals with optional filtering""" results = await get_accruals( skip=skip, limit=limit, service_group=service_group, sales_platform=sales_platform, accrual_type=accrual_type, ) # `results` are `databases` Record objects which are not JSON-serializable # Convert each Record to a plain dict and ensure types (Decimal) are encoded try: list_of_dicts = [dict(r) for r in results] except Exception: # Fallback: try jsonable_encoder directly (can still fail for unknown types) list_of_dicts = jsonable_encoder(results) return jsonable_encoder(list_of_dicts) @app.get("/stats/summary", response_model=dict) async def api_get_summary_stats( service_group: Optional[str] = None, sales_platform: Optional[str] = None, accrual_type: Optional[str] = None, ): """Get overall statistics for all accruals (with optional filters)""" return await get_accrual_stats( service_group=service_group, sales_platform=sales_platform, accrual_type=accrual_type, ) @app.get("/stats/accruals", response_model=dict) async def api_get_filtered_stats( service_group: Optional[str] = None, sales_platform: Optional[str] = None, accrual_type: Optional[str] = None, ): """Get statistics for accruals with optional filters""" return await get_accrual_stats( service_group=service_group, sales_platform=sales_platform, accrual_type=accrual_type, )