API Reference

pdum.aws

AWS utils.

Small, reusable helpers over boto3:

• :mod:pdum.aws.core — sessions and clients on the ambient credential chain.
• :mod:pdum.aws.secrets — a namespaced view of SSM Parameter Store.
• :mod:pdum.aws.quotas — inspect Service Quotas and request increases.
• :mod:pdum.aws.env — build a process environment from .env and SSM.

Nothing here hardcodes a profile, region, or account. Credentials come from boto3's own resolution, so an account is selected by setting AWS_PROFILE in the environment:

$ AWS_PROFILE=my-account python -m my_script

from pdum import aws

print(aws.whoami()["Account"])

A script that wants the environment pdx would have given it — the project's .env, its .env.local overlay, and the secrets on the SSM search path — calls :func:~pdum.aws.env.load_env in place of dotenv.load_dotenv():

from pdum.aws import load_env

load_env()

EnvReport dataclass

What :func:load_env did, layer by layer.

Attributes:

Name	Type	Description
env_files	LoadReport or None	What the .env and its overlay contributed, or None when there was no such file.
path	str	The SSM search path that was used.
path_source	str	Where that path came from, for reporting.
secrets	LoadReport	What the store contributed.

Source code in src/pdum/aws/env.py

@dataclass(frozen=True, slots=True)
class EnvReport:
    """What :func:`load_env` did, layer by layer.

    Attributes
    ----------
    env_files : LoadReport or None
        What the ``.env`` and its overlay contributed, or ``None`` when there
        was no such file.
    path : str
        The SSM search path that was used.
    path_source : str
        Where that path came from, for reporting.
    secrets : LoadReport
        What the store contributed.
    """

    env_files: LoadReport | None
    path: str
    path_source: str
    secrets: LoadReport

NoSearchPath

Bases: LookupError

No SSM search path could be resolved, so no secrets could be loaded.

Raised rather than passed over: a process that quietly continues without the secrets it asked for fails later, somewhere less informative.

Source code in src/pdum/aws/env.py

class NoSearchPath(LookupError):
    """No SSM search path could be resolved, so no secrets could be loaded.

    Raised rather than passed over: a process that quietly continues without the
    secrets it asked for fails later, somewhere less informative.
    """

client(service, region=None)

Create a service client on the ambient credential chain.

Parameters:

Name	Type	Description	Default
service	str	Service name, e.g. "ssm", "ec2", "service-quotas".	required
region	str	Region to pin the client to. See :func:session.	None

Returns:

Type	Description
Any	A botocore client for service.

Source code in src/pdum/aws/core.py

def client(service: str, region: str | None = None) -> Any:
    """Create a service client on the ambient credential chain.

    Parameters
    ----------
    service : str
        Service name, e.g. ``"ssm"``, ``"ec2"``, ``"service-quotas"``.
    region : str, optional
        Region to pin the client to. See :func:`session`.

    Returns
    -------
    Any
        A ``botocore`` client for *service*.
    """
    return session(region).client(service)

load_env(*, environ=None, env_file='.env', envvar=DEFAULT_ENVVAR, default_path=None, store_factory=SecretStore)

Load the project's .env files and its secrets into the environment.

The drop-in for dotenv.load_dotenv() in a project that keeps its secrets in SSM. Layers are applied most-specific-last-wins-least: whatever is already set stays, then the .env pair fills gaps, then the store fills what remains. See the module docstring for why the order matters.

Parameters:

Name	Type	Description	Default
environ	mutable mapping	Environment to update; defaults to os.environ. Pass a plain dict to compute an environment without touching this process's own.	None
env_file	str	File to search for upward from the working directory, and the base name of the .local overlay beside it. A name with a directory component is used as given.	".env"
envvar	str	Variable naming the SSM search path. None leaves only default_path.	DEFAULT_ENVVAR
default_path	str or callable	Search path to fall back on when the environment, including the .env, has none. A callable is resolved here.	None
store_factory	callable	Called with the resolved path to build the store. Override to pin a region, e.g. lambda path: SecretStore(path, region="us-east-1").	:class:`~pdum.aws.secrets.SecretStore`

Returns:

Type	Description
EnvReport	What each layer contributed. Ignore it for the common case; it is there for reporting and for tests.

Raises:

Type	Description
NoSearchPath	If no search path could be resolved. Loading a .env and silently skipping the secrets would leave a process that fails later, somewhere less informative.
ValueError	If the resolved path is not a usable SSM prefix — a root path, say.

Source code in src/pdum/aws/env.py

def load_env(
    *,
    environ: MutableMapping[str, str] | None = None,
    env_file: str = ".env",
    envvar: str | None = DEFAULT_ENVVAR,
    default_path: PathDefault = None,
    store_factory: Callable[[str], SecretStore] = SecretStore,
) -> EnvReport:
    """Load the project's ``.env`` files and its secrets into the environment.

    The drop-in for ``dotenv.load_dotenv()`` in a project that keeps its secrets
    in SSM. Layers are applied most-specific-last-wins-least: whatever is
    already set stays, then the ``.env`` pair fills gaps, then the store fills
    what remains. See the module docstring for why the order matters.

    Parameters
    ----------
    environ : mutable mapping, optional
        Environment to update; defaults to ``os.environ``. Pass a plain dict to
        compute an environment without touching this process's own.
    env_file : str, default ".env"
        File to search for upward from the working directory, and the base name
        of the ``.local`` overlay beside it. A name with a directory component
        is used as given.
    envvar : str, optional
        Variable naming the SSM search path. ``None`` leaves only
        *default_path*.
    default_path : str or callable, optional
        Search path to fall back on when the environment, including the
        ``.env``, has none. A callable is resolved here.
    store_factory : callable, default :class:`~pdum.aws.secrets.SecretStore`
        Called with the resolved path to build the store. Override to pin a
        region, e.g. ``lambda path: SecretStore(path, region="us-east-1")``.

    Returns
    -------
    EnvReport
        What each layer contributed. Ignore it for the common case; it is there
        for reporting and for tests.

    Raises
    ------
    NoSearchPath
        If no search path could be resolved. Loading a ``.env`` and silently
        skipping the secrets would leave a process that fails later, somewhere
        less informative.
    ValueError
        If the resolved path is not a usable SSM prefix — a root path, say.
    """
    target = os.environ if environ is None else environ
    from_files = load_env_files(env_file, target)
    path, source = resolve_search_path(
        envvar=envvar, default_path=default_path, environ=target, env_file_report=from_files
    )
    if not path:
        raise NoSearchPath(no_search_path_message(envvar, env_file, from_files))
    return EnvReport(
        env_files=from_files,
        path=path,
        path_source=source,
        secrets=load_secrets(store_factory(path), target),
    )

resource(service, region=None)

Create a service resource on the ambient credential chain.

Parameters:

Name	Type	Description	Default
service	str	Service name, e.g. "s3", "dynamodb".	required
region	str	Region to pin the resource to. See :func:session.	None

Returns:

Type	Description
Any	A boto3 resource for service.

Source code in src/pdum/aws/core.py

def resource(service: str, region: str | None = None) -> Any:
    """Create a service resource on the ambient credential chain.

    Parameters
    ----------
    service : str
        Service name, e.g. ``"s3"``, ``"dynamodb"``.
    region : str, optional
        Region to pin the resource to. See :func:`session`.

    Returns
    -------
    Any
        A ``boto3`` resource for *service*.
    """
    return session(region).resource(service)

session(region=None)

Create a session on the ambient credential chain.

A fresh :class:boto3.Session is created per call rather than reusing boto3's module-level default session, which caches the credentials it resolved on first use. Building a new one keeps each call honest about the current environment — it picks up a re-exported AWS_PROFILE or a refreshed SSO token instead of serving stale credentials for the life of the process.

Parameters:

Name	Type	Description	Default
region	str	Region to pin the session to. When None, boto3 resolves it from the environment or the active profile.	None

Returns:

Type	Description
Session	A session bound to whatever credentials the environment provides.

Source code in src/pdum/aws/core.py

def session(region: str | None = None) -> boto3.Session:
    """Create a session on the ambient credential chain.

    A fresh :class:`boto3.Session` is created per call rather than reusing
    ``boto3``'s module-level default session, which caches the credentials it
    resolved on first use. Building a new one keeps each call honest about the
    current environment — it picks up a re-exported ``AWS_PROFILE`` or a
    refreshed SSO token instead of serving stale credentials for the life of the
    process.

    Parameters
    ----------
    region : str, optional
        Region to pin the session to. When ``None``, ``boto3`` resolves it from
        the environment or the active profile.

    Returns
    -------
    boto3.Session
        A session bound to whatever credentials the environment provides.
    """
    return boto3.Session(region_name=region)

whoami(region=None)

Return the identity the current credentials resolve to.

Useful as a cheap credential check before doing anything destructive, and as the fastest way to confirm which account is actually in play.

Parameters:

Name	Type	Description	Default
region	str	Region for the STS call. See :func:session.	None

Returns:

Type	Description
dict of str to str	The sts:GetCallerIdentity response, with Account, Arn and UserId keys.

Source code in src/pdum/aws/core.py

def whoami(region: str | None = None) -> dict[str, str]:
    """Return the identity the current credentials resolve to.

    Useful as a cheap credential check before doing anything destructive, and as
    the fastest way to confirm which account is actually in play.

    Parameters
    ----------
    region : str, optional
        Region for the STS call. See :func:`session`.

    Returns
    -------
    dict of str to str
        The ``sts:GetCallerIdentity`` response, with ``Account``, ``Arn`` and
        ``UserId`` keys.
    """
    return client("sts", region).get_caller_identity()

pdum.aws.core

Ambient AWS sessions and clients.

This module deliberately knows nothing about named profiles. boto3 already resolves credentials from AWS_PROFILE, AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY, SSO token caches, ECS/EC2 instance roles and ~/.aws/config — so the correct behaviour for a reusable library is to let it, and to select an account by setting AWS_PROFILE in the environment:

$ AWS_PROFILE=my-account python -m my_script

Nothing here hardcodes a profile, a region, or an account. A library that picks a default profile is a library that silently talks to the wrong account when it is reused somewhere else.

Region is likewise left to boto3, which reads AWS_REGION, AWS_DEFAULT_REGION and the profile's region. Pass region= only for APIs that are pinned to one region regardless of where the caller lives — Route 53 Domains and CloudFront ACM certificates being the usual examples.

client(service, region=None)

Create a service client on the ambient credential chain.

Parameters:

Name	Type	Description	Default
service	str	Service name, e.g. "ssm", "ec2", "service-quotas".	required
region	str	Region to pin the client to. See :func:session.	None

Returns:

Type	Description
Any	A botocore client for service.

Source code in src/pdum/aws/core.py

def client(service: str, region: str | None = None) -> Any:
    """Create a service client on the ambient credential chain.

    Parameters
    ----------
    service : str
        Service name, e.g. ``"ssm"``, ``"ec2"``, ``"service-quotas"``.
    region : str, optional
        Region to pin the client to. See :func:`session`.

    Returns
    -------
    Any
        A ``botocore`` client for *service*.
    """
    return session(region).client(service)

resource(service, region=None)

Create a service resource on the ambient credential chain.

Parameters:

Name	Type	Description	Default
service	str	Service name, e.g. "s3", "dynamodb".	required
region	str	Region to pin the resource to. See :func:session.	None

Returns:

Type	Description
Any	A boto3 resource for service.

Source code in src/pdum/aws/core.py

def resource(service: str, region: str | None = None) -> Any:
    """Create a service resource on the ambient credential chain.

    Parameters
    ----------
    service : str
        Service name, e.g. ``"s3"``, ``"dynamodb"``.
    region : str, optional
        Region to pin the resource to. See :func:`session`.

    Returns
    -------
    Any
        A ``boto3`` resource for *service*.
    """
    return session(region).resource(service)

session(region=None)

Create a session on the ambient credential chain.

A fresh :class:boto3.Session is created per call rather than reusing boto3's module-level default session, which caches the credentials it resolved on first use. Building a new one keeps each call honest about the current environment — it picks up a re-exported AWS_PROFILE or a refreshed SSO token instead of serving stale credentials for the life of the process.

Parameters:

Name	Type	Description	Default
region	str	Region to pin the session to. When None, boto3 resolves it from the environment or the active profile.	None

Returns:

Type	Description
Session	A session bound to whatever credentials the environment provides.

Source code in src/pdum/aws/core.py

def session(region: str | None = None) -> boto3.Session:
    """Create a session on the ambient credential chain.

    A fresh :class:`boto3.Session` is created per call rather than reusing
    ``boto3``'s module-level default session, which caches the credentials it
    resolved on first use. Building a new one keeps each call honest about the
    current environment — it picks up a re-exported ``AWS_PROFILE`` or a
    refreshed SSO token instead of serving stale credentials for the life of the
    process.

    Parameters
    ----------
    region : str, optional
        Region to pin the session to. When ``None``, ``boto3`` resolves it from
        the environment or the active profile.

    Returns
    -------
    boto3.Session
        A session bound to whatever credentials the environment provides.
    """
    return boto3.Session(region_name=region)

whoami(region=None)

Return the identity the current credentials resolve to.

Useful as a cheap credential check before doing anything destructive, and as the fastest way to confirm which account is actually in play.

Parameters:

Name	Type	Description	Default
region	str	Region for the STS call. See :func:session.	None

Returns:

Type	Description
dict of str to str	The sts:GetCallerIdentity response, with Account, Arn and UserId keys.

Source code in src/pdum/aws/core.py

def whoami(region: str | None = None) -> dict[str, str]:
    """Return the identity the current credentials resolve to.

    Useful as a cheap credential check before doing anything destructive, and as
    the fastest way to confirm which account is actually in play.

    Parameters
    ----------
    region : str, optional
        Region for the STS call. See :func:`session`.

    Returns
    -------
    dict of str to str
        The ``sts:GetCallerIdentity`` response, with ``Account``, ``Arn`` and
        ``UserId`` keys.
    """
    return client("sts", region).get_caller_identity()

pdum.aws.secrets

Secrets in AWS SSM Parameter Store, resolved along a search path.

A :class:SecretStore is built from a search path of SSM prefixes, most specific first — "/myapp/:/org/" — and maps a name like STRIPE_KEY onto the parameter <prefix>STRIPE_KEY in each layer. Reads resolve in this order:

1. a real environment variable of that name, so a shell export, a .env loaded by the caller, or a test fixture can override without touching AWS;
2. each SSM prefix in path order (decrypted), first hit wins;
3. the supplied default, or a :class:KeyError when required=True.

Layered prefixes work like layered config files (project, then org, then global): a project stores only what is truly its own and inherits the rest, so a secret shared by many projects lives in exactly one place. Writes and deletes always target the first prefix — the layer the store belongs to — never a fallback layer; mutating a shared layer requires constructing a store whose path starts there.

Why Parameter Store rather than Secrets Manager: standard-tier parameters are free where Secrets Manager is roughly $0.40 per secret per month, and both give KMS encryption, IAM gating and CloudTrail audit. This module is the only thing that knows which backend is in use, so switching later is a one-file change.

The path is required and has no default — a shared default would let two unrelated projects collide in the same namespace, and would make :meth:SecretStore.names return parameters the caller does not own.

from pdum.aws.secrets import SecretStore

store = SecretStore("/myapp/:/org/")
store.put("STRIPE_KEY", "sk_live_...")   # written to /myapp/STRIPE_KEY
store.get("GOOGLE_CLIENT_ID")            # found at /org/GOOGLE_CLIENT_ID

This module does not read .env files, and nothing here touches the filesystem as a side effect of being imported. Env-first resolution looks at os.environ only. When you want the .env layer too, call :func:pdum.aws.env.load_env explicitly — it applies the files and then this store, in that order, and returns a report of what each contributed.

SecretStore

A namespaced view of SSM Parameter Store, layered along a search path.

Parameters:

Name	Type	Description	Default
path	str or sequence of str	SSM search path, most specific prefix first: a colon-separated string ("/myapp/:/org/") or a sequence (["/myapp/", "/org/"]). Required; see the module docstring for why there is no default.	required
region	str	Region for the SSM client. When None, boto3 resolves it from the environment or active profile.	None

Attributes:

Name	Type	Description
prefixes	tuple of str	The normalised search path, each element leading- and trailing-slashed.

Source code in src/pdum/aws/secrets.py

class SecretStore:
    """A namespaced view of SSM Parameter Store, layered along a search path.

    Parameters
    ----------
    path : str or sequence of str
        SSM search path, most specific prefix first: a colon-separated string
        (``"/myapp/:/org/"``) or a sequence (``["/myapp/", "/org/"]``).
        Required; see the module docstring for why there is no default.
    region : str, optional
        Region for the SSM client. When ``None``, ``boto3`` resolves it from the
        environment or active profile.

    Attributes
    ----------
    prefixes : tuple of str
        The normalised search path, each element leading- and trailing-slashed.
    """

    def __init__(self, path: str | Sequence[str], *, region: str | None = None) -> None:
        self.prefixes = _normalise_path(path)
        self._region = region
        self._cache: dict[str, str | None] = {}
        self._ssm: Any = None

    def __repr__(self) -> str:
        return f"{type(self).__name__}(path={':'.join(self.prefixes)!r})"

    @property
    def prefix(self) -> str:
        """The first prefix on the path — the layer writes and deletes target."""
        return self.prefixes[0]

    @property
    def ssm(self) -> Any:
        """The SSM client, created lazily on first use.

        Deferred so that constructing a store, or resolving a secret that is
        already present in the environment, never requires credentials.
        """
        if self._ssm is None:
            self._ssm = client("ssm", self._region)
        return self._ssm

    def path(self, name: str) -> str:
        """Return the full SSM parameter path for *name* in the write layer.

        Parameters
        ----------
        name : str
            Secret name without a prefix.

        Returns
        -------
        str
            The full path, e.g. ``"/myapp/STRIPE_KEY"``.
        """
        return self.prefix + name

    def _read_layer(self, prefix: str, name: str) -> str | None:
        """Fetch ``<prefix><name>`` from SSM, or ``None`` when absent."""
        from botocore.exceptions import ClientError

        try:
            response = self.ssm.get_parameter(Name=prefix + name, WithDecryption=True)
        except ClientError as exc:
            if exc.response["Error"]["Code"] == "ParameterNotFound":
                return None
            raise
        return response["Parameter"]["Value"]

    def get(self, name: str, *, default: str | None = None, required: bool = False) -> str | None:
        """Resolve a secret from the environment, then each prefix, then *default*.

        Successful and missing lookups are both cached for the life of the
        store, so repeated reads cost one walk of the path at most.

        Parameters
        ----------
        name : str
            Secret name without a prefix.
        default : str, optional
            Returned when the secret exists nowhere and ``required`` is false.
        required : bool, default False
            Raise instead of returning *default* when the secret is missing.

        Returns
        -------
        str or None
            The secret value, or *default* when absent.

        Raises
        ------
        KeyError
            If the secret is missing and ``required`` is true.
        """
        env_value = os.environ.get(name)
        if env_value:
            return env_value

        if name not in self._cache:
            value = None
            for prefix in self.prefixes:
                value = self._read_layer(prefix, name)
                if value is not None:
                    break
            self._cache[name] = value

        value = self._cache[name]
        if value is None:
            if required:
                searched = ", ".join(p + name for p in self.prefixes)
                raise KeyError(f"secret {name!r} not found in the environment or SSM at {searched}")
            return default
        return value

    def get_json(self, name: str, *, default: Any = None, required: bool = False) -> Any:
        """Resolve a secret and parse it as JSON.

        Parameters
        ----------
        name : str
            Secret name without a prefix.
        default : Any, optional
            Returned when the secret is absent and ``required`` is false.
        required : bool, default False
            Raise instead of returning *default* when the secret is missing.

        Returns
        -------
        Any
            The parsed JSON value, or *default* when absent.

        Raises
        ------
        KeyError
            If the secret is missing and ``required`` is true.
        json.JSONDecodeError
            If the stored value is not valid JSON.
        """
        raw = self.get(name, default=None, required=required)
        return json.loads(raw) if raw else default

    def put(self, name: str, value: str, *, secure: bool = True) -> None:
        """Create or overwrite a secret in the write layer, updating the cache.

        Parameters
        ----------
        name : str
            Secret name without a prefix.
        value : str
            Value to store.
        secure : bool, default True
            Store as a KMS-encrypted ``SecureString``. Pass ``False`` for
            non-sensitive configuration that benefits from the same namespace.
        """
        self.ssm.put_parameter(
            Name=self.path(name),
            Value=value,
            Type="SecureString" if secure else "String",
            Overwrite=True,
        )
        self._cache[name] = value

    def delete(self, name: str) -> None:
        """Delete a secret from the write layer and drop it from the cache.

        Only the first prefix is ever deleted from. When the name exists solely
        in a fallback layer, this raises rather than reaching down: deleting
        from a shared layer must be asked for explicitly, with a store whose
        path starts there.

        Parameters
        ----------
        name : str
            Secret name without a prefix.

        Raises
        ------
        KeyError
            If the secret is not in the write layer. The message names the
            fallback layer holding it, when there is one.
        """
        from botocore.exceptions import ClientError

        try:
            self.ssm.delete_parameter(Name=self.path(name))
        except ClientError as exc:
            if exc.response["Error"]["Code"] != "ParameterNotFound":
                raise
            for fallback in self.prefixes[1:]:
                if self._read_layer(fallback, name) is not None:
                    raise KeyError(
                        f"secret {name!r} is not in the write layer {self.prefix}; it lives at "
                        f"{fallback + name} — use a store with path {fallback!r} to delete it there"
                    ) from exc
            raise KeyError(f"secret {name!r} not found under {':'.join(self.prefixes)}") from exc
        self._cache.pop(name, None)

    def names(self) -> list[str]:
        """List the secret names visible to this store, without prefixes.

        Names present in several layers appear once.

        Returns
        -------
        list of str
            Secret names, unsorted.
        """
        return [row["name"] for row in self.describe() if not row["shadowed"]]

    def read_all(self) -> dict[str, str]:
        """Read every secret on the search path in one pass, first layer winning.

        Costs one paginated call per layer rather than one call per name, which
        is what makes materialising a whole store — into a process environment,
        say — cheap enough to do on every invocation. Unlike :meth:`describe`
        this decrypts, so it does pull values over the wire.

        The environment is deliberately **not** consulted, unlike :meth:`get`:
        this is the store's own view, leaving the caller to decide how it should
        interact with variables that are already set.

        Values are cached as if each name had been fetched individually, so a
        later :meth:`get` for any of them costs nothing.

        Returns
        -------
        dict of str to str
            Secret name to value, shadowed layers excluded.
        """
        paginator = self.ssm.get_paginator("get_parameters_by_path")
        out: dict[str, str] = {}
        for prefix in self.prefixes:
            for page in paginator.paginate(Path=prefix, Recursive=True, WithDecryption=True):
                for p in page["Parameters"]:
                    out.setdefault(p["Name"][len(prefix) :], p["Value"])
        self._cache.update(out)
        return out

    def describe(self) -> list[dict[str, Any]]:
        """List metadata for each secret on the search path.

        Deliberately does not decrypt: this reads names, types and timestamps
        only, so rendering a listing never pulls secret values over the wire.

        Returns
        -------
        list of dict
            One dict per parameter per layer, with ``name``, ``type``,
            ``version``, ``modified``, ``origin`` (the prefix holding it) and
            ``shadowed`` (true when an earlier layer also has the name, so
            resolution never reaches this row).
        """
        paginator = self.ssm.get_paginator("get_parameters_by_path")
        out: list[dict[str, Any]] = []
        seen: set[str] = set()
        for prefix in self.prefixes:
            for page in paginator.paginate(Path=prefix, Recursive=True):
                for p in page["Parameters"]:
                    name = p["Name"][len(prefix) :]
                    out.append(
                        {
                            "name": name,
                            "type": p["Type"],
                            "version": p["Version"],
                            "modified": p["LastModifiedDate"],
                            "origin": prefix,
                            "shadowed": name in seen,
                        }
                    )
            seen.update(row["name"] for row in out)
        return out

    def clear_cache(self) -> None:
        """Drop cached values so the next read goes back to SSM.

        Needed when another process has rotated a secret during this one's
        lifetime.
        """
        self._cache.clear()

prefix property

The first prefix on the path — the layer writes and deletes target.

ssm property

The SSM client, created lazily on first use.

Deferred so that constructing a store, or resolving a secret that is already present in the environment, never requires credentials.

clear_cache()

Drop cached values so the next read goes back to SSM.

Needed when another process has rotated a secret during this one's lifetime.

Source code in src/pdum/aws/secrets.py

def clear_cache(self) -> None:
    """Drop cached values so the next read goes back to SSM.

    Needed when another process has rotated a secret during this one's
    lifetime.
    """
    self._cache.clear()

delete(name)

Delete a secret from the write layer and drop it from the cache.

Only the first prefix is ever deleted from. When the name exists solely in a fallback layer, this raises rather than reaching down: deleting from a shared layer must be asked for explicitly, with a store whose path starts there.

Parameters:

Name	Type	Description	Default
name	str	Secret name without a prefix.	required

Raises:

Type	Description
KeyError	If the secret is not in the write layer. The message names the fallback layer holding it, when there is one.

Source code in src/pdum/aws/secrets.py

def delete(self, name: str) -> None:
    """Delete a secret from the write layer and drop it from the cache.

    Only the first prefix is ever deleted from. When the name exists solely
    in a fallback layer, this raises rather than reaching down: deleting
    from a shared layer must be asked for explicitly, with a store whose
    path starts there.

    Parameters
    ----------
    name : str
        Secret name without a prefix.

    Raises
    ------
    KeyError
        If the secret is not in the write layer. The message names the
        fallback layer holding it, when there is one.
    """
    from botocore.exceptions import ClientError

    try:
        self.ssm.delete_parameter(Name=self.path(name))
    except ClientError as exc:
        if exc.response["Error"]["Code"] != "ParameterNotFound":
            raise
        for fallback in self.prefixes[1:]:
            if self._read_layer(fallback, name) is not None:
                raise KeyError(
                    f"secret {name!r} is not in the write layer {self.prefix}; it lives at "
                    f"{fallback + name} — use a store with path {fallback!r} to delete it there"
                ) from exc
        raise KeyError(f"secret {name!r} not found under {':'.join(self.prefixes)}") from exc
    self._cache.pop(name, None)

describe()

List metadata for each secret on the search path.

Deliberately does not decrypt: this reads names, types and timestamps only, so rendering a listing never pulls secret values over the wire.

Returns:

Type	Description
list of dict	One dict per parameter per layer, with name, type, version, modified, origin (the prefix holding it) and shadowed (true when an earlier layer also has the name, so resolution never reaches this row).

Source code in src/pdum/aws/secrets.py

def describe(self) -> list[dict[str, Any]]:
    """List metadata for each secret on the search path.

    Deliberately does not decrypt: this reads names, types and timestamps
    only, so rendering a listing never pulls secret values over the wire.

    Returns
    -------
    list of dict
        One dict per parameter per layer, with ``name``, ``type``,
        ``version``, ``modified``, ``origin`` (the prefix holding it) and
        ``shadowed`` (true when an earlier layer also has the name, so
        resolution never reaches this row).
    """
    paginator = self.ssm.get_paginator("get_parameters_by_path")
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    for prefix in self.prefixes:
        for page in paginator.paginate(Path=prefix, Recursive=True):
            for p in page["Parameters"]:
                name = p["Name"][len(prefix) :]
                out.append(
                    {
                        "name": name,
                        "type": p["Type"],
                        "version": p["Version"],
                        "modified": p["LastModifiedDate"],
                        "origin": prefix,
                        "shadowed": name in seen,
                    }
                )
        seen.update(row["name"] for row in out)
    return out

get(name, *, default=None, required=False)

Resolve a secret from the environment, then each prefix, then default.

Successful and missing lookups are both cached for the life of the store, so repeated reads cost one walk of the path at most.

Parameters:

Name	Type	Description	Default
name	str	Secret name without a prefix.	required
default	str	Returned when the secret exists nowhere and required is false.	None
required	bool	Raise instead of returning default when the secret is missing.	False

Returns:

Type	Description
str or None	The secret value, or default when absent.

Raises:

Type	Description
KeyError	If the secret is missing and required is true.

Source code in src/pdum/aws/secrets.py

def get(self, name: str, *, default: str | None = None, required: bool = False) -> str | None:
    """Resolve a secret from the environment, then each prefix, then *default*.

    Successful and missing lookups are both cached for the life of the
    store, so repeated reads cost one walk of the path at most.

    Parameters
    ----------
    name : str
        Secret name without a prefix.
    default : str, optional
        Returned when the secret exists nowhere and ``required`` is false.
    required : bool, default False
        Raise instead of returning *default* when the secret is missing.

    Returns
    -------
    str or None
        The secret value, or *default* when absent.

    Raises
    ------
    KeyError
        If the secret is missing and ``required`` is true.
    """
    env_value = os.environ.get(name)
    if env_value:
        return env_value

    if name not in self._cache:
        value = None
        for prefix in self.prefixes:
            value = self._read_layer(prefix, name)
            if value is not None:
                break
        self._cache[name] = value

    value = self._cache[name]
    if value is None:
        if required:
            searched = ", ".join(p + name for p in self.prefixes)
            raise KeyError(f"secret {name!r} not found in the environment or SSM at {searched}")
        return default
    return value

get_json(name, *, default=None, required=False)

Resolve a secret and parse it as JSON.

Parameters:

Name	Type	Description	Default
name	str	Secret name without a prefix.	required
default	Any	Returned when the secret is absent and required is false.	None
required	bool	Raise instead of returning default when the secret is missing.	False

Returns:

Type	Description
Any	The parsed JSON value, or default when absent.

Raises:

Type	Description
KeyError	If the secret is missing and required is true.
JSONDecodeError	If the stored value is not valid JSON.

Source code in src/pdum/aws/secrets.py

def get_json(self, name: str, *, default: Any = None, required: bool = False) -> Any:
    """Resolve a secret and parse it as JSON.

    Parameters
    ----------
    name : str
        Secret name without a prefix.
    default : Any, optional
        Returned when the secret is absent and ``required`` is false.
    required : bool, default False
        Raise instead of returning *default* when the secret is missing.

    Returns
    -------
    Any
        The parsed JSON value, or *default* when absent.

    Raises
    ------
    KeyError
        If the secret is missing and ``required`` is true.
    json.JSONDecodeError
        If the stored value is not valid JSON.
    """
    raw = self.get(name, default=None, required=required)
    return json.loads(raw) if raw else default

names()

List the secret names visible to this store, without prefixes.

Names present in several layers appear once.

Returns:

Type	Description
list of str	Secret names, unsorted.

Source code in src/pdum/aws/secrets.py

def names(self) -> list[str]:
    """List the secret names visible to this store, without prefixes.

    Names present in several layers appear once.

    Returns
    -------
    list of str
        Secret names, unsorted.
    """
    return [row["name"] for row in self.describe() if not row["shadowed"]]

path(name)

Return the full SSM parameter path for name in the write layer.

Parameters:

Name	Type	Description	Default
name	str	Secret name without a prefix.	required

Returns:

Type	Description
str	The full path, e.g. "/myapp/STRIPE_KEY".

Source code in src/pdum/aws/secrets.py

def path(self, name: str) -> str:
    """Return the full SSM parameter path for *name* in the write layer.

    Parameters
    ----------
    name : str
        Secret name without a prefix.

    Returns
    -------
    str
        The full path, e.g. ``"/myapp/STRIPE_KEY"``.
    """
    return self.prefix + name

put(name, value, *, secure=True)

Create or overwrite a secret in the write layer, updating the cache.

Parameters:

Name	Type	Description	Default
name	str	Secret name without a prefix.	required
value	str	Value to store.	required
secure	bool	Store as a KMS-encrypted SecureString. Pass False for non-sensitive configuration that benefits from the same namespace.	True

Source code in src/pdum/aws/secrets.py

def put(self, name: str, value: str, *, secure: bool = True) -> None:
    """Create or overwrite a secret in the write layer, updating the cache.

    Parameters
    ----------
    name : str
        Secret name without a prefix.
    value : str
        Value to store.
    secure : bool, default True
        Store as a KMS-encrypted ``SecureString``. Pass ``False`` for
        non-sensitive configuration that benefits from the same namespace.
    """
    self.ssm.put_parameter(
        Name=self.path(name),
        Value=value,
        Type="SecureString" if secure else "String",
        Overwrite=True,
    )
    self._cache[name] = value

read_all()

Read every secret on the search path in one pass, first layer winning.

Costs one paginated call per layer rather than one call per name, which is what makes materialising a whole store — into a process environment, say — cheap enough to do on every invocation. Unlike :meth:describe this decrypts, so it does pull values over the wire.

The environment is deliberately not consulted, unlike :meth:get: this is the store's own view, leaving the caller to decide how it should interact with variables that are already set.

Values are cached as if each name had been fetched individually, so a later :meth:get for any of them costs nothing.

Returns:

Type	Description
dict of str to str	Secret name to value, shadowed layers excluded.

Source code in src/pdum/aws/secrets.py

def read_all(self) -> dict[str, str]:
    """Read every secret on the search path in one pass, first layer winning.

    Costs one paginated call per layer rather than one call per name, which
    is what makes materialising a whole store — into a process environment,
    say — cheap enough to do on every invocation. Unlike :meth:`describe`
    this decrypts, so it does pull values over the wire.

    The environment is deliberately **not** consulted, unlike :meth:`get`:
    this is the store's own view, leaving the caller to decide how it should
    interact with variables that are already set.

    Values are cached as if each name had been fetched individually, so a
    later :meth:`get` for any of them costs nothing.

    Returns
    -------
    dict of str to str
        Secret name to value, shadowed layers excluded.
    """
    paginator = self.ssm.get_paginator("get_parameters_by_path")
    out: dict[str, str] = {}
    for prefix in self.prefixes:
        for page in paginator.paginate(Path=prefix, Recursive=True, WithDecryption=True):
            for p in page["Parameters"]:
                out.setdefault(p["Name"][len(prefix) :], p["Value"])
    self._cache.update(out)
    return out

pdum.aws.env

Building a process environment from .env files and SSM, in layers.

This is what pdx does, available to any script that wants the same environment without being launched under it:

from pdum.aws import load_env

load_env()   # os.environ now carries the project's variables and its secrets

Use it where you would otherwise call dotenv.load_dotenv(). It does strictly more: the same .env handling, plus the .env.local overlay, plus the secrets from the SSM search path that the .env usually names.

Three layers, most specific first

1. the environment you already have — never overwritten, so a variable exported for one run wins, and a CI job's injected variables are never undone;
2. the nearest .env, overlaid with an adjacent .env.local — searched for upward from the working directory, so this works from anywhere inside a project. The whole file is loaded, not just the SSM path: a project's .env is where AWS_PROFILE and AWS_REGION live too, and without those the store cannot be reached at all. That is why these are applied before the store is opened;
3. the SSM store, on the search path named by PDUM_SSM_PATH — which is itself usually set by that .env.

Nothing here runs on import. :mod:pdum.aws.secrets says that a library must not read .env behind a plain import, and that still holds — this module provides the function and lets the application decide when to call it. Being explicit is also what makes the ordering above something you can rely on.

.env and .env.local

Only .env is searched for; the overlay is taken from beside whatever was found, never searched for separately, so one directory always wins and there is never a question of which .env.local applies. A .env.local with no .env to anchor it is ignored.

The two are read as one document rather than as two dicts to merge, so that a ${VAR} in the local file can refer to a name defined in the committed one — which is most of the point of having a local file. Merging parsed results would leave that reference unresolved.

Parsing goes through python-dotenv rather than a hand-rolled KEY=VALUE split, because the format has more corners than it appears to: export prefixes, inline comments, single versus double quoting, and quoted values spanning several lines. A naive split reads

PDUM_SSM_PATH=/myapp/:/org/   # the project's own layer

as a prefix with a comment glued onto the end, and SSM would take that at face value.

Interpolation of ${VAR} is the caller's choice, because the two callers want opposite things. Loading an environment should expand, as every other .env consumer does. secrets import should not: a value on its way into permanent storage must arrive exactly as written, or the password p@ss${word}word silently becomes p@ssword and nothing records that it ever said more.

EnvReport dataclass

What :func:load_env did, layer by layer.

Attributes:

Name	Type	Description
env_files	LoadReport or None	What the .env and its overlay contributed, or None when there was no such file.
path	str	The SSM search path that was used.
path_source	str	Where that path came from, for reporting.
secrets	LoadReport	What the store contributed.

Source code in src/pdum/aws/env.py

@dataclass(frozen=True, slots=True)
class EnvReport:
    """What :func:`load_env` did, layer by layer.

    Attributes
    ----------
    env_files : LoadReport or None
        What the ``.env`` and its overlay contributed, or ``None`` when there
        was no such file.
    path : str
        The SSM search path that was used.
    path_source : str
        Where that path came from, for reporting.
    secrets : LoadReport
        What the store contributed.
    """

    env_files: LoadReport | None
    path: str
    path_source: str
    secrets: LoadReport

LoadReport dataclass

What one layer contributed to the environment.

Attributes:

Name	Type	Description
source	str	Where the values came from, for reporting: a file path, or a search path.
applied	list of str	Names this layer set.
already_set	list of str	Names it offered that the environment already had, so were kept. A more specific layer had already won.
skipped	list of str	Names unusable as environment variables, e.g. a nested db/PASSWORD.

Source code in src/pdum/aws/env.py

@dataclass(frozen=True, slots=True)
class LoadReport:
    """What one layer contributed to the environment.

    Attributes
    ----------
    source : str
        Where the values came from, for reporting: a file path, or a search
        path.
    applied : list of str
        Names this layer set.
    already_set : list of str
        Names it offered that the environment already had, so were kept. A more
        specific layer had already won.
    skipped : list of str
        Names unusable as environment variables, e.g. a nested ``db/PASSWORD``.
    """

    source: str
    applied: list[str]
    already_set: list[str]
    skipped: list[str]

NoSearchPath

Bases: LookupError

No SSM search path could be resolved, so no secrets could be loaded.

Raised rather than passed over: a process that quietly continues without the secrets it asked for fails later, somewhere less informative.

Source code in src/pdum/aws/env.py

class NoSearchPath(LookupError):
    """No SSM search path could be resolved, so no secrets could be loaded.

    Raised rather than passed over: a process that quietly continues without the
    secrets it asked for fails later, somewhere less informative.
    """

find_env_file(name='.env')

Find name in the working directory or the nearest ancestor holding one.

Searching upward is what lets this run from anywhere inside a project and still find that project's file, the way git finds its repository. A name with a directory component is taken literally instead, so an explicit config/.env is never second-guessed.

Parameters:

Name	Type	Description	Default
name	str	File name to search for, or a path to use as given.	".env"

Returns:

Type	Description
Path or None	The file found, or None when there is none.

Source code in src/pdum/aws/env.py

def find_env_file(name: str = ".env") -> Path | None:
    """Find *name* in the working directory or the nearest ancestor holding one.

    Searching upward is what lets this run from anywhere inside a project and
    still find that project's file, the way ``git`` finds its repository. A
    *name* with a directory component is taken literally instead, so an explicit
    ``config/.env`` is never second-guessed.

    Parameters
    ----------
    name : str, default ".env"
        File name to search for, or a path to use as given.

    Returns
    -------
    pathlib.Path or None
        The file found, or ``None`` when there is none.
    """
    candidate = Path(name)
    if candidate.parent != Path():
        return candidate if candidate.is_file() else None
    # usecwd=True is essential: python-dotenv otherwise walks up from the file
    # that called it, which here would be this library rather than the project.
    found = find_dotenv(candidate.name, usecwd=True)
    return Path(found) if found else None

find_env_files(name='.env')

Find name and its adjacent .local overlay, least specific first.

Parameters:

Name	Type	Description	Default
name	str	File name to search for, or a path to use as given.	".env"

Returns:

Type	Description
list of pathlib.Path	Empty when name was not found, otherwise the file and, if it exists, its .local neighbour.

Source code in src/pdum/aws/env.py

def find_env_files(name: str = ".env") -> list[Path]:
    """Find *name* and its adjacent ``.local`` overlay, least specific first.

    Parameters
    ----------
    name : str, default ".env"
        File name to search for, or a path to use as given.

    Returns
    -------
    list of pathlib.Path
        Empty when *name* was not found, otherwise the file and, if it exists,
        its ``.local`` neighbour.
    """
    base = find_env_file(name)
    if base is None:
        return []
    local = base.with_name(base.name + LOCAL_SUFFIX)
    return [base, local] if local.is_file() else [base]

is_env_name(name)

Whether name can be used as an environment variable.

A store may hold nested parameters, whose names carry a slash — a secret at /myapp/db/PASSWORD is named db/PASSWORD. Exporting that would make a variable no program could name, so such names are reported and skipped rather than set.

Parameters:

Name	Type	Description	Default
name	str	Candidate variable name.	required

Returns:

Type	Description
bool	True when a shell could name it.

Source code in src/pdum/aws/env.py

def is_env_name(name: str) -> bool:
    """Whether *name* can be used as an environment variable.

    A store may hold nested parameters, whose names carry a slash — a secret at
    ``/myapp/db/PASSWORD`` is named ``db/PASSWORD``. Exporting that would make a
    variable no program could name, so such names are reported and skipped
    rather than set.

    Parameters
    ----------
    name : str
        Candidate variable name.

    Returns
    -------
    bool
        True when a shell could name it.
    """
    return _ENV_NAME.match(name) is not None

load_env(*, environ=None, env_file='.env', envvar=DEFAULT_ENVVAR, default_path=None, store_factory=SecretStore)

Load the project's .env files and its secrets into the environment.

The drop-in for dotenv.load_dotenv() in a project that keeps its secrets in SSM. Layers are applied most-specific-last-wins-least: whatever is already set stays, then the .env pair fills gaps, then the store fills what remains. See the module docstring for why the order matters.

Parameters:

Name	Type	Description	Default
environ	mutable mapping	Environment to update; defaults to os.environ. Pass a plain dict to compute an environment without touching this process's own.	None
env_file	str	File to search for upward from the working directory, and the base name of the .local overlay beside it. A name with a directory component is used as given.	".env"
envvar	str	Variable naming the SSM search path. None leaves only default_path.	DEFAULT_ENVVAR
default_path	str or callable	Search path to fall back on when the environment, including the .env, has none. A callable is resolved here.	None
store_factory	callable	Called with the resolved path to build the store. Override to pin a region, e.g. lambda path: SecretStore(path, region="us-east-1").	:class:`~pdum.aws.secrets.SecretStore`

Returns:

Type	Description
EnvReport	What each layer contributed. Ignore it for the common case; it is there for reporting and for tests.

Raises:

Type	Description
NoSearchPath	If no search path could be resolved. Loading a .env and silently skipping the secrets would leave a process that fails later, somewhere less informative.
ValueError	If the resolved path is not a usable SSM prefix — a root path, say.

Source code in src/pdum/aws/env.py

def load_env(
    *,
    environ: MutableMapping[str, str] | None = None,
    env_file: str = ".env",
    envvar: str | None = DEFAULT_ENVVAR,
    default_path: PathDefault = None,
    store_factory: Callable[[str], SecretStore] = SecretStore,
) -> EnvReport:
    """Load the project's ``.env`` files and its secrets into the environment.

    The drop-in for ``dotenv.load_dotenv()`` in a project that keeps its secrets
    in SSM. Layers are applied most-specific-last-wins-least: whatever is
    already set stays, then the ``.env`` pair fills gaps, then the store fills
    what remains. See the module docstring for why the order matters.

    Parameters
    ----------
    environ : mutable mapping, optional
        Environment to update; defaults to ``os.environ``. Pass a plain dict to
        compute an environment without touching this process's own.
    env_file : str, default ".env"
        File to search for upward from the working directory, and the base name
        of the ``.local`` overlay beside it. A name with a directory component
        is used as given.
    envvar : str, optional
        Variable naming the SSM search path. ``None`` leaves only
        *default_path*.
    default_path : str or callable, optional
        Search path to fall back on when the environment, including the
        ``.env``, has none. A callable is resolved here.
    store_factory : callable, default :class:`~pdum.aws.secrets.SecretStore`
        Called with the resolved path to build the store. Override to pin a
        region, e.g. ``lambda path: SecretStore(path, region="us-east-1")``.

    Returns
    -------
    EnvReport
        What each layer contributed. Ignore it for the common case; it is there
        for reporting and for tests.

    Raises
    ------
    NoSearchPath
        If no search path could be resolved. Loading a ``.env`` and silently
        skipping the secrets would leave a process that fails later, somewhere
        less informative.
    ValueError
        If the resolved path is not a usable SSM prefix — a root path, say.
    """
    target = os.environ if environ is None else environ
    from_files = load_env_files(env_file, target)
    path, source = resolve_search_path(
        envvar=envvar, default_path=default_path, environ=target, env_file_report=from_files
    )
    if not path:
        raise NoSearchPath(no_search_path_message(envvar, env_file, from_files))
    return EnvReport(
        env_files=from_files,
        path=path,
        path_source=source,
        secrets=load_secrets(store_factory(path), target),
    )

load_env_files(env_file, environ)

Apply the nearest .env, and its .local overlay, to environ.

Every variable in the files is loaded, not only the SSM search path. Call this before opening a store, or the store may be opened against the wrong account, or no account at all — see the module docstring.

Parameters:

Name	Type	Description	Default
env_file	str	File name to search for upward from the working directory, or a path with a directory component to use as given.	required
environ	mutable mapping	Environment to update, normally os.environ.	required

Returns:

Type	Description
LoadReport or None	What the files contributed, or None when there is no such file.

Source code in src/pdum/aws/env.py

def load_env_files(env_file: str, environ: MutableMapping[str, str]) -> LoadReport | None:
    """Apply the nearest ``.env``, and its ``.local`` overlay, to *environ*.

    Every variable in the files is loaded, not only the SSM search path. Call
    this before opening a store, or the store may be opened against the wrong
    account, or no account at all — see the module docstring.

    Parameters
    ----------
    env_file : str
        File name to search for upward from the working directory, or a path
        with a directory component to use as given.
    environ : mutable mapping
        Environment to update, normally ``os.environ``.

    Returns
    -------
    LoadReport or None
        What the files contributed, or ``None`` when there is no such file.
    """
    found = find_env_files(env_file)
    if not found:
        return None
    base, *overlay = found
    source = f"{base} + {overlay[0].name}" if overlay else str(base)
    return _apply(read_env_files(found, interpolate=True), environ, source)

load_secrets(store, environ)

Apply store to environ, in one bulk read, keeping what is already set.

Parameters:

Name	Type	Description	Default
store	SecretStore	Store to read.	required
environ	mutable mapping	Environment to update, normally os.environ.	required

Returns:

Type	Description
LoadReport	What the store contributed.

Source code in src/pdum/aws/env.py

def load_secrets(store: SecretStore, environ: MutableMapping[str, str]) -> LoadReport:
    """Apply *store* to *environ*, in one bulk read, keeping what is already set.

    Parameters
    ----------
    store : SecretStore
        Store to read.
    environ : mutable mapping
        Environment to update, normally ``os.environ``.

    Returns
    -------
    LoadReport
        What the store contributed.
    """
    return _apply(store.read_all(), environ, ":".join(store.prefixes))

no_search_path_message(envvar, env_file, found)

Explain every way the search path could have been supplied.

Source code in src/pdum/aws/env.py

def no_search_path_message(envvar: str | None, env_file: str, found: LoadReport | None) -> str:
    """Explain every way the search path could have been supplied."""
    if not envvar:
        return "no SSM search path configured, and no default was supplied"
    where = f"add {envvar}= to {found.source}" if found else f"add {envvar}= to a {env_file} in this project"
    return f"no SSM search path configured: set {envvar}, or {where}"

read_env_file(path, *, interpolate=True)

Read one .env file. See :func:read_env_files.

Source code in src/pdum/aws/env.py

def read_env_file(path: Path | str, *, interpolate: bool = True) -> dict[str, str]:
    """Read one ``.env`` file. See :func:`read_env_files`."""
    return read_env_files([path], interpolate=interpolate)

read_env_files(paths, *, interpolate=True)

Read several .env files as though they were one, later files winning.

The files are concatenated and parsed once, so interpolation reaches across them and a repeated name resolves the way a repeated name inside a single file does — the last one wins.

Parameters:

Name	Type	Description	Default
paths	sequence of pathlib.Path or str	Files to read, least specific first.	required
interpolate	bool	Expand ${VAR} against the environment and earlier entries, as python-dotenv does by default. Pass False when the values are headed somewhere permanent; see the module docstring.	True

Returns:

Type	Description
dict of str to str	Names mapped to values. A bare name declared with no = at all is dropped rather than reported as None, so every value here is a string the caller can use directly.

Source code in src/pdum/aws/env.py

def read_env_files(paths: Sequence[Path | str], *, interpolate: bool = True) -> dict[str, str]:
    """Read several ``.env`` files as though they were one, later files winning.

    The files are concatenated and parsed once, so interpolation reaches across
    them and a repeated name resolves the way a repeated name inside a single
    file does — the last one wins.

    Parameters
    ----------
    paths : sequence of pathlib.Path or str
        Files to read, least specific first.
    interpolate : bool, default True
        Expand ``${VAR}`` against the environment and earlier entries, as
        ``python-dotenv`` does by default. Pass ``False`` when the values are
        headed somewhere permanent; see the module docstring.

    Returns
    -------
    dict of str to str
        Names mapped to values. A bare name declared with no ``=`` at all is
        dropped rather than reported as ``None``, so every value here is a
        string the caller can use directly.
    """
    # Explicit utf-8, and an explicit separator: Path.read_text would otherwise
    # use the locale encoding, and a file with no trailing newline would glue
    # its last line onto the next file's first.
    text = "\n".join(Path(path).read_text(encoding="utf-8") for path in paths)
    values = dotenv_values(stream=io.StringIO(text), interpolate=interpolate)
    return {key: value for key, value in values.items() if value is not None}

resolve_search_path(*, envvar=DEFAULT_ENVVAR, default_path=None, environ, env_file_report=None)

Work out which SSM search path to use, and say where it came from.

Call after :func:load_env_files, so that a path set by a file is already in environ and needs no separate lookup — it is one environment variable among all the others by this point. env_file_report is used only to describe the source precisely.

Parameters:

Name	Type	Description	Default
envvar	str	Variable holding the path. None leaves only default_path.	DEFAULT_ENVVAR
default_path	str or callable	Fallback when the environment has none; a callable is resolved here.	None
environ	mapping	Environment to consult, normally os.environ.	required
env_file_report	LoadReport	Result of :func:load_env_files, to tell a path that came from a file from one that was already exported.	None

Returns:

Type	Description
tuple of (str or None, str)	The path, and a short description of its source for reporting.

Source code in src/pdum/aws/env.py

def resolve_search_path(
    *,
    envvar: str | None = DEFAULT_ENVVAR,
    default_path: PathDefault = None,
    environ: Mapping[str, str],
    env_file_report: LoadReport | None = None,
) -> tuple[str | None, str]:
    """Work out which SSM search path to use, and say where it came from.

    Call after :func:`load_env_files`, so that a path set by a file is already
    in *environ* and needs no separate lookup — it is one environment variable
    among all the others by this point. *env_file_report* is used only to
    describe the source precisely.

    Parameters
    ----------
    envvar : str, optional
        Variable holding the path. ``None`` leaves only *default_path*.
    default_path : str or callable, optional
        Fallback when the environment has none; a callable is resolved here.
    environ : mapping
        Environment to consult, normally ``os.environ``.
    env_file_report : LoadReport, optional
        Result of :func:`load_env_files`, to tell a path that came from a file
        from one that was already exported.

    Returns
    -------
    tuple of (str or None, str)
        The path, and a short description of its source for reporting.
    """
    if envvar and environ.get(envvar):
        if env_file_report is not None and envvar in env_file_report.applied:
            return environ[envvar], env_file_report.source
        return environ[envvar], f"${envvar}"
    resolved = default_path() if callable(default_path) else default_path
    return (resolved, "built-in default") if resolved else (None, "")

pdum.aws.quotas

Service Quotas: inspect current limits and request increases.

A fresh AWS account can launch almost nothing — typically 5 vCPUs of standard on-demand EC2 and zero of every accelerator family. Raising that is a per-region, per-quota request process with several non-obvious traps, which this module encodes.

from pdum.aws import quotas

statuses = quotas.report(quotas.EC2_VCPU_TARGETS, region="us-east-1")
for s in statuses:
    print(s.target.label, s.current, s.state)

results = quotas.submit(quotas.EC2_VCPU_TARGETS, region="us-east-1")

Functions here return data and never print, so callers own presentation.

Things that surprise people

EC2 quotas count vCPUs, not instances. "I want 4 GPUs" becomes "I want 768 vCPUs of P", because p5.48xlarge is 192 vCPUs. Convert through the instance type you actually intend to run.

Instance generation is not a quota dimension. P spans p3 (V100), p4 (A100) and p5 (H100); G spans g4dn (T4), g5 (A10G), g6 (L4) and g6e (L40S). Old and new hardware cannot be requested separately.

Nor is CPU architecture. Graviton (c7g, c8g, m7g, r7g) draws from the Standard pool, whose name lists instance-family letters (A, C, D, H, I, M, R, T, Z), not architectures — the A is a1, not "ARM".

Quota is a ceiling, not a commitment — and not capacity. Holding a large quota costs nothing, so asking high has no downside but refusal. But an approved P quota does not mean p5 will launch; on-demand H100 is frequently unavailable regardless of quota, and Capacity Blocks for ML is the mechanism that actually reserves that class of hardware.

Rate limits on requesting

Two meta-quotas govern the requests themselves:

• L-36BDD542 "Active requests per quota" is 1, so an in-flight request cannot be revised — you wait for it to be decided.
• An undocumented account-wide cap of :data:ACCOUNT_OPEN_REQUEST_CAP open requests, which is not readable from any API and surfaces only as QuotaExceededException. It held at exactly 20 on two separate accounts.

So a large batch cannot be submitted in one sitting. :func:submit stops cleanly at the cap and is idempotent, making the workflow: submit, wait for cases to be decided, submit again.

Looking up quota codes

Never hand-type a quota code; a wrong one is a wasted support case.

$ aws service-quotas list-service-quotas --service-code ec2 \
    --query 'Quotas[].[QuotaCode,QuotaName,Value]' --output text

Note that list-service-quotas returns only quotas with an applied value, so ones never modified can be missing. Use list-aws-default-service-quotas to enumerate the full catalogue, and check Adjustable before requesting — some quotas cannot be raised at all.

QuotaRequest dataclass

One quota-increase request as Service Quotas recorded it.

Attributes:

Name	Type	Description
code	str	Quota code the request targets.
name	str	Human-readable quota name from the API.
desired	float	Value that was asked for.
status	str	PENDING, CASE_OPENED, APPROVED, DENIED, CASE_CLOSED, NOT_APPROVED or INVALID_REQUEST.
request_id	str or None	Service Quotas request id.
case_id	str or None	Support case id, present once a request has been escalated to a human.
created	datetime or None	When the request was raised.
last_updated	datetime or None	When it last changed status.

Source code in src/pdum/aws/quotas.py

@dataclass(frozen=True, slots=True)
class QuotaRequest:
    """One quota-increase request as Service Quotas recorded it.

    Attributes
    ----------
    code : str
        Quota code the request targets.
    name : str
        Human-readable quota name from the API.
    desired : float
        Value that was asked for.
    status : str
        ``PENDING``, ``CASE_OPENED``, ``APPROVED``, ``DENIED``, ``CASE_CLOSED``,
        ``NOT_APPROVED`` or ``INVALID_REQUEST``.
    request_id : str or None
        Service Quotas request id.
    case_id : str or None
        Support case id, present once a request has been escalated to a human.
    created : datetime or None
        When the request was raised.
    last_updated : datetime or None
        When it last changed status.
    """

    code: str
    name: str
    desired: float
    status: str
    request_id: str | None = None
    case_id: str | None = None
    created: datetime | None = None
    last_updated: datetime | None = None

    @property
    def is_open(self) -> bool:
        """Whether this request still awaits a decision."""
        return self.status in OPEN_STATUSES

    @property
    def is_approved(self) -> bool:
        """Whether this request was granted.

        An unapproved request does not imply the quota is unchanged. AWS also
        raises quotas on young accounts automatically, independently of any
        request — observed raising standard EC2 vCPU limits in a region where no
        request had been submitted at all. Always compare against the applied
        value from :func:`current_values` rather than inferring it from status.
        """
        return self.status == "APPROVED"

is_approved property

Whether this request was granted.

An unapproved request does not imply the quota is unchanged. AWS also raises quotas on young accounts automatically, independently of any request — observed raising standard EC2 vCPU limits in a region where no request had been submitted at all. Always compare against the applied value from :func:current_values rather than inferring it from status.

is_open property

Whether this request still awaits a decision.

QuotaStatus dataclass

Where one quota stands relative to its target.

Attributes:

Name	Type	Description
target	QuotaTarget	The quota and desired value.
current	float or None	Currently applied value, or None if the quota does not exist in this region.
pending_value	float or None	Desired value of an in-flight request, if any.
pending_status	str or None	Status of that in-flight request, e.g. "CASE_OPENED".

Source code in src/pdum/aws/quotas.py

@dataclass(frozen=True, slots=True)
class QuotaStatus:
    """Where one quota stands relative to its target.

    Attributes
    ----------
    target : QuotaTarget
        The quota and desired value.
    current : float or None
        Currently applied value, or ``None`` if the quota does not exist in this
        region.
    pending_value : float or None
        Desired value of an in-flight request, if any.
    pending_status : str or None
        Status of that in-flight request, e.g. ``"CASE_OPENED"``.
    """

    target: QuotaTarget
    current: float | None
    pending_value: float | None = None
    pending_status: str | None = None

    @property
    def satisfied(self) -> bool:
        """Whether the applied value already meets the target."""
        return self.current is not None and self.current >= self.target.value

    @property
    def has_open_request(self) -> bool:
        """Whether a request for this quota is awaiting a decision."""
        return self.pending_status is not None

    @property
    def needs_submit(self) -> bool:
        """Whether this quota still needs a request raised."""
        return not self.satisfied and not self.has_open_request

    @property
    def state(self) -> str:
        """A one-word summary: ``satisfied``, ``open`` or ``to-submit``."""
        if self.satisfied:
            return "satisfied"
        return "open" if self.has_open_request else "to-submit"

has_open_request property

Whether a request for this quota is awaiting a decision.

needs_submit property

Whether this quota still needs a request raised.

satisfied property

Whether the applied value already meets the target.

state property

A one-word summary: satisfied, open or to-submit.

QuotaTarget dataclass

A quota and the value to request for it.

Attributes:

Name	Type	Description
label	str	Human-readable name, used only for display.
code	str	Service Quotas code, e.g. "L-417A185B".
value	float	Desired value. For EC2 compute quotas this is a vCPU count.

Source code in src/pdum/aws/quotas.py

@dataclass(frozen=True, slots=True)
class QuotaTarget:
    """A quota and the value to request for it.

    Attributes
    ----------
    label : str
        Human-readable name, used only for display.
    code : str
        Service Quotas code, e.g. ``"L-417A185B"``.
    value : float
        Desired value. For EC2 compute quotas this is a vCPU count.
    """

    label: str
    code: str
    value: float

SubmitResult dataclass

The outcome of trying to raise one quota.

Attributes:

Name	Type	Description
target	QuotaTarget	The quota that was attempted.
outcome	str	One of "submitted", "skipped", "dry-run", "cap-reached" or "error".
request_id	str or None	Service Quotas request id, when one was created.
request_status	str or None	Status the request came back with, typically "PENDING".
error	str or None	Error message when outcome is "error" or "cap-reached".

Source code in src/pdum/aws/quotas.py

@dataclass(frozen=True, slots=True)
class SubmitResult:
    """The outcome of trying to raise one quota.

    Attributes
    ----------
    target : QuotaTarget
        The quota that was attempted.
    outcome : str
        One of ``"submitted"``, ``"skipped"``, ``"dry-run"``, ``"cap-reached"``
        or ``"error"``.
    request_id : str or None
        Service Quotas request id, when one was created.
    request_status : str or None
        Status the request came back with, typically ``"PENDING"``.
    error : str or None
        Error message when ``outcome`` is ``"error"`` or ``"cap-reached"``.
    """

    target: QuotaTarget
    outcome: str
    request_id: str | None = None
    request_status: str | None = None
    error: str | None = None

current_values(targets, *, service_code='ec2', region=None)

Look up the currently applied value for each target's quota.

Parameters:

Name	Type	Description	Default
targets	list of QuotaTarget	Quotas to look up.	required
service_code	str	Service Quotas service code.	"ec2"
region	str	Region to query. Quotas are per-region.	None

Returns:

Type	Description
dict of str to (float or None)	Quota code mapped to its applied value, or None where the quota does not exist in this region.

Source code in src/pdum/aws/quotas.py

def current_values(
    targets: list[QuotaTarget],
    *,
    service_code: str = "ec2",
    region: str | None = None,
) -> dict[str, float | None]:
    """Look up the currently applied value for each target's quota.

    Parameters
    ----------
    targets : list of QuotaTarget
        Quotas to look up.
    service_code : str, default "ec2"
        Service Quotas service code.
    region : str, optional
        Region to query. Quotas are per-region.

    Returns
    -------
    dict of str to (float or None)
        Quota code mapped to its applied value, or ``None`` where the quota does
        not exist in this region.
    """
    quotas = _quotas_client(region)
    out: dict[str, float | None] = {}
    for target in targets:
        try:
            response = quotas.get_service_quota(ServiceCode=service_code, QuotaCode=target.code)
            out[target.code] = response["Quota"]["Value"]
        except ClientError:
            # Not every quota code exists in every region; treat as absent.
            out[target.code] = None
    return out

open_requests(*, service_code='ec2', region=None)

Find quota-increase requests that are still awaiting a decision.

Parameters:

Name	Type	Description	Default
service_code	str	Service Quotas service code.	"ec2"
region	str	Region to query.	None

Returns:

Type	Description
dict of str to tuple	Quota code mapped to (desired_value, status) for each open request.

Source code in src/pdum/aws/quotas.py

def open_requests(*, service_code: str = "ec2", region: str | None = None) -> dict[str, tuple[float, str]]:
    """Find quota-increase requests that are still awaiting a decision.

    Parameters
    ----------
    service_code : str, default "ec2"
        Service Quotas service code.
    region : str, optional
        Region to query.

    Returns
    -------
    dict of str to tuple
        Quota code mapped to ``(desired_value, status)`` for each open request.
    """
    return {
        request.code: (request.desired, request.status)
        for request in request_history(service_code=service_code, region=region)
        if request.is_open
    }

report(targets, *, service_code='ec2', region=None)

Describe where each target stands, without changing anything.

Parameters:

Name	Type	Description	Default
targets	list of QuotaTarget	Quotas to inspect.	required
service_code	str	Service Quotas service code.	"ec2"
region	str	Region to query.	None

Returns:

Type	Description
list of QuotaStatus	One entry per target, in the order given.

Source code in src/pdum/aws/quotas.py

def report(
    targets: list[QuotaTarget],
    *,
    service_code: str = "ec2",
    region: str | None = None,
) -> list[QuotaStatus]:
    """Describe where each target stands, without changing anything.

    Parameters
    ----------
    targets : list of QuotaTarget
        Quotas to inspect.
    service_code : str, default "ec2"
        Service Quotas service code.
    region : str, optional
        Region to query.

    Returns
    -------
    list of QuotaStatus
        One entry per target, in the order given.
    """
    applied = current_values(targets, service_code=service_code, region=region)
    pending = open_requests(service_code=service_code, region=region)
    out: list[QuotaStatus] = []
    for target in targets:
        value, status = pending.get(target.code, (None, None))
        out.append(
            QuotaStatus(
                target=target,
                current=applied.get(target.code),
                pending_value=value,
                pending_status=status,
            )
        )
    return out

request_history(*, service_code='ec2', region=None)

List every quota-increase request, decided or not.

This is how you find out what AWS actually did with a batch: statuses include APPROVED, DENIED and CASE_CLOSED alongside the open PENDING and CASE_OPENED.

Parameters:

Name	Type	Description	Default
service_code	str	Service Quotas service code.	"ec2"
region	str	Region to query.	None

Returns:

Type	Description
list of QuotaRequest	Requests newest-updated first.

Source code in src/pdum/aws/quotas.py

def request_history(*, service_code: str = "ec2", region: str | None = None) -> list[QuotaRequest]:
    """List every quota-increase request, decided or not.

    This is how you find out what AWS actually did with a batch: statuses
    include ``APPROVED``, ``DENIED`` and ``CASE_CLOSED`` alongside the open
    ``PENDING`` and ``CASE_OPENED``.

    Parameters
    ----------
    service_code : str, default "ec2"
        Service Quotas service code.
    region : str, optional
        Region to query.

    Returns
    -------
    list of QuotaRequest
        Requests newest-updated first.
    """
    quotas = _quotas_client(region)
    paginator = quotas.get_paginator("list_requested_service_quota_change_history")
    out: list[QuotaRequest] = []
    for page in paginator.paginate(ServiceCode=service_code):
        for request in page["RequestedQuotas"]:
            out.append(
                QuotaRequest(
                    code=request["QuotaCode"],
                    name=request.get("QuotaName", ""),
                    desired=request["DesiredValue"],
                    status=request["Status"],
                    request_id=request.get("Id"),
                    case_id=request.get("CaseId"),
                    created=request.get("Created"),
                    last_updated=request.get("LastUpdated"),
                )
            )
    out.sort(key=lambda r: (r.last_updated is None, r.last_updated), reverse=True)
    return out

submit(targets, *, service_code='ec2', region=None, dry_run=False)

Request increases for every target that needs one.

Idempotent: targets already satisfied, or already carrying an open request, are skipped rather than resubmitted. Stops at the first QuotaExceededException, since once the account's open-request cap is hit every subsequent call would fail the same way.

Parameters:

Name	Type	Description	Default
targets	list of QuotaTarget	Quotas to raise.	required
service_code	str	Service Quotas service code.	"ec2"
region	str	Region to act on.	None
dry_run	bool	Report what would be submitted without calling the write API.	False

Returns:

Type	Description
list of SubmitResult	One entry per target considered. Iteration stops early on "cap-reached", so the list may be shorter than targets; check for that outcome to know whether to re-run later.

Source code in src/pdum/aws/quotas.py

def submit(
    targets: list[QuotaTarget],
    *,
    service_code: str = "ec2",
    region: str | None = None,
    dry_run: bool = False,
) -> list[SubmitResult]:
    """Request increases for every target that needs one.

    Idempotent: targets already satisfied, or already carrying an open request,
    are skipped rather than resubmitted. Stops at the first
    ``QuotaExceededException``, since once the account's open-request cap is hit
    every subsequent call would fail the same way.

    Parameters
    ----------
    targets : list of QuotaTarget
        Quotas to raise.
    service_code : str, default "ec2"
        Service Quotas service code.
    region : str, optional
        Region to act on.
    dry_run : bool, default False
        Report what would be submitted without calling the write API.

    Returns
    -------
    list of SubmitResult
        One entry per target considered. Iteration stops early on
        ``"cap-reached"``, so the list may be shorter than *targets*; check for
        that outcome to know whether to re-run later.
    """
    quotas = _quotas_client(region)
    statuses = report(targets, service_code=service_code, region=region)
    out: list[SubmitResult] = []

    for status in statuses:
        target = status.target
        if not status.needs_submit:
            out.append(SubmitResult(target=target, outcome="skipped"))
            continue
        if dry_run:
            out.append(SubmitResult(target=target, outcome="dry-run"))
            continue
        try:
            response = quotas.request_service_quota_increase(
                ServiceCode=service_code,
                QuotaCode=target.code,
                DesiredValue=float(target.value),
            )
            requested = response["RequestedQuota"]
            out.append(
                SubmitResult(
                    target=target,
                    outcome="submitted",
                    request_id=requested.get("Id"),
                    request_status=requested.get("Status"),
                )
            )
        except ClientError as exc:
            error = exc.response["Error"]
            if error["Code"] == "QuotaExceededException":
                out.append(SubmitResult(target=target, outcome="cap-reached", error=error["Message"]))
                break
            out.append(SubmitResult(target=target, outcome="error", error=f"{error['Code']}: {error['Message']}"))
    return out

pdum.aws.cli

pdum-aws — manage SSM secrets and service quotas from the shell.

$ pdum-aws whoami
$ pdum-aws secrets --path /myapp/ list
$ pdum-aws quotas status --region us-east-1
$ pdum-aws pdx-doctor

Installing the package also provides pdx, a short form of pdum-aws pdx: it loads a project's secrets into the environment and hands the process over to a command. See :mod:pdum.aws.cli.pdx.

$ pdx npm run dev

Run inside a project, the CLI first loads the nearest .env and its .env.local overlay — the whole file, exactly as pdx reads it, never overriding what the shell already set. A project whose .env names AWS_PROFILE and PDUM_SSM_PATH therefore gets every command bare: no exports, no flags. Only the files are loaded, never the SSM store — fetching secrets in order to manage secrets would be circular. pdx and pdx-doctor are excluded: they assemble the environment themselves and report where every layer came from, and a pre-loaded .env would blur that report.

Beyond that, credentials come from the ambient environment, exactly as in the library. There is deliberately no --profile flag: select an account the standard way — AWS_PROFILE in the project's .env or in the shell — so this behaves like every other AWS tool on the box.

$ AWS_PROFILE=my-account pdum-aws quotas status

The secrets commands need a search path — one or more SSM prefixes, colon- separated, most specific first — which has no default: PDUM_SSM_PATH in the .env or the shell, or --path per invocation.

Embedding these commands elsewhere

Every group here is produced by a factory, so another application can mount the same commands under its own name, carrying its own defaults and rendering through its own console. That is the supported way to give an application a secrets subcommand without asking its users to type a search path:

import typer
from rich.console import Console

from pdum.aws.cli import add_whoami, aws_errors, build_quotas_app, build_secrets_app

console = Console()
app = typer.Typer()
add_whoami(app, console=console)
app.add_typer(
    build_secrets_app(console=console, default_path="/acme/:/org/", envvar="ACME_SSM_PATH"),
    name="secrets",
)
app.add_typer(
    build_quotas_app(console=console, default_targets=ACME_TARGETS, default_regions=["us-east-1"]),
    name="quotas",
)


def main() -> None:
    with aws_errors(console):
        app()

See :func:~pdum.aws.cli.secrets.build_secrets_app and :func:~pdum.aws.cli.quotas.build_quotas_app for the full set of knobs. The groups keep their per-invocation state in ctx.meta under namespaced keys rather than in ctx.obj, so mounting them never disturbs what a host app stores there for its own commands.

add_pdx(app, *, console=None, default_path=None, envvar=DEFAULT_ENVVAR, env_file='.env', store_factory=SecretStore, name='pdx')

Register a pdx command on app.

Parameters:

Name	Type	Description	Default
app	Typer	Application to register the command on.	required
console	Console	Console for this command's own diagnostics. Defaults to :data:~pdum.aws.cli.output.DEFAULT_STDERR_CONSOLE; whatever you pass should write to stderr, since stdout belongs to the wrapped program.	None
default_path	str or callable	Search path used when the environment, including the .env, has none. A callable is resolved at invocation.	None
envvar	str	Environment variable holding the search path. None leaves only default_path; the .env is still loaded either way.	DEFAULT_ENVVAR
env_file	str	File searched for upward from the working directory, and the base name of the .local overlay beside it. A name with a directory component is used as given.	".env"
store_factory	callable	Called with the resolved path to build the store.	:class:`~pdum.aws.secrets.SecretStore`
name	str	Name to register the command under.	"pdx"

Source code in src/pdum/aws/cli/pdx.py

def add_pdx(
    app: typer.Typer,
    *,
    console: Console | None = None,
    default_path: PathDefault = None,
    envvar: str | None = DEFAULT_ENVVAR,
    env_file: str = ".env",
    store_factory: Callable[[str], SecretStore] = SecretStore,
    name: str = "pdx",
) -> None:
    """Register a ``pdx`` command on *app*.

    Parameters
    ----------
    app : typer.Typer
        Application to register the command on.
    console : rich.console.Console, optional
        Console for this command's own diagnostics. Defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_STDERR_CONSOLE`; whatever you pass
        should write to stderr, since stdout belongs to the wrapped program.
    default_path : str or callable, optional
        Search path used when the environment, including the ``.env``, has
        none. A callable is resolved at invocation.
    envvar : str, optional
        Environment variable holding the search path. ``None`` leaves only
        *default_path*; the ``.env`` is still loaded either way.
    env_file : str, default ".env"
        File searched for upward from the working directory, and the base name
        of the ``.local`` overlay beside it. A name with a directory component
        is used as given.
    store_factory : callable, default :class:`~pdum.aws.secrets.SecretStore`
        Called with the resolved path to build the store.
    name : str, default "pdx"
        Name to register the command under.
    """
    out = console or DEFAULT_STDERR_CONSOLE

    @app.command(name, context_settings=_PASSTHROUGH)
    def pdx(
        command: list[str] = typer.Argument(None, metavar="COMMAND [ARGS]...", help="program to run, with arguments"),
    ) -> None:
        """Run a command with the project's environment and secrets loaded.

        Takes no options of its own: everything here is the command to run.
        Variables already set are left alone.
        """
        argv = list(command or [])
        if not argv:
            out.print("[red]nothing to run:[/] pdx takes the command to exec, e.g. [bold]pdx npm start[/]")
            raise typer.Exit(2)

        try:
            load_env(env_file=env_file, envvar=envvar, default_path=default_path, store_factory=store_factory)
        except (NoSearchPath, ValueError) as exc:
            out.print(f"[red]{exc}[/]")
            raise typer.Exit(2) from exc

        _exec(argv, out)

add_pdx_doctor(app, *, console=None, default_path=None, envvar=DEFAULT_ENVVAR, env_file='.env', store_factory=SecretStore, name='pdx-doctor')

Register a command explaining what :func:add_pdx would do.

Answers the questions secrets list cannot, because they are about this process rather than about the store: which .env was found and what it contributed, which search path resolved and from where, and which names the environment already holds — the reason a program run under pdx can see a value that is not the one in SSM.

It applies the .env exactly as pdx does, since that is what decides the account the store is read from. Secrets themselves are reported by name and layer only: this works from :meth:~pdum.aws.secrets.SecretStore.describe, which does not decrypt, so running it pulls no secret value over the wire. Use secrets get when the value is what you want.

Parameters are as for :func:add_pdx, except that console defaults to stdout — here the report is the output.

Source code in src/pdum/aws/cli/pdx.py

def add_pdx_doctor(
    app: typer.Typer,
    *,
    console: Console | None = None,
    default_path: PathDefault = None,
    envvar: str | None = DEFAULT_ENVVAR,
    env_file: str = ".env",
    store_factory: Callable[[str], SecretStore] = SecretStore,
    name: str = "pdx-doctor",
) -> None:
    """Register a command explaining what :func:`add_pdx` would do.

    Answers the questions ``secrets list`` cannot, because they are about this
    process rather than about the store: which ``.env`` was found and what it
    contributed, which search path resolved and from where, and which names the
    environment already holds — the reason a program run under ``pdx`` can see a
    value that is not the one in SSM.

    It applies the ``.env`` exactly as ``pdx`` does, since that is what decides
    the account the store is read from. Secrets themselves are reported by name
    and layer only: this works from
    :meth:`~pdum.aws.secrets.SecretStore.describe`, which does not decrypt, so
    running it pulls no secret value over the wire. Use ``secrets get`` when the
    value is what you want.

    Parameters are as for :func:`add_pdx`, except that *console* defaults to
    stdout — here the report is the output.
    """
    out = console or DEFAULT_CONSOLE

    @app.command(name)
    def pdx_doctor() -> None:
        """Explain what pdx would put in the environment, without running anything."""
        from_file = load_env_files(env_file, os.environ)
        if from_file is None:
            out.print(f"[dim]no {env_file} found in this directory or above it[/]")
        else:
            out.print(f"[bold].env[/] {from_file.source}")
            out.print(f"  [green]sets[/] {', '.join(from_file.applied) or '—'}")
            if from_file.already_set:
                out.print(f"  [cyan]kept from the environment[/] {', '.join(from_file.already_set)}")

        path, source = resolve_search_path(
            envvar=envvar, default_path=default_path, environ=os.environ, env_file_report=from_file
        )
        if not path:
            out.print(f"[red]{no_search_path_message(envvar, env_file, from_file)}[/]")
            raise typer.Exit(2)
        out.print(f"[bold]search path[/] {path}  [dim](from {source})[/]")
        try:
            store = store_factory(path)
        except ValueError as exc:
            out.print(f"[red]{exc}[/]")
            raise typer.Exit(2) from exc

        rows = sorted((r for r in store.describe() if not r["shadowed"]), key=lambda r: r["name"])
        if not rows:
            out.print(f"[yellow]no secrets under {path}[/]")
            return

        table = Table(box=None, pad_edge=False)
        table.add_column("NAME", style="bold")
        table.add_column("LAYER")
        table.add_column("PDX WOULD")
        counts = {"set it": 0, "keep the environment's": 0, "skip it": 0}
        for row in rows:
            secret = row["name"]
            if not is_env_name(secret):
                verdict, style = "skip it", "yellow"
            elif os.environ.get(secret):
                verdict, style = "keep the environment's", "cyan"
            else:
                verdict, style = "set it", "green"
            counts[verdict] += 1
            table.add_row(secret, row["origin"], f"[{style}]{verdict}[/]")
        out.print(table)
        out.print("  [dim]" + ", ".join(f"{n} to {verdict}" for verdict, n in counts.items() if n) + "[/]")

add_whoami(app, *, console=None, name='whoami')

Register a whoami command on app.

Parameters:

Name	Type	Description	Default
app	Typer	Application to register the command on.	required
console	Console	Console to render through; defaults to :data:~pdum.aws.cli.output.DEFAULT_CONSOLE.	None
name	str	Name to register the command under, in case the host already has one.	"whoami"

Source code in src/pdum/aws/cli/identity.py

def add_whoami(app: typer.Typer, *, console: Console | None = None, name: str = "whoami") -> None:
    """Register a ``whoami`` command on *app*.

    Parameters
    ----------
    app : typer.Typer
        Application to register the command on.
    console : rich.console.Console, optional
        Console to render through; defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_CONSOLE`.
    name : str, default "whoami"
        Name to register the command under, in case the host already has one.
    """
    out = console or DEFAULT_CONSOLE

    @app.command(name)
    def whoami() -> None:
        """Print the AWS identity the current credentials resolve to."""
        identity = aws.whoami()
        out.print(f"[bold]account[/] {identity['Account']}", soft_wrap=True)
        out.print(f"[bold]arn[/]     {identity['Arn']}", soft_wrap=True)

aws_errors(console=None)

Turn credential and API failures into one readable line.

A missing profile or an expired SSO session is the single most common way these commands fail, and a botocore traceback is a poor way to say so. Host applications embedding a group should wrap their own entry point in this to get the same treatment:

def main() -> None:
    with aws_errors(my_console):
        app()

Parameters:

Name	Type	Description	Default
console	Console	Console to report through; defaults to :data:DEFAULT_CONSOLE.	None

Raises:

Type	Description
SystemExit	With status 2, in place of the original botocore exception.

Source code in src/pdum/aws/cli/output.py

@contextmanager
def aws_errors(console: Console | None = None) -> Iterator[None]:
    """Turn credential and API failures into one readable line.

    A missing profile or an expired SSO session is the single most common way
    these commands fail, and a botocore traceback is a poor way to say so. Host
    applications embedding a group should wrap their own entry point in this to
    get the same treatment:

    ```python
    def main() -> None:
        with aws_errors(my_console):
            app()
    ```

    Parameters
    ----------
    console : rich.console.Console, optional
        Console to report through; defaults to :data:`DEFAULT_CONSOLE`.

    Raises
    ------
    SystemExit
        With status 2, in place of the original ``botocore`` exception.
    """
    try:
        yield
    except (BotoCoreError, ClientError) as exc:
        (console or DEFAULT_CONSOLE).print(f"[red]{exc}[/]")
        raise SystemExit(2) from exc

build_quotas_app(*, console=None, default_service='ec2', default_targets=None, default_regions=None, show_service_option=True, show_targets_option=True, help='Inspect and request AWS service quotas.')

Build a quotas command group.

Parameters:

Name	Type	Description	Default
console	Console	Console every command in the group renders through; defaults to :data:~pdum.aws.cli.output.DEFAULT_CONSOLE.	None
default_service	str	Service Quotas service code used when --service is absent.	"ec2"
default_targets	list of QuotaTarget or callable	Target plan used when --targets is absent. Pass a zero-argument callable to defer building it. None means :data:~pdum.aws.quotas.EC2_VCPU_TARGETS, which is an example to copy and edit rather than a recommendation — a host with a real workload should pass its own.	None
default_regions	list of str	Regions acted on when --region is absent. None means the single region boto3 resolves from the environment.	None
show_service_option	bool	Whether --service appears in --help. It keeps working either way; hiding it is for hosts whose plan only makes sense for one service.	True
show_targets_option	bool	Whether --targets appears in --help, on the same terms.	True
help	str	Group help text.	'Inspect and request AWS service quotas.'

Returns:

Type	Description
Typer	A new application, ready to pass to add_typer.

Source code in src/pdum/aws/cli/quotas.py

def build_quotas_app(
    *,
    console: Console | None = None,
    default_service: str = "ec2",
    default_targets: TargetsDefault = None,
    default_regions: list[str] | None = None,
    show_service_option: bool = True,
    show_targets_option: bool = True,
    help: str = "Inspect and request AWS service quotas.",
) -> typer.Typer:
    """Build a ``quotas`` command group.

    Parameters
    ----------
    console : rich.console.Console, optional
        Console every command in the group renders through; defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_CONSOLE`.
    default_service : str, default "ec2"
        Service Quotas service code used when ``--service`` is absent.
    default_targets : list of QuotaTarget or callable, optional
        Target plan used when ``--targets`` is absent. Pass a zero-argument
        callable to defer building it. ``None`` means
        :data:`~pdum.aws.quotas.EC2_VCPU_TARGETS`, which is an example to copy
        and edit rather than a recommendation — a host with a real workload
        should pass its own.
    default_regions : list of str, optional
        Regions acted on when ``--region`` is absent. ``None`` means the single
        region ``boto3`` resolves from the environment.
    show_service_option : bool, default True
        Whether ``--service`` appears in ``--help``. It keeps working either
        way; hiding it is for hosts whose plan only makes sense for one service.
    show_targets_option : bool, default True
        Whether ``--targets`` appears in ``--help``, on the same terms.
    help : str
        Group help text.

    Returns
    -------
    typer.Typer
        A new application, ready to pass to ``add_typer``.
    """
    out = console or DEFAULT_CONSOLE
    app = typer.Typer(help=help, no_args_is_help=True)

    def resolve_targets(path: Path | None) -> list[quotas_lib.QuotaTarget]:
        """Targets from the flag, else the host's plan, else the bundled one."""
        if path is not None:
            return _load_targets_file(path)
        if default_targets is None:
            # Read late so the module attribute stays the single source of truth.
            return quotas_lib.EC2_VCPU_TARGETS
        return default_targets() if callable(default_targets) else default_targets

    def resolve_regions(given: list[str] | None) -> list[str | None]:
        """Regions to act on; ``[None]`` means whatever the environment resolves."""
        return list(given) if given else [None]

    def region_option(verb: str) -> Any:
        return typer.Option(
            list(default_regions) if default_regions else None,
            "--region",
            "-r",
            help=f"region to {verb} (repeatable)",
        )

    def service_option() -> Any:
        return typer.Option(
            default_service,
            "--service",
            hidden=not show_service_option,
            help="Service Quotas service code",
        )

    def targets_option() -> Any:
        source = "the bundled EC2 plan" if default_targets is None else "this application's plan"
        return typer.Option(
            None,
            "--targets",
            hidden=not show_targets_option,
            help=f"JSON target list; defaults to {source}",
        )

    @app.command("status")
    def quotas_status(
        region: list[str] = region_option("query"),
        service: str = service_option(),
        targets_file: Path = targets_option(),
    ) -> None:
        """Show current values against target values."""
        targets = resolve_targets(targets_file)
        for reg in resolve_regions(region):
            _heading(out, reg)
            statuses = quotas_lib.report(targets, service_code=service, region=reg)
            out.print(_status_table(statuses))
            pending = sum(1 for s in statuses if s.has_open_request)
            todo = sum(1 for s in statuses if s.needs_submit)
            out.print(f"  [dim]{pending} open, {todo} to submit[/]")

    @app.command("history")
    def quotas_history(
        region: list[str] = region_option("query"),
        service: str = service_option(),
        open_only: bool = typer.Option(False, "--open", help="show only requests still awaiting a decision"),
    ) -> None:
        """Show what AWS decided about past quota-increase requests."""
        for reg in resolve_regions(region):
            _heading(out, reg)
            requests = quotas_lib.request_history(service_code=service, region=reg)
            if open_only:
                requests = [r for r in requests if r.is_open]
            if not requests:
                out.print("  [dim]no requests[/]")
                continue

            out.print(_history_table(requests))
            counts: dict[str, int] = {}
            for request in requests:
                counts[request.status] = counts.get(request.status, 0) + 1
            out.print("  [dim]" + ", ".join(f"{n} {s}" for s, n in sorted(counts.items())) + "[/]")

    @app.command("request")
    def quotas_request(
        region: list[str] = region_option("act on"),
        service: str = service_option(),
        targets_file: Path = targets_option(),
        dry_run: bool = typer.Option(False, "--dry-run", help="show what would be requested, write nothing"),
    ) -> None:
        """Request increases for every target that needs one.

        Idempotent: quotas already satisfied, or already carrying an open
        request, are skipped. Stops when the account hits its open-request cap,
        so re-run once earlier cases are decided.
        """
        targets = resolve_targets(targets_file)
        for reg in resolve_regions(region):
            _heading(out, reg)
            results = quotas_lib.submit(targets, service_code=service, region=reg, dry_run=dry_run)
            capped = False
            for result in results:
                if result.outcome == "skipped":
                    continue
                if result.outcome == "cap-reached":
                    capped = True
                    out.print(f"  [yellow]{result.target.label}[/] -> blocked: account is at its open-request cap")
                    break
                colour = {"submitted": "green", "dry-run": "cyan", "error": "red"}[result.outcome]
                detail = f" {result.error}" if result.error else ""
                out.print(f"  [{colour}]{result.outcome}[/] {result.target.label} -> {result.target.value:.0f}{detail}")
            submitted = sum(1 for r in results if r.outcome in {"submitted", "dry-run"})
            skipped = sum(1 for r in results if r.outcome == "skipped")
            out.print(f"  [dim]{submitted} submitted, {skipped} skipped[/]")
            if capped:
                out.print("  [dim]re-run once some cases are decided[/]")

    @app.command("targets")
    def quotas_targets() -> None:
        """Print this group's default target plan as JSON, ready to edit and pass to --targets."""
        payload: list[dict[str, Any]] = [
            {"label": t.label, "code": t.code, "value": t.value} for t in resolve_targets(None)
        ]
        emit(out, json.dumps(payload, indent=2))

    return app

build_secrets_app(*, console=None, default_path=None, envvar='PDUM_SSM_PATH', expose_path_option=True, store_factory=SecretStore, help='Manage secrets in AWS SSM Parameter Store.')

Build a secrets command group.

Parameters:

Name	Type	Description	Default
console	Console	Console every command in the group renders through; defaults to :data:~pdum.aws.cli.output.DEFAULT_CONSOLE.	None
default_path	str or callable	SSM search path to use when neither the flag nor the environment supplies one. Pass a zero-argument callable to defer resolving it — a host reading the path from a config file wants that read to happen on invocation, not on import.	None
envvar	str	Environment variable consulted before default_path. Pass None to disable environment lookup entirely.	'PDUM_SSM_PATH'
expose_path_option	bool	Whether to offer --path. False removes the flag, pinning the group to default_path.	True
store_factory	callable	Called with the resolved path to build the store. Override to thread host configuration through, e.g. lambda path: SecretStore(path, region=cfg.region).	:class:`~pdum.aws.secrets.SecretStore`
help	str	Group help text.	'Manage secrets in AWS SSM Parameter Store.'

Returns:

Type	Description
Typer	A new application, ready to pass to add_typer.

Source code in src/pdum/aws/cli/secrets.py

def build_secrets_app(
    *,
    console: Console | None = None,
    default_path: PathDefault = None,
    envvar: str | None = "PDUM_SSM_PATH",
    expose_path_option: bool = True,
    store_factory: Callable[[str], SecretStore] = SecretStore,
    help: str = "Manage secrets in AWS SSM Parameter Store.",
) -> typer.Typer:
    """Build a ``secrets`` command group.

    Parameters
    ----------
    console : rich.console.Console, optional
        Console every command in the group renders through; defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_CONSOLE`.
    default_path : str or callable, optional
        SSM search path to use when neither the flag nor the environment
        supplies one. Pass a zero-argument callable to defer resolving it — a
        host reading the path from a config file wants that read to happen on
        invocation, not on import.
    envvar : str, optional
        Environment variable consulted before *default_path*. Pass ``None``
        to disable environment lookup entirely.
    expose_path_option : bool, default True
        Whether to offer ``--path``. ``False`` removes the flag, pinning the
        group to *default_path*.
    store_factory : callable, default :class:`~pdum.aws.secrets.SecretStore`
        Called with the resolved path to build the store. Override to thread
        host configuration through, e.g.
        ``lambda path: SecretStore(path, region=cfg.region)``.
    help : str
        Group help text.

    Returns
    -------
    typer.Typer
        A new application, ready to pass to ``add_typer``.
    """
    out = console or DEFAULT_CONSOLE
    app = typer.Typer(help=help, no_args_is_help=True)

    def configure(ctx: typer.Context, path: str | None) -> None:
        """Resolve the search path and put the store where subcommands find it."""
        if not path:
            path = default_path() if callable(default_path) else default_path
        if not path:
            source = f" or set {envvar}" if envvar else ""
            out.print(f"[red]no SSM search path configured: pass --path{source}[/]")
            raise typer.Exit(2)
        try:
            ctx.meta[STORE_KEY] = store_factory(path)
        except ValueError as exc:
            out.print(f"[red]{exc}[/]")
            raise typer.Exit(2) from exc

    if expose_path_option:

        @app.callback()
        def _configure(
            ctx: typer.Context,
            path: str = typer.Option(
                _flag_default(default_path),
                "--path",
                "-p",
                envvar=envvar,
                help="SSM search path: colon-separated prefixes, most specific first, e.g. /myapp/:/org/",
            ),
        ) -> None:
            """Build the store every secrets subcommand works against."""
            configure(ctx, path)

    else:

        @app.callback()
        def _configure_fixed(ctx: typer.Context) -> None:
            """Build the store every secrets subcommand works against."""
            configure(ctx, None)

    @app.command("list")
    def secrets_list(
        ctx: typer.Context,
        long: bool = typer.Option(False, "--long", "-l", help="show type, version, last-modified and origin"),
    ) -> None:
        """List secret names visible on the search path."""
        store = store_from(ctx)
        if not long:
            for name in sorted(store.names()):
                emit(out, name)
            return

        rows = sorted(store.describe(), key=lambda r: (r["name"], r["shadowed"]))
        if not rows:
            out.print(f"[yellow]no secrets under {':'.join(store.prefixes)}[/]")
            return
        out.print(_describe_table(rows, show_origin=len(store.prefixes) > 1))

    @app.command("get")
    def secrets_get(
        ctx: typer.Context, name: str = typer.Argument(..., help="secret name, without any prefix")
    ) -> None:
        """Print one secret's value to stdout, verbatim."""
        store = store_from(ctx)
        try:
            emit(out, store.get(name, required=True))
        except KeyError as exc:
            out.print(f"[red]{exc.args[0]}[/]")
            raise typer.Exit(1) from exc

    @app.command("set")
    def secrets_set(
        ctx: typer.Context,
        name: str = typer.Argument(...),
        value: str = typer.Argument(None, help="value; omit or pass '-' to read stdin"),
        plain: bool = typer.Option(False, "--plain", help="store as String rather than SecureString"),
    ) -> None:
        """Set a secret in the first prefix, taking the value from the argument or stdin.

        Prefer stdin: a value passed as an argument lands in your shell history.
        """
        store = store_from(ctx)
        if value in (None, "-"):
            value = sys.stdin.read().rstrip("\n")
        if not value:
            out.print("[red]refusing to store an empty value[/]")
            raise typer.Exit(1)
        store.put(name, value, secure=not plain)
        out.print(f"[green]set[/] {store.path(name)}")

    @app.command("rm")
    def secrets_rm(
        ctx: typer.Context,
        name: str = typer.Argument(...),
        yes: bool = typer.Option(False, "--yes", "-y", help="skip the confirmation prompt"),
    ) -> None:
        """Delete a secret from the first prefix.

        A secret living only in a fallback layer is refused, naming where it
        is: deleting from a shared layer takes an explicit ``--path`` there.
        """
        store = store_from(ctx)
        if not yes:
            typer.confirm(f"delete {store.path(name)}?", abort=True)
        try:
            store.delete(name)
        except KeyError as exc:
            out.print(f"[red]{exc.args[0]}[/]")
            raise typer.Exit(1) from exc
        out.print(f"[green]deleted[/] {store.path(name)}")

    @app.command("import")
    def secrets_import(
        ctx: typer.Context,
        path: Path = typer.Argument(Path(".secrets"), help="a KEY=VALUE file"),
        dry_run: bool = typer.Option(False, "--dry-run", help="show what would be pushed, write nothing"),
    ) -> None:
        """Push secrets from a KEY=VALUE file into the first prefix.

        AWS bootstrap keys are skipped; see :data:`SKIP_ON_IMPORT`.
        """
        store = store_from(ctx)
        if not path.exists():
            out.print(f"[red]no such file:[/] {path}")
            raise typer.Exit(1)
        out.print(f"[dim]reading {path}[/]")

        pushed: list[str] = []
        skipped: list[str] = []
        # Not interpolated: these values are going into permanent storage, where
        # a silently expanded ``${...}`` could not be recovered.
        for key, value in read_env_file(path, interpolate=False).items():
            if key in SKIP_ON_IMPORT:
                skipped.append(key)
                continue
            if not value:
                continue
            if not dry_run:
                store.put(key, value)
            pushed.append(key)

        verb = "would import" if dry_run else "imported"
        out.print(f"[green]{verb}[/] {len(pushed)} secret(s): {', '.join(pushed) or '—'}")
        if skipped:
            out.print(f"[dim]skipped AWS bootstrap keys: {', '.join(skipped)}[/]")

    @app.command("export")
    def secrets_export(ctx: typer.Context) -> None:
        """Print every visible secret as KEY=VALUE lines, resolved along the path."""
        store = store_from(ctx)
        for name in sorted(store.names()):
            emit(out, f"{name}={store.get(name, required=True)}")

    return app

emit(console, text)

Print text verbatim, as data rather than as a message.

Markup, syntax highlighting and wrapping are all disabled for this one call, so a secret containing square brackets is never mangled, a long value is never broken across lines, and piping stays clean. Use this for anything a caller might redirect into a file or another process; use console.print directly for messages addressed to a human.

Parameters:

Name	Type	Description	Default
console	Console	Console to write through.	required
text	str	The value to write.	required

Source code in src/pdum/aws/cli/output.py

def emit(console: Console, text: str) -> None:
    """Print *text* verbatim, as data rather than as a message.

    Markup, syntax highlighting and wrapping are all disabled for this one call,
    so a secret containing square brackets is never mangled, a long value is
    never broken across lines, and piping stays clean. Use this for anything a
    caller might redirect into a file or another process; use
    ``console.print`` directly for messages addressed to a human.

    Parameters
    ----------
    console : rich.console.Console
        Console to write through.
    text : str
        The value to write.
    """
    console.print(text, markup=False, highlight=False, soft_wrap=True)

main()

Console-script entry point.

Wraps the app so credential and API problems print one readable line instead of a botocore traceback — a missing profile or an expired SSO session is the single most common way this tool fails.

Source code in src/pdum/aws/cli/__init__.py

def main() -> None:
    """Console-script entry point.

    Wraps the app so credential and API problems print one readable line instead
    of a botocore traceback — a missing profile or an expired SSO session is the
    single most common way this tool fails.
    """
    with aws_errors(console):
        app()

pdum.aws.cli.secrets

The secrets command group, as a factory other applications can embed.

:func:build_secrets_app returns a fresh :class:typer.Typer each call, so a host CLI can mount these commands under its own name, render them through its own console, and — the point of the exercise — supply the SSM search path its users should not have to type:

import typer

from pdum.aws.cli import build_secrets_app

app = typer.Typer()
app.add_typer(build_secrets_app(default_path="/acme/:/org/", envvar="ACME_SSM_PATH"), name="secrets")

acme secrets list now works bare. The search path resolves in this order:

1. --path on the command line;
2. the environment variable named by envvar;
3. default_path, which may be a callable when the host reads it from a config file and wants that read deferred to invocation time;
4. otherwise an error — the library itself ships no default, because a shared one would let unrelated projects collide in one namespace.

A path is one or more SSM prefixes, colon-separated, most specific first — reads fall back along it, writes and deletes target the first prefix only (see :mod:pdum.aws.secrets). Pass expose_path_option=False to drop the flag entirely and pin the namespace to default_path.

build_secrets_app(*, console=None, default_path=None, envvar='PDUM_SSM_PATH', expose_path_option=True, store_factory=SecretStore, help='Manage secrets in AWS SSM Parameter Store.')

Build a secrets command group.

Parameters:

Name	Type	Description	Default
console	Console	Console every command in the group renders through; defaults to :data:~pdum.aws.cli.output.DEFAULT_CONSOLE.	None
default_path	str or callable	SSM search path to use when neither the flag nor the environment supplies one. Pass a zero-argument callable to defer resolving it — a host reading the path from a config file wants that read to happen on invocation, not on import.	None
envvar	str	Environment variable consulted before default_path. Pass None to disable environment lookup entirely.	'PDUM_SSM_PATH'
expose_path_option	bool	Whether to offer --path. False removes the flag, pinning the group to default_path.	True
store_factory	callable	Called with the resolved path to build the store. Override to thread host configuration through, e.g. lambda path: SecretStore(path, region=cfg.region).	:class:`~pdum.aws.secrets.SecretStore`
help	str	Group help text.	'Manage secrets in AWS SSM Parameter Store.'

Returns:

Type	Description
Typer	A new application, ready to pass to add_typer.

Source code in src/pdum/aws/cli/secrets.py

def build_secrets_app(
    *,
    console: Console | None = None,
    default_path: PathDefault = None,
    envvar: str | None = "PDUM_SSM_PATH",
    expose_path_option: bool = True,
    store_factory: Callable[[str], SecretStore] = SecretStore,
    help: str = "Manage secrets in AWS SSM Parameter Store.",
) -> typer.Typer:
    """Build a ``secrets`` command group.

    Parameters
    ----------
    console : rich.console.Console, optional
        Console every command in the group renders through; defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_CONSOLE`.
    default_path : str or callable, optional
        SSM search path to use when neither the flag nor the environment
        supplies one. Pass a zero-argument callable to defer resolving it — a
        host reading the path from a config file wants that read to happen on
        invocation, not on import.
    envvar : str, optional
        Environment variable consulted before *default_path*. Pass ``None``
        to disable environment lookup entirely.
    expose_path_option : bool, default True
        Whether to offer ``--path``. ``False`` removes the flag, pinning the
        group to *default_path*.
    store_factory : callable, default :class:`~pdum.aws.secrets.SecretStore`
        Called with the resolved path to build the store. Override to thread
        host configuration through, e.g.
        ``lambda path: SecretStore(path, region=cfg.region)``.
    help : str
        Group help text.

    Returns
    -------
    typer.Typer
        A new application, ready to pass to ``add_typer``.
    """
    out = console or DEFAULT_CONSOLE
    app = typer.Typer(help=help, no_args_is_help=True)

    def configure(ctx: typer.Context, path: str | None) -> None:
        """Resolve the search path and put the store where subcommands find it."""
        if not path:
            path = default_path() if callable(default_path) else default_path
        if not path:
            source = f" or set {envvar}" if envvar else ""
            out.print(f"[red]no SSM search path configured: pass --path{source}[/]")
            raise typer.Exit(2)
        try:
            ctx.meta[STORE_KEY] = store_factory(path)
        except ValueError as exc:
            out.print(f"[red]{exc}[/]")
            raise typer.Exit(2) from exc

    if expose_path_option:

        @app.callback()
        def _configure(
            ctx: typer.Context,
            path: str = typer.Option(
                _flag_default(default_path),
                "--path",
                "-p",
                envvar=envvar,
                help="SSM search path: colon-separated prefixes, most specific first, e.g. /myapp/:/org/",
            ),
        ) -> None:
            """Build the store every secrets subcommand works against."""
            configure(ctx, path)

    else:

        @app.callback()
        def _configure_fixed(ctx: typer.Context) -> None:
            """Build the store every secrets subcommand works against."""
            configure(ctx, None)

    @app.command("list")
    def secrets_list(
        ctx: typer.Context,
        long: bool = typer.Option(False, "--long", "-l", help="show type, version, last-modified and origin"),
    ) -> None:
        """List secret names visible on the search path."""
        store = store_from(ctx)
        if not long:
            for name in sorted(store.names()):
                emit(out, name)
            return

        rows = sorted(store.describe(), key=lambda r: (r["name"], r["shadowed"]))
        if not rows:
            out.print(f"[yellow]no secrets under {':'.join(store.prefixes)}[/]")
            return
        out.print(_describe_table(rows, show_origin=len(store.prefixes) > 1))

    @app.command("get")
    def secrets_get(
        ctx: typer.Context, name: str = typer.Argument(..., help="secret name, without any prefix")
    ) -> None:
        """Print one secret's value to stdout, verbatim."""
        store = store_from(ctx)
        try:
            emit(out, store.get(name, required=True))
        except KeyError as exc:
            out.print(f"[red]{exc.args[0]}[/]")
            raise typer.Exit(1) from exc

    @app.command("set")
    def secrets_set(
        ctx: typer.Context,
        name: str = typer.Argument(...),
        value: str = typer.Argument(None, help="value; omit or pass '-' to read stdin"),
        plain: bool = typer.Option(False, "--plain", help="store as String rather than SecureString"),
    ) -> None:
        """Set a secret in the first prefix, taking the value from the argument or stdin.

        Prefer stdin: a value passed as an argument lands in your shell history.
        """
        store = store_from(ctx)
        if value in (None, "-"):
            value = sys.stdin.read().rstrip("\n")
        if not value:
            out.print("[red]refusing to store an empty value[/]")
            raise typer.Exit(1)
        store.put(name, value, secure=not plain)
        out.print(f"[green]set[/] {store.path(name)}")

    @app.command("rm")
    def secrets_rm(
        ctx: typer.Context,
        name: str = typer.Argument(...),
        yes: bool = typer.Option(False, "--yes", "-y", help="skip the confirmation prompt"),
    ) -> None:
        """Delete a secret from the first prefix.

        A secret living only in a fallback layer is refused, naming where it
        is: deleting from a shared layer takes an explicit ``--path`` there.
        """
        store = store_from(ctx)
        if not yes:
            typer.confirm(f"delete {store.path(name)}?", abort=True)
        try:
            store.delete(name)
        except KeyError as exc:
            out.print(f"[red]{exc.args[0]}[/]")
            raise typer.Exit(1) from exc
        out.print(f"[green]deleted[/] {store.path(name)}")

    @app.command("import")
    def secrets_import(
        ctx: typer.Context,
        path: Path = typer.Argument(Path(".secrets"), help="a KEY=VALUE file"),
        dry_run: bool = typer.Option(False, "--dry-run", help="show what would be pushed, write nothing"),
    ) -> None:
        """Push secrets from a KEY=VALUE file into the first prefix.

        AWS bootstrap keys are skipped; see :data:`SKIP_ON_IMPORT`.
        """
        store = store_from(ctx)
        if not path.exists():
            out.print(f"[red]no such file:[/] {path}")
            raise typer.Exit(1)
        out.print(f"[dim]reading {path}[/]")

        pushed: list[str] = []
        skipped: list[str] = []
        # Not interpolated: these values are going into permanent storage, where
        # a silently expanded ``${...}`` could not be recovered.
        for key, value in read_env_file(path, interpolate=False).items():
            if key in SKIP_ON_IMPORT:
                skipped.append(key)
                continue
            if not value:
                continue
            if not dry_run:
                store.put(key, value)
            pushed.append(key)

        verb = "would import" if dry_run else "imported"
        out.print(f"[green]{verb}[/] {len(pushed)} secret(s): {', '.join(pushed) or '—'}")
        if skipped:
            out.print(f"[dim]skipped AWS bootstrap keys: {', '.join(skipped)}[/]")

    @app.command("export")
    def secrets_export(ctx: typer.Context) -> None:
        """Print every visible secret as KEY=VALUE lines, resolved along the path."""
        store = store_from(ctx)
        for name in sorted(store.names()):
            emit(out, f"{name}={store.get(name, required=True)}")

    return app

store_from(ctx)

Return the store this invocation resolved.

Useful to host applications adding their own commands to a group built by :func:build_secrets_app, which then read the same search path as the rest.

Parameters:

Name	Type	Description	Default
ctx	Context	Context of a command inside the group.	required

Returns:

Type	Description
SecretStore	The store built by the group's callback.

Source code in src/pdum/aws/cli/secrets.py

def store_from(ctx: typer.Context) -> SecretStore:
    """Return the store this invocation resolved.

    Useful to host applications adding their own commands to a group built by
    :func:`build_secrets_app`, which then read the same search path as the rest.

    Parameters
    ----------
    ctx : typer.Context
        Context of a command inside the group.

    Returns
    -------
    SecretStore
        The store built by the group's callback.
    """
    return ctx.meta[STORE_KEY]

pdum.aws.cli.quotas

The quotas command group, as a factory other applications can embed.

:func:build_quotas_app mirrors :func:~pdum.aws.cli.secrets.build_secrets_app: a fresh :class:typer.Typer per call, rendering through a console the host supplies, with the defaults its users should not have to type. Where the secrets group needs a search path, this one needs a service code, a target plan and a set of regions:

import typer

from pdum.aws.cli import build_quotas_app

app = typer.Typer()
app.add_typer(
    build_quotas_app(
        default_service="ec2",
        default_targets=ACME_GPU_TARGETS,
        default_regions=["us-east-1", "us-west-2"],
    ),
    name="quotas",
)

acme quotas status now reports the plan that application cares about, in the regions it runs in. The flags remain available to override any of it; pass show_service_option=False / show_targets_option=False to keep them out of --help when the host's users have no business changing them.

build_quotas_app(*, console=None, default_service='ec2', default_targets=None, default_regions=None, show_service_option=True, show_targets_option=True, help='Inspect and request AWS service quotas.')

Build a quotas command group.

Parameters:

Name	Type	Description	Default
console	Console	Console every command in the group renders through; defaults to :data:~pdum.aws.cli.output.DEFAULT_CONSOLE.	None
default_service	str	Service Quotas service code used when --service is absent.	"ec2"
default_targets	list of QuotaTarget or callable	Target plan used when --targets is absent. Pass a zero-argument callable to defer building it. None means :data:~pdum.aws.quotas.EC2_VCPU_TARGETS, which is an example to copy and edit rather than a recommendation — a host with a real workload should pass its own.	None
default_regions	list of str	Regions acted on when --region is absent. None means the single region boto3 resolves from the environment.	None
show_service_option	bool	Whether --service appears in --help. It keeps working either way; hiding it is for hosts whose plan only makes sense for one service.	True
show_targets_option	bool	Whether --targets appears in --help, on the same terms.	True
help	str	Group help text.	'Inspect and request AWS service quotas.'

Returns:

Type	Description
Typer	A new application, ready to pass to add_typer.

Source code in src/pdum/aws/cli/quotas.py

def build_quotas_app(
    *,
    console: Console | None = None,
    default_service: str = "ec2",
    default_targets: TargetsDefault = None,
    default_regions: list[str] | None = None,
    show_service_option: bool = True,
    show_targets_option: bool = True,
    help: str = "Inspect and request AWS service quotas.",
) -> typer.Typer:
    """Build a ``quotas`` command group.

    Parameters
    ----------
    console : rich.console.Console, optional
        Console every command in the group renders through; defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_CONSOLE`.
    default_service : str, default "ec2"
        Service Quotas service code used when ``--service`` is absent.
    default_targets : list of QuotaTarget or callable, optional
        Target plan used when ``--targets`` is absent. Pass a zero-argument
        callable to defer building it. ``None`` means
        :data:`~pdum.aws.quotas.EC2_VCPU_TARGETS`, which is an example to copy
        and edit rather than a recommendation — a host with a real workload
        should pass its own.
    default_regions : list of str, optional
        Regions acted on when ``--region`` is absent. ``None`` means the single
        region ``boto3`` resolves from the environment.
    show_service_option : bool, default True
        Whether ``--service`` appears in ``--help``. It keeps working either
        way; hiding it is for hosts whose plan only makes sense for one service.
    show_targets_option : bool, default True
        Whether ``--targets`` appears in ``--help``, on the same terms.
    help : str
        Group help text.

    Returns
    -------
    typer.Typer
        A new application, ready to pass to ``add_typer``.
    """
    out = console or DEFAULT_CONSOLE
    app = typer.Typer(help=help, no_args_is_help=True)

    def resolve_targets(path: Path | None) -> list[quotas_lib.QuotaTarget]:
        """Targets from the flag, else the host's plan, else the bundled one."""
        if path is not None:
            return _load_targets_file(path)
        if default_targets is None:
            # Read late so the module attribute stays the single source of truth.
            return quotas_lib.EC2_VCPU_TARGETS
        return default_targets() if callable(default_targets) else default_targets

    def resolve_regions(given: list[str] | None) -> list[str | None]:
        """Regions to act on; ``[None]`` means whatever the environment resolves."""
        return list(given) if given else [None]

    def region_option(verb: str) -> Any:
        return typer.Option(
            list(default_regions) if default_regions else None,
            "--region",
            "-r",
            help=f"region to {verb} (repeatable)",
        )

    def service_option() -> Any:
        return typer.Option(
            default_service,
            "--service",
            hidden=not show_service_option,
            help="Service Quotas service code",
        )

    def targets_option() -> Any:
        source = "the bundled EC2 plan" if default_targets is None else "this application's plan"
        return typer.Option(
            None,
            "--targets",
            hidden=not show_targets_option,
            help=f"JSON target list; defaults to {source}",
        )

    @app.command("status")
    def quotas_status(
        region: list[str] = region_option("query"),
        service: str = service_option(),
        targets_file: Path = targets_option(),
    ) -> None:
        """Show current values against target values."""
        targets = resolve_targets(targets_file)
        for reg in resolve_regions(region):
            _heading(out, reg)
            statuses = quotas_lib.report(targets, service_code=service, region=reg)
            out.print(_status_table(statuses))
            pending = sum(1 for s in statuses if s.has_open_request)
            todo = sum(1 for s in statuses if s.needs_submit)
            out.print(f"  [dim]{pending} open, {todo} to submit[/]")

    @app.command("history")
    def quotas_history(
        region: list[str] = region_option("query"),
        service: str = service_option(),
        open_only: bool = typer.Option(False, "--open", help="show only requests still awaiting a decision"),
    ) -> None:
        """Show what AWS decided about past quota-increase requests."""
        for reg in resolve_regions(region):
            _heading(out, reg)
            requests = quotas_lib.request_history(service_code=service, region=reg)
            if open_only:
                requests = [r for r in requests if r.is_open]
            if not requests:
                out.print("  [dim]no requests[/]")
                continue

            out.print(_history_table(requests))
            counts: dict[str, int] = {}
            for request in requests:
                counts[request.status] = counts.get(request.status, 0) + 1
            out.print("  [dim]" + ", ".join(f"{n} {s}" for s, n in sorted(counts.items())) + "[/]")

    @app.command("request")
    def quotas_request(
        region: list[str] = region_option("act on"),
        service: str = service_option(),
        targets_file: Path = targets_option(),
        dry_run: bool = typer.Option(False, "--dry-run", help="show what would be requested, write nothing"),
    ) -> None:
        """Request increases for every target that needs one.

        Idempotent: quotas already satisfied, or already carrying an open
        request, are skipped. Stops when the account hits its open-request cap,
        so re-run once earlier cases are decided.
        """
        targets = resolve_targets(targets_file)
        for reg in resolve_regions(region):
            _heading(out, reg)
            results = quotas_lib.submit(targets, service_code=service, region=reg, dry_run=dry_run)
            capped = False
            for result in results:
                if result.outcome == "skipped":
                    continue
                if result.outcome == "cap-reached":
                    capped = True
                    out.print(f"  [yellow]{result.target.label}[/] -> blocked: account is at its open-request cap")
                    break
                colour = {"submitted": "green", "dry-run": "cyan", "error": "red"}[result.outcome]
                detail = f" {result.error}" if result.error else ""
                out.print(f"  [{colour}]{result.outcome}[/] {result.target.label} -> {result.target.value:.0f}{detail}")
            submitted = sum(1 for r in results if r.outcome in {"submitted", "dry-run"})
            skipped = sum(1 for r in results if r.outcome == "skipped")
            out.print(f"  [dim]{submitted} submitted, {skipped} skipped[/]")
            if capped:
                out.print("  [dim]re-run once some cases are decided[/]")

    @app.command("targets")
    def quotas_targets() -> None:
        """Print this group's default target plan as JSON, ready to edit and pass to --targets."""
        payload: list[dict[str, Any]] = [
            {"label": t.label, "code": t.code, "value": t.value} for t in resolve_targets(None)
        ]
        emit(out, json.dumps(payload, indent=2))

    return app

pdum.aws.cli.pdx

pdx — run a command with the project's environment and secrets loaded.

$ pdx npm run dev
$ pdx python -m my_service --port 8080
$ pdx -- ls -la

pdx has no options of its own. Everything after the name is the command to run, so no flag of yours can collide with one of its own, and -- is available but never required. The only exception is a leading --help, which click reserves; pdx python --help still reaches Python, because parsing stops at the first bare word.

The environment it hands over is built by :func:pdum.aws.env.load_env, in three layers — what you already have, the nearest .env and its .env.local overlay, then the SSM store. That module is the place to read about the ordering and why it matters; this one is only the command around it, and a script wanting the same environment should call load_env directly rather than shelling out to pdx.

Then the process is replaced by the command, a real execvp: same PID, so signals, job control, exit status and the terminal all belong to the program you asked for, with nothing left in the middle to forward them.

pdx-doctor shows what that adds up to without running anything.

Both commands are attachable, so an application can offer the same thing under its own name and defaults:

from pdum.aws.cli import add_pdx, add_pdx_doctor

add_pdx(app, default_path="/acme/:/org/", envvar="ACME_SSM_PATH")
add_pdx_doctor(app, default_path="/acme/:/org/", envvar="ACME_SSM_PATH")

add_pdx(app, *, console=None, default_path=None, envvar=DEFAULT_ENVVAR, env_file='.env', store_factory=SecretStore, name='pdx')

Register a pdx command on app.

Parameters:

Name	Type	Description	Default
app	Typer	Application to register the command on.	required
console	Console	Console for this command's own diagnostics. Defaults to :data:~pdum.aws.cli.output.DEFAULT_STDERR_CONSOLE; whatever you pass should write to stderr, since stdout belongs to the wrapped program.	None
default_path	str or callable	Search path used when the environment, including the .env, has none. A callable is resolved at invocation.	None
envvar	str	Environment variable holding the search path. None leaves only default_path; the .env is still loaded either way.	DEFAULT_ENVVAR
env_file	str	File searched for upward from the working directory, and the base name of the .local overlay beside it. A name with a directory component is used as given.	".env"
store_factory	callable	Called with the resolved path to build the store.	:class:`~pdum.aws.secrets.SecretStore`
name	str	Name to register the command under.	"pdx"

Source code in src/pdum/aws/cli/pdx.py

def add_pdx(
    app: typer.Typer,
    *,
    console: Console | None = None,
    default_path: PathDefault = None,
    envvar: str | None = DEFAULT_ENVVAR,
    env_file: str = ".env",
    store_factory: Callable[[str], SecretStore] = SecretStore,
    name: str = "pdx",
) -> None:
    """Register a ``pdx`` command on *app*.

    Parameters
    ----------
    app : typer.Typer
        Application to register the command on.
    console : rich.console.Console, optional
        Console for this command's own diagnostics. Defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_STDERR_CONSOLE`; whatever you pass
        should write to stderr, since stdout belongs to the wrapped program.
    default_path : str or callable, optional
        Search path used when the environment, including the ``.env``, has
        none. A callable is resolved at invocation.
    envvar : str, optional
        Environment variable holding the search path. ``None`` leaves only
        *default_path*; the ``.env`` is still loaded either way.
    env_file : str, default ".env"
        File searched for upward from the working directory, and the base name
        of the ``.local`` overlay beside it. A name with a directory component
        is used as given.
    store_factory : callable, default :class:`~pdum.aws.secrets.SecretStore`
        Called with the resolved path to build the store.
    name : str, default "pdx"
        Name to register the command under.
    """
    out = console or DEFAULT_STDERR_CONSOLE

    @app.command(name, context_settings=_PASSTHROUGH)
    def pdx(
        command: list[str] = typer.Argument(None, metavar="COMMAND [ARGS]...", help="program to run, with arguments"),
    ) -> None:
        """Run a command with the project's environment and secrets loaded.

        Takes no options of its own: everything here is the command to run.
        Variables already set are left alone.
        """
        argv = list(command or [])
        if not argv:
            out.print("[red]nothing to run:[/] pdx takes the command to exec, e.g. [bold]pdx npm start[/]")
            raise typer.Exit(2)

        try:
            load_env(env_file=env_file, envvar=envvar, default_path=default_path, store_factory=store_factory)
        except (NoSearchPath, ValueError) as exc:
            out.print(f"[red]{exc}[/]")
            raise typer.Exit(2) from exc

        _exec(argv, out)

add_pdx_doctor(app, *, console=None, default_path=None, envvar=DEFAULT_ENVVAR, env_file='.env', store_factory=SecretStore, name='pdx-doctor')

Register a command explaining what :func:add_pdx would do.

Answers the questions secrets list cannot, because they are about this process rather than about the store: which .env was found and what it contributed, which search path resolved and from where, and which names the environment already holds — the reason a program run under pdx can see a value that is not the one in SSM.

It applies the .env exactly as pdx does, since that is what decides the account the store is read from. Secrets themselves are reported by name and layer only: this works from :meth:~pdum.aws.secrets.SecretStore.describe, which does not decrypt, so running it pulls no secret value over the wire. Use secrets get when the value is what you want.

Parameters are as for :func:add_pdx, except that console defaults to stdout — here the report is the output.

Source code in src/pdum/aws/cli/pdx.py

def add_pdx_doctor(
    app: typer.Typer,
    *,
    console: Console | None = None,
    default_path: PathDefault = None,
    envvar: str | None = DEFAULT_ENVVAR,
    env_file: str = ".env",
    store_factory: Callable[[str], SecretStore] = SecretStore,
    name: str = "pdx-doctor",
) -> None:
    """Register a command explaining what :func:`add_pdx` would do.

    Answers the questions ``secrets list`` cannot, because they are about this
    process rather than about the store: which ``.env`` was found and what it
    contributed, which search path resolved and from where, and which names the
    environment already holds — the reason a program run under ``pdx`` can see a
    value that is not the one in SSM.

    It applies the ``.env`` exactly as ``pdx`` does, since that is what decides
    the account the store is read from. Secrets themselves are reported by name
    and layer only: this works from
    :meth:`~pdum.aws.secrets.SecretStore.describe`, which does not decrypt, so
    running it pulls no secret value over the wire. Use ``secrets get`` when the
    value is what you want.

    Parameters are as for :func:`add_pdx`, except that *console* defaults to
    stdout — here the report is the output.
    """
    out = console or DEFAULT_CONSOLE

    @app.command(name)
    def pdx_doctor() -> None:
        """Explain what pdx would put in the environment, without running anything."""
        from_file = load_env_files(env_file, os.environ)
        if from_file is None:
            out.print(f"[dim]no {env_file} found in this directory or above it[/]")
        else:
            out.print(f"[bold].env[/] {from_file.source}")
            out.print(f"  [green]sets[/] {', '.join(from_file.applied) or '—'}")
            if from_file.already_set:
                out.print(f"  [cyan]kept from the environment[/] {', '.join(from_file.already_set)}")

        path, source = resolve_search_path(
            envvar=envvar, default_path=default_path, environ=os.environ, env_file_report=from_file
        )
        if not path:
            out.print(f"[red]{no_search_path_message(envvar, env_file, from_file)}[/]")
            raise typer.Exit(2)
        out.print(f"[bold]search path[/] {path}  [dim](from {source})[/]")
        try:
            store = store_factory(path)
        except ValueError as exc:
            out.print(f"[red]{exc}[/]")
            raise typer.Exit(2) from exc

        rows = sorted((r for r in store.describe() if not r["shadowed"]), key=lambda r: r["name"])
        if not rows:
            out.print(f"[yellow]no secrets under {path}[/]")
            return

        table = Table(box=None, pad_edge=False)
        table.add_column("NAME", style="bold")
        table.add_column("LAYER")
        table.add_column("PDX WOULD")
        counts = {"set it": 0, "keep the environment's": 0, "skip it": 0}
        for row in rows:
            secret = row["name"]
            if not is_env_name(secret):
                verdict, style = "skip it", "yellow"
            elif os.environ.get(secret):
                verdict, style = "keep the environment's", "cyan"
            else:
                verdict, style = "set it", "green"
            counts[verdict] += 1
            table.add_row(secret, row["origin"], f"[{style}]{verdict}[/]")
        out.print(table)
        out.print("  [dim]" + ", ".join(f"{n} to {verdict}" for verdict, n in counts.items() if n) + "[/]")

build_pdx_app(**kwargs)

Build a standalone one-command application around :func:add_pdx.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Forwarded to :func:add_pdx.	{}

Returns:

Type	Description
Typer	An application whose only command is pdx, so it is invoked as pdx COMMAND [ARGS]... rather than pdx pdx ....

Source code in src/pdum/aws/cli/pdx.py

def build_pdx_app(**kwargs: Any) -> typer.Typer:
    """Build a standalone one-command application around :func:`add_pdx`.

    Parameters
    ----------
    **kwargs
        Forwarded to :func:`add_pdx`.

    Returns
    -------
    typer.Typer
        An application whose only command is ``pdx``, so it is invoked as
        ``pdx COMMAND [ARGS]...`` rather than ``pdx pdx ...``.
    """
    app = typer.Typer(add_completion=False)
    add_pdx(app, **kwargs)
    return app

main()

Console-script entry point for pdx.

Source code in src/pdum/aws/cli/pdx.py

def main() -> None:
    """Console-script entry point for ``pdx``."""
    with aws_errors(DEFAULT_STDERR_CONSOLE):
        app()

pdum.aws.cli.identity

The whoami command, attachable to any Typer application.

A single command rather than a group, so it is registered onto the host's app directly instead of being added as a sub-app:

import typer

from pdum.aws.cli import add_whoami

app = typer.Typer()
add_whoami(app, name="aws-identity")

add_whoami(app, *, console=None, name='whoami')

Register a whoami command on app.

Parameters:

Name	Type	Description	Default
app	Typer	Application to register the command on.	required
console	Console	Console to render through; defaults to :data:~pdum.aws.cli.output.DEFAULT_CONSOLE.	None
name	str	Name to register the command under, in case the host already has one.	"whoami"

Source code in src/pdum/aws/cli/identity.py

def add_whoami(app: typer.Typer, *, console: Console | None = None, name: str = "whoami") -> None:
    """Register a ``whoami`` command on *app*.

    Parameters
    ----------
    app : typer.Typer
        Application to register the command on.
    console : rich.console.Console, optional
        Console to render through; defaults to
        :data:`~pdum.aws.cli.output.DEFAULT_CONSOLE`.
    name : str, default "whoami"
        Name to register the command under, in case the host already has one.
    """
    out = console or DEFAULT_CONSOLE

    @app.command(name)
    def whoami() -> None:
        """Print the AWS identity the current credentials resolve to."""
        identity = aws.whoami()
        out.print(f"[bold]account[/] {identity['Account']}", soft_wrap=True)
        out.print(f"[bold]arn[/]     {identity['Arn']}", soft_wrap=True)

pdum.aws.cli.output

Console plumbing shared by every command group.

Each group is produced by a factory taking a console, so a host CLI can pass its own themed :class:~rich.console.Console and have these commands render through it — one theme, one width, one output stream for the whole application. Omitting it falls back to :data:DEFAULT_CONSOLE.

from rich.console import Console

from pdum.aws.cli import build_secrets_app

app.add_typer(build_secrets_app(console=Console(stderr=True)), name="secrets")

aws_errors(console=None)

Turn credential and API failures into one readable line.

A missing profile or an expired SSO session is the single most common way these commands fail, and a botocore traceback is a poor way to say so. Host applications embedding a group should wrap their own entry point in this to get the same treatment:

def main() -> None:
    with aws_errors(my_console):
        app()

Parameters:

Name	Type	Description	Default
console	Console	Console to report through; defaults to :data:DEFAULT_CONSOLE.	None

Raises:

Type	Description
SystemExit	With status 2, in place of the original botocore exception.

Source code in src/pdum/aws/cli/output.py

@contextmanager
def aws_errors(console: Console | None = None) -> Iterator[None]:
    """Turn credential and API failures into one readable line.

    A missing profile or an expired SSO session is the single most common way
    these commands fail, and a botocore traceback is a poor way to say so. Host
    applications embedding a group should wrap their own entry point in this to
    get the same treatment:

    ```python
    def main() -> None:
        with aws_errors(my_console):
            app()
    ```

    Parameters
    ----------
    console : rich.console.Console, optional
        Console to report through; defaults to :data:`DEFAULT_CONSOLE`.

    Raises
    ------
    SystemExit
        With status 2, in place of the original ``botocore`` exception.
    """
    try:
        yield
    except (BotoCoreError, ClientError) as exc:
        (console or DEFAULT_CONSOLE).print(f"[red]{exc}[/]")
        raise SystemExit(2) from exc

emit(console, text)

Print text verbatim, as data rather than as a message.

Markup, syntax highlighting and wrapping are all disabled for this one call, so a secret containing square brackets is never mangled, a long value is never broken across lines, and piping stays clean. Use this for anything a caller might redirect into a file or another process; use console.print directly for messages addressed to a human.

Parameters:

Name	Type	Description	Default
console	Console	Console to write through.	required
text	str	The value to write.	required

Source code in src/pdum/aws/cli/output.py

def emit(console: Console, text: str) -> None:
    """Print *text* verbatim, as data rather than as a message.

    Markup, syntax highlighting and wrapping are all disabled for this one call,
    so a secret containing square brackets is never mangled, a long value is
    never broken across lines, and piping stays clean. Use this for anything a
    caller might redirect into a file or another process; use
    ``console.print`` directly for messages addressed to a human.

    Parameters
    ----------
    console : rich.console.Console
        Console to write through.
    text : str
        The value to write.
    """
    console.print(text, markup=False, highlight=False, soft_wrap=True)