vendor_connectors.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
Section titled “Module Contents”Classes
Section titled “Classes”GoogleServicesMixin | Mixin providing Google Cloud services discovery operations. |
|---|
class vendor_connectors.google.services.GoogleServicesMixin
Section titled “class vendor_connectors.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) → list[dict[str, Any]]
Section titled “list_compute_instances(project_id: str, zone: str | None = None, unhump_instances: bool = False) → list[dict[str, Any]]”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) → list[dict[str, Any]]
Section titled “list_gke_clusters(project_id: str, location: str = ’-’, unhump_clusters: bool = False) → list[dict[str, Any]]”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) → dict[str, Any] | None
Section titled “get_gke_cluster(project_id: str, location: str, cluster_id: str) → dict[str, Any] | 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) → list[dict[str, Any]]
Section titled “list_storage_buckets(project_id: str, unhump_buckets: bool = False) → list[dict[str, Any]]”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) → list[dict[str, Any]]
Section titled “list_sql_instances(project_id: str, unhump_instances: bool = False) → list[dict[str, Any]]”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) → list[dict[str, Any]]
Section titled “list_pubsub_topics(project_id: str, unhump_topics: bool = False) → list[dict[str, Any]]”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) → list[dict[str, Any]]
Section titled “list_pubsub_subscriptions(project_id: str, unhump_subscriptions: bool = False) → list[dict[str, Any]]”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) → list[dict[str, Any]]
Section titled “list_enabled_services(project_id: str, unhump_services: bool = False) → list[dict[str, Any]]”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) → dict[str, Any]
Section titled “enable_service(project_id: str, service_name: str) → dict[str, Any]”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) → dict[str, Any]
Section titled “disable_service(project_id: str, service_name: str, force: bool = False) → dict[str, Any]”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]) → dict[str, Any]
Section titled “batch_enable_services(project_id: str, service_names: list[str]) → dict[str, Any]”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) → list[dict[str, Any]]
Section titled “list_kms_keyrings(project_id: str, location: str, unhump_keyrings: bool = False) → list[dict[str, Any]]”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) → dict[str, Any]
Section titled “create_kms_keyring(project_id: str, location: str, keyring_id: str) → dict[str, Any]”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’) → dict[str, Any]
Section titled “create_kms_key(project_id: str, location: str, keyring_id: str, key_id: str, purpose: str = ‘ENCRYPT_DECRYPT’, algorithm: str = ‘GOOGLE_SYMMETRIC_ENCRYPTION’) → dict[str, Any]”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
Section titled “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) → dict[str, dict[str, Any]]
Section titled “get_project_iam_users(project_id: str) → dict[str, dict[str, Any]]”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) → dict[str, Any]
Section titled “get_pubsub_resources_for_project(project_id: str, include_subscriptions: bool = True, unhump_resources: bool = False) → dict[str, Any]”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: dict[str, dict[str, Any]] | None = None, check_resources: bool = True, days_since_activity: int = 90) → list[dict[str, Any]]
Section titled “find_inactive_projects(projects: dict[str, dict[str, Any]] | None = None, check_resources: bool = True, days_since_activity: int = 90) → list[dict[str, Any]]”Find projects that appear to be inactive or dead.
A project is considered inactive if:
- It has no resources (compute, GKE, storage, etc.)
- Its lifecycle state is not ACTIVE
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 activity (not implemented yet).
Returns: List of inactive project dictionaries.