Securely store typed configuration

Prefect blocks store typed configuration that can be used across workflows and deployments. The most common use case for blocks is storing credentials used to access external systems such as AWS or GCP. Prefect supports a large number of common blocks and it’s also easy to create your own!

​

How blocks work

To get the most out of blocks it’s important to understand how they work.

There are three layers to a block: its type, a document, and a Python class.

​

Block type

A block type is essentially a schema registered with the Prefect API. This schema can be inspected and discovered in the UI on the Blocks page.

To see block types available for configuration, use prefect block type ls from the CLI or navigate to the Blocks page in the UI and click +.

These types separate blocks from Prefect variables, which are unstructured JSON documents. In addition, block schemas allow for fields of SecretStr type which are stored with additional encryption and not displayed by default in the UI.

Block types are identified by a slug that is not configurable.

from prefect.blocks.core import Block


class Cube(Block):
    edge_length_inches: float


# register the block type under the slug 'cube'
Cube.register_type_and_schema()

​

Block document

A block document is an instantiation of the schema, or block type. A document contains specific values for each field defined in the schema.

All block types allow for the creation of as many documents as you wish. Building on our example above:

from prefect.blocks.core import Block


class Cube(Block):
    edge_length_inches: float


# instantiate the type with specific values
rubiks_cube = Cube(edge_length_inches=2.25)

# store those values in a block document 
# on the server for future use
rubiks_cube.save("rubiks-cube")


# instantiate and save another block document
tiny_cube = Cube(edge_length_inches=0.001)
tiny_cube.save("tiny")

Block documents can also be created and updated in the UI or API for easy change management. This allows you to work with slowly changing configuration without having to redeploy all workflows that rely on it; for example, you may use this to rotate credentials on a regular basis without touching your deployments.

​

Block class

A block class is the primary user-facing object; it is a Python class whose attributes are loaded from a block document. Most Prefect blocks encapsulate additional functionality built on top of the block document.

For example, an S3Bucket block contains methods for downloading data from, or upload data to, an S3 bucket; a SnowflakeConnector block contains methods for querying Snowflake databases.

Returning to our Cube example from above:

from prefect.blocks.core import Block

class Cube(Block):
    edge_length_inches: float

    def get_volume(self):
        return self.edge_length_inches ** 3

    def get_surface_area(self):
        return 6 * self.edge_length_inches ** 2


rubiks_cube = Cube.load("rubiks-cube")
rubiks_cube.get_volume() # 11.390625

​

Using blocks

Using existing or built-in Prefect blocks is as simple as importing the relevant class and calling its save and load methods. Note that many Prefect blocks are packaged into one of Prefect’s integration libraries.

To instantiate a block document that stores an S3 bucket value, use the S3Bucket block:

from prefect_aws.s3 import S3Bucket

# note that the S3Bucket contains many other fields
bucket_block = S3Bucket(bucket_name="data-bucket")

To store this block document for future use, use the .save() method:

bucket_block.save(name="my-data-bucket-block")

To update a previously saved block document, overwrite the existing document by passing overwrite=True:

bucket_block.save(overwrite=True)

You can now use the block document name to load the block:

from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow
def what_is_the_bucket():
    bucket = S3Bucket.load("my-data-bucket-block")
    print(bucket.bucket_name)

if __name__ == "__main__":
    what_is_the_bucket() # data-bucket

Alternatively, load a block with the unique slug that is a combination of the block type slug and the block name.

To load the S3Bucket block from above, run the following:

from prefect_aws.s3 import S3Bucket

s3_block = Block.load("s3-bucket/my_s3_bucket_block")
print(s3_bucket.bucket_name) # data-bucket

You can also delete a block document with the .delete() method:

from prefect.blocks.core import Block

Block.delete("s3-bucket/my-data-bucket-block")

Alternatively, use the CLI to delete specific blocks with a given slug or id:

prefect block delete s3-bucket/my-data-bucket-block

prefect block delete --id <my-id>

​

Nested blocks

Blocks are composable: a block can be used within other blocks. You can create a block type that uses capabilities from another block type by giving it the appropriate type.

Nested blocks are loosely coupled, and configuration can be changed for each block independently. This allows sharing configuration across multiple block documents.

For example, we can expand our basic S3Bucket example above using the AwsCredentials block:

from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

credentials = AwsCredentials(aws_access_key_id="AKIAJKLJKLJKLJKLJKLJK", 
                             aws_secret_access_key="password")
my_bucket_block = S3Bucket(bucket_name="my-bucket", credentials=credentials)
my_bucket_block.save("my_s3_bucket")

Alternatively, we can also load previously saved credentials:

my_bucket_block = S3Bucket(
    bucket_name="my-bucket",
    credentials=AwsCredentials.load("my_aws_credentials")
)

my_bucket_block.save("my_s3_bucket")

Both of these create a reference from the AwsCredentials block document my_aws_credentials to the S3Bucket block document my_s3_bucket, so that changes to the values in my_aws_credentials propagate to my_s3_bucket.

​

Create a new block type

To create a custom block type, define a class that subclasses Block. The Block base class builds on Pydantic’s BaseModel, so you can declare custom fields just like a Pydantic model.

We’ve already seen an example of a Cube block that represents a cube and holds information about the length of each edge in inches:

from prefect.blocks.core import Block

class Cube(Block):
    edge_length_inches: float


Cube.register_type_and_schema()

​

Register custom blocks

In addition to the register_type_and_schema method shown above, you can register blocks from a Python module with a CLI command:

prefect block register --module prefect_aws.credentials

This command is useful for registering all blocks found within a module in a Prefect Integration library.

Alternatively, if a custom block was created in a .py file, you can register the block with the CLI command:

prefect block register --file my_block.py

Block documents can now be created with the registered block schema.

​

Secret fields

All block values are encrypted before being stored. If you have values that you would not like visible in the UI or in logs, use the SecretStr field type provided by Pydantic to automatically obfuscate those values. You can use this capability for fields that store credentials such as passwords and API tokens.

Here’s an example of an AwsCredentials block that uses SecretStr:

from typing import Optional

from prefect.blocks.core import Block
from pydantic import SecretStr  # if pydantic version >= 2.0, use: from pydantic.v1 import SecretStr

class AwsCredentials(Block):
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[SecretStr] = None
    aws_session_token: Optional[str] = None
    profile_name: Optional[str] = None
    region_name: Optional[str] = None

Since aws_secret_access_key has the SecretStr type hint assigned to it, the value of that field is not exposed if the object is logged:

aws_credentials_block = AwsCredentials(
    aws_access_key_id="AKIAJKLJKLJKLJKLJKLJK",
    aws_secret_access_key="secret_access_key"
)

print(aws_credentials_block)
# aws_access_key_id='AKIAJKLJKLJKLJKLJKLJK' aws_secret_access_key=SecretStr('**********') aws_session_token=None profile_name=None region_name=None

Prefect’s SecretDict field type allows you to add a dictionary field to your block that automatically obfuscates values at all levels in the UI or in logs. This capability is useful for blocks where typing or structure of secret fields is not known until configuration time.

Here’s an example of a block that uses SecretDict:

from typing import Dict

from prefect.blocks.core import Block
from prefect.blocks.fields import SecretDict


class SystemConfiguration(Block):
    system_secrets: SecretDict
    system_variables: Dict


system_configuration_block = SystemConfiguration(
    system_secrets={
        "password": "p@ssw0rd",
        "api_token": "token_123456789",
        "private_key": "<private key here>",
    },
    system_variables={
        "self_destruct_countdown_seconds": 60,
        "self_destruct_countdown_stop_time": 7,
    },
)

system_secrets is obfuscated when system_configuration_block is displayed, but system_variables show up in plain-text:

print(system_configuration_block)
# SystemConfiguration(
#   system_secrets=SecretDict('{'password': '**********', 'api_token': '**********', 'private_key': '**********'}'), 
#   system_variables={'self_destruct_countdown_seconds': 60, 'self_destruct_countdown_stop_time': 7}
# )

​

Customize a block’s display

You can set metadata fields on a block type’s subclass to control how a block displays.

Available metadata fields include:

​

Update custom Block types

Here’s an example of how to add a bucket_folder field to your custom S3Bucket block; it represents the default path to read and write objects from (this field exists on our implementation).

Add the new field to the class definition:

class S3Bucket(Block):
    bucket_name: str
    credentials: AwsCredentials
    bucket_folder: Optional[str] = None
    ...

Then register the updated block type with either your Prefect Cloud account or your self-hosted Prefect server instance.

If you have any existing blocks of this type that were created before the update and you’d prefer to not re-create them, migrate them to the new version of your block type by adding the missing values:

# Bypass Pydantic validation to allow your local Block class to load the old block version
my_s3_bucket_block = S3Bucket.load("my-s3-bucket", validate=False)

# Set the new field to an appropriate value
my_s3_bucket_block.bucket_path = "my-default-bucket-path"

# Overwrite the old block values and update the expected fields on the block
my_s3_bucket_block.save("my-s3-bucket", overwrite=True)

​

Pre-registered blocks

​

Built-in blocks

Commonly used block types come built-in with Prefect. You can create and use these block types through the UI without installing any additional packages.

​

Blocks in Prefect integration libraries

Some block types that appear in the UI can be created immediately, with the corresponding integration library installed for use. For example, an AWS Secret block can be created, but not used until the prefect-aws library is installed.

Find available block types in many of the published Prefect integrations libraries. If a block type is not available in the UI, you can register it through the CLI.

Anyone can create a custom block type and, optionally, share it with the community.