Skip to content

Third-party Secrets: Connect to services without storing credentials in blocks

Credentials 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. Organizations whose security policies make such storage infeasible can still use Prefect to connect to third-party services securely.

Any sensitive information that is not stored in a block can be read from the environment.

For example, to find AWS credentials for authentication, any attributes not provided to an AWS Credentials block are sourced at runtime in the order shown in the Boto3 docs. Prefect-aws creates the session object using the values in the block and then, any missing values follow the sequence in the Boto3 docs.

Prefect-gcp and prefect-azure follow similar patterns.

In the example below, we interact with a Snowflake database using credentials stored in AWS Secrets Manager. This example can be generalized to 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

  1. Install prefect-aws and prefect-snowflake integration libraries.
  2. Store Snowflake password in AWS Secrets Manager.
  3. Create AwsSecret block to access the Snowflake password.
  4. Create AwsCredentials block for authentication.
  5. Ensure the compute environment has access to AWS credentials that are authorized to access the secret in AWS.
  6. Create and use SnowflakeCredentials and SnowflakeConnector blocks in Python code to interact with Snowflake.

Install prefect-aws and prefect-snowflake libraries

The following code will install and upgrade the necessary libraries and their dependencies.

pip install -U prefect-aws prefect-snowflake

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, we won't rotate the key, so click Next.
  7. Click Store.

Create AwsSecret block to access your Snowflake password

You can create blocks with Python code or via the Prefect UI. Block creation through the UI can help you visualize how the pieces fit together, so let's use it here.

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.

Note that if you're using a self-hosted Prefect server instance, you'll need to register the block types in the newly installed modules before creating blocks.

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

Create AwsCredentials block

Under the hood, Prefect is using 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.

Values for Access Key ID and Secret Access Key will be read from the compute environment. My AWS Access Key ID and Secret Access Key values with permissions to read the AWS Secret are stored locally in my ~/.aws/credentials file, so I'll leave those fields blank. You could enter those values at block creation, but then they would be saved to the database, and that's what we're trying to avoid. By leaving those attributes blank, Prefect knows to look to the compute environment. If the compute environment contains the necessary credentials, Prefect will use them to authenticate in the order shown in the Boto3 docs.

The same order is followed to resolve the AWS region. Let's specify the region in our AWSCredentials block so that our connection works regardless of the contents of our local AWS config file or whether we run our code on AWS compute located in anther region than our secret.

Click Create to save the blocks.

Ensure the compute environment has access to AWS credentials

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

Create and use SnowflakeCredentials and SnowflakeConnector blocks in Python code

Let's use Prefect's blocks for convenient access to Snowflake. We won't save the blocks, to ensure the credentials are not stored in Prefect Cloud.

We'll 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.

Note that 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.

Check out the prefect-snowflake docs for more examples of working with Snowflake.

Next steps

Now you can turn your flow into a deployment so that 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 that they are available at runtime.

Also ensure your compute has the AWS credentials for accessing 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.