{"id":4689,"library":"prefect-gcp","title":"Prefect GCP","description":"Prefect GCP is an integration library that allows Prefect workflows to interact seamlessly with various Google Cloud Platform (GCP) services. It provides blocks and tasks for services like Google Cloud Storage, BigQuery, Secret Manager, Vertex AI, and Cloud Run. The library is currently at version 0.6.17 and generally follows the release cadence of the main Prefect orchestration engine.","status":"active","version":"0.6.17","language":"en","source_language":"en","source_url":"https://github.com/PrefectHQ/prefect/tree/main/src/integrations/prefect-gcp","tags":["prefect","gcp","google cloud","orchestration","workflow","data pipeline","cloud storage","bigquery","secret manager","vertex ai"],"install":[{"cmd":"pip install -U \"prefect-gcp\"","lang":"bash","label":"Base installation"},{"cmd":"pip install -U \"prefect-gcp[all_extras]\"","lang":"bash","label":"Installation with all extras (BigQuery, Cloud Storage, Secret Manager, Vertex AI)"},{"cmd":"prefect block register -m prefect_gcp","lang":"bash","label":"Registering Blocks after installation"}],"dependencies":[{"reason":"Core orchestration engine for which prefect-gcp provides integrations.","package":"prefect","optional":false},{"reason":"Required for Cloud Storage functionality (installed with `[cloud_storage]` extra).","package":"google-cloud-storage","optional":true},{"reason":"Required for BigQuery functionality (installed with `[bigquery]` extra).","package":"google-cloud-bigquery","optional":true},{"reason":"Required for Secret Manager functionality (installed with `[secret_manager]` extra).","package":"google-cloud-secret-manager","optional":true},{"reason":"Required for Vertex AI functionality (installed with `[aiplatform]` extra).","package":"google-cloud-aiplatform","optional":true}],"imports":[{"note":"While `prefect_gcp.credentials` is technically correct, the top-level import `from prefect_gcp import GcpCredentials` is more idiomatic and usually sufficient, as `GcpCredentials` is directly exposed.","wrong":"from prefect_gcp.credentials import GcpCredentials","symbol":"GcpCredentials","correct":"from prefect_gcp import GcpCredentials"},{"note":"The `prefect.filesystems.GCS` block is part of Prefect Core, but `prefect_gcp.cloud_storage.GcsBucket` in `prefect-gcp` uses the `google-cloud-storage` package directly and offers more robust functionality and configuration options specific to GCP.","wrong":"from prefect.filesystems import GCS","symbol":"GcsBucket","correct":"from prefect_gcp.cloud_storage import GcsBucket"},{"symbol":"BigQueryWarehouse","correct":"from prefect_gcp.bigquery import BigQueryWarehouse"},{"symbol":"GcpSecret","correct":"from prefect_gcp.secret_manager import GcpSecret"}],"quickstart":{"code":"import os\nimport asyncio\nfrom prefect import flow, task\nfrom prefect_gcp import GcpCredentials\nfrom prefect_gcp.cloud_storage import GcsBucket\nfrom dotenv import load_dotenv\n\n# Load environment variables from .env file (if present)\nload_dotenv()\n\n@task\nasync def upload_and_download_file(bucket_name: str, file_content: str, source_path: str, destination_path: str):\n    \"\"\"Uploads a string to GCS and downloads it to a different path.\"\"\"\n    # Ensure GcpCredentials block exists, or create one for testing\n    # In a real scenario, you'd load a pre-configured block: GcpCredentials.load(\"my-gcp-creds\")\n    # For quickstart, we use env vars for service account info if available\n    service_account_info_str = os.environ.get('GCP_SERVICE_ACCOUNT_INFO')\n    service_account_info = None\n    if service_account_info_str:\n        import json\n        service_account_info = json.loads(service_account_info_str)\n\n    gcp_credentials = GcpCredentials(service_account_info=service_account_info)\n    await gcp_credentials.save('gcp-quickstart-creds', overwrite=True)\n\n    # Ensure GcsBucket block exists, or create one for testing\n    gcs_bucket_block = GcsBucket(bucket_name=bucket_name, gcp_credentials=gcp_credentials)\n    await gcs_bucket_block.save('gcs-quickstart-bucket', overwrite=True)\n\n    # Use the saved block in the flow\n    gcs_block = await GcsBucket.load(\"gcs-quickstart-bucket\")\n\n    print(f\"Uploading content to gs://{bucket_name}/{source_path}\")\n    await gcs_block.upload_from_bytes(file_content.encode('utf-8'), source_path)\n    print(\"Upload complete.\")\n\n    print(f\"Downloading content from gs://{bucket_name}/{source_path} to {destination_path}\")\n    downloaded_content = await gcs_block.read_bytes(source_path)\n    print(f\"Downloaded content: {downloaded_content.decode('utf-8')}\")\n\n    # Clean up the file (optional)\n    # await gcs_block.rm(source_path)\n    # print(f\"Cleaned up gs://{bucket_name}/{source_path}\")\n\n@flow(log_prints=True)\nasync def gcp_storage_flow(bucket_name: str = \"your-prefect-gcp-bucket-name\"):\n    file_content = \"Hello from Prefect GCP!\"\n    source_path = \"prefect-gcp-test.txt\"\n    destination_path = \"downloaded-prefect-gcp-test.txt\"\n    await upload_and_download_file(bucket_name, file_content, source_path, destination_path)\n\nif __name__ == \"__main__\":\n    # Set this environment variable or use actual service account info for testing\n    # export GCP_SERVICE_ACCOUNT_INFO='{\"type\": \"service_account\", ...}'\n    # Ensure you have a GCP bucket created and permissions for the service account\n    \n    # To run:\n    # 1. pip install -U \"prefect-gcp[cloud_storage]\" python-dotenv\n    # 2. prefect block register -m prefect_gcp\n    # 3. Create a GCS bucket (e.g., 'my-prefect-gcp-bucket')\n    # 4. Set GCP_SERVICE_ACCOUNT_INFO env var with your service account JSON, or rely on ADC.\n    # 5. python your_script_name.py\n    asyncio.run(gcp_storage_flow(bucket_name=os.environ.get(\"GCS_BUCKET_NAME\", \"your-prefect-gcp-bucket-name\")))","lang":"python","description":"This quickstart demonstrates how to use the `GcpCredentials` block for authentication and the `GcsBucket` block to interact with Google Cloud Storage. It uploads a simple text string to a specified bucket and then downloads it. It's designed to be runnable by either using Application Default Credentials or by providing service account JSON via an environment variable."},"warnings":[{"fix":"Migrate deployments to use Prefect workers (Cloud Run Worker or Vertex AI Worker) instead of the deprecated infrastructure blocks. Refer to Prefect's upgrade guides for detailed instructions.","message":"The `CloudRunJob` and `VertexAICustomTrainingJob` infrastructure blocks have been deprecated in favor of Cloud Run and Vertex AI workers. These blocks will not be available after September 2024. Users should migrate to the new worker-based execution model for deploying flows to these services.","severity":"breaking","affected_versions":"<=0.6.x"},{"fix":"Run `prefect block register -m prefect_gcp` in your environment after installing the library or any time new blocks are added.","message":"Always register `prefect-gcp` blocks after installation using `prefect block register -m prefect_gcp` for them to be available in the Prefect UI and for `Block.load()` operations.","severity":"gotcha","affected_versions":"All"},{"fix":"Provide `service_account_file` (path to JSON key) or `service_account_info` (JSON content as dict) to `GcpCredentials` or ensure `GOOGLE_APPLICATION_CREDENTIALS` environment variable is set for ADC. Verify the service account has necessary roles (e.g., Storage Admin for GCS, BigQuery Data Editor for BigQuery).","message":"Authentication with GCP services via `GcpCredentials` relies on Google's Application Default Credentials (ADC) by default if `service_account_file` or `service_account_info` are not provided. Ensure your environment (e.g., local machine, VM, Cloud Run service) has proper ADC configured or explicitly provide service account credentials. Missing or incorrect permissions on the service account are a common source of errors.","severity":"gotcha","affected_versions":"All"},{"fix":"For robust Google Cloud Storage interactions, prefer `from prefect_gcp.cloud_storage import GcsBucket`. If simple file system operations suffice, `prefect.filesystems.GCS` might be acceptable.","message":"There's a distinction between `prefect.filesystems.GCS` (from Prefect Core) and `prefect_gcp.cloud_storage.GcsBucket`. The latter, part of `prefect-gcp`, utilizes the `google-cloud-storage` package for more comprehensive functionality and is generally recommended for deeper GCP Cloud Storage integrations.","severity":"gotcha","affected_versions":"All"},{"fix":"Specify exact versions for `prefect` and `prefect-gcp` in your dependency management files or Dockerfiles.","message":"For production deployments, it's highly recommended to pin specific versions of `prefect` and `prefect-gcp` (e.g., `prefecthq/prefect-gcp:0.6.17-python3.12-prefect3.6.19`) in your Docker images or `requirements.txt` to ensure stability and avoid unexpected behavior from automatic updates.","severity":"gotcha","affected_versions":"All"}],"env_vars":null,"last_verified":"2026-04-12T00:00:00.000Z","next_check":"2026-07-11T00:00:00.000Z"}