Connect to external systems

About blocks

Prefect blocks store configuration and provide an interface for interacting with external systems. Blocks expose methods that provide functionality specific to these systems. For example, you can use blocks to:

Download data from, or upload data to, an S3 bucket
Query data from, or write data to, a database
Send a message to a Slack channel.

Block types are Python classes with a UI webform for configuration. Blocks are instantiation of these classes with specific values. Configure blocks through Python code or through the UI. Access blocks for use in Python code. Block values are stored in Prefect Cloud or your self-hosted Prefect server instance. You can share blocks with other users in your Prefect Cloud workspace. 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 +. The block catalogue in the UI

Blocks and parametersBlocks are useful for sharing configuration across flow runs and between flows.For configuration that will change between flow runs, we recommend using parameters.

Prefect built-in blocks

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

Block	Slug	Description
Azure	`azure`	Stores data as a file on Azure Data Lake and Azure Blob Storage.
Custom Webhook	`custom-webhook`	Calls custom webhooks.
Discord Webhook	`discord-webhook`	Calls Discord webhooks.
GCS	`gcs`	Store data as a file on Google Cloud Storage.
GitHub	`github`	Interacts with files stored on public GitHub repositories.
Kubernetes Cluster Config	`kubernetes-cluster-config`	Stores configuration for interaction with Kubernetes clusters.
Local File System	`local-file-system`	Stores data as a file on a local file system.
Mattermost Webhook	`mattermost-webhook`	Sends notifications through a provided Mattermost webhook.
Microsoft Teams Webhook	`ms-teams-webhook`	Sends notifications through a provided Microsoft Teams webhook.
Opsgenie Webhook	`opsgenie-webhook`	Sends notifications through a provided Opsgenie webhook.
Pager Duty Webhook	`pager-duty-webhook`	Sends notifications through a provided PagerDuty webhook.
Remote File System	`remote-file-system`	Stores data as a file on any remote file system that supports `fsspec`.
S3	`s3`	Stores data as a file on AWS S3.
Secret	`secret`	Stores a secret value. The value will be obfuscated when this block is logged or shown in the UI.
Sendgrid Email	`sendgrid-email`	Sends notifications through Sendgrid email.
Slack Webhook	`slack-webhook`	Sends notifications through a provided Slack webhook.
SMB	`smb`	Stores data as a file on a SMB share.
Twilio SMS	`twilio-sms`	Sends notifications through Twilio SMS.

The S3, Azure, GCS, and GitHub blocks are deprecated in favor of the the corresponding S3Bucket, AzureBlobStorageCredentials, GCSBucket, and GitHubRepository blocks found in the Prefect integration libraries. The JSON, DateTime, and String blocks are deprecated in favor of Variables.

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. Anyone can create block types and optionally share them with the community. 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.

Integration	Block	Slug
prefect-aws	ECS Task	`ecs-task`
prefect-aws	MinIO Credentials	`minio-credentials`
prefect-aws	S3 Bucket	`s3-bucket`
prefect-azure	Azure Blob Storage Credentials	`azure-blob-storage-credentials`
prefect-azure	Azure Container Instance Credentials	`azure-container-instance-credentials`
prefect-azure	Azure Container Instance Job	`azure-container-instance-job`
prefect-azure	Azure Cosmos DB Credentials	`azure-cosmos-db-credentials`
prefect-azure	AzureML Credentials	`azureml-credentials`
prefect-bitbucket	BitBucket Credentials	`bitbucket-credentials`
prefect-bitbucket	BitBucket Repository	`bitbucket-repository`
prefect-databricks	Databricks Credentials	`databricks-credentials`
prefect-dbt	dbt CLI BigQuery Target Configs	`dbt-cli-bigquery-target-configs`
prefect-dbt	dbt CLI Profile	`dbt-cli-profile`
prefect-dbt	dbt Cloud Credentials	`dbt-cloud-credentials`
prefect-dbt	dbt CLI Global Configs	`dbt-cli-global-configs`
prefect-dbt	dbt CLI Postgres Target Configs	`dbt-cli-postgres-target-configs`
prefect-dbt	dbt CLI Snowflake Target Configs	`dbt-cli-snowflake-target-configs`
prefect-dbt	dbt CLI Target Configs	`dbt-cli-target-configs`
prefect-docker	Docker Host	`docker-host`
prefect-docker	Docker Registry Credentials	`docker-registry-credentials`
prefect-email	Email Server Credentials	`email-server-credentials`
prefect-gcp	BigQuery Warehouse	`bigquery-warehouse`
prefect-gcp	GCP Cloud Run Job	`cloud-run-job`
prefect-gcp	GCP Credentials	`gcp-credentials`
prefect-gcp	GcpSecret	`gcpsecret`
prefect-gcp	GCS Bucket	`gcs-bucket`
prefect-gcp	Vertex AI Custom Training Job	`vertex-ai-custom-training-job`
prefect-github	GitHub Credentials	`github-credentials`
prefect-github	GitHub Repository	`github-repository`
prefect-gitlab	GitLab Credentials	`gitlab-credentials`
prefect-gitlab	GitLab Repository	`gitlab-repository`
prefect-kubernetes	Kubernetes Credentials	`kubernetes-credentials`
prefect-shell	Shell Operation	`shell-operation`
prefect-slack	Slack Credentials	`slack-credentials`
prefect-slack	Slack Incoming Webhook	`slack-incoming-webhook`
prefect-snowflake	Snowflake Connector	`snowflake-connector`
prefect-snowflake	Snowflake Credentials	`snowflake-credentials`
prefect-sqlalchemy	Database Credentials	`database-credentials`
prefect-sqlalchemy	SQLAlchemy Connector	`sqlalchemy-connector`

Use existing block types

Blocks are classes that subclass the Block base class. You can instantiate and use them like normal classes.

Instantiate blocks

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

from prefect_aws.s3 import S3Bucket

bucket_block = S3Bucket(bucket_name="data-bucket")

Save blocks

To retrieve this saved value use the .save() method:

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

To update saved block value stored for a given block, overwrite the existing block by passing overwrite=True:

bucket_block.save(overwrite=True)

Create a new S3Bucket by setting the name parameter to a new value:

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

Load blocks

You can use the block 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("S3Bucket/my_s3_bucket_block")
print(s3_bucket.bucket_name) # data-bucket

Delete blocks

Delete a block 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>

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 blocks just like a Pydantic model. Here’s a 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

You can include methods on a block to provide functionality. Here’s the same cube block with methods to calculate the volume and surface area of the cube:

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

Use the new Cube block type in a flow:

from prefect import flow

rubiks_cube = Cube(edge_length_inches=2.25)
rubiks_cube.save("rubiks-cube")

@flow
def calculate_cube_surface_area(cube_name):
    cube = Cube.load(cube_name)
    print(cube.get_surface_area())

if __name__ == "__main__":
    calculate_cube_surface_area("rubiks-cube") # 30.375

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 functionality 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 functionality 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}
# )

Block type metadata

Set metadata fields on a block type’s subclass to control how a block displays. Available metadata fields include:

Property	Description
_block_type_name	Display name of the block in the UI. Defaults to the class name.
_block_type_slug	Unique slug used to reference the block type in the API. Defaults to a lowercase, dash-delimited version of the block type name.
_logo_url	URL pointing to an image that should be displayed for the block type in the UI. Default to `None`.
_description	Short description of block type. Defaults to docstring, if provided.
_code_example	Short code snippet shown in UI for how to load/use block type. Default to first example provided in the docstring of the class, if provided.

Nested blocks

Blocks are composable: a block can be used within other blocks. You can create a block type that uses functionality from another block type by declaring it as an attribute. Nestable blocks are loosely coupled, and configuration can be changed for each block independently. This allows sharing configuration across multiple use cases. For example, here’s an expanded AWSCredentials block that enables an authenticated session through the boto3 library:

from typing import Optional

import boto3
from prefect.blocks.core import Block
from pydantic 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

    def get_boto3_session(self):
        return boto3.Session(
            aws_access_key_id = self.aws_access_key_id
            aws_secret_access_key = self.aws_secret_access_key
            aws_session_token = self.aws_session_token
            profile_name = self.profile_name
            region_name = self.region
        )

You can use the AWSCredentials block within an S3Bucket block to provide authentication when interacting with an S3 bucket:

import io

class S3Bucket(Block):
    bucket_name: str
    credentials: AWSCredentials

    def read(self, key: str) -> bytes:
        s3_client = self.credentials.get_boto3_session().client("s3")

        stream = io.BytesIO()
        s3_client.download_fileobj(Bucket=self.bucket_name, key=key, Fileobj=stream)

        stream.seek(0)
        output = stream.read()

        return output

    def write(self, key: str, data: bytes) -> None:
        s3_client = self.credentials.get_boto3_session().client("s3")
        stream = io.BytesIO(data)
        s3_client.upload_fileobj(stream, Bucket=self.bucket_name, Key=key)

You can use this S3Bucket block with previously saved AWSCredentials block values in order to interact with the configured S3 bucket:

my_s3_bucket = S3Bucket(
    bucket_name="my_s3_bucket",
    credentials=AWSCredentials.load("my_aws_credentials")
)

my_s3_bucket.save("my_s3_bucket")

This creates a reference from the AWSCredentials block my_aws_credentials to the S3Bucket block my_s3_bucket, so that changes to the values in my_aws_credentials propogate to my_s3_bucket. Values for nested blocks can also be specified in place:

my_s3_bucket = S3Bucket(
    bucket_name="my_s3_bucket",
    credentials=AWSCredentials(
        aws_access_key_id="AKIAJKLJKLJKLJKLJKLJK",
        aws_secret_access_key="secret_access_key"
    )
)

my_s3_bucket.save("my_s3_bucket")

In the above example, the values for AWSCredentials are saved with my_s3_bucket and are not usable with any other blocks.

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 Prefect Cloud 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)

Register blocks

Prefect comes with many blocks pre-registered and ready to use. If you do not have a block available for use, you can register it. Register blocks from a Python module available in the current environment 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

The registered block is now available for configuration.

Get started

Develop

Deploy

Automate

Cloud and Server

Resources

Connect to external systems

About blocks

Prefect built-in blocks

Blocks in Prefect integration libraries

Use existing block types

Instantiate blocks

Save blocks

Load blocks

Delete blocks

Create a new block type

Secret fields

Block type metadata

Nested blocks

Update custom `Block` types

Register blocks

​About blocks

​Prefect built-in blocks

​Blocks in Prefect integration libraries

​Use existing block types

​Instantiate blocks

​Save blocks

​Load blocks

​Delete blocks

​Create a new block type

​Secret fields

​Block type metadata

​Nested blocks

​Update custom Block types

​Register blocks

About blocks

Prefect built-in blocks

Blocks in Prefect integration libraries

Use existing block types

Instantiate blocks

Save blocks

Load blocks

Delete blocks

Create a new block type

Secret fields

Block type metadata

Nested blocks

Update custom `Block` types

Register blocks