Skip to main content

​
prefect_gcp.credentials

Module handling GCP credentials.

​
Classes

​
ClientType

​
GcpCredentials

Block used to manage authentication with GCP. Google authentication is handled via the google.oauth2 module or through the CLI. Specify either one of service account_file or service_account_info; if both are not specified, the client will try to detect the credentials following Google’s Application Default Credentials. See Google’s Authentication documentation for details on inference and recommended authentication patterns. Attributes:
  • service_account_file: Path to the service account JSON keyfile.
  • service_account_info: The contents of the keyfile as a dict.
Methods:

​
block_initialization 

block_initialization(self)

​
get_access_token 

get_access_token(self)
See: https://stackoverflow.com/a/69107745 Also: https://www.jhanley.com/google-cloud-creating-oauth-access-tokens-for-rest-api-calls/

​
get_bigquery_client 

get_bigquery_client(self, project: Optional[str] = None, location: Optional[str] = None) -> 'BigQueryClient'
Gets an authenticated BigQuery client. Args:
  • project: Name of the project to use; overrides the base class’s project if provided.
  • location: Location to use.
Returns:
  • An authenticated BigQuery client.
Examples: Gets a GCP BigQuery client from a path. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_bigquery_client()
example_get_client_flow()
Gets a GCP BigQuery client from a dictionary. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_bigquery_client()

example_get_client_flow()

​
get_client 

get_client(self, client_type: Union[str, ClientType], **get_client_kwargs: Dict[str, Any]) -> Any
Helper method to dynamically get a client type. Args:
  • client_type: The name of the client to get.
  • **get_client_kwargs: Additional keyword arguments to pass to the get_*_client method.
Returns:
  • An authenticated client.
Raises:
  • ValueError: if the client is not supported.

​
get_cloud_storage_client 

get_cloud_storage_client(self, project: Optional[str] = None) -> 'StorageClient'
Gets an authenticated Cloud Storage client. Args:
  • project: Name of the project to use; overrides the base class’s project if provided.
Returns:
  • An authenticated Cloud Storage client.
Examples: Gets a GCP Cloud Storage client from a path. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_cloud_storage_client()
example_get_client_flow()
Gets a GCP Cloud Storage client from a dictionary. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_cloud_storage_client()
example_get_client_flow()

​
get_credentials_from_service_account 

get_credentials_from_service_account(self) -> Credentials
Helper method to serialize credentials by using either service_account_file or service_account_info.

​
get_job_service_async_client 

get_job_service_async_client(self, client_options: Union[Dict[str, Any], ClientOptions] = None) -> 'JobServiceAsyncClient'
Gets an authenticated Job Service async client for Vertex AI. Returns:
  • An authenticated Job Service async client.
Examples: Gets a GCP Job Service client from a path. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_job_service_async_client()

example_get_client_flow()
Gets a GCP Cloud Storage client from a dictionary. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_job_service_async_client()

example_get_client_flow()

​
get_job_service_client 

get_job_service_client(self, client_options: Union[Dict[str, Any], ClientOptions] = None) -> 'JobServiceClient'
Gets an authenticated Job Service client for Vertex AI. Returns:
  • An authenticated Job Service client.
Examples: Gets a GCP Job Service client from a path. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_job_service_client()

example_get_client_flow()
Gets a GCP Cloud Storage client from a dictionary. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_job_service_client()

example_get_client_flow()

​
get_secret_manager_client 

get_secret_manager_client(self) -> 'SecretManagerServiceClient'
Gets an authenticated Secret Manager Service client. Returns:
  • An authenticated Secret Manager Service client.
Examples: Gets a GCP Secret Manager client from a path. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_secret_manager_client()
example_get_client_flow()
Gets a GCP Cloud Storage client from a dictionary. 
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_secret_manager_client()
example_get_client_flow()