vendor_fabric.google.services

Google Cloud services discovery operations.

This module provides operations for discovering resources across Google Cloud services like GKE, Compute Engine, Cloud Storage, Cloud SQL, Pub/Sub, etc.

Module Contents

Classes

GoogleServicesMixin

Mixin providing Google Cloud services discovery operations.

API

class vendor_fabric.google.services.GoogleServicesMixin

Mixin providing Google Cloud services discovery operations.

This mixin requires the base GoogleConnector class to provide:

  • get_compute_service()

  • get_container_service()

  • get_storage_service()

  • get_sqladmin_service()

  • get_pubsub_service()

  • get_serviceusage_service()

  • get_cloudkms_service()

  • logger

list_compute_instances(project_id: str, zone: str | None = None, unhump_instances: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List Compute Engine instances in a project.

Args: project_id: The project ID. zone: Optional zone filter. If not provided, lists all zones. unhump_instances: Convert keys to snake_case. Defaults to False.

Returns: List of instance dictionaries.

list_gke_clusters(project_id: str, location: str = '-', unhump_clusters: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List GKE clusters in a project.

Args: project_id: The project ID. location: Zone or region. Use ‘-’ for all locations. unhump_clusters: Convert keys to snake_case. Defaults to False.

Returns: List of cluster dictionaries.

get_gke_cluster(project_id: str, location: str, cluster_id: str) → extended_data.containers.ExtendedDict | None

Get a specific GKE cluster.

Args: project_id: The project ID. location: Zone or region. cluster_id: The cluster ID.

Returns: Cluster dictionary or None if not found.

list_storage_buckets(project_id: str, unhump_buckets: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List Cloud Storage buckets in a project.

Args: project_id: The project ID. unhump_buckets: Convert keys to snake_case. Defaults to False.

Returns: List of bucket dictionaries.

list_sql_instances(project_id: str, unhump_instances: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List Cloud SQL instances in a project.

Args: project_id: The project ID. unhump_instances: Convert keys to snake_case. Defaults to False.

Returns: List of SQL instance dictionaries.

list_pubsub_topics(project_id: str, unhump_topics: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List Pub/Sub topics in a project.

Args: project_id: The project ID. unhump_topics: Convert keys to snake_case. Defaults to False.

Returns: List of topic dictionaries.

list_pubsub_subscriptions(project_id: str, unhump_subscriptions: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List Pub/Sub subscriptions in a project.

Args: project_id: The project ID. unhump_subscriptions: Convert keys to snake_case. Defaults to False.

Returns: List of subscription dictionaries.

list_enabled_services(project_id: str, unhump_services: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List enabled APIs/services in a project.

Args: project_id: The project ID. unhump_services: Convert keys to snake_case. Defaults to False.

Returns: List of service dictionaries.

enable_service(project_id: str, service_name: str) → extended_data.containers.ExtendedDict

Enable an API/service in a project.

Args: project_id: The project ID. service_name: Service name (e.g., ‘compute.googleapis.com’).

Returns: Operation response dictionary.

disable_service(project_id: str, service_name: str, force: bool = False) → extended_data.containers.ExtendedDict

Disable an API/service in a project.

Args: project_id: The project ID. service_name: Service name (e.g., ‘compute.googleapis.com’). force: Force disable even if dependencies exist.

Returns: Operation response dictionary.

batch_enable_services(project_id: str, service_names: list[str]) → extended_data.containers.ExtendedDict

Enable multiple APIs/services in a project.

Args: project_id: The project ID. service_names: List of service names to enable.

Returns: Operation response dictionary.

list_kms_keyrings(project_id: str, location: str, unhump_keyrings: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List KMS key rings in a project location.

Args: project_id: The project ID. location: The location (e.g., ‘us-central1’, ‘global’). unhump_keyrings: Convert keys to snake_case. Defaults to False.

Returns: List of key ring dictionaries.

create_kms_keyring(project_id: str, location: str, keyring_id: str) → extended_data.containers.ExtendedDict

Create a KMS key ring.

Args: project_id: The project ID. location: The location (e.g., ‘us-central1’, ‘global’). keyring_id: Unique key ring ID.

Returns: Created key ring dictionary.

create_kms_key(project_id: str, location: str, keyring_id: str, key_id: str, purpose: str = 'ENCRYPT_DECRYPT', algorithm: str = 'GOOGLE_SYMMETRIC_ENCRYPTION') → extended_data.containers.ExtendedDict

Create a KMS crypto key.

Args: project_id: The project ID. location: The location. keyring_id: The key ring ID. key_id: Unique key ID. purpose: Key purpose (ENCRYPT_DECRYPT, ASYMMETRIC_SIGN, etc.). algorithm: Key algorithm.

Returns: Created crypto key dictionary.

is_project_empty(project_id: str, check_compute: bool = True, check_gke: bool = True, check_storage: bool = True, check_sql: bool = True, check_pubsub: bool = True) → bool

Check if a project has no resources.

Args: project_id: The project ID. check_compute: Check for Compute Engine instances. check_gke: Check for GKE clusters. check_storage: Check for Cloud Storage buckets. check_sql: Check for Cloud SQL instances. check_pubsub: Check for Pub/Sub topics.

Returns: True if the project has no resources.

get_project_iam_users(project_id: str) → extended_data.containers.ExtendedDict

Get IAM users (members) with access to a project.

Args: project_id: The project ID.

Returns: Dictionary mapping member identifiers to their roles.

get_pubsub_resources_for_project(project_id: str, include_subscriptions: bool = True, unhump_resources: bool = False) → extended_data.containers.ExtendedDict

Get all Pub/Sub topics and subscriptions for a project.

Args: project_id: The project ID. include_subscriptions: Include subscription details. Defaults to True. unhump_resources: Convert keys to snake_case. Defaults to False.

Returns: Dictionary with ‘topics’ and ‘subscriptions’ lists.

find_inactive_projects(projects: collections.abc.MutableMapping[str, collections.abc.MutableMapping[str, Any]] | None = None, check_resources: bool = True, days_since_activity: int = 90) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

Find projects that appear to be inactive or dead.

A project is considered inactive if:

  • Its lifecycle state is not ACTIVE

  • It has no resources and no recent activity timestamp

Args: projects: Pre-fetched projects dict. Fetched if not provided. check_resources: Check if projects have resources. Defaults to True. days_since_activity: Days threshold for available project activity timestamps. Empty projects with recent timestamps are not marked inactive. Empty projects without activity timestamps are treated as inactive.

Returns: List of inactive project dictionaries.