Skip to content

rs_server_frontend/main.md

<< Back to index

The frontend application.

Frontend

The frontend application.

Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class Frontend:
    """The frontend application."""

    def __init__(self):
        """Create a frontend application.

        The frontend serves the rs-server REST API documentation.
        This documentation is an openapi specification loaded from a json file.
        This file location is given by the RSPY_OPENAPI_FILE environment variable.

        This file is loaded during the frontend application initialization
        and is kept in memory cache for the entire life of the application.

        A specific FrontendFailed exception is raised if the openapi loading failed.
        """

        # For cluster deployment: override the swagger /docs URL from an environment variable.
        # Also set the openapi.json URL under the same path.
        try:
            docs_url = env["RSPY_DOCS_URL"].strip("/")
            docs_params = {"docs_url": f"/{docs_url}", "openapi_url": f"/{docs_url}/openapi.json"}
        except KeyError:
            docs_params = {}

        try:
            self.openapi_spec: dict = self.load_openapi_spec()
            self.app: FastAPI = FastAPI(
                title="RS-Server",
                version=__version__,
                **docs_params,  # type: ignore
                # Same hardcoded values than in the apikey manager
                # (they don't appear in the openapi.json)
                swagger_ui_init_oauth={
                    "clientId": "(this value is not used)",
                    "appName": "API-Key Manager",
                    "usePkceWithAuthorizationCodeGrant": True,
                },
            )
            self.app.openapi = self.get_openapi  # type: ignore
        except BaseException as e:
            raise FrontendFailed("Unable to serve openapi specification.") from e

        # include_in_schema=False: hide this endpoint from the swagger
        @self.app.get("/health", response_model=HealthSchema, name="Check service health", include_in_schema=False)
        async def health() -> HealthSchema:
            """
            Always return a flag set to 'true' when the service is up and running.
            \f
            Otherwise this code won't be run anyway and the caller will have other sorts of errors.
            """
            return HealthSchema(healthy=True)

    @staticmethod
    def load_openapi_spec() -> dict:
        """Load the openapi specification.

        The openapi is loaded from a json file.
        This json file location is given by the environment variable RSPY_OPENAPI_FILE.

        An IOError is raised in case of errors during the file reading.
        A ValueError is raised in case of errors during the json parsing.

        Returns:
            the loaded openapi specification

        """
        openapi_location = os.getenv("RSPY_OPENAPI_FILE", "")
        try:
            with open(openapi_location, encoding="utf-8") as file:
                return json.load(file)
        except (FileNotFoundError, OSError) as e:
            raise type(e)(
                f"openapi spec was not found at {openapi_location!r}. "
                "Is the 'RSPY_OPENAPI_FILE' environment variable correctly set ?",
            ) from e
        except ValueError as e:
            raise ValueError(
                f"openapi spec was found at {openapi_location!r} but the file is not valid.",
            ) from e

    def get_openapi(self) -> dict:
        """Returns the openapi specification.

        Returns:
            the openapi specification as a dict.
        """
        return self.openapi_spec

__init__()

Create a frontend application.

The frontend serves the rs-server REST API documentation. This documentation is an openapi specification loaded from a json file. This file location is given by the RSPY_OPENAPI_FILE environment variable.

This file is loaded during the frontend application initialization and is kept in memory cache for the entire life of the application.

A specific FrontendFailed exception is raised if the openapi loading failed.

Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
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
def __init__(self):
    """Create a frontend application.

    The frontend serves the rs-server REST API documentation.
    This documentation is an openapi specification loaded from a json file.
    This file location is given by the RSPY_OPENAPI_FILE environment variable.

    This file is loaded during the frontend application initialization
    and is kept in memory cache for the entire life of the application.

    A specific FrontendFailed exception is raised if the openapi loading failed.
    """

    # For cluster deployment: override the swagger /docs URL from an environment variable.
    # Also set the openapi.json URL under the same path.
    try:
        docs_url = env["RSPY_DOCS_URL"].strip("/")
        docs_params = {"docs_url": f"/{docs_url}", "openapi_url": f"/{docs_url}/openapi.json"}
    except KeyError:
        docs_params = {}

    try:
        self.openapi_spec: dict = self.load_openapi_spec()
        self.app: FastAPI = FastAPI(
            title="RS-Server",
            version=__version__,
            **docs_params,  # type: ignore
            # Same hardcoded values than in the apikey manager
            # (they don't appear in the openapi.json)
            swagger_ui_init_oauth={
                "clientId": "(this value is not used)",
                "appName": "API-Key Manager",
                "usePkceWithAuthorizationCodeGrant": True,
            },
        )
        self.app.openapi = self.get_openapi  # type: ignore
    except BaseException as e:
        raise FrontendFailed("Unable to serve openapi specification.") from e

    # include_in_schema=False: hide this endpoint from the swagger
    @self.app.get("/health", response_model=HealthSchema, name="Check service health", include_in_schema=False)
    async def health() -> HealthSchema:
        """
        Always return a flag set to 'true' when the service is up and running.
        \f
        Otherwise this code won't be run anyway and the caller will have other sorts of errors.
        """
        return HealthSchema(healthy=True)

get_openapi()

Returns the openapi specification.

Returns:

Type Description
dict

the openapi specification as a dict.
Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
116
117
118
119
120
121
122
def get_openapi(self) -> dict:
    """Returns the openapi specification.

    Returns:
        the openapi specification as a dict.
    """
    return self.openapi_spec

load_openapi_spec() staticmethod

Load the openapi specification.

The openapi is loaded from a json file. This json file location is given by the environment variable RSPY_OPENAPI_FILE.

An IOError is raised in case of errors during the file reading. A ValueError is raised in case of errors during the json parsing.

Returns:

Type Description
dict

the loaded openapi specification
Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@staticmethod
def load_openapi_spec() -> dict:
    """Load the openapi specification.

    The openapi is loaded from a json file.
    This json file location is given by the environment variable RSPY_OPENAPI_FILE.

    An IOError is raised in case of errors during the file reading.
    A ValueError is raised in case of errors during the json parsing.

    Returns:
        the loaded openapi specification

    """
    openapi_location = os.getenv("RSPY_OPENAPI_FILE", "")
    try:
        with open(openapi_location, encoding="utf-8") as file:
            return json.load(file)
    except (FileNotFoundError, OSError) as e:
        raise type(e)(
            f"openapi spec was not found at {openapi_location!r}. "
            "Is the 'RSPY_OPENAPI_FILE' environment variable correctly set ?",
        ) from e
    except ValueError as e:
        raise ValueError(
            f"openapi spec was found at {openapi_location!r} but the file is not valid.",
        ) from e

FrontendFailed

Bases: BaseException

Exception raised if the frontend initialization failed.

Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
26
27
class FrontendFailed(BaseException):
    """Exception raised if the frontend initialization failed."""

HealthSchema

Bases: BaseModel

Health status flag.

Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
30
31
32
33
class HealthSchema(BaseModel):
    """Health status flag."""

    healthy: bool

start_app()

Start the starlette app.

Factory function that starts the application.

Returns:

Type Description
FastAPI

the initialized application
Source code in docs/rs-server/services/frontend/rs_server_frontend/main.py 
125
126
127
128
129
130
131
132
133
134
def start_app() -> FastAPI:
    """Start the starlette app.

    Factory function that starts the application.

    Returns:
        the initialized application

    """
    return Frontend().app