Untitled
unknown
plain_text
4 months ago
5.5 kB
4
Indexable
import sys from fastapi.encoders import jsonable_encoder sys.path.append(".") import os os.environ["XNODE_NAVI_APP"] = "navi_prompt" # Make these changes as necessary from navi_lib.utils.logging import logger from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from fastapi import FastAPI, HTTPException, status, WebSocket, Body, BackgroundTasks, Request, WebSocketDisconnect from fastapi.exceptions import HTTPException from fastapi import FastAPI, HTTPException, status, WebSocket from routes.router import all_routes from services.conversation.conversation_svc import get_bot_response from navi_lib.schema.prompt import * from fastapi import APIRouter, Body, BackgroundTasks, Request, WebSocket, WebSocketDisconnect, FastAPI from services.conversation.conversation_svc import get_bot_response from navi_lib.schema.prompt import * from websocket_connection_manager import WebSocketConnectionManager # Import your EpicWrapper class from epic_wrapper import EpicWrapper description = """ Navi REST calls """ app = FastAPI( docs_url="/docs", openapi_url="/docs/openapi.json", title="Navi - An AI Enabed X-Node Bot", description=description, version="v1", openapi_tags=[] ) app.add_middleware( CORSMiddleware, allow_origins=["*"], # for now allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(all_routes) @app.get("/healthcheck", status_code=status.HTTP_200_OK) async def healthcheck(): """Endpoint to check if the application is up and running and the database is accessible.""" try: return JSONResponse({"status": "OK"}) except Exception as ex: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE , detail='Connection Unavailable') from ex @app.get("/") def read_root(): return {"message": "Hello, i'm Navi Prompt!"} @app.exception_handler(HTTPException) async def validation_exception_handler(request, exc): return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}), ) @app.exception_handler(Exception) async def general_exception_handler(request, exc): return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}), ) @app.websocket("/stream_bot_response/{user_id}") async def bot_response(websocket: WebSocket, user_id: str): await websocket_manager.connect(websocket, user_id) async for data in websocket.iter_json(): parameter_list = ["conversation_history", "user_message", "conversation_type", "conversation_id", "email", "is_navi_active", "agent_team_type", "access_token", "participants"] missing_keys = set(parameter_list) - set(data.keys()) if len(missing_keys) > 0: logger.error(f"missing keys in request body {missing_keys}") error_res = { "usr_msg_response": f"missing keys in request body {missing_keys}", "stream_status": "error", } if "conversation_id" in data.keys(): error_res["conversation_id"] = data['conversation_id'] if "participants" in data.keys(): error_res["participants"] = data['participants'] else: return await websocket_manager.broadcast(error_res, type_='json') else: await get_bot_response(msgs=data, websocket_manager=websocket_manager) websocket_manager = WebSocketConnectionManager() @app.websocket("/stream_bot_response_group/{user_id}") async def bot_response(websocket: WebSocket, user_id: str): await websocket_manager.connect(websocket, user_id) async for data in websocket.iter_json(): await get_bot_response(msgs=data, websocket_manager=websocket_manager) # ------------------- # New code begins here # ------------------- # Instantiate the EpicWrapper class epic_api = EpicWrapper(private_key_file="privatekey.pem", client_id="6d3a397b-a801-4734-9c22-6c93aed5e00a") @app.get("/patient_data_with_id") def patient_data_with_id(patient_id: str): """ Get patient data using an existing patient_id. """ try: data = epic_api.get_data(patient_id=patient_id) if data == "No patient data available": raise HTTPException(status_code=404, detail="Patient not found") return JSONResponse(content={"data": data}, status_code=200) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/patient_data_without_id") def patient_data_without_id( first_name: str, family_name: str, birthdate: str ): """ Get patient data using first_name, family_name, and birthdate (no patient_id provided). """ try: data = epic_api.get_data(first_name=first_name, family_name=family_name, birthdate=birthdate) if data == "No patient data available": raise HTTPException(status_code=404, detail="Patient not found") return JSONResponse(content={"data": data}, status_code=200) except Exception as e: raise HTTPException(status_code=500, detail=str(e))
Editor is loading...
Leave a Comment