@Marvin how to create custom block?

To create a custom block in Prefect, you define a class that subclasses

Block

, which is built on Pydantic's

BaseModel

. Below is the example from the Prefect documentation:

python
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

You can then use this new

Cube

block to store configurations that can be used later in a flow:

python
from prefect import flow

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

For further details, you may refer to the Prefect documentation on creating new block types.

@Marvin can a block be "hinted" in a decorator

The search results from the Prefect documentation do not provide any explicit information about using a block as a hint in task decorators or specifying blocks directly within a decorator syntax. For further information or clarifications, it would be worthwhile to either check additional documentation on blocks or look into community discussions or GitHub issues for examples and best practices. If necessary, I can help search GitHub issues for relevant discussions on this topic. Would you like me to do that?

@Marvin is there a way to auto-initiate db connection (block) pre task/flow?

The excerpts provided from the documentation do not present a clear example of how to auto-initiate a database connection (block) pre-task or pre-flow automatically. The typical usage pattern involves instantiating a block within a flow or a task, using a context manager to manage connection lifetimes or calling explicit methods to establish a connection. However, Prefect 2.x has a mechanism to load blocks from storage spontaneously. For instance, with a

SnowflakeConnector

block, you can do the following:

python
from prefect_snowflake.database import SnowflakeConnector

with SnowflakeConnector.load("BLOCK_NAME") as conn:
    # Use `conn` to interact with the database

This approach is manual and explicit, requiring the developer to handle the connector within the context of a task. If you need a more automatic approach, you might consider writing a flow-level startup task that establishes the connection, and then making subsequent tasks depend on it. Alternatively, you could create your initialization logic as a global object that gets imported and reused across tasks. For more tailored solutions, you might want to explore the Prefect Community Slack Channel or GitHub Issues, where you could get specific advice from the community or find code samples that fit your scenario. Would you like assistance searching GitHub or Community discussions for more information?

@Marvin can I use env vars to instantiate block?

To use environment variables for instantiating blocks in Prefect, you can leverage the environment variable syntax within your configuration or deployment files, though the documentation doesn't provide a clear example of this being done directly for blocks. However, for related configurations, environment variables can be referenced, but you need to be cautious about sensitive information and prefer to use blocks to handle that. You can reference environment variables with the following syntax:

yaml
{{ $MY_ENV_VAR }}

It is also worth noting that when using Prefect deployments, you can reference blocks within your

prefect.yaml

file like so:

yaml
{{ prefect.blocks.block_type.block_slug }}

While the documentation excerpts do not offer a step-by-step guide on how to use environment variables to directly instantiate blocks, it does suggest that environment variables can be referenced in the context of Prefect configuration. For sensitive information, blocks and variables are the recommended approach. For instance, you might reference a Prefect Secret block that is populated with an environment variable, ensuring your secrets are not exposed. For more specific guidance, refer to the Prefect documentation on using environment variables in a deployment definition as well as the general information about Prefect deployments.

@Marvin is it possible to set DaskTaskRunner as the default runner?

The documentation excerpts provided do not explicitly describe how to set the

DaskTaskRunner

as the default task runner globally. However, you can specify a task runner when you define the flow, which allows you to use

DaskTaskRunner

for that particular flow. Here's an example of how to specify a task runner for a flow:

python
from prefect import flow, task
from prefect.task_runners import DaskTaskRunner

@task
def example_task():
    # Task implementation here

@flow(task_runner=DaskTaskRunner())
def my_flow():
    example_task.submit()

Using the

task_runner

argument when creating the flow, you instantiate

DaskTaskRunner

and assign it as the task runner for the flow. If you want to use

DaskTaskRunner

for all your flows by default, one approach is to set it directly in each flow definition, as shown above. If you're looking to configure this at a global level, you might consider setting the configuration in your user configuration file (

~/.prefect/config.toml

). However, this was not explicitly found in the documentation and may not be directly supported. Keep in mind that task runner configurations can be sensitive to execution environments and specific use cases. So even if you manage to set a global default, it's worth considering on a per-flow basis whether

DaskTaskRunner

is the most appropriate choice. For more detailed information, please refer to the Prefect Task Runners documentation and the Dask and Ray Task Runners guide.