Source code for

from import DatabaseUser
from cobald.daemon import service
from cobald.daemon.plugins import yaml_tag
import threading

from uvicorn.config import Config
from uvicorn.server import Server

from functools import lru_cache
from typing import List, Optional
import asyncio

[docs]@service(flavour=asyncio) @yaml_tag(eager=True) class RestService(object): def __init__( self, users: List = None, **fast_api_args, ): self._users = users or [] # necessary to avoid that the TARDIS' logger configuration is overwritten! if "log_config" not in fast_api_args: fast_api_args["log_config"] = None self._config = Config("", **fast_api_args)
[docs] @lru_cache(maxsize=16) def get_user(self, user_name: str) -> Optional[DatabaseUser]: for user in self._users: if user["user_name"] == user_name: return DatabaseUser(**user) return None
[docs] async def run(self) -> None: server = Server(config=self._config) await server.serve() # See # The server has shut down after receiving *and suppressing* a signal. # Explicitly raise the corresponding shutdown exception as a workaround. if server.should_exit and threading.current_thread() is threading.main_thread(): raise KeyboardInterrupt