Skip to content

Lifetime

register_shutdown_event(app)

Actions to run on application's shutdown.

Parameters:

Name Type Description Default
app FastAPI

fastAPI application.

 required

Returns:

Type Description
Callable[[], Awaitable[None]]

function that actually performs actions.
Source code in hestia/web/lifetime.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def register_shutdown_event(
    app: FastAPI,
) -> Callable[[], Awaitable[None]]:  # pragma: no cover
    """
    Actions to run on application's shutdown.

    :param app: fastAPI application.
    :return: function that actually performs actions.
    """

    @app.on_event("shutdown")
    async def _shutdown() -> None:  # noqa: WPS430
        if not broker.is_worker_process:
            await broker.shutdown()
        await app.state.db_engine.dispose()

        await shutdown_redis(app)
        await shutdown_rabbit(app)
        stop_opentelemetry(app)
        pass  # noqa: WPS420

    return _shutdown

register_startup_event(app)

Actions to run on application startup.

This function uses fastAPI app to store data in the state, such as db_engine.

Parameters:

Name Type Description Default
app FastAPI

the fastAPI application.

 required

Returns:

Type Description
Callable[[], Awaitable[None]]

function that actually performs actions.
Source code in hestia/web/lifetime.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def register_startup_event(
    app: FastAPI,
) -> Callable[[], Awaitable[None]]:  # pragma: no cover
    """
    Actions to run on application startup.

    This function uses fastAPI app to store data
    in the state, such as db_engine.

    :param app: the fastAPI application.
    :return: function that actually performs actions.
    """

    @app.on_event("startup")
    async def _startup() -> None:  # noqa: WPS430
        if not broker.is_worker_process:
            await broker.startup()
        _setup_db(app)
        setup_opentelemetry(app)
        init_redis(app)
        init_rabbit(app)
        setup_prometheus(app)

    return _startup

setup_opentelemetry(app)

Enables opentelemetry instrumentation.

Parameters:

Name Type Description Default
app FastAPI

current application.

 required
Source code in hestia/web/lifetime.py 
 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
def setup_opentelemetry(app: FastAPI) -> None:  # pragma: no cover
    """
    Enables opentelemetry instrumentation.

    :param app: current application.
    """
    if not settings.opentelemetry_endpoint:
        return

    tracer_provider = TracerProvider(
        resource=Resource(
            attributes={
                SERVICE_NAME: "hestia",
                TELEMETRY_SDK_LANGUAGE: "python",
                DEPLOYMENT_ENVIRONMENT: settings.environment,
            },
        ),
    )

    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(
                endpoint=settings.opentelemetry_endpoint,
                insecure=True,
            ),
        ),
    )

    excluded_endpoints = [
        app.url_path_for("health_check"),
        app.url_path_for("openapi"),
        app.url_path_for("swagger_ui_html"),
        app.url_path_for("swagger_ui_redirect"),
        app.url_path_for("redoc_html"),
        "/metrics",
    ]

    FastAPIInstrumentor().instrument_app(
        app,
        tracer_provider=tracer_provider,
        excluded_urls=",".join(excluded_endpoints),
    )
    RedisInstrumentor().instrument(
        tracer_provider=tracer_provider,
    )
    SQLAlchemyInstrumentor().instrument(
        tracer_provider=tracer_provider,
        engine=app.state.db_engine.sync_engine,
    )
    AioPikaInstrumentor().instrument(
        tracer_provider=tracer_provider,
    )

    set_tracer_provider(tracer_provider=tracer_provider)

setup_prometheus(app)

Enables prometheus integration.

Parameters:

Name Type Description Default
app FastAPI

current application.

 required
Source code in hestia/web/lifetime.py 
119
120
121
122
123
124
125
126
127
def setup_prometheus(app: FastAPI) -> None:  # pragma: no cover
    """
    Enables prometheus integration.

    :param app: current application.
    """
    PrometheusFastApiInstrumentator(should_group_status_codes=False).instrument(
        app,
    ).expose(app, should_gzip=True, name="prometheus_metrics")

stop_opentelemetry(app)

Disables opentelemetry instrumentation.

Parameters:

Name Type Description Default
app FastAPI

current application.

 required
Source code in hestia/web/lifetime.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
def stop_opentelemetry(app: FastAPI) -> None:  # pragma: no cover
    """
    Disables opentelemetry instrumentation.

    :param app: current application.
    """
    if not settings.opentelemetry_endpoint:
        return

    FastAPIInstrumentor().uninstrument_app(app)
    RedisInstrumentor().uninstrument()
    SQLAlchemyInstrumentor().uninstrument()
    AioPikaInstrumentor().uninstrument()