StrictDoc Documentation
strictdoc/server/app.py
Source file coverage
Path:
strictdoc/server/app.py
Lines:
153
Non-empty lines:
122
Non-empty lines covered with requirements:
122 / 122 (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
        " Subscribe to the StrictDoc mailing list for news about features,",
59
        " breaking changes, and other updates:",
60
        " https://groups.io/g/strictdoc",
61
        "",
62
        " Share feedback or report issues:",
63
        " https://github.com/strictdoc-project/strictdoc/issues",
64
    ]
65
 
66
    banner = (
67
        "\n\n"
68
        f"╔{border}╗\n"
69
        + "\n".join(f"║{line.ljust(width)}║" for line in lines)
70
        + f"\n╚{border}╝\n"
71
    )
72
 
73
    LOGGER.info(banner)
74
 
75
 
76
def create_app(*, project_config: ProjectConfig) -> FastAPI:
77
    def lifespan(_: FastAPI) -> Generator[None, None, None]:
78
        print_welcome_message(project_config)
79
        yield
80
 
81
    app = FastAPI(lifespan=lifespan)
82
 
83
    origins = [
84
        "http://localhost",
85
        "http://localhost:8081",
86
        "http://localhost:3000",
87
    ]
88
 
89
    # Uncomment this to enable performance measurements.
90
    @app.middleware("http")
91
    async def add_process_time_header(  # pylint: disable=unused-variable
92
        request: Request, call_next: Callable[[Request], Awaitable[Response]]
93
    ) -> Response:
94
        start_time = time.time()
95
        response: Response = await call_next(request)
96
        time_passed = round(time.time() - start_time, 3)
97
 
98
        request_path = request.url.path
99
        if len(request.url.query) > 0:
100
            request_path += f"?{request.url.query}"
101
 
102
        print(  # noqa: T201
103
            f"PERF:     {request.method} {request_path} {time_passed}s"
104
        )
105
        return response
106
 
107
    app.add_middleware(
108
        CORSMiddleware,
109
        allow_origins=origins,
110
        allow_credentials=True,
111
        allow_methods=["*"],
112
        allow_headers=["*"],
113
    )
114
 
115
    lock_manager = HierarchicalRWLockManager()
116
 
117
    app.include_router(
118
        create_other_router(
119
            project_config=project_config,
120
            lock_manager=lock_manager,
121
        )
122
    )
123
    app.include_router(
124
        create_main_router(
125
            project_config=project_config,
126
            app=app,
127
            lock_manager=lock_manager,
128
        )
129
    )
130
 
131
    return app
132
 
133
 
134
def strictdoc_production_app() -> FastAPI:
135
    register_code_coverage_hook()
136
 
137
    # This is a work-around to allow opening a file created with
138
    # NamedTemporaryFile on Windows.
139
    # See https://stackoverflow.com/a/15235559
140
    def temp_opener(name: str, flag: int, mode: int = 0o777) -> int:
141
        flag |= O_TEMPORARY
142
        return os.open(name, flag, mode)
143
 
144
    path_to_tmp_config = os.environ[SDocServerEnvVariable.PATH_TO_CONFIG]
145
    with open(path_to_tmp_config, "rb", opener=temp_opener) as tmp_config_file:
146
        tmp_config_bytes = tmp_config_file.read()
147
 
148
    project_config = pickle_load(tmp_config_bytes)
149
    assert isinstance(project_config, ProjectConfig), project_config
150
 
151
    return create_app(
152
        project_config=project_config,
153
    )
1 - 153 | entire file