"""Application configuration.""" import logging import os from os.path import abspath, dirname, join, pardir from typing import Any, Dict from audience_common.logger import LogFormat, configure_logging from environs import Env env = Env() # Load environment variables from a .env file if present dotenv_path = abspath(join(dirname(__file__), pardir, ".env")) env.read_env(dotenv_path, recurse=False) # Service information SERVICE_NAME = "ows-pdp" SERVICE_VERSION = "1.0.0" # Production environment PROD_ENVIRONMENT = "prod" DEV_ENVIRONMENT = "dev" QA_ENVIRONMENT = "qa" TEST_ENVIRONMENT = "test" ENVIRONMENT = os.environ.get("Environment") or DEV_ENVIRONMENT # Cerbos CERBOS_SERVER_HOST = os.environ.get("CERBOS_SERVER_HOST") or "http://localhost:3592" CERBOS_BATCH_SIZE = env.int("CERBOS_BATCH_SIZE", default=50) CERBOS_TIMEOUT_SECS = env.int("CERBOS_TIMEOUT_SECS", default=5) CERBOS_REQUEST_RETRIES = env.int("CERBOS_REQUEST_RETRIES", default=1) # DynamoDB DYNAMODB_ENDPOINT_URL = os.environ.get("DYNAMODB_ENDPOINT_URL") or None DYNAMODB_TABLE_IDENTITY = os.environ.get("DYNAMODB_TABLE_IDENTITY") or "dev_pp_identity" DYNAMODB_BATCH_UNPROCESSED_ITEMS_MAX_ATTEMPTS = env.int( "DYNAMODB_MAX_BATCH_UNPROCESSED_ITEMS_ITERATIONS", 3 ) DYNAMODB_TCP_KEEP_ALIVE = env.bool("DYNAMODB_TCP_KEEP_ALIVE", default=False) IDENTITY_HASH_KEY = "identity_uuid" # also known as Partition key IDENTITY_RANGE_KEY = "tenant_uuid" # also known as Sort key # Errors and loggers SENTRY = os.environ.get("SENTRY_DSN") or None LOGGER_LEVEL = logging.INFO LOGGER_NAME = "ows1" LOGGING_CONFIG: Dict[str, Any] = { "version": 1, "disable_existing_loggers": False, "handlers": { "stream": { "level": "INFO", "class": "audience_common.logger.OwsStreamHandler", }, "null": { "level": "INFO", "class": "logging.NullHandler", }, }, "loggers": { "pdp": { "handlers": ["stream"], "level": "INFO", }, "audience_common.logger.asgi.middleware": { "handlers": ["stream"], "level": "INFO", }, "uvicorn": { "handlers": ["stream"], "level": "INFO", }, "uvicorn.access": { "handlers": ["null"], "level": "INFO", }, "splitio": { "handlers": ["null"], "level": "INFO", }, }, } configure_logging( LOGGING_CONFIG, environment=ENVIRONMENT, service_name=SERVICE_NAME, service_version=SERVICE_VERSION, log_format=LogFormat.JSON, logger_name=LOGGER_NAME, debug=False, # Use debug=True to set all handlers level to DEBUG ) # Prevent third-party modules from generating sentry alerts. # This config will ignore ALL ERRORs for modules listed in IGNORED_SENTRY_LOGGERS. # To ignore specific module paths only, use ignore_logger in sentry_sdk. IGNORED_SENTRY_LOGGERS = {"splitio"} # Infra handlers HEALTH_CHECK = "/hello/" CONNECTIVITY_OWS_ACCOUNT = "/infra/connectivity/ows-account/" CONNECTIVITY_OWS_PERMISSIONS = "/infra/connectivity/ows-permissions/" CONNECTIVITY_CERBOS_FARGATE = "/infra/connectivity/cerbos/" CONNECTIVITY_REDIS = "/infra/connectivity/redis/" CONNECTIVITY_SPLITIO = "/infra/connectivity/splitio/" ID_TO_UUID_EXCHANGE = "/infra/tenant/id-to-uuid-exchange/" UUID_TO_ID_EXCHANGE = "/infra/tenant/uuid-to-id-exchange/" # Swagger documentation path SWAGGER_FILE_PATH = "spec/ows_pdp-1.0.0.yaml" AUTH0_ALGORITHMS = ["RS256"] JWT_USER_METADATA = "https://grass.theorchard.com/user_metadata" IS_MACHINE = "isMachine" JWT_IMPERSONATED_BY = "https://grass.theorchard.com/impersonated_by" # NOTE: Setting "APP_DEBUG" in the env will disable the # default_error_handler. Omit it from the .env if to test # uncaught errors in PDP APP_DEBUG: bool = os.environ.get("APP_DEBUG", default=False) # type:ignore SPLITIO_LOCALHOST_ENV = "localhost" DEFAULT_SPLIT_FILE_PATH = "tests/.split" SPLIT_FILE_PATH = os.environ.get("SPLIT_FILE_PATH", DEFAULT_SPLIT_FILE_PATH) SPLITIO_API_KEY = os.environ.get("SPLITIO_API_KEY") SPLITIO_IMPRESSIONS_MODE = os.environ.get("SPLITIO_IMPRESSIONS_MODE", "optimized") SPLITIO_PREFORKED_INIT = env.bool("SPLITIO_PREFORKED_INIT", default=False) REDIS_URL = os.environ.get("REDIS_URL", "redis://127.0.0.1:50052") CACHE_USE_REDIS = env.bool("CACHE_USE_REDIS", default=False) REDIS_CACHE_TTL = env.int("REDIS_CACHE_TTL_SECONDS", default=60 * 5) REDIS_SOCKET_TIMEOUT = env.float("REDIS_SOCKET_TIMEOUT", default=1.0) REDIS_SOCKET_CONNECT_TIMEOUT = env.float("REDIS_SOCKET_CONNECT_TIMEOUT", 10.0) CACHE_TTL_SECONDS_TENANT_HIERARCHY_PARENT_COMPANY = env.int( "CACHE_TTL_SECONDS_TENANT_HIERARCHY_PARENT_COMPANY", default=None ) CACHE_TTL_SECONDS_TENANT_HIERARCHY_COMPANY_BRAND = env.int( "CACHE_TTL_SECONDS_TENANT_HIERARCHY_COMPANY_BRAND", default=None ) CACHE_TTL_SECONDS_TENANT_HIERARCHY_ACCOUNT = env.int( "CACHE_TTL_SECONDS_TENANT_HIERARCHY_ACCOUNT", default=86400 ) CACHE_TTL_SECONDS_TENANT_HIERARCHY_SUBACCOUNT = env.int( "CACHE_TTL_SECONDS_TENANT_HIERARCHY_SUBACCOUNT", default=86400 ) CACHE_TTL_SECONDS_TENANT_HIERARCHY_LABEL_PARTICIPANT = env.int( "CACHE_TTL_SECONDS_TENANT_HIERARCHY_LABEL_PARTICIPANT", default=86400 ) CACHE_TTL_SECONDS_TENANT_HIERARCHY_COLLABORATOR = env.int( "CACHE_TTL_SECONDS_TENANT_HIERARCHY_COLLABORATOR", default=86400 ) TOMBSTONE_TTL_SECONDS = env.int("TOMBSTONE_TTL_SECONDS", default=1) CACHE_TTL_SECONDS_IDENTITY_TENANT_ROLES: int | None = env.int( "CACHE_TTL_SECONDS_IDENTITY_TENANT_ROLES", default=86400 ) CACHE_TTL_SECONDS_IDENTITY_ROLES_RESPONSE: int | None = env.int( "CACHE_TTL_SECONDS_IDENTITY_ROLES_RESPONSE", default=86400 ) CACHE_TTL_SECONDS_IDENTITY_ALLOWED_TENANTS: int | None = env.int( "CACHE_TTL_SECONDS_IDENTITY_ALLOWED_TENANTS", default=86400 )