from __future__ import annotations import logging import os import uuid from os import path from typing import Optional from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration from sentry_sdk.utils import BadDsn, Dsn from smelog.entities import LoggerConfig, SentryConfig from smelog.factory import LoggerFactory from structlog import BoundLogger from src.config import APP_FULL_NAME, IS_LOCAL, SENTRY_DSN from src.constants import ENV_PROD log: Optional[BoundLogger] = None def _get_version() -> str: """Get version from file. Returns: Version. """ version_file = os.environ.get("VERSION_FILE", "../VERSION") if not path.isfile(version_file): return "unknown" with open(version_file) as f: return f.readline().strip() def _validate_dsn() -> bool: """Validate sentry DSN.""" try: Dsn(SENTRY_DSN) return True except BadDsn: return False def get_logger(trace_id: str | None = None, recreate: bool = False) -> BoundLogger: """Get logger. Args: trace_id: Run ID. recreate: Recreate if exists. Returns: Logger. """ global log if log and not recreate: return log if not trace_id: trace_id = str(uuid.uuid4()) environment = os.environ.get("ENVIRONMENT", ENV_PROD).lower() is_production = environment == ENV_PROD log_config = LoggerConfig( name=APP_FULL_NAME, version=_get_version(), level=logging.INFO if is_production else logging.DEBUG, environment=environment, is_local=IS_LOCAL, ) if not IS_LOCAL and _validate_dsn(): log_config.sentry = SentryConfig(dsn=SENTRY_DSN, integrations=[AwsLambdaIntegration(timeout_warning=True)]) log = LoggerFactory(log_config).get_logger(APP_FULL_NAME) log = log.bind(request_id=trace_id) return log