|
| 1 | +from fastapi import FastAPI, Header, HTTPException, Request, status |
| 2 | +from fastapi.exceptions import RequestValidationError |
| 3 | +from fastapi.responses import JSONResponse |
| 4 | +from fastapi.encoders import jsonable_encoder |
| 5 | +import uvicorn |
| 6 | +from typing import Dict, Optional, List |
| 7 | +from linuxforhealth.x12.config import get_x12_api_config, X12ApiConfig |
| 8 | +from linuxforhealth.x12.io import X12SegmentReader, X12ModelReader |
| 9 | +from linuxforhealth.x12.parsing import X12ParseException |
| 10 | +from pydantic import ValidationError, BaseModel, Field |
| 11 | + |
| 12 | +app = FastAPI() |
| 13 | + |
| 14 | + |
| 15 | +@app.exception_handler(RequestValidationError) |
| 16 | +async def request_validation_handler(request: Request, exc: RequestValidationError): |
| 17 | + return JSONResponse( |
| 18 | + status_code=status.HTTP_400_BAD_REQUEST, |
| 19 | + content=jsonable_encoder( |
| 20 | + {"detail": "Invalid request. Expected {'x12': <x12 message string>}"} |
| 21 | + ), |
| 22 | + ) |
| 23 | + |
| 24 | + |
| 25 | +class X12Request(BaseModel): |
| 26 | + """ |
| 27 | + The X12 Request object |
| 28 | + """ |
| 29 | + |
| 30 | + x12: str = Field(description="The X12 payload to process, conveyed as a string") |
| 31 | + |
| 32 | + class Config: |
| 33 | + schema_extra = { |
| 34 | + "example": { |
| 35 | + "x12": "ISA*03*9876543210*01*9876543210*30*000000005 *30*12345 *131031*1147*^*00501*000000907*1*T*:~GS*HS*000000005*54321*20131031*1147*1*X*005010X279A1~ST*270*1234*005010X279A1~BHT*0022*13*10001234*20060501*1319~HL*1**20*1~NM1*PR*2*ABC COMPANY*****PI*842610001~HL*2*1*21*1~NM1*1P*2*BONE AND JOINT CLINIC*****SV*2000035~HL*3*2*22*0~TRN*1*93175-012547*9877281234~NM1*IL*1*SMITH*ROBERT****MI*11122333301~DMG*D8*19430519~DTP*291*D8*20060501~EQ*30~SE*13*1234~GE*1*1~IEA*1*000000907~", |
| 36 | + } |
| 37 | + } |
| 38 | + |
| 39 | + |
| 40 | +@app.post("/x12") |
| 41 | +async def post_x12( |
| 42 | + x12_request: X12Request, |
| 43 | + lfh_x12_response: Optional[str] = Header(default="models"), |
| 44 | +) -> List[Dict]: |
| 45 | + """ |
| 46 | + Processes an incoming X12 payload. |
| 47 | +
|
| 48 | + Requests are submitted as: |
| 49 | +
|
| 50 | + { |
| 51 | + "x12": <x12 message string> |
| 52 | + } |
| 53 | +
|
| 54 | + The response payload is a JSON document containing either the "raw" X12 segments, or a rich |
| 55 | + X12 domain model. The response type defaults to the domain model and is configured using the |
| 56 | + LFH-X12-RESPONSE header. Valid values include: "segments" or "models". |
| 57 | +
|
| 58 | + :param x12_request: The X12 request model/object. |
| 59 | + :param lfh_x12_response: A header value used to drive processing. |
| 60 | +
|
| 61 | + :return: The X12 response - List[List] (segments) or List[Dict] (models) |
| 62 | + """ |
| 63 | + if lfh_x12_response.lower() not in ("models", "segments"): |
| 64 | + lfh_x12_response = "models" |
| 65 | + |
| 66 | + try: |
| 67 | + if lfh_x12_response.lower() == "models": |
| 68 | + with X12ModelReader(x12_request.x12) as r: |
| 69 | + api_results = [m.dict() for m in r.models()] |
| 70 | + else: |
| 71 | + with X12SegmentReader(x12_request.x12) as r: |
| 72 | + api_results = [] |
| 73 | + for segment_name, segment in r.segments(): |
| 74 | + segment_data = { |
| 75 | + f"{segment_name}{str(i).zfill(2)}": v |
| 76 | + for i, v in enumerate(segment) |
| 77 | + } |
| 78 | + api_results.append(segment_data) |
| 79 | + |
| 80 | + except (X12ParseException, ValidationError) as error: |
| 81 | + raise HTTPException( |
| 82 | + status_code=status.HTTP_400_BAD_REQUEST, |
| 83 | + detail="Invalid X12 payload. To troubleshoot please run the LFH X12 CLI", |
| 84 | + ) |
| 85 | + else: |
| 86 | + return api_results |
| 87 | + |
| 88 | + |
| 89 | +def run_server(): |
| 90 | + """Launches the API server""" |
| 91 | + config: X12ApiConfig = get_x12_api_config() |
| 92 | + |
| 93 | + uvicorn_params = { |
| 94 | + "app": config.x12_uvicorn_app, |
| 95 | + "host": config.x12_uvicorn_host, |
| 96 | + "port": config.x12_uvicorn_port, |
| 97 | + "reload": config.x12_uvicorn_reload, |
| 98 | + } |
| 99 | + |
| 100 | + uvicorn.run(**uvicorn_params) |
| 101 | + |
| 102 | + |
| 103 | +if __name__ == "__main__": |
| 104 | + run_server() |
0 commit comments