> ## Documentation Index
> Fetch the complete documentation index at: https://prefect-bd373955-pytest-markdown.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Manage cloud-hosted data

> Move data to and from cloud provider storage.

Learn how to use Prefect to move data to and from AWS, Azure, and GCP blob storage.

## Prerequisites

* Prefect [installed](/3.0rc/get-started/install/)
* Authenticated with [Prefect Cloud](/3.0rc/manage/cloud/connect-to-cloud/) (or self-hosted
  [Prefect server](/3.0rc/manage/self-host/) instance)
* A cloud provider account

## Install relevant Prefect integration library

In the CLI, install the Prefect integration library for your cloud provider:

<Tabs>
  <Tab title="AWS">
    [prefect-aws](https://prefecthq.github.io/prefect-aws/) provides blocks for interacting with AWS services.

    <div class="terminal">
      ```bash
      pip install -U prefect-aws
      ```
    </div>
  </Tab>

  <Tab title="Azure">
    [prefect-azure](https://prefecthq.github.io/prefect-azure/) provides blocks for interacting with Azure services.

    <div class="terminal">
      ```bash
       pip install -U prefect-azure
      ```
    </div>
  </Tab>

  <Tab title="GCP">
    [prefect-gcp](https://prefecthq.github.io/prefect-gcp/) provides blocks for interacting with GCP services.

    <div class="terminal">
      ```bash
       pip install -U prefect-gcp
      ```
    </div>
  </Tab>
</Tabs>

## Register the block types

Register the new block types with Prefect Cloud (or with your self-hosted Prefect server instance):

<Tabs>
  <Tab title="AWS">
    <div class="terminal">
      ```bash
      prefect block register -m prefect_aws  
      ```
    </div>
  </Tab>

  <Tab title="Azure">
    <div class="terminal">
      ```bash
      prefect block register -m prefect_azure 
      ```
    </div>
  </Tab>

  <Tab title="GCP">
    <div class="terminal">
      ```bash
      prefect block register -m prefect_gcp
      ```
    </div>
  </Tab>
</Tabs>

A confirmation message in the CLI shows that several block types were registered.
The UI shows the new block types listed.

## Create a storage bucket

Create a storage bucket in the cloud provider account.
Ensure the bucket is publicly accessible, or create a user or service account with the appropriate
permissions to fetch and write data to the bucket.

## Create a credentials block

If the bucket is private, there are several options to authenticate:

* At deployment runtime, ensure the runtime environment is authenticated.
* Create a block with configuration details and reference it when creating the storage block.

If you saved credential details in a block, you can use a credentials block specific to the cloud provider
or use a more generic secret block.
You can create [blocks](/3.0rc/develop/blocks/) through the UI or Python code.

The example below uses Python code to create a credentials block for your cloud provider.

<Warning>
  **Credentials safety**

  Don't store credential values in public locations such as public git platform repositories.
  The examples below use environment variables to store credential values.
</Warning>

<Tabs>
  <Tab title="AWS">
    ```python
    import os
    from prefect_aws import AwsCredentials

    my_aws_creds = AwsCredentials(
        aws_access_key_id="123abc",
        aws_secret_access_key=os.environ.get("MY_AWS_SECRET_ACCESS_KEY"),
    )
    my_aws_creds.save(name="my-aws-creds-block", overwrite=True)
    ```
  </Tab>

  <Tab title="Azure">
    ```python
    import os
    from prefect_azure import AzureBlobStorageCredentials

    my_azure_creds = AzureBlobStorageCredentials(
        connection_string=os.environ.get("MY_AZURE_CONNECTION_STRING"),
    )
    my_azure_creds.save(name="my-azure-creds-block", overwrite=True)
    ```
  </Tab>

  <Tab title="GCP">
    We recommend specifying the service account key file contents as a string, rather than the path to the file, because that file might not be available in your production environments.

    ```python
    import os
    from prefect_gcp import GCPCredentials

    my_gcp_creds = GCPCredentials(
        service_account_info=os.environ.get("GCP_SERVICE_ACCOUNT_KEY_FILE_CONTENTS"), 
    )
    my_gcp_creds.save(name="my-gcp-creds-block", overwrite=True)
    ```
  </Tab>
</Tabs>

Run the code to create the block. You should see a message that the block was created.

## Create a storage block

You can create a block for the chosen cloud provider using Python code or the UI.
This example uses Python code.

<Tabs>
  <Tab title="AWS">
    Note that the `S3Bucket` block is not the same as the `S3` block that ships with Prefect.
    The `S3Bucket` block used in this example is part of the `prefect-aws` library and provides additional functionality.

    Next, reference the credentials block created above.

    ```python
    from prefect_aws import S3Bucket

    s3bucket = S3Bucket.create(
        bucket="my-bucket-name",
        credentials="my-aws-creds-block"
        )
    s3bucket.save(name="my-s3-bucket-block", overwrite=True)
    ```
  </Tab>

  <Tab title="Azure">
    Note that the `AzureBlobStorageCredentials` block is not the same as the Azure block that ships with Prefect.
    The `AzureBlobStorageCredentials` block used in this example is part of the `prefect-azure` library and provides additional functionality.

    Azure blob storage doesn't require a separate block. The connection string used in the `AzureBlobStorageCredentials` block can encode the information needed.
  </Tab>

  <Tab title="GCP">
    Note that the `GcsBucket` block is not the same as the `GCS` block that ships with Prefect.
    The `GcsBucket` block is part of the `prefect-gcp` library and provides additional functionality.

    Next, reference the credentials block created above.

    ```python
    from prefect_gcp.cloud_storage import GcsBucket

    gcsbucket = GcsBucket(
        bucket="my-bucket-name", 
        credentials="my-gcp-creds-block"
        )
    gcsbucket.save(name="my-gcs-bucket-block", overwrite=True)
    ```
  </Tab>
</Tabs>

Run the code to create the block. You should see a message that the block was created.

## Write data

Use your new block inside a flow to write data to your cloud provider.

<Tabs>
  <Tab title="AWS">
    ```python
    from pathlib import Path
    from prefect import flow
    from prefect_aws.s3 import S3Bucket

    @flow()
    def upload_to_s3():
        """Flow function to upload data"""
        path = Path("my_path_to/my_file.parquet")
        aws_block = S3Bucket.load("my-s3-bucket-block")
        aws_block.upload_from_path(from_path=path, to_path=path)

    if __name__ == "__main__":
        upload_to_s3()
    ```
  </Tab>

  <Tab title="Azure">
    ```python
    from prefect import flow
    from prefect_azure import AzureBlobStorageCredentials
    from prefect_azure.blob_storage import blob_storage_upload

    @flow
    def upload_to_azure():
        """Flow function to upload data"""
        blob_storage_credentials = AzureBlobStorageCredentials.load(
            name="my-azure-creds-block"
        )

        with open("my_path_to/my_file.parquet", "rb") as f:
            blob_storage_upload(
                data=f.read(),
                container="my_container",
                blob="my_path_to/my_file.parquet",
                blob_storage_credentials=blob_storage_credentials,
            )

    if __name__ == "__main__":
        upload_to_azure()
    ```
  </Tab>

  <Tab title="GCP">
    ```python
    from pathlib import Path
    from prefect import flow
    from prefect_gcp.cloud_storage import GcsBucket

    @flow()
    def upload_to_gcs():
        """Flow function to upload data"""
        path = Path("my_path_to/my_file.parquet")
        gcs_block = GcsBucket.load("my-gcs-bucket-block")
        gcs_block.upload_from_path(from_path=path, to_path=path)

    if __name__ == "__main__":
        upload_to_gcs()
    ```
  </Tab>
</Tabs>

## Read data

Use your block to read data from your cloud provider inside a flow.

<Tabs>
  <Tab title="AWS">
    ```python

    from prefect import flow
    from prefect_aws import S3Bucket
    from prefect_aws.s3 import S3Bucket

    @flow
    def download_from_s3():
        """Flow function to download data"""
        s3_block = S3Bucket.load("my-s3-bucket-block")
        s3_block.get_directory(
            from_path="my_path_to/my_file.parquet", 
            local_path="my_path_to/my_file.parquet"
        )

    if __name__ == "__main__":
        download_from_s3()
    ```
  </Tab>

  <Tab title="Azure">
    ```python
    from prefect import flow
    from prefect_azure import AzureBlobStorageCredentials
    from prefect_azure.blob_storage import blob_storage_download

    @flow
    def download_from_azure():
        """Flow function to download data"""
        blob_storage_credentials = AzureBlobStorageCredentials.load(
            name="my-azure-creds-block"
        )
        blob_storage_download(
            blob="my_path_to/my_file.parquet",
            container="my_container",
            blob_storage_credentials=blob_storage_credentials,
        )

    if __name__ == "__main__":
        download_from_azure()
    ```
  </Tab>

  <Tab title="GCP">
    ```python
    from prefect import flow
    from prefect_gcp.cloud_storage import GcsBucket

    @flow
    def download_from_gcs():
        gcs_block = GcsBucket.load("my-gcs-bucket-block")
        gcs_block.get_directory(
            from_path="my_path_to/my_file.parquet", 
            local_path="my_path_to/my_file.parquet"
        )

    if __name__ == "__main__":
        download_from_gcs()

    ```
  </Tab>
</Tabs>

## Next steps

Check out the [`prefect-aws`](https://prefecthq.github.io/prefect-aws/), [`prefect-azure`](https://prefecthq.github.io/prefect-azure/), and [`prefect-gcp`](https://prefecthq.github.io/prefect-gcp/) docs to see additional methods for interacting with cloud storage providers.
Each library also contains blocks for interacting with other cloud-provider services.