> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Deploy Dags from Google Cloud Storage to Astro

## Prerequisites

* A Google Cloud Storage (GCS) bucket.
* An [Astro Deployment](/docs/astro/create-deployment) with [Dag-only deploys enabled](/docs/astro/deploy-dags#enable-or-disable-dag-only-deploys-on-a-deployment).
* A [Deployment API token](/docs/astro/deployment-api-tokens), [Workspace API token](/docs/astro/workspace-api-tokens), or [Organization API token](/docs/astro/organization-api-tokens).
* An [Astro project](/docs/cli/v1.43/develop-project#create-an-astro-project) containing your project configurations.

## Dag deploy template

This CI/CD template can be used to deploy Dags from a single GCS bucket to a single Astro Deployment. When you create or modify a Dag in the GCS bucket, a Cloud Function triggers and initializes an Astro project to deploy your Dags using Astro CLI.

<Info>
  To deploy any non-Dag code changes to Astro, you need to trigger a standard image deploy with your Astro project. When you do this, your Astro project must include the latest version of your Dags from your GCS bucket. If your Astro project `dags` folder isn't up to date with your GCS Dags bucket when you trigger this deploy, you revert your Dags back to the version hosted in your Astro project.
</Info>

1. Download the latest Astro CLI binary from [GitHub releases](https://github.com/astronomer/astro-cli/releases), then rename the file to `astro_cli.tar.gz`. For example, to use Astro CLI version 1.40.0 in your template, download `astro_1.40.0_linux_amd64.tar.gz` and rename it to `astro_cli.tar.gz`.

2. In your GCS bucket, create the following new folders:

   * `dags`
   * `cli_binary`

3. Add `astro_cli.tar.gz` to `cli_binary`.

4. Create a [Cloud Run Function](https://docs.cloud.google.com/run/docs/quickstarts/functions/deploy-functions-console) with the Python 3.12 runtime in the same region as your storage bucket. Use the inline editor to create your function.

5. Create a [Cloud Storage trigger](https://docs.cloud.google.com/run/docs/triggering/storage-triggers) with the following configuration:

   * **Event provider**: Select **Cloud Storage**.
   * **Event**: Select **google.cloud.storage.object.v1.finalized**.
   * **Bucket**: Select your storage bucket.
   * **Service Account**: Ensure the service account you use has the Cloud Run Invoker role.

6. Choose the runtime service account on the **Security** tab of the Cloud Run Functions settings. Ensure that the service account has **Storage Object Viewer** (`storage.objects.list`) access to the Google Cloud Storage bucket.

7. Under the **Containers** settings, set the following [environment variables](https://docs.cloud.google.com/run/docs/configuring/services/environment-variables#setting_runtime_environment_variables) for your Cloud Function:

   * `ASTRO_HOME` = `/tmp`
   * `ASTRO_API_TOKEN`: The value for your Workspace or Organization API token.
   * `ASTRO_DEPLOYMENT_ID`: Your Deployment ID.
   * `BUCKET`: Your GCS bucket.

   For production Deployments, ensure that you store the value for your API token in a secrets backend. See [Secret Manager overview](https://cloud.google.com/secret-manager/docs/overview).

8. When editing the function source, change the function entry point to `astro_deploy`.

9. Add the following code to `main.py`:

```python title="main.py" expandable wrap theme={null}
import os
import shutil
import subprocess
import tarfile

from google.cloud import storage
import functions_framework

BUCKET = os.environ.get("BUCKET", "missing-bucket")
deploymentId = os.environ.get("ASTRO_DEPLOYMENT_ID", "missing-deployment-id")


def clear_dir(path: str) -> None:
    if os.path.exists(path):
        print(f"Clearing directory: {path}")
        shutil.rmtree(path)
    os.makedirs(path, exist_ok=True)
    print(f"Re-created directory: {path}")

def untar(filename: str, destination: str) -> None:
    with tarfile.open(filename) as file:
        file.extractall(destination)

def run_command(cmd: str) -> None:
    print(f'running command: {cmd}')
    p = subprocess.Popen("set -x; " + cmd, shell=True)
    p.communicate()

def download_to_local(bucket_name: str, gcs_folder: str, local_dir: str = None) -> None:
    """Download the contents of a folder directory
    :param bucket_name: the name of the gcs bucket
    :param gcs_folder: the folder path in the gcs bucket
    :param local_dir: a relative or absolute directory path in the local file system
    """

    ## create a storage client to access GCS objects
    storage_client = storage.Client()
    source_bucket = storage_client.bucket(bucket_name)

    ## get a list of all the files in the bucket folder
    blobs = source_bucket.list_blobs(prefix=gcs_folder)

    ## download each of the dag to local
    for blob in blobs:
        if blob.name.endswith('/'):
            continue

        target = blob.name if local_dir is None \
            else os.path.join(local_dir, os.path.relpath(blob.name, gcs_folder))
        print(target)
        if not os.path.exists(os.path.dirname(target)):
            os.makedirs(os.path.dirname(target))

        blob.download_to_filename(target)
        print("downloaded file")

@functions_framework.cloud_event
def astro_deploy(cloud_event) -> None:
    base_dir = '/tmp/astro'
    dags_dir = f'{base_dir}/dags'

    clear_dir(dags_dir)

    # --- Download DAGs ---
    print('downloading dags')
    download_to_local(BUCKET, 'dags/', f'{base_dir}/dags')  # NOTE: use "dags/" prefix

    # --- Download CLI ---
    print('downloading cli')
    download_to_local(BUCKET, 'cli_binary/', base_dir)

    # --- Initialize project ---
    os.chdir(base_dir)
    untar('./astro_cli.tar.gz', '.')
    run_command('echo y | ./astro dev init')

    # --- Remove generated example DAG(s) ---
    example_paths = [
        "dags/example_dag.py",
        "dags/exampledag.py",
    ]

    for path in example_paths:
        full_path = os.path.join(base_dir, path)
        if os.path.exists(full_path):
            print(f"Removing generated example DAG: {full_path}")
            os.remove(full_path)
        else:
            print(f"Example DAG not found: {full_path}")

    # --- Deploy ----
    run_command(f'./astro deploy {deploymentId} --dags')
```

10. Add the dependency `google-cloud-storage` to the `requirements.txt` file for your Cloud Function. See [Specifying Dependencies in Python](https://docs.cloud.google.com/run/docs/runtimes/python-dependencies).

11. (Optional) If you want the function to trigger when Dags are deleted as well as created/modified, create another [Cloud Storage trigger](https://docs.cloud.google.com/run/docs/triggering/storage-triggers) with the following configuration:

    * **Event provider**: Select **Cloud Storage**.
    * **Event**: Select **google.cloud.storage.object.v1.deleted**.
    * **Bucket**: Select your storage bucket.
    * **Service Account**: Ensure the service account you use has the Cloud Run Invoker role.

12. If you haven't already, deploy your complete Astro project to your Deployment. See [Deploy code](/docs/astro/deploy-code).

13. Add your Dags to the `dags` folder in your storage bucket.

14. In the Astro UI, click **Deployments**, then select your Deployment. Confirm that your deploy worked by checking the Deployment **Dag bundle version**. The version's name should include the time that you added the Dags to your GCS bucket.