StrictDoc Documentation
strictdoc/server/app.py
Source file coverage
Path:
strictdoc/server/app.py
Lines:
149
Non-empty lines:
118
Non-empty lines covered with requirements:
118 / 118 (100.0%)
Functions:
6
Functions covered by requirements:
6 / 6 (100.0%)
1
"""
2
@relation(SDOC-SRS-126, scope=file)
3
"""
4
 
5
import logging
6
import os
7
import sys
8
import time
9
from typing import Awaitable, Callable, Generator
10
 
11
from fastapi import FastAPI
12
from fastapi.middleware.cors import CORSMiddleware
13
from starlette.requests import Request
14
from starlette.responses import Response
15
 
16
from strictdoc import __version__
17
from strictdoc.core.project_config import ProjectConfig
18
from strictdoc.helpers.coverage import register_code_coverage_hook
19
from strictdoc.helpers.pickle import pickle_load
20
from strictdoc.server.config import SDocServerEnvVariable
21
from strictdoc.server.helpers.hierarchical_rw_lock_manager import (
22
    HierarchicalRWLockManager,
23
)
24
from strictdoc.server.routers.main_router import create_main_router
25
from strictdoc.server.routers.other_router import create_other_router
26
 
27
# Define O_TEMPORARY for Windows only
28
if sys.platform == "win32":
29
    O_TEMPORARY = os.O_TEMPORARY  # pragma: no cover
30
else:
31
    O_TEMPORARY = 0
32
 
33
 
34
LOGGER = logging.getLogger("uvicorn.error")
35
 
36
 
37
def print_welcome_message(project_config: ProjectConfig) -> None:
38
    strictdoc_version = f"StrictDoc Web Server v{__version__}"
39
 
40
    host = (
41
        project_config.server_host
42
        if project_config.server_host.startswith("http")
43
        else f"http://{project_config.server_host}"
44
    )
45
 
46
    url = f"{host}:{project_config.server_port}"
47
 
48
    width = 72
49
    border = "═" * width
50
 
51
    lines = [
52
        f" {strictdoc_version.center(width - 2)} ",
53
        "",
54
        f" Server URL: {url}",
55
        "",
56
        " Documentation: https://strictdoc.readthedocs.io/",
57
        "",
58
        " Share feedback or report issues:",
59
        " https://github.com/strictdoc-project/strictdoc/issues",
60
    ]
61
 
62
    banner = (
63
        "\n\n"
64
        f"╔{border}╗\n"
65
        + "\n".join(f"║{line.ljust(width)}║" for line in lines)
66
        + f"\n╚{border}╝\n"
67
    )
68
 
69
    LOGGER.info(banner)
70
 
71
 
72
def create_app(*, project_config: ProjectConfig) -> FastAPI:
73
    def lifespan(_: FastAPI) -> Generator[None, None, None]:
74
        print_welcome_message(project_config)
75
        yield
76
 
77
    app = FastAPI(lifespan=lifespan)
78
 
79
    origins = [
80
        "http://localhost",
81
        "http://localhost:8081",
82
        "http://localhost:3000",
83
    ]
84
 
85
    # Uncomment this to enable performance measurements.
86
    @app.middleware("http")
87
    async def add_process_time_header(  # pylint: disable=unused-variable
88
        request: Request, call_next: Callable[[Request], Awaitable[Response]]
89
    ) -> Response:
90
        start_time = time.time()
91
        response: Response = await call_next(request)
92
        time_passed = round(time.time() - start_time, 3)
93
 
94
        request_path = request.url.path
95
        if len(request.url.query) > 0:
96
            request_path += f"?{request.url.query}"
97
 
98
        print(  # noqa: T201
99
            f"PERF:     {request.method} {request_path} {time_passed}s"
100
        )
101
        return response
102
 
103
    app.add_middleware(
104
        CORSMiddleware,
105
        allow_origins=origins,
106
        allow_credentials=True,
107
        allow_methods=["*"],
108
        allow_headers=["*"],
109
    )
110
 
111
    lock_manager = HierarchicalRWLockManager()
112
 
113
    app.include_router(
114
        create_other_router(
115
            project_config=project_config,
116
            lock_manager=lock_manager,
117
        )
118
    )
119
    app.include_router(
120
        create_main_router(
121
            project_config=project_config,
122
            app=app,
123
            lock_manager=lock_manager,
124
        )
125
    )
126
 
127
    return app
128
 
129
 
130
def strictdoc_production_app() -> FastAPI:
131
    register_code_coverage_hook()
132
 
133
    # This is a work-around to allow opening a file created with
134
    # NamedTemporaryFile on Windows.
135
    # See https://stackoverflow.com/a/15235559
136
    def temp_opener(name: str, flag: int, mode: int = 0o777) -> int:
137
        flag |= O_TEMPORARY
138
        return os.open(name, flag, mode)
139
 
140
    path_to_tmp_config = os.environ[SDocServerEnvVariable.PATH_TO_CONFIG]
141
    with open(path_to_tmp_config, "rb", opener=temp_opener) as tmp_config_file:
142
        tmp_config_bytes = tmp_config_file.read()
143
 
144
    project_config = pickle_load(tmp_config_bytes)
145
    assert isinstance(project_config, ProjectConfig), project_config
146
 
147
    return create_app(
148
        project_config=project_config,
149
    )
1 - 149 | entire file