Coverage for skema/rest/morae_proxy.py: 68%

1# -*- coding: utf-8 -*- 

2""" 

3Proxies requests to skema_rs API to provide a unified API. 

4""" 

5 

6 

7from typing import Any, Dict, List, Text 

8from skema.rest.proxies import SKEMA_RS_ADDESS 

9from fastapi import APIRouter, Depends 

10from skema.rest import utils 

11# TODO: replace use of requests with httpx 

12import httpx 

13 

14 

15router = APIRouter() 

16 

17 

18# FIXME: make GrometFunctionModuleCollection a pydantic model via code gen 

19@router.post("/model", summary="Pushes gromet (function network) to the graph database", include_in_schema=False) 

20async def post_model(gromet: Dict[Text, Any], client: httpx.AsyncClient = Depends(utils.get_client)): 

21 res = await client.post(f"{SKEMA_RS_ADDESS}/models", json=gromet) 

22 return res.json() 

23 

24 

25@router.get("/models", summary="Gets function network IDs from the graph database") 

26async def get_models(client: httpx.AsyncClient = Depends(utils.get_client)) -> List[int]: 

27 res = await client.get(f"{SKEMA_RS_ADDESS}/models") 

28 print(f"request: {res}") 

29 return res.json() 

30 

31 

32@router.get("/ping", summary="Status of MORAE service") 

33async def healthcheck(client: httpx.AsyncClient = Depends(utils.get_client)) -> int: 

34 res = await client.get(f"{SKEMA_RS_ADDESS}/ping") 

35 return res.status_code 

36 

37 

38@router.get("/version", summary="Status of MORAE service") 

39async def versioncheck(client: httpx.AsyncClient = Depends(utils.get_client)) -> str: 

40 res = await client.get(f"{SKEMA_RS_ADDESS}/version") 

41 return res.text 

42 

43@router.post("/mathml/decapodes", summary="Gets Decapodes from a list of MathML strings") 

44async def get_decapodes(mathml: List[str], client: httpx.AsyncClient = Depends(utils.get_client)) -> Dict[Text, Any]: 

45 res = await client.put(f"{SKEMA_RS_ADDESS}/mathml/decapodes", json=mathml) 

46 return res.json()