Env¶

Environment configuration and namespace management.

Overview¶

The env module provides the EnvBase class which allows for creating isolated namespaces based on the type and name of environment.

Usage¶

from aibs_informatics_core.env import EnvBase

env_base = EnvBase('dev-projectX')
env_base.prefixed('my_resource', 'blue')  # 'dev-projectX-my_resource-blue'

EnvBase ¶

EnvBase(*args, **kwargs)

Bases: ValidatedStr

Environment namespace for generating prefixed resource names.

Parses an environment string (e.g. "dev-myproject") into an environment type and optional label. Provides methods to generate consistently prefixed names for AWS resources such as stacks, tables, functions, and buckets.

Source code in src/aibs_informatics_core/collections.py

def __init__(self, *args, **kwargs):
    """Placeholder for subclass to override"""
    super().__init__()

env_label `property` ¶

env_label

The environment label, or None if not specified.

env_type `property` ¶

env_type

The environment type (dev, test, or prod).

concat `classmethod` ¶

concat(*names, delim='-')

Join name parts with a delimiter.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to concatenate.	`()`
`delim`	`SupportedDelim`	Delimiter between parts.	`'-'`

Returns:

Type	Description
`str`	Concatenated string.

Source code in src/aibs_informatics_core/env.py

@classmethod
def concat(cls, *names: str, delim: SupportedDelim = "-") -> str:
    """Join name parts with a delimiter.

    Args:
        *names: Name parts to concatenate.
        delim: Delimiter between parts.

    Returns:
        Concatenated string.
    """
    return delim.join(names)

from_env `classmethod` ¶

from_env()

Get value from environment variables

Checks for env base variables.
Checks for env type and label:

Environment Variable Keys: env base : "env_base" / "ENV_BASE" env type : "env_type" / "ENV_TYPE" env label : "env_label" / "ENV_LABEL" / "label" / "LABEL" Raises: ApplicationException: if variables are not found.

Returns:

Type	Description
`E`	EnvBase

Source code in src/aibs_informatics_core/env.py

@classmethod
def from_env(cls: type[E]) -> E:
    """Get value from environment variables

    1. Checks for env base variables.
    2. Checks for env type and label:

    Environment Variable Keys:
    env base  : "env_base" / "ENV_BASE"
    env type  : "env_type" / "ENV_TYPE"
    env label : "env_label" / "ENV_LABEL" / "label" / "LABEL"
    Raises:
        ApplicationException: if variables are not found.

    Returns:
        EnvBase
    """
    env_base = get_env_var(ENV_BASE_KEY, ENV_BASE_KEY_ALIAS)
    if env_base:
        return cls(env_base)
    else:
        try:
            env_type = cls.load_env_type__from_env()
            env_label = cls.load_env_label__from_env()
        except Exception:
            raise ApplicationException("Could not resolve Env Base or Env Type from env")

        env_base = cls.from_type_and_label(EnvType(env_type), env_label=env_label)
        return env_base

from_type_and_label `classmethod` ¶

from_type_and_label(env_type, env_label=None)

Construct an EnvBase from an environment type and optional label.

Parameters:

Name	Type	Description	Default
`env_type`	`EnvType`	The environment type.	required
`env_label`	`str \| None`	Optional label to append after the type.	`None`

Returns:

Type	Description
`E`	A new EnvBase instance in the form `<env_type>[-<env_label>]`.

Source code in src/aibs_informatics_core/env.py

@classmethod
def from_type_and_label(cls: type[E], env_type: EnvType, env_label: str | None = None) -> E:
    """Construct an EnvBase from an environment type and optional label.

    Args:
        env_type: The environment type.
        env_label: Optional label to append after the type.

    Returns:
        A new EnvBase instance in the form ``<env_type>[-<env_label>]``.
    """
    return cls(f"{env_type.value}-{env_label}" if env_label else env_type.value)

get_bucket_name ¶

get_bucket_name(base_name, account_id, region)

Generate an S3 bucket name.

Parameters:

Name	Type	Description	Default
`base_name`	`str`	Bucket base name.	required
`account_id`	`str`	AWS account ID.	required
`region`	`str`	AWS region.	required

Returns:

Type	Description
`str`	Bucket name in the form `<env_base>-<base_name>-<region>-<account_id>`.

Source code in src/aibs_informatics_core/env.py

def get_bucket_name(self, base_name: str, account_id: str, region: str) -> str:
    """Generate an S3 bucket name.

    Args:
        base_name: Bucket base name.
        account_id: AWS account ID.
        region: AWS region.

    Returns:
        Bucket name in the form ``<env_base>-<base_name>-<region>-<account_id>``.
    """
    return self.concat(self, base_name, region, account_id, delim="-")

get_construct_id ¶

get_construct_id(*names)

Generate a CDK construct ID with the env base prefix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`

Returns:

Type	Description
`str`	Prefixed construct ID string.

Source code in src/aibs_informatics_core/env.py

def get_construct_id(self, *names: str) -> str:
    """Generate a CDK construct ID with the env base prefix.

    Args:
        *names: Name parts to join after the prefix.

    Returns:
        Prefixed construct ID string.
    """
    return self.prefixed(*names)

get_function_name ¶

get_function_name(name)

Generate a Lambda function name with the env base prefix.

Parameters:

Name	Type	Description	Default
`name`	`str`	Function base name.	required

Returns:

Type	Description
`str`	Prefixed function name string.

Source code in src/aibs_informatics_core/env.py

def get_function_name(self, name: str) -> str:
    """Generate a Lambda function name with the env base prefix.

    Args:
        name: Function base name.

    Returns:
        Prefixed function name string.
    """
    return self.prefixed(name)

get_job_name ¶

get_job_name(name, hash_value)

Generate a Batch job name with the env base prefix.

Parameters:

Name	Type	Description	Default
`name`	`str`	Job base name.	required
`hash_value`	`str`	Hash value to append for uniqueness.	required

Returns:

Type	Description
`str`	Prefixed job name string.

Source code in src/aibs_informatics_core/env.py

def get_job_name(self, name: str, hash_value: str) -> str:
    """Generate a Batch job name with the env base prefix.

    Args:
        name: Job base name.
        hash_value: Hash value to append for uniqueness.

    Returns:
        Prefixed job name string.
    """
    return self.get_resource_name(name, hash_value)

get_metric_namespace ¶

get_metric_namespace(name)

Generate a CloudWatch metric namespace.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric namespace base name.	required

Returns:

Type	Description
`str`	Metric namespace in the form `AIBS/<env_base>/<name>`.

Source code in src/aibs_informatics_core/env.py

def get_metric_namespace(self, name: str) -> str:
    """Generate a CloudWatch metric namespace.

    Args:
        name: Metric namespace base name.

    Returns:
        Metric namespace in the form ``AIBS/<env_base>/<name>``.
    """
    return self.concat("AIBS", self, name, delim="/")

get_pipeline_name ¶

get_pipeline_name(*names)

Generate a pipeline name with the env base prefix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`

Returns:

Type	Description
`str`	Prefixed pipeline name string.

Source code in src/aibs_informatics_core/env.py

def get_pipeline_name(self, *names: str) -> str:
    """Generate a pipeline name with the env base prefix.

    Args:
        *names: Name parts to join after the prefix.

    Returns:
        Prefixed pipeline name string.
    """
    return self.prefixed(*names)

get_repository_name ¶

get_repository_name(*names)

Generate a repository name using / as delimiter.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`

Returns:

Type	Description
`str`	Repository name in the form `<env_base>/<name1>/<name2>/...`.

Source code in src/aibs_informatics_core/env.py

def get_repository_name(self, *names: str) -> str:
    """Generate a repository name using ``/`` as delimiter.

    Args:
        *names: Name parts to join after the prefix.

    Returns:
        Repository name in the form ``<env_base>/<name1>/<name2>/...``.
    """
    return self.concat(self, self.concat(*names), delim="/")

get_resource_name ¶

get_resource_name(*names)

Generate a generic resource name with the env base prefix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`

Returns:

Type	Description
`str`	Prefixed resource name string.

Source code in src/aibs_informatics_core/env.py

def get_resource_name(self, *names: str) -> str:
    """Generate a generic resource name with the env base prefix.

    Args:
        *names: Name parts to join after the prefix.

    Returns:
        Prefixed resource name string.
    """
    return self.prefixed(*names)

get_ssm_param_name ¶

get_ssm_param_name(*names)

Generate an SSM parameter name.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Parameter name parts.	`()`

Returns:

Type	Description
`str`	SSM parameter name in the form `/<env_base>/<name1>/<name2>/...`.

Source code in src/aibs_informatics_core/env.py

def get_ssm_param_name(self, *names: str) -> str:
    """Generate an SSM parameter name.

    Args:
        *names: Parameter name parts.

    Returns:
        SSM parameter name in the form ``/<env_base>/<name1>/<name2>/...``.
    """
    return self.concat("", self, self.concat(*names), delim="/")

get_stack_name ¶

get_stack_name(*names)

Generate a CloudFormation stack name with the env base prefix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`

Returns:

Type	Description
`str`	Prefixed stack name string.

Source code in src/aibs_informatics_core/env.py

def get_stack_name(self, *names: str) -> str:
    """Generate a CloudFormation stack name with the env base prefix.

    Args:
        *names: Name parts to join after the prefix.

    Returns:
        Prefixed stack name string.
    """
    return self.prefixed(*names)

get_stage_name ¶

get_stage_name(*names)

Generate a pipeline stage name with the env base prefix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`

Returns:

Type	Description
`str`	Prefixed stage name string.

Source code in src/aibs_informatics_core/env.py

def get_stage_name(self, *names: str) -> str:
    """Generate a pipeline stage name with the env base prefix.

    Args:
        *names: Name parts to join after the prefix.

    Returns:
        Prefixed stage name string.
    """
    return self.prefixed(*names)

get_state_machine_log_group_name ¶

get_state_machine_log_group_name(name)

Generate a CloudWatch log group name for a Step Functions state machine.

Parameters:

Name	Type	Description	Default
`name`	`str`	State machine base name.	required

Returns:

Type	Description
`str`	Log group name in the format `/aws/vendedlogs/<env_base>/states/<name>`.

Source code in src/aibs_informatics_core/env.py

def get_state_machine_log_group_name(self, name: str) -> str:
    """Generate a CloudWatch log group name for a Step Functions state machine.

    Args:
        name: State machine base name.

    Returns:
        Log group name in the format ``/aws/vendedlogs/<env_base>/states/<name>``.
    """
    return self.concat(
        # https://docs.aws.amazon.com/step-functions/latest/dg/bp-cwl.html
        "/aws/vendedlogs",
        self,
        "states",
        name,
        delim="/",
    )

get_state_machine_name ¶

get_state_machine_name(name)

Generate a Step Functions state machine name with the env base prefix.

Parameters:

Name	Type	Description	Default
`name`	`str`	State machine base name.	required

Returns:

Type	Description
`str`	Prefixed state machine name string.

Source code in src/aibs_informatics_core/env.py

def get_state_machine_name(self, name: str) -> str:
    """Generate a Step Functions state machine name with the env base prefix.

    Args:
        name: State machine base name.

    Returns:
        Prefixed state machine name string.
    """
    return self.get_resource_name(name)

get_table_name ¶

get_table_name(name)

Generate a DynamoDB table name with the env base prefix.

Parameters:

Name	Type	Description	Default
`name`	`str`	Table base name.	required

Returns:

Type	Description
`str`	Prefixed table name string.

Source code in src/aibs_informatics_core/env.py

def get_table_name(self, name: str) -> str:
    """Generate a DynamoDB table name with the env base prefix.

    Args:
        name: Table base name.

    Returns:
        Prefixed table name string.
    """
    return self.prefixed(name)

load_env_label__from_env `classmethod` ¶

load_env_label__from_env()

Load the environment label from environment variables.

Checks env_label, ENV_LABEL, label, and LABEL environment variables.

Returns:

Type	Description
`str \| None`	The label string, or None if not set.

Source code in src/aibs_informatics_core/env.py

@classmethod
def load_env_label__from_env(cls) -> str | None:
    """Load the environment label from environment variables.

    Checks ``env_label``, ``ENV_LABEL``, ``label``, and ``LABEL``
    environment variables.

    Returns:
        The label string, or None if not set.
    """
    return get_env_var(ENV_LABEL_KEY, ENV_LABEL_KEY_ALIAS, LABEL_KEY, LABEL_KEY_ALIAS)

load_env_type__from_env `classmethod` ¶

load_env_type__from_env()

Load the environment type from environment variables.

Checks env_type and ENV_TYPE environment variables.

Returns:

Type	Description
`EnvType`	The resolved EnvType.

Raises:

Type	Description
`ApplicationException`	If no environment type variable is found.

Source code in src/aibs_informatics_core/env.py

@classmethod
def load_env_type__from_env(cls) -> EnvType:
    """Load the environment type from environment variables.

    Checks ``env_type`` and ``ENV_TYPE`` environment variables.

    Returns:
        The resolved EnvType.

    Raises:
        ApplicationException: If no environment type variable is found.
    """
    env_type = get_env_var(ENV_TYPE_KEY, ENV_TYPE_KEY_ALIAS)
    if not env_type:
        raise ApplicationException(
            f"{cls} is not found as {ENV_TYPE_KEY} or {ENV_TYPE_KEY_ALIAS}"
        )
    return EnvType(env_type)

prefixed ¶

prefixed(*names, delim='-')

Return a string with the env base as prefix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join after the prefix.	`()`
`delim`	`SupportedDelim`	Delimiter between parts.	`'-'`

Returns:

Type	Description
`str`	String in the form `<env_base><delim><name1><delim>...`.

Source code in src/aibs_informatics_core/env.py

def prefixed(self, *names: str, delim: SupportedDelim = "-") -> str:
    """Return a string with the env base as prefix.

    Args:
        *names: Name parts to join after the prefix.
        delim: Delimiter between parts.

    Returns:
        String in the form ``<env_base><delim><name1><delim>...``.
    """
    # Don't add if self in names already
    if names and (names[0] == self or names[0].startswith(self + delim)):
        return self.concat(*names, delim=delim)
    return self.concat(self, *names, delim=delim)

suffixed ¶

suffixed(*names, delim='-')

Return a string with the env base as suffix.

Parameters:

Name	Type	Description	Default
`*names`	`str`	Name parts to join before the suffix.	`()`
`delim`	`SupportedDelim`	Delimiter between parts.	`'-'`

Returns:

Type	Description
`str`	String in the form `<name1><delim>...<nameN><delim><env_base>`.

Source code in src/aibs_informatics_core/env.py

def suffixed(self, *names: str, delim: SupportedDelim = "-") -> str:
    """Return a string with the env base as suffix.

    Args:
        *names: Name parts to join before the suffix.
        delim: Delimiter between parts.

    Returns:
        String in the form ``<name1><delim>...<nameN><delim><env_base>``.
    """
    # Don't add if self in names already
    if names and (names[-1] == self or names[-1].endswith(delim + self)):
        return self.concat(*names, delim=delim)
    return self.concat(*names, self, delim=delim)

to_env ¶

to_env()

Write the env base value to the env_base environment variable.

Source code in src/aibs_informatics_core/env.py

def to_env(self):
    """Write the env base value to the ``env_base`` environment variable."""
    set_env_var(ENV_BASE_KEY, self)

to_type_and_label ¶

to_type_and_label()

Parse into environment type and optional label.

Returns:

Type	Description
`tuple[EnvType, str \| None]`	A tuple of `(EnvType, label)` where label may be None.

Source code in src/aibs_informatics_core/env.py

def to_type_and_label(self) -> tuple[EnvType, str | None]:
    """Parse into environment type and optional label.

    Returns:
        A tuple of ``(EnvType, label)`` where label may be None.
    """
    env_type, env_label = self.get_match_groups()
    return (EnvType(env_type), env_label)

EnvBaseEnumMixins ¶

Mixin for enums that provides environment-prefixed name generation.

prefix_with ¶

prefix_with(env_base=None, env_base_class=None)

Prefix the enum value with an environment base.

Parameters:

Name	Type	Description	Default
`env_base`	`E \| None`	Optional EnvBase instance to use.	`None`
`env_base_class`	`type[E] \| None`	Optional EnvBase subclass for loading from environment.	`None`

Returns:

Type	Description
`str`	The prefixed string.

Source code in src/aibs_informatics_core/env.py

def prefix_with(self, env_base: E | None = None, env_base_class: type[E] | None = None) -> str:
    """Prefix the enum value with an environment base.

    Args:
        env_base: Optional EnvBase instance to use.
        env_base_class: Optional EnvBase subclass for loading from environment.

    Returns:
        The prefixed string.
    """
    env_base_ = get_env_base(env_base, env_base_class)
    if isinstance(self, Enum):
        return env_base_.prefixed(self.value)
    else:
        return env_base_.prefixed(str(self))

EnvBaseMixins ¶

Bases: EnvBaseMixinsBase[EnvBase]

Concrete mixin providing env_base property bound to EnvBase.

EnvBaseMixinsBase ¶

Bases: Generic[E]

Generic mixin providing env_base property with lazy environment loading.

env_base `property` `writable` ¶

env_base

returns env base

If env base has not been set, it sets the value using environment variables

Returns:

Type	Description
`E`	env base

EnvType ¶

Bases: StrEnum

Enumeration of supported environment types.

ResourceNameBaseEnum ¶

Bases: str, EnvBaseEnumMixins, Enum

Enum for defining resource names that can be prefixed with an environment base.

get_name ¶

get_name(env_base)

Get the environment-prefixed resource name.

Parameters:

Name	Type	Description	Default
`env_base`	`EnvBase`	The environment base to use as prefix.	required

Returns:

Type	Description
`str`	The prefixed resource name string.

Source code in src/aibs_informatics_core/env.py

def get_name(self, env_base: EnvBase) -> str:
    """Get the environment-prefixed resource name.

    Args:
        env_base: The environment base to use as prefix.

    Returns:
        The prefixed resource name string.
    """
    return self.prefix_with(env_base)

get_env_base ¶

get_env_base() -> EnvBase

get_env_base(env_base: str | EnvBase) -> EnvBase

get_env_base(env_base: Literal[None]) -> EnvBase

get_env_base(
    env_base: Literal[None], env_base_class: Literal[None]
) -> EnvBase

get_env_base(
    env_base: Literal[None], env_base_class: type[E]
) -> E

get_env_base(
    env_base: str | E, env_base_class: type[E]
) -> E

get_env_base(
    env_base: str | E, env_base_class: Literal[None]
) -> EnvBase

get_env_base(env_base=None, env_base_class=None)

Resolve an EnvBase, loading from environment if not provided.

Parameters:

Name	Type	Description	Default
`env_base`	`str \| E \| None`	An existing env base string/instance, or None to load from environment.	`None`
`env_base_class`	`type[E \| EnvBase] \| None`	The EnvBase subclass to use for construction. Defaults to EnvBase.	`None`

Returns:

Type	Description
`E \| EnvBase`	The resolved EnvBase instance.

Source code in src/aibs_informatics_core/env.py

def get_env_base(
    env_base: str | E | None = None,
    env_base_class: type[E | EnvBase] | None = None,
) -> E | EnvBase:
    """Resolve an EnvBase, loading from environment if not provided.

    Args:
        env_base: An existing env base string/instance, or None to load from environment.
        env_base_class: The EnvBase subclass to use for construction. Defaults to EnvBase.

    Returns:
        The resolved EnvBase instance.
    """
    env_base_cls: type[E] = env_base_class or EnvBase  # type: ignore[assignment]
    if env_base:
        if isinstance(env_base, env_base_cls):
            return env_base
        else:
            return env_base_cls(env_base)
    return env_base_cls.from_env()

get_env_label ¶

get_env_label(env_label=MISSING, env_base_class=None)

Get Environment label

If not specified, it will be loaded from envirionment (First checking for env base, then label)

Parameters:

Name	Type	Description	Default
`env_label`	`Union[Optional[str], _Missing]`	env label. Defaults to MISSING.	`MISSING`

Returns:

Type	Description
`str \| None`	env label string if exists, else None

Source code in src/aibs_informatics_core/env.py

def get_env_label(
    env_label: str | None | _Missing = MISSING,
    env_base_class: E | None = None,
) -> str | None:
    """Get Environment label

    If not specified, it will be loaded from envirionment (First checking for env base, then label)

    Args:
        env_label (Union[Optional[str], _Missing], optional): env label. Defaults to MISSING.

    Returns:
        env label string if exists, else None
    """
    env_base_cls = env_base_class or EnvBase
    if isinstance(env_label, _Missing):
        try:
            # First check if EnvBase exists and
            return env_base_cls.from_env().env_label
        except Exception:
            # next check for env label.
            return env_base_cls.load_env_label__from_env()
    # Right now env label regex is only baked into EnvBase, so let's
    # create an EnvBase to validate
    return env_base_cls.from_type_and_label(EnvType.DEV, env_label=env_label).env_label

get_env_type ¶

get_env_type(
    env_type=None,
    default_env_type=None,
    env_base_class=None,
)

Resolve an EnvType, loading from environment if not provided.

If loading from environment, the EnvBase variable takes precedence over the EnvType variable.

Parameters:

Name	Type	Description	Default
`env_type`	`str \| EnvType \| None`	An existing env type string/instance, or None to load from environment.	`None`
`default_env_type`	`EnvType \| None`	Fallback EnvType if loading from environment fails.	`None`
`env_base_class`	`E \| None`	Optional EnvBase subclass for environment resolution.	`None`

Returns:

Type	Description
`EnvType`	The resolved EnvType.

Raises:

Type	Description
`Exception`	If no env type can be resolved and no default is provided.

Source code in src/aibs_informatics_core/env.py

def get_env_type(
    env_type: str | EnvType | None = None,
    default_env_type: EnvType | None = None,
    env_base_class: E | None = None,
) -> EnvType:
    """Resolve an EnvType, loading from environment if not provided.

    If loading from environment, the ``EnvBase`` variable takes precedence
    over the ``EnvType`` variable.

    Args:
        env_type: An existing env type string/instance, or None to load from environment.
        default_env_type: Fallback EnvType if loading from environment fails.
        env_base_class: Optional EnvBase subclass for environment resolution.

    Returns:
        The resolved EnvType.

    Raises:
        Exception: If no env type can be resolved and no default is provided.
    """
    if env_type:
        return env_type if isinstance(env_type, EnvType) else EnvType(env_type)
    try:
        return (env_base_class or EnvBase).from_env().env_type
    except Exception as e:
        if default_env_type:
            return default_env_type
        raise e

Env¶

Overview¶

Usage¶

EnvBase ¶

env_label property ¶

env_type property ¶

concat classmethod ¶

from_env classmethod ¶

from_type_and_label classmethod ¶

get_bucket_name ¶

get_construct_id ¶

get_function_name ¶

get_job_name ¶

get_metric_namespace ¶

get_pipeline_name ¶

get_repository_name ¶

get_resource_name ¶

get_ssm_param_name ¶

get_stack_name ¶

get_stage_name ¶

get_state_machine_log_group_name ¶

get_state_machine_name ¶

get_table_name ¶

load_env_label__from_env classmethod ¶

load_env_type__from_env classmethod ¶

prefixed ¶

suffixed ¶

to_env ¶

to_type_and_label ¶

EnvBaseEnumMixins ¶

prefix_with ¶

EnvBaseMixins ¶

EnvBaseMixinsBase ¶

env_base property writable ¶

EnvType ¶

ResourceNameBaseEnum ¶

get_name ¶

get_env_base ¶

get_env_label ¶

get_env_type ¶

env_label `property` ¶

env_type `property` ¶

concat `classmethod` ¶

from_env `classmethod` ¶

from_type_and_label `classmethod` ¶

load_env_label__from_env `classmethod` ¶

load_env_type__from_env `classmethod` ¶

env_base `property` `writable` ¶