File size: 2,505 Bytes
c7ee8c4 b26b1fd c7ee8c4 1b0a7c7 c7ee8c4 1b0a7c7 b26b1fd c7ee8c4 b26b1fd 1b0a7c7 b26b1fd c7ee8c4 1b0a7c7 c7ee8c4 8cceab7 c7ee8c4 3b9a486 c7ee8c4 1b0a7c7 c7ee8c4 cc7e275 c7ee8c4 cc7e275 c7ee8c4 cc7e275 1b0a7c7 cc7e275 1b0a7c7 c7ee8c4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
import time
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from app.dependencies import get_api_logger, get_category_service
from app.schemas.categories import CategorizeRequest, CategorizeResponse
from app.services.autocategorizer import AutoCategoryService
from app.services.api_logger import ApiLogger
router = APIRouter()
@router.post("/categorize", response_model=CategorizeResponse, summary="Categorize a transaction note")
async def categorize_transaction(
payload: CategorizeRequest,
service: AutoCategoryService = Depends(get_category_service),
api_logger: ApiLogger = Depends(get_api_logger),
) -> CategorizeResponse | JSONResponse:
started_at = time.monotonic()
try:
result = await service.categorize(payload.notes, payload.user_id)
await api_logger.log_categorization(
name="Auto Expense Categorization",
status="success",
response_time=time.monotonic() - started_at,
user_id=payload.user_id,
)
return CategorizeResponse(status="success", data=result, message="Categorization logged in database.")
except HTTPException as exc:
await api_logger.log_categorization(
name="Auto Expense Categorization",
status="fail",
response_time=time.monotonic() - started_at,
user_id=payload.user_id,
error_message=_normalize_error_detail(exc.detail),
)
return JSONResponse(
status_code=exc.status_code,
content={"status": "fail", "message": _normalize_error_detail(exc.detail)},
)
except Exception as exc:
error_message = str(exc)
error_type = type(exc).__name__
await api_logger.log_categorization(
name="Auto Expense Categorization",
status="fail",
response_time=time.monotonic() - started_at,
user_id=payload.user_id,
error_message=f"{error_type}: {error_message}",
)
# Include error details for debugging (in production, you might want to hide this)
return JSONResponse(
status_code=500,
content={
"status": "fail",
"message": f"Internal server error: {error_type} - {error_message}",
},
)
def _normalize_error_detail(detail: Any) -> str:
if isinstance(detail, str):
return detail
return str(detail)
|