Skip to content

Service

Service

A service wrapper which registers routes and starts a FastAPI service

Parameters: endpoints (Dict[str, Enpoint]): the list of endpoints to register, must be a dictionary of Endpoint objects. Should have human-readable keys e.g. ["info", "service_mount"]

Source code in healthchain/service/service.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class Service:
    """
    A service wrapper which registers routes and starts a FastAPI service

    Parameters:
    endpoints (Dict[str, Enpoint]): the list of endpoints to register, must be a dictionary
    of Endpoint objects. Should have human-readable keys e.g. ["info", "service_mount"]

    """

    def __init__(self, endpoints: Dict[str, Endpoint] = None):
        self.app = FastAPI(lifespan=self.lifespan)
        self.endpoints: Endpoint = endpoints

        if self.endpoints is not None:
            self._register_routes()

        # Router to handle stopping the server
        self.stop_router = APIRouter()
        self.stop_router.add_api_route(
            "/shutdown", self._shutdown, methods=["GET"], include_in_schema=False
        )
        self.app.include_router(self.stop_router)

    def _register_routes(self) -> None:
        # TODO: add kwargs
        for endpoint in self.endpoints.values():
            if endpoint.api_protocol == ApiProtocol.soap:
                wsgi_app = start_wsgi(endpoint.function)
                self.app.mount(endpoint.path, WSGIMiddleware(wsgi_app))
            else:
                self.app.add_api_route(
                    endpoint.path,
                    endpoint.function,
                    methods=[endpoint.method],
                    response_model_exclude_none=True,
                )

    @asynccontextmanager
    async def lifespan(self, app: FastAPI):
        self._startup()
        yield
        self._shutdown()

    def _startup(self) -> None:
        healthchain_ascii = r"""

    __  __           ____  __    ________          _
   / / / /__  ____ _/ / /_/ /_  / ____/ /_  ____ _(_)___
  / /_/ / _ \/ __ `/ / __/ __ \/ /   / __ \/ __ `/ / __ \
 / __  /  __/ /_/ / / /_/ / / / /___/ / / / /_/ / / / / /
/_/ /_/\___/\__,_/_/\__/_/ /_/\____/_/ /_/\__,_/_/_/ /_/

"""  # noqa: E501

        colors = ["red", "yellow", "green", "cyan", "blue", "magenta"]
        for i, line in enumerate(healthchain_ascii.split("\n")):
            color = colors[i % len(colors)]
            print(colored(line, color))
        for endpoint in self.endpoints.values():
            print(
                f"{colored('HEALTHCHAIN', 'green')}: {endpoint.method} endpoint at {endpoint.path}/"
            )
        print(
            f"{colored('HEALTHCHAIN', 'green')}: See more details at {colored(self.app.docs_url, 'magenta')}"
        )

    def _shutdown(self):
        """
        Shuts down server
        """
        os.kill(os.getpid(), signal.SIGTERM)
        return JSONResponse(content={"message": "Server is shutting down..."})

    def run(self, config: Dict = None) -> None:
        """
        Starts server on uvicorn.

        Parameters:
        config (Dict): kwargs to pass into uvicorn.

        """
        if config is None:
            config = {}

        uvicorn.run(self.app, **config)

run(config=None)

Starts server on uvicorn.

Parameters: config (Dict): kwargs to pass into uvicorn.

Source code in healthchain/service/service.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def run(self, config: Dict = None) -> None:
    """
    Starts server on uvicorn.

    Parameters:
    config (Dict): kwargs to pass into uvicorn.

    """
    if config is None:
        config = {}

    uvicorn.run(self.app, **config)