Coverage for api/app.py: 84%
113 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1import logging
2from contextlib import asynccontextmanager
4import aiohttp
5import sentry_sdk
6import uvicorn
7from config import ENVIRONMENT, SentryConfig
8from endpoints import organization, users
9from endpoints.admin import ads
10from endpoints.admin.api_keys import api_keys_router
11from endpoints.admin.settings import settings_router
12from endpoints.admin.utils import util_router
13from endpoints.cleanup import cleanup_router
14from endpoints.int1c import get_events, integration_router, updates_statuses
15from endpoints.misc import clients, documents, verifications
16from endpoints.notification_webhook import notifications_router
17from endpoints.orders import (
18 active_orders,
19 archive,
20 bids,
21 buffer_orders,
22 confirmed_orders,
23 exchange_orders,
24)
25from endpoints.orders.appointment import appointment_router
26from endpoints.orders.canceled import canceled_router
27from endpoints.orders.external_orders import scraped_router
28from endpoints.orders.orders import orders_router
29from endpoints.orders.recommendations import recommendations_router
30from endpoints.orders.trash import trash_router
31from endpoints.organization.subsidiaries import subsidiary_router
32from endpoints.reports import report_router
33from endpoints.resources import drivers, trailers, trucks
34from endpoints.statistics import statistics_router
35from endpoints.suggest import suggestions_router
36from errors import app_error_handler, keycloak_error_handler
37from fastapi import Depends, FastAPI
38from fastapi.openapi.utils import get_openapi
39from handlers.authorization.by_api_key import check_access
40from indexes import IDX_VERSION, run_indexes
41from keycloak import idp
42from services.location_connector import geo_connector
43from services.notifications.director import ScrapedNotificationContainer
44from services.opensearch_client import opensearch_adapter
45from services.recommendations import suggestion_api
46from sotrans_fastapi_keycloak import KeycloakError
47from sotrans_models.models.responses import ErrorMessage, ErrorRepr
48from starlette.middleware.cors import CORSMiddleware
49from stubs import scraped_container
51sentry_sdk.init(
52 dsn=SentryConfig.DSN,
53 # Set traces_sample_rate to 1.0 to capture 100%
54 # of transactions for performance monitoring.
55 # We recommend adjusting this value in production.
56 traces_sample_rate=1.0,
57 environment=ENVIRONMENT,
58)
61def startup_event_logger():
62 logger = logging.getLogger("uvicorn.access")
64 stream_handler = logging.StreamHandler()
65 stream_handler.setFormatter(
66 logging.Formatter(
67 "%(asctime)s.%(msecs)03d; %(levelname)s; SOTRANS MAIN; module=%(name)s; %(funcName)s=%(lineno)d; msg=%(message)s",
68 "%Y-%m-%d %H:%M:%S",
69 )
70 )
71 logger.addHandler(stream_handler)
74@asynccontextmanager
75async def lifespan(_: FastAPI):
76 startup_event_logger()
77 timeout = aiohttp.ClientTimeout(connect=2)
78 geo_connector.session = aiohttp.ClientSession()
79 suggestion_api.connector.session = aiohttp.ClientSession(timeout=timeout)
80 yield
81 await suggestion_api.connector.session.close()
82 if idp._async_client:
83 await idp.close_session()
84 await geo_connector.session.close()
85 if hasattr(opensearch_adapter, "client"):
86 await opensearch_adapter.client.close()
89run_indexes(IDX_VERSION)
90app = FastAPI(
91 lifespan=lifespan,
92 responses={
93 500: {"model": ErrorMessage},
94 401: {"model": ErrorMessage},
95 403: {"model": ErrorRepr}, # everywhere where KC_lib used is needed
96 },
97)
98idp.add_swagger_config(app)
100app.get("/updates", dependencies=[Depends(check_access)])(get_events)
101app.patch("/updates", dependencies=[Depends(check_access)])(updates_statuses)
103app.include_router(users.users_router)
104app.include_router(documents.documents_col_router)
105app.include_router(organization.organizations_router)
106app.include_router(drivers.drivers_router)
107app.include_router(trucks.trucks_router)
108app.include_router(trailers.trailers_router)
109app.include_router(documents.documents_router)
110app.include_router(verifications.verifications_router)
111app.include_router(clients.clients_router)
112app.include_router(clients.clients_assignment_router)
113app.include_router(buffer_orders.buffer_orders_router)
114app.include_router(appointment_router)
115app.include_router(exchange_orders.exchange_orders_router)
116app.include_router(bids.bids_router)
117app.include_router(confirmed_orders.confirmed_orders_router)
118app.include_router(active_orders.active_orders_router)
119app.include_router(scraped_router)
120app.include_router(archive.archive_router)
121app.include_router(suggestions_router)
122app.include_router(ads.ads_router)
123app.include_router(cleanup_router)
124app.include_router(trash_router)
125app.include_router(recommendations_router)
126app.include_router(statistics_router)
127app.include_router(canceled_router)
128app.include_router(orders_router)
129app.include_router(subsidiary_router)
130app.include_router(api_keys_router)
131app.include_router(settings_router)
132app.include_router(notifications_router)
133app.include_router(report_router)
134app.include_router(util_router)
136app.include_router(integration_router)
137app.add_exception_handler(KeycloakError, keycloak_error_handler)
138app.add_exception_handler(Exception, app_error_handler)
139app.add_middleware(
140 CORSMiddleware,
141 allow_origins=["*"],
142 allow_credentials=True,
143 allow_methods=["*"],
144 allow_headers=["*"],
145)
147scraped_notifications_container = ScrapedNotificationContainer(
148 order_by_clients=[]
149)
150app.dependency_overrides[
151 scraped_container
152] = lambda: scraped_notifications_container
155tags_desc_list = [
156 {"name": "organizations", "description": "Carrier organizations"},
157]
160def custom_openapi():
161 if app.openapi_schema:
162 return app.openapi_schema
163 openapi_schema = get_openapi(
164 title="Sotrans Main Service",
165 version="0.1",
166 description="Sotrans Main Service swagger",
167 routes=app.routes,
168 )
169 openapi_schema["info"]["x-logo"] = {
170 "url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"
171 }
172 openapi_schema["tags"] = tags_desc_list
174 app.openapi_schema = openapi_schema
176 return app.openapi_schema
179app.openapi = custom_openapi
181# uvicorn.run(app)