Access third-party secrets

Connect to services without storing credentials in blocks.

Credential blocks and secret blocks are popular ways to store and retrieve sensitive information for connecting to third-party services. In Prefect Cloud, these block values are stored in encrypted format, allowing you to use Prefect to connect to third-party services securely.

This example interacts with a Snowflake database and stores the credentials required to connect in AWS Secrets Manager. This example uses Prefect Cloud and generally works with other third-party services that require credentials.

Prerequisites

1. Prefect installed
2. CLI authenticated to your Prefect Cloud account
3. Snowflake account
4. AWS account

Steps

Step 1: Install prefect-aws and prefect-snowflake libraries

The following code installs and upgrades the necessary libraries and their dependencies:

pip install -U prefect-aws prefect-snowflake

Step 2: Store Snowflake password in AWS Secrets Manager

Go to the AWS Secrets Manager console and create a new secret. Alternatively, create a secret using the AWS CLI or a script.

1. In the UI, choose Store a new secret.
2. Select Other type of secret.
3. Input the key-value pair for your Snowflake password where the key is any string and the value is your Snowflake password.
4. Copy the key for future reference and click Next.
5. Enter a name for your secret, copy the name, and click Next.
6. For this demo, you won’t rotate the key, so click Next.
7. Click Store.

Step 3: Create AwsSecret block to access your Snowflake password

Create blocks with Python code or through the Prefect UI. Block creation through the UI helps you visualize how the pieces fit together.

On the Blocks page, click on + to add a new block and select AWS Secret from the list of block types. Enter a name for your block and enter the secret name from AWS Secrets Manager.

If using self-hosted Prefect server, you must register the block types in the newly installed modules before you can create blocks.

prefect block register -m prefect_aws && prefect block register -m prefect_snowflake

Step 4: Create AwsCredentials block

In the AwsCredentials section, click Add + to create an AWS Credentials block.

Values for Access Key ID and Secret Access Key are read from the compute environment. Your AWS Access Key ID and Secret Access Key values with permissions to read the AWS Secret are stored locally in your ~/.aws/credentials file, so leave those fields blank. By leaving those attributes blank, Prefect knows to look to the compute environment.

Specify a region in your AWSCredentials block instead of your local AWS config file. The AwsCredentials block takes precedence and is more portable.

Under the hood, Prefect uses the AWS boto3 client to create a session.

In the AwsCredentials section of the form, click Add + and create an AWS Credentials block by entering the necessary values.

If the compute environment contains the necessary credentials, Prefect will use them to authenticate in the order shown in the Boto3 docs.

Follow the same order to resolve the AWS region. Specify the region in your AWSCredentials block so your connection works regardless of the contents of your local AWS config file—or whether you run your code on AWS compute located in anther region than your secret.

Click Create to save the blocks.

Step 5: Ensure the compute environment has access to AWS credentials

Ensure the compute environment contains AWS credentials with authorization to access AWS Secrets Manager. When you connect to Snowflake, Prefect automatically uses these credentials to authenticate and access the AWS secret that contains the Snowflake password.

Step 6: Create and use SnowflakeCredentials and SnowflakeConnector blocks in Python code

Use Prefect’s blocks for convenient access to Snowflake. Don’t save the blocks to ensure the credentials are not stored in Prefect Cloud.

Create a flow that connects to Snowflake and calls two tasks. The first task creates a table and inserts some data. The second task reads the data out.

import json
from prefect import flow, task
from prefect_aws import AwsSecret
from prefect_snowflake import SnowflakeConnector, SnowflakeCredentials


@task
def setup_table(snow_connector: SnowflakeConnector) -> None:
    with snow_connector as connector:
        connector.execute(
            "CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);"
        )
        connector.execute_many(
            "INSERT INTO customers (name, address) VALUES (%(name)s, %(address)s);",
            seq_of_parameters=[
                {"name": "Ford", "address": "Highway 42"},
                {"name": "Unknown", "address": "Space"},
                {"name": "Me", "address": "Myway 88"},
            ],
        )


@task
def fetch_data(snow_connector: SnowflakeConnector) -> list:
    all_rows = []
    with snow_connector as connector:
        while True:
            new_rows = connector.fetch_many("SELECT * FROM customers", size=2)
            if len(new_rows) == 0:
                break
            all_rows.append(new_rows)
    return all_rows


@flow(log_prints=True)
def snowflake_flow():
    aws_secret_block = AwsSecret.load("my-snowflake-pw")

    snow_connector = SnowflakeConnector(
        schema="MY_SCHEMA",
        database="MY_DATABASE",
        warehouse="COMPUTE_WH",
        fetch_size=1,
        credentials=SnowflakeCredentials(
            role="MYROLE",
            user="MYUSERNAME",
            account="ab12345.us-east-2.aws",
            password=json.loads(aws_secret_block.read_secret()).get("my-snowflake-pw"),
        ),
        poll_frequency_s=1,
    )

    setup_table(snow_connector)
    all_rows = fetch_data(snow_connector)
    print(all_rows)


if __name__ == "__main__":
    snowflake_flow()

Fill in the relevant details for your Snowflake account and run the script.

The flow reads the Snowflake password from the AWS Secret Manager and uses it in the SnowflakeCredentials block.

The SnowflakeConnector block uses the nested SnowflakeCredentials block to connect to Snowflake. Again, neither of the Snowflake blocks are saved, so the credentials are not stored in Prefect Cloud.

See prefect-snowflake for more examples of working with Snowflake.

Next steps

Now you can turn your flow into a deployment so you and your team can run it remotely on a schedule, in response to an event, or manually.

Make sure to specify the prefect-aws and prefect-snowflake dependencies in your work pool or deployment so they are available at runtime.

Also ensure your compute has the proper AWS credentials to access the secret in AWS Secrets Manager.

You’ve seen how to use Prefect blocks to store non-sensitive configuration and fetch sensitive configuration values from the environment. You can use this pattern to connect to other third-party services that require credentials, such as databases and APIs. You can use a similar pattern with any secret manager, or extend it to work with environment variables.