bedrock.endpoints.runner

 1import traceback  # pragma: unit
 2from bedrock.config import internals  # pragma: unit
 3from bedrock.db.alchemy_helper import get_session  # pragma: unit
 4from bedrock.endpoints.response import respond, err  # pragma: unit
 5from bedrock.log import log_config  # pragma: unit
 6from bedrock.exceptions import BedrockException  # pragma: unit
 7from bedrock.config.models import get_models  # pragma: unit
 8
 9log = log_config("runner")  # pragma: unit
10
11# load all models
12get_models(internals["models_location"], internals["models_prefix"])  # pragma: unit
13
14
15def cleanup_session():  # pragma: no cover
16    try:
17        get_session().close()
18        log.info("Database session closed successfully")
19    except Exception as e:
20        log.warning(f"Session close failed or was already closed: {e}")
21
22
23def handle_with_endpoint(endpoint, event, context, is_local=False):  # pragma: unit
24    try:
25        response = respond(*endpoint._handle_with_context(event, context))
26    except BedrockException as e:
27        exception = traceback.format_exc()
28        log.error(exception, exc_info=True)
29        response = err(e, exception, event=event)
30    except Exception:
31        exception = traceback.format_exc()
32        log.error(exception, exc_info=True)
33        response = err("An unexpected error has occurred", exception, event=event)
34    finally:
35        if not is_local:
36            cleanup_session()
37
38    return response