import re import os from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, Query, Request from fastapi.responses import JSONResponse from pydantic import BaseModel from mayo.exceptions import ( MayoAuthError, MayoConnectionError, MayoError, MayoOrderNotFound, MayoParseError, MayoResponseError, MayoSessionError, ) from mayo import MayoClient, MayoResponse app = FastAPI(title="Mayo Integration API") MAYO_BASE_URL = os.getenv("MAYO_BASE_URL", "http://10.8.0.6/mayo2") MAYO_LOGIN = os.getenv("MAYO_LOGIN", "nowakb") MAYO_PASSWORD = os.getenv("MAYO_PASSWORD", "def") ORDER_ID_RE = re.compile( r"^(?P\d{4})/(?P\d{4})/(?P\d{1,2})$" ) class OrderId(BaseModel): order_num: str year: str item_idx: str @app.exception_handler(MayoOrderNotFound) async def mayo_order_not_found_handler(request: Request, exc: MayoOrderNotFound): return JSONResponse(status_code=404, content={"detail": str(exc)}) @app.exception_handler(MayoConnectionError) async def mayo_connection_error_handler(request: Request, exc: MayoConnectionError): return JSONResponse( status_code=503, content={"detail": "Mayo service is currently unavailable."}, ) @app.exception_handler(MayoAuthError) async def mayo_auth_error_handler(request: Request, exc: MayoAuthError): return JSONResponse( status_code=502, content={"detail": "Authentication with Mayo failed."}, ) @app.exception_handler(MayoSessionError) async def mayo_session_error_handler(request: Request, exc: MayoSessionError): return JSONResponse( status_code=502, content={"detail": "Mayo session is invalid or has expired."}, ) @app.exception_handler(MayoResponseError) async def mayo_response_error_handler(request: Request, exc: MayoResponseError): return JSONResponse( status_code=502, content={"detail": "Mayo returned an unexpected response."}, ) @app.exception_handler(MayoParseError) async def mayo_parse_error_handler(request: Request, exc: MayoParseError): return JSONResponse( status_code=502, content={"detail": "Failed to parse data returned by Mayo."}, ) @app.exception_handler(MayoError) async def mayo_error_handler(request: Request, exc: MayoError): return JSONResponse( status_code=500, content={"detail": "Unexpected Mayo integration error."}, ) def parse_order_id( order_id: Annotated[str, Query(description="Format: XXXX/YYYY/ZZ")], ) -> OrderId: match = ORDER_ID_RE.fullmatch(order_id) if not match: raise HTTPException( status_code=422, detail="order_id must have format XXXX/YYYY/ZZ", ) return OrderId(**match.groupdict()) async def get_mayo_client(): async with MayoClient(MAYO_BASE_URL, MAYO_LOGIN, MAYO_PASSWORD) as client: yield client @app.get("/orders/", response_model=MayoResponse) async def get_order(order: Annotated[OrderId, Depends(parse_order_id)], client: MayoClient = Depends(get_mayo_client)): return await client.get_full_order_data(order.order_num, order.year, order.item_idx) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8001)