API Reference

Utilities and tools for Google Cloud

APIResolutionError

Bases: Exception

Raised when an API display name cannot be uniquely resolved to a service ID.

Source code in src/pdum/gcp/types/exceptions.py

class APIResolutionError(Exception):
    """Raised when an API display name cannot be uniquely resolved to a service ID."""

    __slots__ = ()

BillingAccount dataclass

Information about a GCP billing account.

Attributes:

Name	Type	Description
id	str	Billing account ID (for example "012345-567890-ABCDEF").
display_name	str	Human-friendly billing account name.
status	str	Status indicator such as "OPEN" or "CLOSED" (defaults to "OPEN").

Source code in src/pdum/gcp/types/billing_account.py

@dataclass
class BillingAccount:
    """Information about a GCP billing account.

    Attributes
    ----------
    id : str
        Billing account ID (for example ``"012345-567890-ABCDEF"``).
    display_name : str
        Human-friendly billing account name.
    status : str
        Status indicator such as ``"OPEN"`` or ``"CLOSED"`` (defaults to ``"OPEN"``).
    """

    id: str
    display_name: str
    status: str = "OPEN"

    def __bool__(self) -> bool:
        """Return True for regular billing accounts.

        Notes
        -----
        Truthiness does not reflect the ``status`` field; even a ``CLOSED`` account
        is truthy. Use ``status`` to inspect openness if needed.
        """

        return True

__bool__()

Return True for regular billing accounts.

Notes

Truthiness does not reflect the status field; even a CLOSED account is truthy. Use status to inspect openness if needed.

Source code in src/pdum/gcp/types/billing_account.py

def __bool__(self) -> bool:
    """Return True for regular billing accounts.

    Notes
    -----
    Truthiness does not reflect the ``status`` field; even a ``CLOSED`` account
    is truthy. Use ``status`` to inspect openness if needed.
    """

    return True

Container dataclass

Bases: Resource

Base class for GCP resource containers (Organizations, Folders, and NO_ORG).

This base class provides common functionality for all container types that can hold projects and folders.

Attributes:

Name	Type	Description
id	str	Container identifier (numeric for organizations/folders, empty for NO_ORG).
resource_name	str	Fully-qualified resource name (e.g., "organizations/123" or "folders/456").
display_name	str	Human-readable label for the container.
_credentials	(Credentials, optional)	Cached credentials used for API calls when provided.

Source code in src/pdum/gcp/types/container.py

@dataclass
class Container(Resource):
    """Base class for GCP resource containers (Organizations, Folders, and NO_ORG).

    This base class provides common functionality for all container types that can
    hold projects and folders.

    Attributes
    ----------
    id : str
        Container identifier (numeric for organizations/folders, empty for ``NO_ORG``).
    resource_name : str
        Fully-qualified resource name (e.g., ``"organizations/123"`` or ``"folders/456"``).
    display_name : str
        Human-readable label for the container.
    _credentials : Credentials, optional
        Cached credentials used for API calls when provided.
    """

    id: str
    resource_name: str
    display_name: str
    _credentials: Optional[Credentials] = field(default=None, repr=False, compare=False)

    def full_resource_name(self) -> str:
        return self.resource_name

    def parent(self, *, credentials: Optional[Credentials] = None) -> Optional[Container]:
        """Get the parent container.

        Parameters
        ----------
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored credentials or ADC are used.

        Returns
        -------
        Container or None
            The parent container (organization or folder), or ``None`` if no parent exists.

        Raises
        ------
        NotImplementedError
            Always raised on the base class; subclasses must implement this method.
        """
        raise NotImplementedError("Subclasses must implement parent()")

    def folders(self, *, credentials: Optional[Credentials] = None) -> list[Folder]:
        """List folders that are direct children of this container.

        Parameters
        ----------
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored credentials or ADC are used.

        Returns
        -------
        list[Folder]
            Direct child folders of this container.

        Raises
        ------
        NotImplementedError
            Always raised on the base class; subclasses must implement this method.
        """
        raise NotImplementedError("Subclasses must implement folders()")

    def projects(self, *, credentials: Optional[Credentials] = None) -> list[Project]:
        """List projects that are direct children of this container.

        Parameters
        ----------
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored credentials or ADC are used.

        Returns
        -------
        list[Project]
            Direct child projects of this container.

        Raises
        ------
        NotImplementedError
            Always raised on the base class; subclasses must implement this method.
        """
        raise NotImplementedError("Subclasses must implement projects()")

    def create_folder(self, display_name: str, *, credentials: Optional[Credentials] = None) -> Folder:
        """Create a new folder as a child of this container.

        Parameters
        ----------
        display_name : str
            Human-readable name for the folder.
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored credentials or ADC are used.

        Returns
        -------
        Folder
            The newly created folder.

        Raises
        ------
        NotImplementedError
            Always raised on the base class; subclasses must implement this method.
        """
        raise NotImplementedError("Subclasses must implement create_folder()")

    def list_roles(
        self,
        *,
        credentials: Optional[Credentials] = None,
        user_email: str | None = None,
    ) -> list[Role]:
        """List IAM roles for a user on this container.

        Parameters
        ----------
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored credentials or ADC are used.
        user_email : str, optional
            Identity to query. If omitted, the email associated with the credentials is used.

        Returns
        -------
        list[Role]
            Roles that directly bind the user on this container.
        """
        creds = self._get_credentials(credentials=credentials)
        return _list_roles(credentials=creds, resource_name=self.resource_name, user_email=user_email)

    def create_project(
        self,
        project_id: str,
        display_name: str,
        *,
        billing_account: BillingAccount | str | None = None,
        credentials: Optional[Credentials] = None,
        timeout: float = 600.0,
        polling_interval: float = 5.0,
    ) -> Project:
        """Create a new project under this container and optionally attach billing.

        Parameters
        ----------
        project_id : str
            The new project's ID (must satisfy GCP constraints).
        display_name : str
            Human-friendly display name for the project.
        billing_account : BillingAccount | str | None, optional
            Billing account to attach after creation. If omitted or falsy (e.g., ``NO_BILLING_ACCOUNT``),
            billing is not attached.
        credentials : Credentials, optional
            Explicit credentials to use. If ``None``, uses stored credentials or ADC.
        timeout : float, default 600.0
            Max seconds to wait for the long-running create operation.
        polling_interval : float, default 5.0
            Seconds between operation polls.

        Returns
        -------
        Project
            The created project materialized as a Project instance.

        Raises
        ------
        googleapiclient.errors.HttpError
            If any API call fails.
        TimeoutError
            If creation does not complete within ``timeout`` seconds.
        RuntimeError
            If the create operation completes with an error.

        Notes
        -----
        This method mutates GCP estate (creates resources, may attach billing).
        Do not run in CI. Prefer invoking manually with appropriate credentials.
        """
        from .billing_account import NO_BILLING_ACCOUNT
        from .no_org import NO_ORG
        from .project import Project

        if billing_account is None:
            billing_account = NO_BILLING_ACCOUNT

        creds = self._get_credentials(credentials=credentials)
        crm = crm_v3(creds)

        body = {
            "projectId": project_id,
            "displayName": display_name,
        }

        is_no_org = (self is NO_ORG) or (getattr(self, "resource_name", "") == "NO_ORG")
        parent_name = None if is_no_org else self.resource_name
        if parent_name:
            body["parent"] = parent_name

        operation = crm.projects().create(body=body).execute()

        import time

        op_name = operation.get("name")
        start = time.time()
        while not operation.get("done", False):
            if time.time() - start > timeout:
                raise TimeoutError(f"Project create operation timed out after {timeout}s (operation: {op_name})")
            time.sleep(polling_interval)
            operation = crm.operations().get(name=op_name).execute()

        if "error" in operation:
            err = operation["error"]
            raise RuntimeError(f"Project creation failed: {err.get('code')}: {err.get('message')}")

        get_start = time.time()
        while True:
            try:
                crm.projects().get(name=f"projects/{project_id}").execute()
                break
            except Exception:
                if time.time() - get_start > timeout:
                    raise TimeoutError(f"Project get timed out after {timeout}s for 'projects/{project_id}'")
                time.sleep(polling_interval)

        if billing_account:
            Project.update_billing_account_for_id(project_id, billing_account, credentials=creds)

        search_start = time.time()
        while True:
            try:
                return Project.lookup(project_id, credentials=creds)
            except FileNotFoundError:
                if time.time() - search_start > timeout:
                    return Project(
                        id=project_id,
                        name=display_name,
                        project_number="",
                        lifecycle_state="",
                        parent=self,
                        _credentials=creds,
                    )
                time.sleep(polling_interval)

    def walk_projects(
        self,
        *,
        credentials: Optional[Credentials] = None,
        active_only: bool = True,
    ) -> Generator[Project, None, None]:
        """Recursively yield all projects within this container and its subfolders.

        Parameters
        ----------
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored credentials or ADC are used.
        active_only : bool, default True
            If ``True``, yield only ``ACTIVE`` projects. If ``False``, yield all lifecycle states.

        Yields
        ------
        Project
            Projects discovered in this container and all nested folders.
        """
        creds = self._get_credentials(credentials=credentials)

        for project in self.projects(credentials=creds):
            if active_only and project.lifecycle_state != "ACTIVE":
                continue
            yield project

        for folder in self.folders(credentials=creds):
            yield from folder.walk_projects(credentials=creds, active_only=active_only)

    def tree(self, *, credentials: Optional[Credentials] = None, _prefix: str = "", _is_last: bool = True) -> None:
        """Print a visual tree of this container and its children."""
        from .no_org import NO_ORG

        creds = self._get_credentials(credentials=credentials)

        if self is NO_ORG:
            emoji = "🐞"
        elif self.__class__.__name__ == "Organization":
            emoji = "🌺"
        else:
            emoji = "🎸"

        print(f"{_prefix}{emoji} {self.display_name} ({self.resource_name})")
        self._tree_children(credentials=creds, _prefix=_prefix)

    def _tree_children(self, *, credentials: Optional[Credentials] = None, _prefix: str = "") -> None:
        """Internal helper to print children without printing the parent again."""
        from .project import Project

        creds = self._get_credentials(credentials=credentials)

        folders = self.folders(credentials=creds)
        projects = self.projects(credentials=creds)

        all_children = folders + projects  # type: ignore[arg-type]
        total_children = len(all_children)

        for idx, child in enumerate(all_children):
            is_last_child = idx == total_children - 1

            branch = "└── " if is_last_child else "├── "

            if isinstance(child, Project):
                print(f"{_prefix}{branch}🎵 {child.id} ({child.lifecycle_state})")
            else:
                extension = "    " if is_last_child else "│   "
                new_prefix = _prefix + extension

                print(f"{_prefix}{branch}🎸 {child.display_name} ({child.resource_name})")
                child._tree_children(credentials=creds, _prefix=new_prefix)

    def cd(self, path: str, *, credentials: Optional[Credentials] = None) -> Folder:
        """Navigate to a child folder using a slash-separated path.

        Parameters
        ----------
        path : str
            Path like ``"dev/team-a/project-folder"``. Leading/trailing slashes are ignored.
        credentials : Credentials, optional
            Explicit credentials to use. If ``None``, uses stored credentials or ADC.

        Returns
        -------
        Folder
            The matching folder.

        Raises
        ------
        ValueError
            If the path is empty or a component is not found.
        TypeError
            If invoked on ``NO_ORG`` (which cannot have folders).
        """
        from .no_org import NO_ORG

        if self is NO_ORG:
            raise TypeError(
                "NO_ORG cannot have folders. Projects without an organization parent "
                "cannot contain folders. Use cd() on an organization or folder instead."
            )

        creds = self._get_credentials(credentials=credentials)

        clean_path = path.strip("/")
        if not clean_path:
            raise ValueError("Path cannot be empty")

        components = clean_path.split("/")
        current: Container = self

        for component in components:
            folders = current.folders(credentials=creds)

            matching_folder = next((folder for folder in folders if folder.display_name == component), None)
            if matching_folder is None:
                available = ", ".join(folder.display_name for folder in folders) or "(none)"
                raise ValueError(
                    f"Folder '{component}' not found in {current.display_name}. Available folders: {available}"
                )

            current = matching_folder

        return current  # type: ignore[return-value]

cd(path, *, credentials=None)

Navigate to a child folder using a slash-separated path.

Parameters:

Name	Type	Description	Default
path	str	Path like "dev/team-a/project-folder". Leading/trailing slashes are ignored.	required
credentials	Credentials	Explicit credentials to use. If None, uses stored credentials or ADC.	None

Returns:

Type	Description
Folder	The matching folder.

Raises:

Type	Description
ValueError	If the path is empty or a component is not found.
TypeError	If invoked on NO_ORG (which cannot have folders).

Source code in src/pdum/gcp/types/container.py

def cd(self, path: str, *, credentials: Optional[Credentials] = None) -> Folder:
    """Navigate to a child folder using a slash-separated path.

    Parameters
    ----------
    path : str
        Path like ``"dev/team-a/project-folder"``. Leading/trailing slashes are ignored.
    credentials : Credentials, optional
        Explicit credentials to use. If ``None``, uses stored credentials or ADC.

    Returns
    -------
    Folder
        The matching folder.

    Raises
    ------
    ValueError
        If the path is empty or a component is not found.
    TypeError
        If invoked on ``NO_ORG`` (which cannot have folders).
    """
    from .no_org import NO_ORG

    if self is NO_ORG:
        raise TypeError(
            "NO_ORG cannot have folders. Projects without an organization parent "
            "cannot contain folders. Use cd() on an organization or folder instead."
        )

    creds = self._get_credentials(credentials=credentials)

    clean_path = path.strip("/")
    if not clean_path:
        raise ValueError("Path cannot be empty")

    components = clean_path.split("/")
    current: Container = self

    for component in components:
        folders = current.folders(credentials=creds)

        matching_folder = next((folder for folder in folders if folder.display_name == component), None)
        if matching_folder is None:
            available = ", ".join(folder.display_name for folder in folders) or "(none)"
            raise ValueError(
                f"Folder '{component}' not found in {current.display_name}. Available folders: {available}"
            )

        current = matching_folder

    return current  # type: ignore[return-value]

create_folder(display_name, *, credentials=None)

Create a new folder as a child of this container.

Parameters:

Name	Type	Description	Default
display_name	str	Human-readable name for the folder.	required
credentials	Credentials	Explicit credentials to use. When omitted, stored credentials or ADC are used.	None

Returns:

Type	Description
Folder	The newly created folder.

Raises:

Type	Description
NotImplementedError	Always raised on the base class; subclasses must implement this method.

Source code in src/pdum/gcp/types/container.py

def create_folder(self, display_name: str, *, credentials: Optional[Credentials] = None) -> Folder:
    """Create a new folder as a child of this container.

    Parameters
    ----------
    display_name : str
        Human-readable name for the folder.
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored credentials or ADC are used.

    Returns
    -------
    Folder
        The newly created folder.

    Raises
    ------
    NotImplementedError
        Always raised on the base class; subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement create_folder()")

create_project(project_id, display_name, *, billing_account=None, credentials=None, timeout=600.0, polling_interval=5.0)

Create a new project under this container and optionally attach billing.

Parameters:

Name	Type	Description	Default
project_id	str	The new project's ID (must satisfy GCP constraints).	required
display_name	str	Human-friendly display name for the project.	required
billing_account	BillingAccount | str | None	Billing account to attach after creation. If omitted or falsy (e.g., NO_BILLING_ACCOUNT), billing is not attached.	None
credentials	Credentials	Explicit credentials to use. If None, uses stored credentials or ADC.	None
timeout	float	Max seconds to wait for the long-running create operation.	600.0
polling_interval	float	Seconds between operation polls.	5.0

Returns:

Type	Description
Project	The created project materialized as a Project instance.

Raises:

Type	Description
HttpError	If any API call fails.
TimeoutError	If creation does not complete within timeout seconds.
RuntimeError	If the create operation completes with an error.

Notes

This method mutates GCP estate (creates resources, may attach billing). Do not run in CI. Prefer invoking manually with appropriate credentials.

Source code in src/pdum/gcp/types/container.py

def create_project(
    self,
    project_id: str,
    display_name: str,
    *,
    billing_account: BillingAccount | str | None = None,
    credentials: Optional[Credentials] = None,
    timeout: float = 600.0,
    polling_interval: float = 5.0,
) -> Project:
    """Create a new project under this container and optionally attach billing.

    Parameters
    ----------
    project_id : str
        The new project's ID (must satisfy GCP constraints).
    display_name : str
        Human-friendly display name for the project.
    billing_account : BillingAccount | str | None, optional
        Billing account to attach after creation. If omitted or falsy (e.g., ``NO_BILLING_ACCOUNT``),
        billing is not attached.
    credentials : Credentials, optional
        Explicit credentials to use. If ``None``, uses stored credentials or ADC.
    timeout : float, default 600.0
        Max seconds to wait for the long-running create operation.
    polling_interval : float, default 5.0
        Seconds between operation polls.

    Returns
    -------
    Project
        The created project materialized as a Project instance.

    Raises
    ------
    googleapiclient.errors.HttpError
        If any API call fails.
    TimeoutError
        If creation does not complete within ``timeout`` seconds.
    RuntimeError
        If the create operation completes with an error.

    Notes
    -----
    This method mutates GCP estate (creates resources, may attach billing).
    Do not run in CI. Prefer invoking manually with appropriate credentials.
    """
    from .billing_account import NO_BILLING_ACCOUNT
    from .no_org import NO_ORG
    from .project import Project

    if billing_account is None:
        billing_account = NO_BILLING_ACCOUNT

    creds = self._get_credentials(credentials=credentials)
    crm = crm_v3(creds)

    body = {
        "projectId": project_id,
        "displayName": display_name,
    }

    is_no_org = (self is NO_ORG) or (getattr(self, "resource_name", "") == "NO_ORG")
    parent_name = None if is_no_org else self.resource_name
    if parent_name:
        body["parent"] = parent_name

    operation = crm.projects().create(body=body).execute()

    import time

    op_name = operation.get("name")
    start = time.time()
    while not operation.get("done", False):
        if time.time() - start > timeout:
            raise TimeoutError(f"Project create operation timed out after {timeout}s (operation: {op_name})")
        time.sleep(polling_interval)
        operation = crm.operations().get(name=op_name).execute()

    if "error" in operation:
        err = operation["error"]
        raise RuntimeError(f"Project creation failed: {err.get('code')}: {err.get('message')}")

    get_start = time.time()
    while True:
        try:
            crm.projects().get(name=f"projects/{project_id}").execute()
            break
        except Exception:
            if time.time() - get_start > timeout:
                raise TimeoutError(f"Project get timed out after {timeout}s for 'projects/{project_id}'")
            time.sleep(polling_interval)

    if billing_account:
        Project.update_billing_account_for_id(project_id, billing_account, credentials=creds)

    search_start = time.time()
    while True:
        try:
            return Project.lookup(project_id, credentials=creds)
        except FileNotFoundError:
            if time.time() - search_start > timeout:
                return Project(
                    id=project_id,
                    name=display_name,
                    project_number="",
                    lifecycle_state="",
                    parent=self,
                    _credentials=creds,
                )
            time.sleep(polling_interval)

folders(*, credentials=None)

List folders that are direct children of this container.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. When omitted, stored credentials or ADC are used.	None

Returns:

Type	Description
list[Folder]	Direct child folders of this container.

Raises:

Type	Description
NotImplementedError	Always raised on the base class; subclasses must implement this method.

Source code in src/pdum/gcp/types/container.py

def folders(self, *, credentials: Optional[Credentials] = None) -> list[Folder]:
    """List folders that are direct children of this container.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored credentials or ADC are used.

    Returns
    -------
    list[Folder]
        Direct child folders of this container.

    Raises
    ------
    NotImplementedError
        Always raised on the base class; subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement folders()")

list_roles(*, credentials=None, user_email=None)

List IAM roles for a user on this container.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. When omitted, stored credentials or ADC are used.	None
user_email	str	Identity to query. If omitted, the email associated with the credentials is used.	None

Returns:

Type	Description
list[Role]	Roles that directly bind the user on this container.

Source code in src/pdum/gcp/types/container.py

def list_roles(
    self,
    *,
    credentials: Optional[Credentials] = None,
    user_email: str | None = None,
) -> list[Role]:
    """List IAM roles for a user on this container.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored credentials or ADC are used.
    user_email : str, optional
        Identity to query. If omitted, the email associated with the credentials is used.

    Returns
    -------
    list[Role]
        Roles that directly bind the user on this container.
    """
    creds = self._get_credentials(credentials=credentials)
    return _list_roles(credentials=creds, resource_name=self.resource_name, user_email=user_email)

parent(*, credentials=None)

Get the parent container.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. When omitted, stored credentials or ADC are used.	None

Returns:

Type	Description
Container or None	The parent container (organization or folder), or None if no parent exists.

Raises:

Type	Description
NotImplementedError	Always raised on the base class; subclasses must implement this method.

Source code in src/pdum/gcp/types/container.py

def parent(self, *, credentials: Optional[Credentials] = None) -> Optional[Container]:
    """Get the parent container.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored credentials or ADC are used.

    Returns
    -------
    Container or None
        The parent container (organization or folder), or ``None`` if no parent exists.

    Raises
    ------
    NotImplementedError
        Always raised on the base class; subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement parent()")

projects(*, credentials=None)

List projects that are direct children of this container.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. When omitted, stored credentials or ADC are used.	None

Returns:

Type	Description
list[Project]	Direct child projects of this container.

Raises:

Type	Description
NotImplementedError	Always raised on the base class; subclasses must implement this method.

Source code in src/pdum/gcp/types/container.py

def projects(self, *, credentials: Optional[Credentials] = None) -> list[Project]:
    """List projects that are direct children of this container.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored credentials or ADC are used.

    Returns
    -------
    list[Project]
        Direct child projects of this container.

    Raises
    ------
    NotImplementedError
        Always raised on the base class; subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement projects()")

tree(*, credentials=None, _prefix='', _is_last=True)

Print a visual tree of this container and its children.

Source code in src/pdum/gcp/types/container.py

def tree(self, *, credentials: Optional[Credentials] = None, _prefix: str = "", _is_last: bool = True) -> None:
    """Print a visual tree of this container and its children."""
    from .no_org import NO_ORG

    creds = self._get_credentials(credentials=credentials)

    if self is NO_ORG:
        emoji = "🐞"
    elif self.__class__.__name__ == "Organization":
        emoji = "🌺"
    else:
        emoji = "🎸"

    print(f"{_prefix}{emoji} {self.display_name} ({self.resource_name})")
    self._tree_children(credentials=creds, _prefix=_prefix)

walk_projects(*, credentials=None, active_only=True)

Recursively yield all projects within this container and its subfolders.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. When omitted, stored credentials or ADC are used.	None
active_only	bool	If True, yield only ACTIVE projects. If False, yield all lifecycle states.	True

Yields:

Type	Description
Project	Projects discovered in this container and all nested folders.

Source code in src/pdum/gcp/types/container.py

def walk_projects(
    self,
    *,
    credentials: Optional[Credentials] = None,
    active_only: bool = True,
) -> Generator[Project, None, None]:
    """Recursively yield all projects within this container and its subfolders.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored credentials or ADC are used.
    active_only : bool, default True
        If ``True``, yield only ``ACTIVE`` projects. If ``False``, yield all lifecycle states.

    Yields
    ------
    Project
        Projects discovered in this container and all nested folders.
    """
    creds = self._get_credentials(credentials=credentials)

    for project in self.projects(credentials=creds):
        if active_only and project.lifecycle_state != "ACTIVE":
            continue
        yield project

    for folder in self.folders(credentials=creds):
        yield from folder.walk_projects(credentials=creds, active_only=active_only)

Folder dataclass

Bases: Container

Information about a GCP folder.

Source code in src/pdum/gcp/types/folder.py

@dataclass
class Folder(Container):
    """Information about a GCP folder."""

    parent_resource_name: str = ""

    def parent(self, *, credentials=None) -> Optional[Container]:
        """Return the parent container."""
        if not self.parent_resource_name:
            return None

        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        if self.parent_resource_name.startswith("organizations/"):
            from .organization import Organization

            org_id = self.parent_resource_name.split("/")[1]
            return Organization.lookup(org_id, credentials=creds)

        if self.parent_resource_name.startswith("folders/"):
            folder_resource = crm_service.folders().get(name=self.parent_resource_name).execute()
            return Folder(
                id=folder_resource["name"].split("/")[1],
                resource_name=folder_resource["name"],
                display_name=folder_resource.get("displayName", ""),
                parent_resource_name=folder_resource.get("parent", ""),
                _credentials=creds,
            )

        return None

    def folders(self, *, credentials=None) -> list["Folder"]:
        """List direct child folders of this folder."""
        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        folders: list[Folder] = []
        request = crm_service.folders().list(parent=self.resource_name)

        while request is not None:
            response = request.execute()
            for folder in response.get("folders", []):
                folders.append(
                    Folder(
                        id=folder["name"].split("/")[1],
                        resource_name=folder["name"],
                        display_name=folder.get("displayName", ""),
                        parent_resource_name=self.resource_name,
                        _credentials=creds,
                    )
                )

            request = crm_service.folders().list_next(previous_request=request, previous_response=response)

        return folders

    def projects(self, *, credentials=None) -> list["Project"]:
        """List direct child projects of this folder."""
        from .project import _project_from_api_response

        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        projects: list["Project"] = []
        request = crm_service.projects().list(parent=self.resource_name)

        while request is not None:
            response = request.execute()
            for project in response.get("projects", []):
                projects.append(_project_from_api_response(project, parent=self, credentials=creds))

            request = crm_service.projects().list_next(previous_request=request, previous_response=response)

        return projects

    def create_folder(self, display_name: str, *, credentials=None) -> "Folder":
        """Create a folder directly under this folder."""
        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        folder_body = {"displayName": display_name, "parent": self.resource_name}

        operation = crm_service.folders().create(body=folder_body).execute()

        import time

        while not operation.get("done", False):
            time.sleep(1)
            operation = crm_service.operations().get(name=operation["name"]).execute()

        folder_resource_name = operation["response"]["name"]
        folder_id = folder_resource_name.split("/")[1]

        return Folder(
            id=folder_id,
            resource_name=folder_resource_name,
            display_name=display_name,
            parent_resource_name=self.resource_name,
            _credentials=creds,
        )

create_folder(display_name, *, credentials=None)

Create a folder directly under this folder.

Source code in src/pdum/gcp/types/folder.py

def create_folder(self, display_name: str, *, credentials=None) -> "Folder":
    """Create a folder directly under this folder."""
    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    folder_body = {"displayName": display_name, "parent": self.resource_name}

    operation = crm_service.folders().create(body=folder_body).execute()

    import time

    while not operation.get("done", False):
        time.sleep(1)
        operation = crm_service.operations().get(name=operation["name"]).execute()

    folder_resource_name = operation["response"]["name"]
    folder_id = folder_resource_name.split("/")[1]

    return Folder(
        id=folder_id,
        resource_name=folder_resource_name,
        display_name=display_name,
        parent_resource_name=self.resource_name,
        _credentials=creds,
    )

folders(*, credentials=None)

List direct child folders of this folder.

Source code in src/pdum/gcp/types/folder.py

def folders(self, *, credentials=None) -> list["Folder"]:
    """List direct child folders of this folder."""
    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    folders: list[Folder] = []
    request = crm_service.folders().list(parent=self.resource_name)

    while request is not None:
        response = request.execute()
        for folder in response.get("folders", []):
            folders.append(
                Folder(
                    id=folder["name"].split("/")[1],
                    resource_name=folder["name"],
                    display_name=folder.get("displayName", ""),
                    parent_resource_name=self.resource_name,
                    _credentials=creds,
                )
            )

        request = crm_service.folders().list_next(previous_request=request, previous_response=response)

    return folders

parent(*, credentials=None)

Return the parent container.

Source code in src/pdum/gcp/types/folder.py

def parent(self, *, credentials=None) -> Optional[Container]:
    """Return the parent container."""
    if not self.parent_resource_name:
        return None

    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    if self.parent_resource_name.startswith("organizations/"):
        from .organization import Organization

        org_id = self.parent_resource_name.split("/")[1]
        return Organization.lookup(org_id, credentials=creds)

    if self.parent_resource_name.startswith("folders/"):
        folder_resource = crm_service.folders().get(name=self.parent_resource_name).execute()
        return Folder(
            id=folder_resource["name"].split("/")[1],
            resource_name=folder_resource["name"],
            display_name=folder_resource.get("displayName", ""),
            parent_resource_name=folder_resource.get("parent", ""),
            _credentials=creds,
        )

    return None

projects(*, credentials=None)

List direct child projects of this folder.

Source code in src/pdum/gcp/types/folder.py

def projects(self, *, credentials=None) -> list["Project"]:
    """List direct child projects of this folder."""
    from .project import _project_from_api_response

    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    projects: list["Project"] = []
    request = crm_service.projects().list(parent=self.resource_name)

    while request is not None:
        response = request.execute()
        for project in response.get("projects", []):
            projects.append(_project_from_api_response(project, parent=self, credentials=creds))

        request = crm_service.projects().list_next(previous_request=request, previous_response=response)

    return projects

MultiRegion

Bases: Enum

Canonical set of Google Cloud multi-regions.

Source code in src/pdum/gcp/types/region.py

class MultiRegion(Enum):
    """Canonical set of Google Cloud multi-regions."""

    EUR3 = ("eur3", "Europe", (Region.EUROPE_WEST1, Region.EUROPE_WEST4), Region.EUROPE_NORTH1)
    NAM5 = (
        "nam5",
        "United States (Central)",
        (Region.US_CENTRAL1, Region.US_CENTRAL2),
        Region.US_EAST1,
    )
    NAM7 = (
        "nam7",
        "United States (Central and East)",
        (Region.US_CENTRAL1, Region.US_EAST4),
        Region.US_CENTRAL2,
    )

    def __init__(
        self,
        multi_region_id: str,
        description: str,
        read_write_regions: Tuple[Region, ...],
        witness_region: Region,
    ) -> None:
        self._multi_region_id = multi_region_id
        self._description = description
        self._read_write_regions = tuple(read_write_regions)
        self._witness_region = witness_region

    @property
    def multi_region_id(self) -> str:
        """Identifier for the multi-region (e.g., ``nam5``)."""

        return self._multi_region_id

    @property
    def description(self) -> str:
        """Human-readable description for the multi-region."""

        return self._description

    @property
    def read_write_regions(self) -> Tuple[Region, ...]:
        """Regions that accept read-write traffic."""

        return self._read_write_regions

    @property
    def witness_region(self) -> Region:
        """Witness region used for tie-breaking."""

        return self._witness_region

    @classmethod
    def from_multi_region_id(cls, multi_region_id: str) -> "MultiRegion":
        """Return the enum entry matching ``multi_region_id``."""

        normalized = multi_region_id.lower()
        for multi_region in cls:
            if multi_region.multi_region_id == normalized:
                return multi_region
        raise ValueError(f"Unknown multi-region id: {multi_region_id!r}")

description property

Human-readable description for the multi-region.

multi_region_id property

Identifier for the multi-region (e.g., nam5).

read_write_regions property

Regions that accept read-write traffic.

witness_region property

Witness region used for tie-breaking.

from_multi_region_id(multi_region_id) classmethod

Return the enum entry matching multi_region_id.

Source code in src/pdum/gcp/types/region.py

@classmethod
def from_multi_region_id(cls, multi_region_id: str) -> "MultiRegion":
    """Return the enum entry matching ``multi_region_id``."""

    normalized = multi_region_id.lower()
    for multi_region in cls:
        if multi_region.multi_region_id == normalized:
            return multi_region
    raise ValueError(f"Unknown multi-region id: {multi_region_id!r}")

Organization dataclass

Bases: Container

Information about a GCP organization.

Source code in src/pdum/gcp/types/organization.py

@dataclass
class Organization(Container):
    """Information about a GCP organization."""

    ORGANIZATION_OWNER_ROLES: tuple[str, ...] = (
        "roles/billing.admin",
        "roles/billing.costsManager",
        "roles/billing.projectManager",
        "roles/iam.securityAdmin",
        "roles/orgpolicy.policyAdmin",
        "roles/resourcemanager.folderAdmin",
        "roles/resourcemanager.organizationAdmin",
        "roles/resourcemanager.projectCreator",
        "roles/resourcemanager.projectDeleter",
        "roles/resourcemanager.projectIamAdmin",
    )

    def parent(self, *, credentials: Optional[Credentials] = None) -> Optional[Container]:
        """Return the parent container (organizations are roots, so ``None``)."""
        return None

    def folders(self, *, credentials: Optional[Credentials] = None) -> list[Folder]:
        """List direct child folders of this organization."""
        from .folder import Folder

        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        folders: list[Folder] = []
        request = crm_service.folders().list(parent=self.resource_name)

        while request is not None:
            response = request.execute()
            for folder in response.get("folders", []):
                folders.append(
                    Folder(
                        id=folder["name"].split("/")[1],
                        resource_name=folder["name"],
                        display_name=folder.get("displayName", ""),
                        parent_resource_name=self.resource_name,
                        _credentials=creds,
                    )
                )

            request = crm_service.folders().list_next(previous_request=request, previous_response=response)

        return folders

    def projects(self, *, credentials: Optional[Credentials] = None) -> list["Project"]:
        """List direct child projects of this organization."""
        from .project import _project_from_api_response

        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        projects: list["Project"] = []
        request = crm_service.projects().list(parent=self.resource_name)

        while request is not None:
            response = request.execute()
            for project in response.get("projects", []):
                projects.append(_project_from_api_response(project, parent=self, credentials=creds))

            request = crm_service.projects().list_next(previous_request=request, previous_response=response)

        return projects

    def create_folder(self, display_name: str, *, credentials: Optional[Credentials] = None) -> "Folder":
        """Create a folder directly under this organization."""
        from .folder import Folder

        creds = self._get_credentials(credentials=credentials)
        crm_service = crm_v3(creds)

        folder_body = {
            "displayName": display_name,
            "parent": self.resource_name,
        }

        operation = crm_service.folders().create(body=folder_body).execute()

        import time

        while not operation.get("done", False):
            time.sleep(1)
            operation = crm_service.operations().get(name=operation["name"]).execute()

        folder_resource_name = operation["response"]["name"]
        folder_id = folder_resource_name.split("/")[1]

        return Folder(
            id=folder_id,
            resource_name=folder_resource_name,
            display_name=display_name,
            parent_resource_name=self.resource_name,
            _credentials=creds,
        )

    def add_user_roles(
        self,
        user_email: str,
        roles_to_add: list[str],
        *,
        credentials: Optional[Credentials] = None,
    ) -> dict:
        """Add a user to one or more IAM roles at the Organization level."""
        if "@" not in user_email or not user_email.strip():
            raise ValueError("user_email must be a valid email address")
        if not roles_to_add:
            raise ValueError("roles_to_add must be a non-empty list of role names")

        member = f"user:{user_email.strip()}"
        creds = self._get_credentials(credentials=credentials)
        crm = crm_v3(creds)
        resource = self.resource_name

        policy = (
            crm.organizations()
            .getIamPolicy(resource=resource, body={"options": {"requestedPolicyVersion": 3}})
            .execute()
        )

        if policy.get("version", 0) < 3:
            policy["version"] = 3

        bindings = policy.setdefault("bindings", [])
        changes_made = False

        for role in roles_to_add:
            binding = next((b for b in bindings if b.get("role") == role), None)
            if binding is None:
                bindings.append({"role": role, "members": [member]})
                changes_made = True
            else:
                members = binding.setdefault("members", [])
                if member not in members:
                    members.append(member)
                    changes_made = True

        if not changes_made:
            return policy

        updated = crm.organizations().setIamPolicy(resource=resource, body={"policy": policy}).execute()
        return updated

    def add_user_as_owner(self, user_email: str, *, credentials: Optional[Credentials] = None) -> dict:
        """Grant a user a standard set of high-privilege org roles."""
        return self.add_user_roles(
            user_email,
            roles_to_add=list(self.ORGANIZATION_OWNER_ROLES),
            credentials=credentials,
        )

    def billing_accounts(
        self,
        *,
        credentials: Optional[Credentials] = None,
        open_only: bool = True,
    ) -> list["BillingAccount"]:
        """List billing accounts scoped to this organization."""
        from .billing_account import BillingAccount

        creds = self._get_credentials(credentials=credentials)
        billing_service = cloud_billing(creds)

        accounts: list[BillingAccount] = []
        request = billing_service.billingAccounts().list(parent=self.resource_name)

        while request is not None:
            response = request.execute()

            for account in response.get("billingAccounts", []):
                billing_account_id = account["name"].split("/")[1]
                display_name = account.get("displayName", billing_account_id)
                is_open = account.get("open", False)
                status = "OPEN" if is_open else "CLOSED"

                if open_only and not is_open:
                    continue

                accounts.append(BillingAccount(id=billing_account_id, display_name=display_name, status=status))

            request = billing_service.billingAccounts().list_next(previous_request=request, previous_response=response)

        return accounts

    @classmethod
    def lookup(cls, org_id: str, *, credentials: Optional[Credentials] = None) -> "Organization":
        """Return an Organization by id using CRM v3."""
        if credentials is None:
            credentials, _ = google.auth.default()

        crm_service = crm_v3(credentials)
        resource_name = f"organizations/{org_id}"
        org_resource = crm_service.organizations().get(name=resource_name).execute()

        return cls(
            id=org_id,
            resource_name=resource_name,
            display_name=org_resource.get("displayName", ""),
            _credentials=credentials,
        )

add_user_as_owner(user_email, *, credentials=None)

Grant a user a standard set of high-privilege org roles.

Source code in src/pdum/gcp/types/organization.py

def add_user_as_owner(self, user_email: str, *, credentials: Optional[Credentials] = None) -> dict:
    """Grant a user a standard set of high-privilege org roles."""
    return self.add_user_roles(
        user_email,
        roles_to_add=list(self.ORGANIZATION_OWNER_ROLES),
        credentials=credentials,
    )

add_user_roles(user_email, roles_to_add, *, credentials=None)

Add a user to one or more IAM roles at the Organization level.

Source code in src/pdum/gcp/types/organization.py

def add_user_roles(
    self,
    user_email: str,
    roles_to_add: list[str],
    *,
    credentials: Optional[Credentials] = None,
) -> dict:
    """Add a user to one or more IAM roles at the Organization level."""
    if "@" not in user_email or not user_email.strip():
        raise ValueError("user_email must be a valid email address")
    if not roles_to_add:
        raise ValueError("roles_to_add must be a non-empty list of role names")

    member = f"user:{user_email.strip()}"
    creds = self._get_credentials(credentials=credentials)
    crm = crm_v3(creds)
    resource = self.resource_name

    policy = (
        crm.organizations()
        .getIamPolicy(resource=resource, body={"options": {"requestedPolicyVersion": 3}})
        .execute()
    )

    if policy.get("version", 0) < 3:
        policy["version"] = 3

    bindings = policy.setdefault("bindings", [])
    changes_made = False

    for role in roles_to_add:
        binding = next((b for b in bindings if b.get("role") == role), None)
        if binding is None:
            bindings.append({"role": role, "members": [member]})
            changes_made = True
        else:
            members = binding.setdefault("members", [])
            if member not in members:
                members.append(member)
                changes_made = True

    if not changes_made:
        return policy

    updated = crm.organizations().setIamPolicy(resource=resource, body={"policy": policy}).execute()
    return updated

billing_accounts(*, credentials=None, open_only=True)

List billing accounts scoped to this organization.

Source code in src/pdum/gcp/types/organization.py

def billing_accounts(
    self,
    *,
    credentials: Optional[Credentials] = None,
    open_only: bool = True,
) -> list["BillingAccount"]:
    """List billing accounts scoped to this organization."""
    from .billing_account import BillingAccount

    creds = self._get_credentials(credentials=credentials)
    billing_service = cloud_billing(creds)

    accounts: list[BillingAccount] = []
    request = billing_service.billingAccounts().list(parent=self.resource_name)

    while request is not None:
        response = request.execute()

        for account in response.get("billingAccounts", []):
            billing_account_id = account["name"].split("/")[1]
            display_name = account.get("displayName", billing_account_id)
            is_open = account.get("open", False)
            status = "OPEN" if is_open else "CLOSED"

            if open_only and not is_open:
                continue

            accounts.append(BillingAccount(id=billing_account_id, display_name=display_name, status=status))

        request = billing_service.billingAccounts().list_next(previous_request=request, previous_response=response)

    return accounts

create_folder(display_name, *, credentials=None)

Create a folder directly under this organization.

Source code in src/pdum/gcp/types/organization.py

def create_folder(self, display_name: str, *, credentials: Optional[Credentials] = None) -> "Folder":
    """Create a folder directly under this organization."""
    from .folder import Folder

    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    folder_body = {
        "displayName": display_name,
        "parent": self.resource_name,
    }

    operation = crm_service.folders().create(body=folder_body).execute()

    import time

    while not operation.get("done", False):
        time.sleep(1)
        operation = crm_service.operations().get(name=operation["name"]).execute()

    folder_resource_name = operation["response"]["name"]
    folder_id = folder_resource_name.split("/")[1]

    return Folder(
        id=folder_id,
        resource_name=folder_resource_name,
        display_name=display_name,
        parent_resource_name=self.resource_name,
        _credentials=creds,
    )

folders(*, credentials=None)

List direct child folders of this organization.

Source code in src/pdum/gcp/types/organization.py

def folders(self, *, credentials: Optional[Credentials] = None) -> list[Folder]:
    """List direct child folders of this organization."""
    from .folder import Folder

    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    folders: list[Folder] = []
    request = crm_service.folders().list(parent=self.resource_name)

    while request is not None:
        response = request.execute()
        for folder in response.get("folders", []):
            folders.append(
                Folder(
                    id=folder["name"].split("/")[1],
                    resource_name=folder["name"],
                    display_name=folder.get("displayName", ""),
                    parent_resource_name=self.resource_name,
                    _credentials=creds,
                )
            )

        request = crm_service.folders().list_next(previous_request=request, previous_response=response)

    return folders

lookup(org_id, *, credentials=None) classmethod

Return an Organization by id using CRM v3.

Source code in src/pdum/gcp/types/organization.py

@classmethod
def lookup(cls, org_id: str, *, credentials: Optional[Credentials] = None) -> "Organization":
    """Return an Organization by id using CRM v3."""
    if credentials is None:
        credentials, _ = google.auth.default()

    crm_service = crm_v3(credentials)
    resource_name = f"organizations/{org_id}"
    org_resource = crm_service.organizations().get(name=resource_name).execute()

    return cls(
        id=org_id,
        resource_name=resource_name,
        display_name=org_resource.get("displayName", ""),
        _credentials=credentials,
    )

parent(*, credentials=None)

Return the parent container (organizations are roots, so None).

Source code in src/pdum/gcp/types/organization.py

def parent(self, *, credentials: Optional[Credentials] = None) -> Optional[Container]:
    """Return the parent container (organizations are roots, so ``None``)."""
    return None

projects(*, credentials=None)

List direct child projects of this organization.

Source code in src/pdum/gcp/types/organization.py

def projects(self, *, credentials: Optional[Credentials] = None) -> list["Project"]:
    """List direct child projects of this organization."""
    from .project import _project_from_api_response

    creds = self._get_credentials(credentials=credentials)
    crm_service = crm_v3(creds)

    projects: list["Project"] = []
    request = crm_service.projects().list(parent=self.resource_name)

    while request is not None:
        response = request.execute()
        for project in response.get("projects", []):
            projects.append(_project_from_api_response(project, parent=self, credentials=creds))

        request = crm_service.projects().list_next(previous_request=request, previous_response=response)

    return projects

Project dataclass

Bases: Resource

Information about a GCP project.

Source code in src/pdum/gcp/types/project.py

@dataclass
class Project(Resource):
    """Information about a GCP project."""

    id: str
    name: str
    project_number: str
    lifecycle_state: str
    parent: "Container"
    _credentials: Optional[Credentials] = field(default=None, repr=False, compare=False)

    def full_resource_name(self) -> str:
        return f"projects/{self.id}"

    def enabled_apis(self, *, credentials: Optional[Credentials] = None) -> list[str]:
        """List enabled APIs for this project."""
        creds = self._get_credentials(credentials=credentials)
        service_usage_client = service_usage(creds)

        enabled_apis: list[str] = []
        parent_name = f"projects/{self.id}"
        request = service_usage_client.services().list(parent=parent_name, filter="state:ENABLED")

        while request is not None:
            response = request.execute()

            for service in response.get("services", []):
                api_name = service.get("config", {}).get("name", "")
                if api_name:
                    enabled_apis.append(api_name)

            request = service_usage_client.services().list_next(previous_request=request, previous_response=response)

        return enabled_apis

    def enable_apis(
        self,
        api_list: list[str],
        *,
        credentials: Optional[Credentials] = None,
        timeout: float = 300.0,
        verbose: bool = True,
        polling_interval: float = 5.0,
    ) -> dict:
        """Enable multiple APIs for this project using batch enable."""
        import time

        creds = self._get_credentials(credentials=credentials)
        service_usage_client = service_usage(creds)

        parent_name = f"projects/{self.id}"
        request_body = {"serviceIds": api_list}

        operation = service_usage_client.services().batchEnable(parent=parent_name, body=request_body).execute()

        operation_name = operation.get("name")
        if verbose:
            print(f"Enabling {len(api_list)} APIs for project {self.id}... ", end="", flush=True)

        start_time = time.time()

        while True:
            elapsed = time.time() - start_time
            if elapsed > timeout:
                if verbose:
                    print()
                raise TimeoutError(f"Operation timed out after {timeout} seconds. Operation name: {operation_name}")

            operation = service_usage_client.operations().get(name=operation_name).execute()

            if operation.get("done", False):
                if verbose:
                    print()

                if "error" in operation:
                    error = operation["error"]
                    error_message = error.get("message", "Unknown error")
                    error_code = error.get("code", "Unknown")
                    raise RuntimeError(f"Operation failed with error code {error_code}: {error_message}")

                return operation

            if verbose:
                print(".", end="", flush=True)

            time.sleep(polling_interval)

    def billing_account(self, *, credentials: Optional[Credentials] = None) -> BillingAccount:
        """Return the project's billing account or ``NO_BILLING_ACCOUNT``."""
        creds = self._get_credentials(credentials=credentials)
        billing_service = cloud_billing(creds)

        resource_name = f"projects/{self.id}"
        billing_info = billing_service.projects().getBillingInfo(name=resource_name).execute()

        billing_enabled = billing_info.get("billingEnabled", False)
        billing_account_name = billing_info.get("billingAccountName", "")

        if not billing_enabled or not billing_account_name:
            return NO_BILLING_ACCOUNT

        billing_account_id = billing_account_name.split("/")[1]
        billing_account_info = billing_service.billingAccounts().get(name=billing_account_name).execute()

        display_name = billing_account_info.get("displayName", billing_account_id)
        is_open = billing_account_info.get("open", False)
        status = "OPEN" if is_open else "CLOSED"

        return BillingAccount(id=billing_account_id, display_name=display_name, status=status)

    def ensure_apis(
        self,
        apis: Iterable[str],
        *,
        credentials: Optional[Credentials] = None,
        timeout: float = 300.0,
        verbose: bool = True,
        polling_interval: float = 5.0,
    ) -> dict:
        """Ensure the given APIs are enabled for this project."""
        creds = self._get_credentials(credentials=credentials)
        current = set(self.enabled_apis(credentials=creds))
        required = set(apis)
        to_enable = sorted(required - current)

        if not to_enable:
            return {"done": True, "result": "no-op", "enabled": sorted(current)}

        return self.enable_apis(
            to_enable,
            credentials=creds,
            timeout=timeout,
            verbose=verbose,
            polling_interval=polling_interval,
        )

    def bootstrap_quota_project(
        self,
        *,
        credentials: Optional[Credentials] = None,
        timeout: float = 300.0,
        verbose: bool = True,
        polling_interval: float = 5.0,
    ) -> dict:
        """Enable the required APIs for using this project as a quota project."""

        return self.ensure_apis(
            _REQUIRED_APIS,
            credentials=credentials,
            timeout=timeout,
            verbose=verbose,
            polling_interval=polling_interval,
        )

    def update_billing_account(
        self,
        billing_account: BillingAccount | str | None,
        *,
        credentials: Optional[Credentials] = None,
    ) -> dict:
        """Update this project's billing account."""
        creds = self._get_credentials(credentials=credentials)
        billing = cloud_billing(creds)

        if billing_account is None:
            ba_name = ""
        elif isinstance(billing_account, BillingAccount):
            ba_name = f"billingAccounts/{billing_account.id}"
        elif isinstance(billing_account, str):
            ba_name = f"billingAccounts/{billing_account}"
        else:
            ba_name = ""

        body = {"billingAccountName": ba_name}
        return billing.projects().updateBillingInfo(name=f"projects/{self.id}", body=body).execute()

    @classmethod
    def update_billing_account_for_id(
        cls,
        project_id: str,
        billing_account: BillingAccount | str | None,
        *,
        credentials: Optional[Credentials] = None,
    ) -> dict:
        """Class-level variant to update billing using a project id."""
        temp = cls(
            id=project_id,
            name="",
            project_number="",
            lifecycle_state="",
            parent=cls._dummy_parent(),
            _credentials=credentials,
        )
        return temp.update_billing_account(billing_account, credentials=credentials)

    @classmethod
    def lookup(cls, project_id: str, *, credentials: Optional[Credentials] = None) -> "Project":
        """Return a Project by id using CRM v3 and resolve its parent."""
        from .folder import Folder
        from .no_org import NO_ORG
        from .organization import Organization

        if credentials is None:
            credentials, _ = google.auth.default()

        crm_service = crm_v3(credentials)

        request = crm_service.projects().search(query=f"id:{project_id}")
        response = request.execute()

        projects = response.get("projects", [])
        if not projects:
            raise FileNotFoundError(f"Project with ID '{project_id}' not found.")
        if len(projects) > 1:
            raise ValueError(f"Found multiple projects with ID '{project_id}'.")

        project_resource = projects[0]
        parent_resource_name = project_resource.get("parent")

        if parent_resource_name and parent_resource_name.startswith("organizations/"):
            org_id = parent_resource_name.split("/")[1]
            parent: Container = Organization.lookup(org_id, credentials=credentials)
        elif parent_resource_name and parent_resource_name.startswith("folders/"):
            folder_resource = crm_service.folders().get(name=parent_resource_name).execute()
            parent = Folder(
                id=folder_resource["name"].split("/")[1],
                resource_name=folder_resource["name"],
                display_name=folder_resource.get("displayName", ""),
                parent_resource_name=folder_resource.get("parent", ""),
                _credentials=credentials,
            )
        else:
            parent = NO_ORG

        return cls(
            id=project_resource["projectId"],
            name=project_resource.get("displayName", ""),
            project_number=str(project_resource.get("projectNumber", "")),
            lifecycle_state=project_resource.get("state", ""),
            parent=parent,
            _credentials=credentials,
        )

    @classmethod
    def suggest_name(cls, *, prefix: Optional[str] = None, random_digits: int = 5) -> str:
        """Suggest a valid GCP project id using an optional prefix."""
        import random

        if not 0 <= random_digits <= 10:
            raise ValueError("random_digits must be between 0 and 10")

        if prefix is None:
            prefix = coolname.generate_slug(2)
        else:
            if not prefix or not prefix[0].islower() or not prefix[0].isalpha():
                raise ValueError("prefix must start with a lowercase letter")

        if random_digits > 0:
            digits = "".join(str(random.randint(0, 9)) for _ in range(random_digits))
            name = f"{prefix}-{digits}"
        else:
            name = prefix

        if len(name) < 6 or len(name) > 30:
            raise ValueError(
                f"Generated name '{name}' is {len(name)} characters, but GCP project IDs must be 6-30 characters long"
            )

        return name

    def list_roles(
        self,
        *,
        credentials: Optional[Credentials] = None,
        user_email: str | None = None,
    ) -> list["Role"]:
        """List IAM roles for a user on this project."""
        creds = self._get_credentials(credentials=credentials)
        from pdum.gcp._helpers import _list_roles

        return _list_roles(credentials=creds, resource_name=f"projects/{self.id}", user_email=user_email)

    def give_user_role(
        self,
        role: str,
        user_email: str,
        *,
        credentials: Optional[Credentials] = None,
        verbose: bool = True,
    ) -> dict:
        """Grant an IAM role to a user on this project."""

        if "@" not in user_email or not user_email.strip():
            raise ValueError("user_email must be a valid email address")

        member = f"user:{user_email.strip()}"
        creds = self._get_credentials(credentials=credentials)
        crm = crm_v3(creds)
        resource = f"projects/{self.id}"

        policy = (
            crm.projects().getIamPolicy(resource=resource, body={"options": {"requestedPolicyVersion": 3}}).execute()
        )

        if policy.get("version", 0) < 3:
            policy["version"] = 3

        bindings = policy.setdefault("bindings", [])
        binding = next((b for b in bindings if b.get("role") == role), None)

        if binding is None:
            if verbose:
                print(f"Role '{role}' not present; creating binding and adding {member}.")
            bindings.append({"role": role, "members": [member]})
        else:
            members = binding.setdefault("members", [])
            if member in members:
                if verbose:
                    print(f"{member} already has role '{role}'; no changes made.")
                return policy
            if verbose:
                print(f"Adding {member} to existing role '{role}' binding.")
            members.append(member)

        if verbose:
            print(f"Updating IAM policy for role '{role}'.")
        updated = crm.projects().setIamPolicy(resource=resource, body={"policy": policy}).execute()
        return updated

    def add_user_as_owner(self, user_email: str, *, credentials: Optional[Credentials] = None) -> list[dict]:
        """Add a user to the project's Owners (roles/owner) binding."""
        res = []
        res.append(self.give_user_role("roles/owner", user_email, credentials=credentials))
        res.append(self.give_user_role("roles/datastore.owner", user_email, credentials=credentials))
        return res

    def create_firestore_db(
        self,
        database_id: str = "(default)",
        *,
        region: Union[Region, MultiRegion],
        credentials: Optional[Credentials] = None,
        concurrency_mode=gfa_database.Database.ConcurrencyMode.OPTIMISTIC,
        edition=gfa_database.Database.DatabaseEdition.STANDARD,
    ) -> google.api_core.operation.Operation:
        """Create a Firestore Native database for this project.

        Parameters
        ----------
        database_id : str, default "(default)"
            Identifier for the database. Use the special ``"(default)"`` value for the
            primary Firestore database or supply a 4–63 character slug for additional databases.
        region : Region or MultiRegion
            Target region or multi-region to host the database.
        credentials : Credentials, optional
            Explicit credentials to use. When omitted, stored project credentials or ADC are used.
        concurrency_mode : Database.ConcurrencyMode, optional
            Optional concurrency mode override. Defaults to ``OPTIMISTIC`` per product guidance.
        edition : Database.DatabaseEdition, optional
            Firestore edition to provision. Defaults to ``STANDARD``.

        Returns
        -------
        google.api_core.operation.Operation
            Long-running operation representing the create database request.

        Raises
        ------
        TypeError
            If ``region`` is not a member of ``Region`` or ``MultiRegion``.
        """
        creds = self._get_credentials(credentials=credentials)
        client = firestore_admin(creds)

        if isinstance(region, Region):
            location_id = region.region_id
        elif isinstance(region, MultiRegion):
            location_id = region.multi_region_id
        else:  # pragma: no cover - defensive
            raise TypeError("region must be an instance of Region or MultiRegion")

        project_resource = self.full_resource_name()

        new_db_object = gfa_database.Database(
            type_=gfa_database.Database.DatabaseType.FIRESTORE_NATIVE,
            location_id=location_id,
            concurrency_mode=concurrency_mode,
            database_edition=edition,
        )

        self.ensure_apis(["firestore.googleapis.com"], credentials=creds)
        create_db_request = CreateDatabaseRequest(
            parent=project_resource, database=new_db_object, database_id=database_id
        )
        operation = client.create_database(create_db_request)

        return operation

    @staticmethod
    def _dummy_parent() -> "Container":
        """Return a lightweight placeholder container for helper construction."""
        from .no_org import NO_ORG

        return NO_ORG

add_user_as_owner(user_email, *, credentials=None)

Add a user to the project's Owners (roles/owner) binding.

Source code in src/pdum/gcp/types/project.py

def add_user_as_owner(self, user_email: str, *, credentials: Optional[Credentials] = None) -> list[dict]:
    """Add a user to the project's Owners (roles/owner) binding."""
    res = []
    res.append(self.give_user_role("roles/owner", user_email, credentials=credentials))
    res.append(self.give_user_role("roles/datastore.owner", user_email, credentials=credentials))
    return res

billing_account(*, credentials=None)

Return the project's billing account or NO_BILLING_ACCOUNT.

Source code in src/pdum/gcp/types/project.py

def billing_account(self, *, credentials: Optional[Credentials] = None) -> BillingAccount:
    """Return the project's billing account or ``NO_BILLING_ACCOUNT``."""
    creds = self._get_credentials(credentials=credentials)
    billing_service = cloud_billing(creds)

    resource_name = f"projects/{self.id}"
    billing_info = billing_service.projects().getBillingInfo(name=resource_name).execute()

    billing_enabled = billing_info.get("billingEnabled", False)
    billing_account_name = billing_info.get("billingAccountName", "")

    if not billing_enabled or not billing_account_name:
        return NO_BILLING_ACCOUNT

    billing_account_id = billing_account_name.split("/")[1]
    billing_account_info = billing_service.billingAccounts().get(name=billing_account_name).execute()

    display_name = billing_account_info.get("displayName", billing_account_id)
    is_open = billing_account_info.get("open", False)
    status = "OPEN" if is_open else "CLOSED"

    return BillingAccount(id=billing_account_id, display_name=display_name, status=status)

bootstrap_quota_project(*, credentials=None, timeout=300.0, verbose=True, polling_interval=5.0)

Enable the required APIs for using this project as a quota project.

Source code in src/pdum/gcp/types/project.py

def bootstrap_quota_project(
    self,
    *,
    credentials: Optional[Credentials] = None,
    timeout: float = 300.0,
    verbose: bool = True,
    polling_interval: float = 5.0,
) -> dict:
    """Enable the required APIs for using this project as a quota project."""

    return self.ensure_apis(
        _REQUIRED_APIS,
        credentials=credentials,
        timeout=timeout,
        verbose=verbose,
        polling_interval=polling_interval,
    )

create_firestore_db(database_id='(default)', *, region, credentials=None, concurrency_mode=gfa_database.Database.ConcurrencyMode.OPTIMISTIC, edition=gfa_database.Database.DatabaseEdition.STANDARD)

Create a Firestore Native database for this project.

Parameters:

Name	Type	Description	Default
database_id	str	Identifier for the database. Use the special "(default)" value for the primary Firestore database or supply a 4–63 character slug for additional databases.	"(default)"
region	Region or MultiRegion	Target region or multi-region to host the database.	required
credentials	Credentials	Explicit credentials to use. When omitted, stored project credentials or ADC are used.	None
concurrency_mode	ConcurrencyMode	Optional concurrency mode override. Defaults to OPTIMISTIC per product guidance.	OPTIMISTIC
edition	DatabaseEdition	Firestore edition to provision. Defaults to STANDARD.	STANDARD

Returns:

Type	Description
Operation	Long-running operation representing the create database request.

Raises:

Type	Description
TypeError	If region is not a member of Region or MultiRegion.

Source code in src/pdum/gcp/types/project.py

def create_firestore_db(
    self,
    database_id: str = "(default)",
    *,
    region: Union[Region, MultiRegion],
    credentials: Optional[Credentials] = None,
    concurrency_mode=gfa_database.Database.ConcurrencyMode.OPTIMISTIC,
    edition=gfa_database.Database.DatabaseEdition.STANDARD,
) -> google.api_core.operation.Operation:
    """Create a Firestore Native database for this project.

    Parameters
    ----------
    database_id : str, default "(default)"
        Identifier for the database. Use the special ``"(default)"`` value for the
        primary Firestore database or supply a 4–63 character slug for additional databases.
    region : Region or MultiRegion
        Target region or multi-region to host the database.
    credentials : Credentials, optional
        Explicit credentials to use. When omitted, stored project credentials or ADC are used.
    concurrency_mode : Database.ConcurrencyMode, optional
        Optional concurrency mode override. Defaults to ``OPTIMISTIC`` per product guidance.
    edition : Database.DatabaseEdition, optional
        Firestore edition to provision. Defaults to ``STANDARD``.

    Returns
    -------
    google.api_core.operation.Operation
        Long-running operation representing the create database request.

    Raises
    ------
    TypeError
        If ``region`` is not a member of ``Region`` or ``MultiRegion``.
    """
    creds = self._get_credentials(credentials=credentials)
    client = firestore_admin(creds)

    if isinstance(region, Region):
        location_id = region.region_id
    elif isinstance(region, MultiRegion):
        location_id = region.multi_region_id
    else:  # pragma: no cover - defensive
        raise TypeError("region must be an instance of Region or MultiRegion")

    project_resource = self.full_resource_name()

    new_db_object = gfa_database.Database(
        type_=gfa_database.Database.DatabaseType.FIRESTORE_NATIVE,
        location_id=location_id,
        concurrency_mode=concurrency_mode,
        database_edition=edition,
    )

    self.ensure_apis(["firestore.googleapis.com"], credentials=creds)
    create_db_request = CreateDatabaseRequest(
        parent=project_resource, database=new_db_object, database_id=database_id
    )
    operation = client.create_database(create_db_request)

    return operation

enable_apis(api_list, *, credentials=None, timeout=300.0, verbose=True, polling_interval=5.0)

Enable multiple APIs for this project using batch enable.

Source code in src/pdum/gcp/types/project.py

def enable_apis(
    self,
    api_list: list[str],
    *,
    credentials: Optional[Credentials] = None,
    timeout: float = 300.0,
    verbose: bool = True,
    polling_interval: float = 5.0,
) -> dict:
    """Enable multiple APIs for this project using batch enable."""
    import time

    creds = self._get_credentials(credentials=credentials)
    service_usage_client = service_usage(creds)

    parent_name = f"projects/{self.id}"
    request_body = {"serviceIds": api_list}

    operation = service_usage_client.services().batchEnable(parent=parent_name, body=request_body).execute()

    operation_name = operation.get("name")
    if verbose:
        print(f"Enabling {len(api_list)} APIs for project {self.id}... ", end="", flush=True)

    start_time = time.time()

    while True:
        elapsed = time.time() - start_time
        if elapsed > timeout:
            if verbose:
                print()
            raise TimeoutError(f"Operation timed out after {timeout} seconds. Operation name: {operation_name}")

        operation = service_usage_client.operations().get(name=operation_name).execute()

        if operation.get("done", False):
            if verbose:
                print()

            if "error" in operation:
                error = operation["error"]
                error_message = error.get("message", "Unknown error")
                error_code = error.get("code", "Unknown")
                raise RuntimeError(f"Operation failed with error code {error_code}: {error_message}")

            return operation

        if verbose:
            print(".", end="", flush=True)

        time.sleep(polling_interval)

enabled_apis(*, credentials=None)

List enabled APIs for this project.

Source code in src/pdum/gcp/types/project.py

def enabled_apis(self, *, credentials: Optional[Credentials] = None) -> list[str]:
    """List enabled APIs for this project."""
    creds = self._get_credentials(credentials=credentials)
    service_usage_client = service_usage(creds)

    enabled_apis: list[str] = []
    parent_name = f"projects/{self.id}"
    request = service_usage_client.services().list(parent=parent_name, filter="state:ENABLED")

    while request is not None:
        response = request.execute()

        for service in response.get("services", []):
            api_name = service.get("config", {}).get("name", "")
            if api_name:
                enabled_apis.append(api_name)

        request = service_usage_client.services().list_next(previous_request=request, previous_response=response)

    return enabled_apis

ensure_apis(apis, *, credentials=None, timeout=300.0, verbose=True, polling_interval=5.0)

Ensure the given APIs are enabled for this project.

Source code in src/pdum/gcp/types/project.py

def ensure_apis(
    self,
    apis: Iterable[str],
    *,
    credentials: Optional[Credentials] = None,
    timeout: float = 300.0,
    verbose: bool = True,
    polling_interval: float = 5.0,
) -> dict:
    """Ensure the given APIs are enabled for this project."""
    creds = self._get_credentials(credentials=credentials)
    current = set(self.enabled_apis(credentials=creds))
    required = set(apis)
    to_enable = sorted(required - current)

    if not to_enable:
        return {"done": True, "result": "no-op", "enabled": sorted(current)}

    return self.enable_apis(
        to_enable,
        credentials=creds,
        timeout=timeout,
        verbose=verbose,
        polling_interval=polling_interval,
    )

give_user_role(role, user_email, *, credentials=None, verbose=True)

Grant an IAM role to a user on this project.

Source code in src/pdum/gcp/types/project.py

def give_user_role(
    self,
    role: str,
    user_email: str,
    *,
    credentials: Optional[Credentials] = None,
    verbose: bool = True,
) -> dict:
    """Grant an IAM role to a user on this project."""

    if "@" not in user_email or not user_email.strip():
        raise ValueError("user_email must be a valid email address")

    member = f"user:{user_email.strip()}"
    creds = self._get_credentials(credentials=credentials)
    crm = crm_v3(creds)
    resource = f"projects/{self.id}"

    policy = (
        crm.projects().getIamPolicy(resource=resource, body={"options": {"requestedPolicyVersion": 3}}).execute()
    )

    if policy.get("version", 0) < 3:
        policy["version"] = 3

    bindings = policy.setdefault("bindings", [])
    binding = next((b for b in bindings if b.get("role") == role), None)

    if binding is None:
        if verbose:
            print(f"Role '{role}' not present; creating binding and adding {member}.")
        bindings.append({"role": role, "members": [member]})
    else:
        members = binding.setdefault("members", [])
        if member in members:
            if verbose:
                print(f"{member} already has role '{role}'; no changes made.")
            return policy
        if verbose:
            print(f"Adding {member} to existing role '{role}' binding.")
        members.append(member)

    if verbose:
        print(f"Updating IAM policy for role '{role}'.")
    updated = crm.projects().setIamPolicy(resource=resource, body={"policy": policy}).execute()
    return updated

list_roles(*, credentials=None, user_email=None)

List IAM roles for a user on this project.

Source code in src/pdum/gcp/types/project.py

def list_roles(
    self,
    *,
    credentials: Optional[Credentials] = None,
    user_email: str | None = None,
) -> list["Role"]:
    """List IAM roles for a user on this project."""
    creds = self._get_credentials(credentials=credentials)
    from pdum.gcp._helpers import _list_roles

    return _list_roles(credentials=creds, resource_name=f"projects/{self.id}", user_email=user_email)

lookup(project_id, *, credentials=None) classmethod

Return a Project by id using CRM v3 and resolve its parent.

Source code in src/pdum/gcp/types/project.py

@classmethod
def lookup(cls, project_id: str, *, credentials: Optional[Credentials] = None) -> "Project":
    """Return a Project by id using CRM v3 and resolve its parent."""
    from .folder import Folder
    from .no_org import NO_ORG
    from .organization import Organization

    if credentials is None:
        credentials, _ = google.auth.default()

    crm_service = crm_v3(credentials)

    request = crm_service.projects().search(query=f"id:{project_id}")
    response = request.execute()

    projects = response.get("projects", [])
    if not projects:
        raise FileNotFoundError(f"Project with ID '{project_id}' not found.")
    if len(projects) > 1:
        raise ValueError(f"Found multiple projects with ID '{project_id}'.")

    project_resource = projects[0]
    parent_resource_name = project_resource.get("parent")

    if parent_resource_name and parent_resource_name.startswith("organizations/"):
        org_id = parent_resource_name.split("/")[1]
        parent: Container = Organization.lookup(org_id, credentials=credentials)
    elif parent_resource_name and parent_resource_name.startswith("folders/"):
        folder_resource = crm_service.folders().get(name=parent_resource_name).execute()
        parent = Folder(
            id=folder_resource["name"].split("/")[1],
            resource_name=folder_resource["name"],
            display_name=folder_resource.get("displayName", ""),
            parent_resource_name=folder_resource.get("parent", ""),
            _credentials=credentials,
        )
    else:
        parent = NO_ORG

    return cls(
        id=project_resource["projectId"],
        name=project_resource.get("displayName", ""),
        project_number=str(project_resource.get("projectNumber", "")),
        lifecycle_state=project_resource.get("state", ""),
        parent=parent,
        _credentials=credentials,
    )

suggest_name(*, prefix=None, random_digits=5) classmethod

Suggest a valid GCP project id using an optional prefix.

Source code in src/pdum/gcp/types/project.py

@classmethod
def suggest_name(cls, *, prefix: Optional[str] = None, random_digits: int = 5) -> str:
    """Suggest a valid GCP project id using an optional prefix."""
    import random

    if not 0 <= random_digits <= 10:
        raise ValueError("random_digits must be between 0 and 10")

    if prefix is None:
        prefix = coolname.generate_slug(2)
    else:
        if not prefix or not prefix[0].islower() or not prefix[0].isalpha():
            raise ValueError("prefix must start with a lowercase letter")

    if random_digits > 0:
        digits = "".join(str(random.randint(0, 9)) for _ in range(random_digits))
        name = f"{prefix}-{digits}"
    else:
        name = prefix

    if len(name) < 6 or len(name) > 30:
        raise ValueError(
            f"Generated name '{name}' is {len(name)} characters, but GCP project IDs must be 6-30 characters long"
        )

    return name

update_billing_account(billing_account, *, credentials=None)

Update this project's billing account.

Source code in src/pdum/gcp/types/project.py

def update_billing_account(
    self,
    billing_account: BillingAccount | str | None,
    *,
    credentials: Optional[Credentials] = None,
) -> dict:
    """Update this project's billing account."""
    creds = self._get_credentials(credentials=credentials)
    billing = cloud_billing(creds)

    if billing_account is None:
        ba_name = ""
    elif isinstance(billing_account, BillingAccount):
        ba_name = f"billingAccounts/{billing_account.id}"
    elif isinstance(billing_account, str):
        ba_name = f"billingAccounts/{billing_account}"
    else:
        ba_name = ""

    body = {"billingAccountName": ba_name}
    return billing.projects().updateBillingInfo(name=f"projects/{self.id}", body=body).execute()

update_billing_account_for_id(project_id, billing_account, *, credentials=None) classmethod

Class-level variant to update billing using a project id.

Source code in src/pdum/gcp/types/project.py

@classmethod
def update_billing_account_for_id(
    cls,
    project_id: str,
    billing_account: BillingAccount | str | None,
    *,
    credentials: Optional[Credentials] = None,
) -> dict:
    """Class-level variant to update billing using a project id."""
    temp = cls(
        id=project_id,
        name="",
        project_number="",
        lifecycle_state="",
        parent=cls._dummy_parent(),
        _credentials=credentials,
    )
    return temp.update_billing_account(billing_account, credentials=credentials)

Region

Bases: Enum

Canonical set of Google Cloud regions.

Source code in src/pdum/gcp/types/region.py

class Region(Enum):
    """Canonical set of Google Cloud regions."""

    US_WEST1 = ("North America", "us-west1", "Oregon")
    US_WEST2 = ("North America", "us-west2", "Los Angeles")
    US_WEST3 = ("North America", "us-west3", "Salt Lake City")
    US_WEST4 = ("North America", "us-west4", "Las Vegas")
    US_CENTRAL1 = ("North America", "us-central1", "Iowa")
    US_CENTRAL2 = ("North America", "us-central2", "Oklahoma—private Google Cloud region")
    NORTHAMERICA_NORTHEAST1 = ("North America", "northamerica-northeast1", "Montréal")
    NORTHAMERICA_NORTHEAST2 = ("North America", "northamerica-northeast2", "Toronto")
    NORTHAMERICA_SOUTH1 = ("North America", "northamerica-south1", "Queretaro")
    US_EAST1 = ("North America", "us-east1", "South Carolina")
    US_EAST4 = ("North America", "us-east4", "Northern Virginia")
    US_EAST5 = ("North America", "us-east5", "Columbus")
    US_SOUTH1 = ("North America", "us-south1", "Dallas")
    SOUTHAMERICA_WEST1 = ("South America", "southamerica-west1", "Santiago")
    SOUTHAMERICA_EAST1 = ("South America", "southamerica-east1", "São Paulo")
    EUROPE_WEST2 = ("Europe", "europe-west2", "London")
    EUROPE_WEST1 = ("Europe", "europe-west1", "Belgium")
    EUROPE_WEST4 = ("Europe", "europe-west4", "Netherlands")
    EUROPE_WEST8 = ("Europe", "europe-west8", "Milan")
    EUROPE_SOUTHWEST1 = ("Europe", "europe-southwest1", "Madrid")
    EUROPE_WEST9 = ("Europe", "europe-west9", "Paris")
    EUROPE_WEST12 = ("Europe", "europe-west12", "Turin")
    EUROPE_WEST10 = ("Europe", "europe-west10", "Berlin")
    EUROPE_WEST3 = ("Europe", "europe-west3", "Frankfurt")
    EUROPE_NORTH1 = ("Europe", "europe-north1", "Finland")
    EUROPE_NORTH2 = ("Europe", "europe-north2", "Stockholm")
    EUROPE_CENTRAL2 = ("Europe", "europe-central2", "Warsaw")
    EUROPE_WEST6 = ("Europe", "europe-west6", "Zürich")
    ME_CENTRAL1 = ("Middle East", "me-central1", "Doha")
    ME_CENTRAL2 = ("Middle East", "me-central2", "Dammam")
    ME_WEST1 = ("Middle East", "me-west1", "Tel Aviv")
    ASIA_SOUTH1 = ("Asia", "asia-south1", "Mumbai")
    ASIA_SOUTH2 = ("Asia", "asia-south2", "Delhi")
    ASIA_SOUTHEAST1 = ("Asia", "asia-southeast1", "Singapore")
    ASIA_SOUTHEAST2 = ("Asia", "asia-southeast2", "Jakarta")
    ASIA_EAST2 = ("Asia", "asia-east2", "Hong Kong")
    ASIA_EAST1 = ("Asia", "asia-east1", "Taiwan")
    ASIA_NORTHEAST1 = ("Asia", "asia-northeast1", "Tokyo")
    ASIA_NORTHEAST2 = ("Asia", "asia-northeast2", "Osaka")
    ASIA_NORTHEAST3 = ("Asia", "asia-northeast3", "Seoul")
    AUSTRALIA_SOUTHEAST1 = ("Australia", "australia-southeast1", "Sydney")
    AUSTRALIA_SOUTHEAST2 = ("Australia", "australia-southeast2", "Melbourne")
    AFRICA_SOUTH1 = ("Africa", "africa-south1", "Johannesburg")

    def __init__(self, continent: str, region_id: str, description: str) -> None:
        self._continent = continent
        self._region_id = region_id
        self._description = description

    @property
    def continent(self) -> str:
        """Continent grouping for the region."""

        return self._continent

    @property
    def region_id(self) -> str:
        """Region identifier (e.g., ``us-west1``)."""

        return self._region_id

    @property
    def description(self) -> str:
        """Human-readable description for the region."""

        return self._description

    @classmethod
    def from_region_id(cls, region_id: str) -> "Region":
        """Return the enum entry matching ``region_id``."""

        normalized = region_id.lower()
        for region in cls:
            if region.region_id == normalized:
                return region
        raise ValueError(f"Unknown region id: {region_id!r}")

continent property

Continent grouping for the region.

description property

Human-readable description for the region.

region_id property

Region identifier (e.g., us-west1).

from_region_id(region_id) classmethod

Return the enum entry matching region_id.

Source code in src/pdum/gcp/types/region.py

@classmethod
def from_region_id(cls, region_id: str) -> "Region":
    """Return the enum entry matching ``region_id``."""

    normalized = region_id.lower()
    for region in cls:
        if region.region_id == normalized:
            return region
    raise ValueError(f"Unknown region id: {region_id!r}")

doctor(*, credentials=None, console=None)

Run environment diagnostics for pdum.gcp.

This prints a human-friendly report: 1) Identity and quota project status (ADC). 2) Enabled APIs on the quota project vs. required APIs used by this package. 3) Organization-level role coverage for the current identity vs. a standard high-privilege set used by Organization.add_user_as_owner.

Notes

• Read-only: does not mutate any resources.
• Role detection is based on direct user bindings; group-based grants may not appear. Use IAM policy viewers to validate group grants.

Source code in src/pdum/gcp/admin.py

def doctor(*, credentials: Optional[Credentials] = None, console: Optional[Console] = None) -> None:
    """Run environment diagnostics for pdum.gcp.

    This prints a human-friendly report:
    1) Identity and quota project status (ADC).
    2) Enabled APIs on the quota project vs. required APIs used by this package.
    3) Organization-level role coverage for the current identity vs. a standard
       high-privilege set used by ``Organization.add_user_as_owner``.

    Notes
    -----
    - Read-only: does not mutate any resources.
    - Role detection is based on direct user bindings; group-based grants may
      not appear. Use IAM policy viewers to validate group grants.
    """
    c = console or Console()

    # Identity
    try:
        email = get_email(credentials=credentials)
    except Exception as e:
        c.print(Panel.fit(f"[red]Failed to resolve identity from ADC:[/red] {e}", title="Identity"))
        return

    c.print(Panel.fit(f"[bold]Active identity:[/bold] {email}", title="Identity", border_style="cyan"))

    # Quota project
    try:
        qp = quota_project(credentials=credentials)
    except Exception:
        msg = (
            "Could not determine a quota project from ADC.\n\n"
            "Set a quota project for Application Default Credentials:\n\n"
            "    gcloud auth application-default set-quota-project <PROJECT_ID>\n\n"
            "After setting, re-run doctor()."
        )
        c.print(Panel(msg, title="Quota Project", border_style="red"))
        return

    c.print(Panel.fit(f"[bold]Quota project:[/bold] {qp.id}", title="Quota Project", border_style="green"))

    # Enabled APIs vs requirements
    try:
        enabled = set(qp.enabled_apis(credentials=credentials))
    except Exception as e:
        c.print(Panel.fit(f"[red]Failed to list enabled APIs:[/red] {e}", title="APIs"))
        enabled = set()

    required = set(_REQUIRED_APIS)
    missing = sorted(required - enabled)
    present = sorted(required & enabled)

    apit = Table(title="APIs", show_lines=False)
    apit.add_column("Status", style="bold")
    apit.add_column("Service")
    for svc in present:
        apit.add_row("[green]OK[/green]", svc)
    for svc in missing:
        apit.add_row("[red]Missing[/red]", svc)
    c.print(apit)

    if missing:
        c.print(
            Panel(
                "You can enable missing APIs on the quota project (requires permissions):\n"
                + "\n".join(f"gcloud services enable {svc} --project {qp.id}" for svc in missing),
                title="Enable Missing APIs",
                border_style="yellow",
            )
        )

    # Organization roles
    try:
        orgs = [o for o in list_organizations(credentials=credentials) if isinstance(o, Organization)]
    except Exception as e:
        c.print(Panel.fit(f"[red]Failed to list organizations:[/red] {e}", title="Organizations"))
        orgs = []

    if not orgs:
        c.print(Panel.fit("No organizations visible to this identity.", title="Organizations"))
        return

    # Standard role set used by add_user_as_owner
    required_roles = sorted(Organization.ORGANIZATION_OWNER_ROLES)

    for org in orgs:
        try:
            roles = list_roles(org, user_email=email, credentials=credentials)
            have = sorted(r.name for r in roles)
        except Exception as e:
            c.print(Panel.fit(f"[red]Failed to list roles on {org.resource_name}:[/red] {e}", title=org.display_name))
            continue

        have_set = set(have)
        req_set = set(required_roles)
        missing_roles = sorted(req_set - have_set)
        present_roles = sorted(req_set & have_set)

        t = Table(title=f"Org: {org.display_name} ({org.id})", show_lines=False)
        t.add_column("Status", style="bold")
        t.add_column("Role")
        for r in present_roles:
            t.add_row("[green]OK[/green]", r)
        for r in missing_roles:
            t.add_row("[red]Missing[/red]", r)
        c.print(t)

        if missing_roles:
            commands = "\n".join(
                f'gcloud organizations add-iam-policy-binding {org.id} --member="user:{email}" --role="{r}"'
                for r in missing_roles
            )
            c.print(
                Panel(
                    "Ask an Organization Admin to run the following commands to add this user as owner-level admin:\n\n"
                    + commands,
                    title="Grant Missing Roles",
                    border_style="yellow",
                )
            )

get_email(*, credentials=None)

Return the email for the provided credentials or ADC.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit Google Cloud credentials to use. If None, attempts to load Application Default Credentials (ADC).	None

Returns:

Type	Description
str	The email address associated with the active identity.

Raises:

Type	Description
DefaultCredentialsError	If no credentials can be found.
AttributeError	If an email cannot be extracted from the credential type.

Examples:

>>> from pdum.gcp.admin import get_email
>>> get_email()
'user@example.com'

Source code in src/pdum/gcp/admin.py

def get_email(*, credentials: Optional[Credentials] = None) -> str:
    """Return the email for the provided credentials or ADC.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit Google Cloud credentials to use. If ``None``, attempts to
        load Application Default Credentials (ADC).

    Returns
    -------
    str
        The email address associated with the active identity.

    Raises
    ------
    google.auth.exceptions.DefaultCredentialsError
        If no credentials can be found.
    AttributeError
        If an email cannot be extracted from the credential type.

    Examples
    --------
    >>> from pdum.gcp.admin import get_email
    >>> get_email()  # doctest: +SKIP
    'user@example.com'
    """
    # Get credentials if not provided
    if credentials is None:
        credentials, project_id = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])

    # Try to get email from various credential types
    email = _extract_email_from_credentials(credentials)

    if email:
        return email

    # If we can't extract email directly, try refreshing and getting from token info
    # This works for user credentials
    if hasattr(credentials, "refresh"):
        request = google.auth.transport.requests.Request()
        credentials.refresh(request)

        # After refresh, try to get email from the token
        if hasattr(credentials, "id_token") and credentials.id_token:
            import base64
            import json

            # Decode the JWT payload (second part of the token)
            parts = credentials.id_token.split(".")
            if len(parts) >= 2:
                # Add padding if needed
                payload_encoded = parts[1]
                padding = 4 - len(payload_encoded) % 4
                if padding != 4:
                    payload_encoded += "=" * padding

                payload = json.loads(base64.urlsafe_b64decode(payload_encoded))
                if "email" in payload:
                    return payload["email"]

    # If all else fails, raise an error
    raise AttributeError(
        f"Could not extract email from credentials of type {type(credentials).__name__}. "
        f"The credentials may not have an associated email address, or the credential type "
        f"is not supported by this function."
    )

get_iam_policy(resource, *, credentials=None)

Return the IAM policy for a Resource.

Parameters:

Name	Type	Description	Default
resource	Resource	Any CRM resource implementing full_resource_name() (Organization, Folder, Project).	required
credentials	Credentials	Explicit credentials to use; if omitted, uses the resource's stored creds or ADC.	None

Returns:

Type	Description
dict	The IAM policy for the resource.

Source code in src/pdum/gcp/admin.py

def get_iam_policy(resource: Resource, *, credentials: Optional[Credentials] = None) -> dict:
    """Return the IAM policy for a Resource.

    Parameters
    ----------
    resource : Resource
        Any CRM resource implementing ``full_resource_name()`` (Organization, Folder, Project).
    credentials : Credentials, optional
        Explicit credentials to use; if omitted, uses the resource's stored creds or ADC.

    Returns
    -------
    dict
        The IAM policy for the resource.
    """
    creds = resource._get_credentials(credentials=credentials)
    return _get_iam_policy_internal(credentials=creds, resource_name=resource.full_resource_name())

list_organizations(*, credentials=None)

List organizations visible to the caller.

Uses Cloud Resource Manager v1 to search all organizations the current identity can see. If projects exist outside an organization, NO_ORG is appended to the result.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. If None, uses ADC.	None

Returns:

Type	Description
list[Organization]	Organizations accessible to the caller, plus NO_ORG if applicable.

Raises:

Type	Description
DefaultCredentialsError	If no credentials can be found.
HttpError	If the API call fails.

Notes

Requires the Cloud Resource Manager API and basic permissions on target organizations (e.g., Viewer).

Source code in src/pdum/gcp/admin.py

def list_organizations(*, credentials: Optional[Credentials] = None) -> list[Organization]:
    """List organizations visible to the caller.

    Uses Cloud Resource Manager v1 to search all organizations the current
    identity can see. If projects exist outside an organization, ``NO_ORG`` is
    appended to the result.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. If ``None``, uses ADC.

    Returns
    -------
    list[Organization]
        Organizations accessible to the caller, plus ``NO_ORG`` if applicable.

    Raises
    ------
    google.auth.exceptions.DefaultCredentialsError
        If no credentials can be found.
    googleapiclient.errors.HttpError
        If the API call fails.

    Notes
    -----
    Requires the Cloud Resource Manager API and basic permissions on target
    organizations (e.g., Viewer).
    """
    # Import here to avoid circular dependency
    from pdum.gcp.types import NO_ORG

    # Get credentials if not provided
    if credentials is None:
        credentials, _ = google.auth.default()

    # Build the Resource Manager V1 service client
    # The V1 API is required for listing organizations visible to the user
    crm_service = crm_v1(credentials)

    organizations = []

    # Call the organizations.search() method
    # This method uses the current identity to search for Organizations
    # for which the user has at least basic permissions
    # Empty body means "search all organizations visible to me"
    request = crm_service.organizations().search(body={})

    # Handle pagination
    while request is not None:
        response = request.execute()
        for org in response.get("organizations", []):
            # The 'name' field is in the format 'organizations/ORG_ID'
            org_id = org["name"].split("/")[1]
            organizations.append(
                Organization(
                    id=org_id,
                    resource_name=org["name"],
                    display_name=org.get("displayName", ""),
                    _credentials=credentials,
                )
            )

        request = crm_service.organizations().search_next(previous_request=request, previous_response=response)

    # Check if there are any projects without an organization parent
    # If so, add NO_ORG to the list
    no_org_projects = NO_ORG.projects(credentials=credentials)
    if no_org_projects:
        organizations.append(NO_ORG)

    return organizations

list_roles(resource, *, user_email=None, credentials=None)

List IAM roles for a user on a Resource.

Parameters:

Name	Type	Description	Default
resource	Resource	Any CRM resource implementing full_resource_name() (Organization, Folder, Project).	required
user_email	str	If provided, list roles for this email. If omitted, uses the email derived from the provided ADC/credentials.	None
credentials	Credentials	Explicit credentials to use; if omitted, uses the resource's stored creds or ADC.	None

Returns:

Type	Description
list[Role]	Roles that directly bind the user on the resource.

Source code in src/pdum/gcp/admin.py

def list_roles(
    resource: Resource,
    *,
    user_email: Optional[str] = None,
    credentials: Optional[Credentials] = None,
) -> list[Role]:
    """List IAM roles for a user on a Resource.

    Parameters
    ----------
    resource : Resource
        Any CRM resource implementing ``full_resource_name()`` (Organization, Folder, Project).
    user_email : str, optional
        If provided, list roles for this email. If omitted, uses the email
        derived from the provided ADC/credentials.
    credentials : Credentials, optional
        Explicit credentials to use; if omitted, uses the resource's stored creds or ADC.

    Returns
    -------
    list[Role]
        Roles that directly bind the user on the resource.
    """
    creds = resource._get_credentials(credentials=credentials)
    return _list_roles_internal(credentials=creds, resource_name=resource.full_resource_name(), user_email=user_email)

lookup_api(display_name)

Resolve a human-friendly API name to its service id.

Uses normalization, substring checks (for short queries), and fuzzy matching against the bundled API map.

Parameters:

Name	Type	Description	Default
display_name	str	Human-readable API name, e.g. "Compute Engine".	required

Returns:

Type	Description
str	Service id such as "compute.googleapis.com".

Raises:

Type	Description
APIResolutionError	If no match or multiple ambiguous matches are found.
FileNotFoundError	If the API map data file is missing.

Examples:

>>> from pdum.gcp import lookup_api
>>> lookup_api("Compute Engine")
'compute.googleapis.com'
>>> lookup_api("Big Query")
'bigquery.googleapis.com'

Source code in src/pdum/gcp/admin.py

def lookup_api(display_name: str) -> str:
    """Resolve a human-friendly API name to its service id.

    Uses normalization, substring checks (for short queries), and fuzzy matching
    against the bundled API map.

    Parameters
    ----------
    display_name : str
        Human-readable API name, e.g. ``"Compute Engine"``.

    Returns
    -------
    str
        Service id such as ``"compute.googleapis.com"``.

    Raises
    ------
    APIResolutionError
        If no match or multiple ambiguous matches are found.
    FileNotFoundError
        If the API map data file is missing.

    Examples
    --------
    >>> from pdum.gcp import lookup_api
    >>> lookup_api("Compute Engine")  # doctest: +SKIP
    'compute.googleapis.com'
    >>> lookup_api("Big Query")  # doctest: +SKIP
    'bigquery.googleapis.com'
    """
    # Load the API map from the text file
    api_map = _load_api_map()

    # 1. Standardize the input for better matching
    normalized_input = display_name.strip().lower().replace("cloud", "").strip()

    # 2. Check for an exact match or normalized match first
    # We use a case-insensitive check against normalized keys
    normalized_keys = {k.lower().replace("cloud", "").strip(): v for k, v in api_map.items()}

    if display_name in api_map:
        return api_map[display_name]

    if normalized_input in normalized_keys:
        return normalized_keys[normalized_input]

    # 3. Check for substring matches to detect overly generic short terms
    # This catches cases like "Cloud" which appears in many API names
    # Only apply for short terms (< 10 chars) to avoid catching longer specific terms
    if len(display_name) < 10:
        substring_matches = [name for name in api_map.keys() if display_name.lower() in name.lower()]

        if len(substring_matches) > 1:
            # Too many substring matches - term is too generic
            # Show a few examples
            examples = substring_matches[:5]
            raise APIResolutionError(
                f"The term '{display_name}' is too generic and matches multiple APIs. "
                f"Please be more specific. Found matches in: {', '.join(examples)}"
                + (f" and {len(substring_matches) - 5} more..." if len(substring_matches) > 5 else "")
            )

        if len(substring_matches) == 1:
            # Only one substring match, return it
            return api_map[substring_matches[0]]

    # 4. Perform Fuzzy Matching
    # Use the full display names from the map as possibilities for fuzzy matching
    possibilities = list(api_map.keys())

    # difflib.get_close_matches is part of Python's standard library
    # We use a strict cutoff of 0.6 and look for up to 5 matches.
    close_matches = difflib.get_close_matches(word=display_name, possibilities=possibilities, n=5, cutoff=0.6)

    # 5. Handle Results
    if len(close_matches) == 1:
        # If there is only one "good enough" match, we treat it as the intended target.
        return api_map[close_matches[0]]

    if len(close_matches) > 1:
        # Multiple close matches found
        raise APIResolutionError(
            f"Multiple close matches found for '{display_name}'. Please be more specific. "
            f"Did you mean one of these: {', '.join(close_matches)}?"
        )

    # No matches found
    raise APIResolutionError(
        f"No direct match or close fuzzy match found for API '{display_name}'. "
        f"Please check the spelling or try a different name."
    )

quota_project(*, credentials=None)

Return the active quota (billing) project from ADC credentials.

This reads credentials.quota_project_id from Application Default Credentials (ADC) and looks up the corresponding Project. The quota project determines which project is billed for requests made with ADC.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. If None, materializes ADC.	None

Returns:

Type	Description
Project	The resolved quota project.

Raises:

Type	Description
DefaultCredentialsError	If no credentials can be found.
ValueError	If the credentials do not have a quota project set. Use: gcloud auth application-default set-quota-project <PROJECT_ID>.
HttpError	If the project lookup API call fails.

Examples:

>>> from pdum.gcp.admin import quota_project
>>> qp = quota_project()
>>> qp.id
'my-quota-project'

Source code in src/pdum/gcp/admin.py

def quota_project(*, credentials: Optional[Credentials] = None) -> Project:
    """Return the active quota (billing) project from ADC credentials.

    This reads ``credentials.quota_project_id`` from Application Default
    Credentials (ADC) and looks up the corresponding Project. The quota
    project determines which project is billed for requests made with ADC.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. If ``None``, materializes ADC.

    Returns
    -------
    Project
        The resolved quota project.

    Raises
    ------
    google.auth.exceptions.DefaultCredentialsError
        If no credentials can be found.
    ValueError
        If the credentials do not have a quota project set. Use:
        ``gcloud auth application-default set-quota-project <PROJECT_ID>``.
    googleapiclient.errors.HttpError
        If the project lookup API call fails.

    Examples
    --------
    >>> from pdum.gcp.admin import quota_project
    >>> qp = quota_project()  # doctest: +SKIP
    >>> qp.id  # doctest: +SKIP
    'my-quota-project'
    """
    # Materialize credentials
    if credentials is None:
        credentials, _ = google.auth.default()

    # Prefer the quota project from the credentials object
    quota_id = getattr(credentials, "quota_project_id", None)
    if not quota_id:
        raise ValueError(
            "No quota project is configured on ADC credentials.\n\n"
            "Set a quota project for Application Default Credentials:\n\n"
            "    gcloud auth application-default set-quota-project <PROJECT_ID>\n\n"
            "Then re-run this method."
        )

    return Project.lookup(quota_id, credentials=credentials)

walk_projects(*, credentials=None, active_only=True)

Yield all projects across all accessible organizations.

Parameters:

Name	Type	Description	Default
credentials	Credentials	Explicit credentials to use. If None, uses ADC.	None
active_only	bool	If True, yields only ACTIVE projects. If False, yields all lifecycle states.	True

Yields:

Type	Description
Project	Projects from organizations and nested folders.

Raises:

Type	Description
DefaultCredentialsError	If no credentials can be found.
HttpError	If any API call fails.

Notes

Traversal may take time in estates with many organizations/folders.

Source code in src/pdum/gcp/admin.py

def walk_projects(
    *, credentials: Optional[Credentials] = None, active_only: bool = True
) -> Generator[Project, None, None]:
    """Yield all projects across all accessible organizations.

    Parameters
    ----------
    credentials : Credentials, optional
        Explicit credentials to use. If ``None``, uses ADC.
    active_only : bool, default True
        If ``True``, yields only ACTIVE projects. If ``False``, yields all
        lifecycle states.

    Yields
    ------
    Project
        Projects from organizations and nested folders.

    Raises
    ------
    google.auth.exceptions.DefaultCredentialsError
        If no credentials can be found.
    googleapiclient.errors.HttpError
        If any API call fails.

    Notes
    -----
    Traversal may take time in estates with many organizations/folders.
    """
    # Get credentials if not provided
    if credentials is None:
        credentials, _ = google.auth.default()

    # List all accessible organizations
    organizations = list_organizations(credentials=credentials)

    # Walk through each organization and yield all projects
    for org in organizations:
        yield from org.walk_projects(credentials=credentials, active_only=active_only)