Configuration Management Pattern

Purpose

Centralize all configuration in a single Config class with environment/region matrix support, lazy secret loading, and environment variable overrides.

Environment × Region Matrix

The platform runs across 4 environments and 3 regions. Region-specific configuration (ES hosts, API endpoints) is resolved at import time:

class Environments(Enum):
    DEV = "development"
    STAGE = "staging"
    QA = "qa"
    PROD = "production"

class Regions(Enum):
    US_EAST_1 = "us-east-1"
    US_WEST_1 = "us-west-1"
    CA_CENTRAL_1 = "ca-central-1"

CONFIG_MAP = {
    # Staging
    f"{Environments.STAGE.value}-{Regions.US_EAST_1.value}": {
        "ES_HOST": "http://internal-staging-c2-elasticsearch7-alb-...elb.amazonaws.com:9200",
    },
    f"{Environments.STAGE.value}-{Regions.US_WEST_1.value}": {
        "ES_HOST": "http://internal-stg-prd-c2-elasticsearch-alb-...elb.amazonaws.com:9200",
    },
    f"{Environments.STAGE.value}-{Regions.CA_CENTRAL_1.value}": {
        "ES_HOST": "http://internal-staging-c5-alb-elasticsearch-...elb.amazonaws.com:9200",
    },
    # QA — same pattern
    # Production — same pattern
}

# Resolution at import time
ENV = os.getenv("ENV", Environments.PROD.value)
REGION = os.getenv("AWS_REGION", Regions.US_EAST_1.value)

if ENV == Environments.DEV.value:
    current_config = {"ES_HOST": "http://127.0.0.1:9200"}
else:
    config_key = f"{ENV}-{REGION}"
    current_config = CONFIG_MAP.get(
        config_key,
        CONFIG_MAP[f"{Environments.PROD.value}-{Regions.US_EAST_1.value}"]  # Fallback
    )

Lazy Secret Loading

Secrets are loaded from AWS Secrets Manager on first access and cached for the Lambda invocation lifetime:

_cached_secrets = None

def _get_secrets():
    global _cached_secrets
    if _cached_secrets is None:
        # Prefer ARN (has suffix matching IAM permissions), fall back to name
        secret_id = os.environ.get("RDS_SECRET_ARN") or os.environ["RDS_SECRET_NAME"]
        region = os.getenv("DEPLOY_REGION", "us-east-1")

        client = boto3_client("secretsmanager", region_name=region)
        response = client.get_secret_value(SecretId=secret_id)
        _cached_secrets = json.loads(response["SecretString"])
    return _cached_secrets

# Exposed as callables for backward compatibility
RDS_USERNAME = get_rds_username  # Called as Config.RDS_USERNAME()
RDS_PASSWORD = get_rds_password

Why Lazy?

• Avoids import-time errors when running tests (no AWS credentials needed)
• Avoids unnecessary Secrets Manager calls if the Lambda doesn't need DB access
• Single API call per Lambda invocation, cached globally

Centralized Config Class

class Config:
    # Environment
    ENV = ENV
    REGION = REGION

    # Database (callables — invoke as Config.RDS_USERNAME())
    RDS_USERNAME = get_rds_username
    RDS_PASSWORD = get_rds_password
    RDS_DBNAME = os.environ["RDS_DBNAME"]
    WRITER_PROXY_ENDPOINT = os.environ["WRITER_PROXY_ENDPOINT"]
    READER_PROXY_ENDPOINT = os.environ["READER_PROXY_ENDPOINT"]

    # Elasticsearch (overridable via env vars with sensible defaults)
    ES_HOST = current_config["ES_HOST"]
    ES_TIMEOUT = int(os.environ.get("ES_TIMEOUT", "180"))
    ES_BULK_THRESHOLD = int(os.environ.get("ES_BULK_THRESHOLD", "3"))
    ES_BULK_BATCH_SIZE = int(os.environ.get("ES_BULK_BATCH_SIZE", "50"))
    ES_RETRY_ATTEMPTS = int(os.environ.get("ES_RETRY_ATTEMPTS", "3"))

    # Processing
    ATTACHMENT_CHUNK_SIZE = int(os.environ.get("ATTACHMENT_CHUNK_SIZE", "100"))
    MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "2"))

    # SQS
    DEPLOY_REGION = os.environ.get("DEPLOY_REGION", REGION)
    JOBS_QUEUE_NAME = os.environ.get("JOBS_QUEUE_NAME", "")
    STACK_NAME_PREFIX = os.environ.get("STACK_NAME_PREFIX", "")
    SQS_BASE_DELAY = int(os.environ.get("SQS_BASE_DELAY", "120"))
    SQS_MAX_DELAY = int(os.environ.get("SQS_MAX_DELAY", "900"))
    SQS_BATCH_FAILURE_MAX_RETRIES = int(os.environ.get("SQS_BATCH_FAILURE_MAX_RETRIES", "2"))
    SQS_MAX_DLQ_REDRIVES = int(os.environ.get("SQS_MAX_DLQ_REDRIVES", "2"))
    JOB_REQUEUE_DELAY = int(os.environ.get("JOB_REQUEUE_DELAY", "300"))

    # Debug
    DEBUGGER = os.environ.get("FORCE_DEBUG", "false") == "true" or ENV == Environments.DEV.value

    # Environment checks
    @classmethod
    def is_dev(cls): return cls.ENV == Environments.DEV.value

    @classmethod
    def is_staging(cls): return cls.ENV == Environments.STAGE.value

    @classmethod
    def is_production(cls): return cls.ENV == Environments.PROD.value

    @classmethod
    def is_qa(cls): return cls.ENV == Environments.QA.value

    @classmethod
    def get_environment_info(cls): return f"{cls.ENV}-{cls.REGION}"

    @classmethod
    def validate(cls):
        """Validate required config on import — warn but don't fail."""
        required = ["RDS_DBNAME", "ES_HOST", "WRITER_PROXY_ENDPOINT", "READER_PROXY_ENDPOINT"]
        missing = [k for k in required if not getattr(cls, k)]
        if missing:
            raise ValueError(f"Missing required configuration: {missing}")

Validation at Import Time

Config validates required fields on import but logs a warning instead of crashing — allows tests and development to proceed without full config:

try:
    Config.validate()
except ValueError as e:
    logging.warning(f"Configuration validation failed: {e}")

Dual Database URL Pattern

Separate URL builders for writer and reader endpoints:

def get_writer_database_url():
    return f"mysql+pymysql://{RDS_USERNAME()}:{RDS_PASSWORD()}@{WRITER_PROXY_ENDPOINT}/{RDS_DBNAME}"

def get_reader_database_url():
    return f"mysql+pymysql://{RDS_USERNAME()}:{RDS_PASSWORD()}@{READER_PROXY_ENDPOINT}/{RDS_DBNAME}"

Key Rules

1. All env var access in Config — never os.environ in core/ or shell/
2. Defaults for every tunable — modules work out-of-box, tuning via env vars
3. Lazy secrets — never call Secrets Manager at import time
4. Validate but don't crash — log warning on missing config, fail on use
5. Fallback to prod — if CONFIG_MAP key missing, default to production us-east-1
6. int() cast env vars — all numeric config explicitly cast from string

Sign In

Sign In

Please set a new password.

Set Password

Cancel

?

Ask the Architecture ×

Ask questions about Nextpoint architecture, patterns, rules, or any module. Powered by Claude Opus 4.6.

Ask