115 lines
3.2 KiB
Python
115 lines
3.2 KiB
Python
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<order_num>\d{4})/(?P<year>\d{4})/(?P<item_idx>\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)
|