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