@Marvin I can't pass Snowflake cursor between tasks in Prefect 3, but can I pass the connection? For example:

from prefect_snowflake import SnowflakeCredentials
credentials = SnowflakeCredentials.load(prefect_block_name)
connection = credentials.get_client()

Passing a Snowflake connection between tasks in Prefect 3.x can be tricky due to serialization issues. Prefect tasks need to serialize their inputs and outputs, and connections or cursors often contain non-serializable components, like open sockets. However, you can pass the connection configuration or credentials between tasks and re-establish the connection within each task. Here's how you might do it: 1. Serialize the Connection Details: Instead of passing the connection object, pass the connection details or credentials to each task. 2. Re-establish Connection in Each Task: Within each task, use the credentials to establish a new connection. This approach ensures that your tasks remain stateless and can be executed independently. Here's a rough example:

python
from prefect import flow, task
from prefect_snowflake import SnowflakeCredentials

@task
def create_connection(prefect_block_name):
    credentials = SnowflakeCredentials.load(prefect_block_name)
    return credentials

@task
def use_connection(credentials):
    connection = credentials.get_client()
    # Use the connection here
    # Make sure to close the connection if needed
    connection.close()

@flow
def my_flow(prefect_block_name):
    credentials = create_connection(prefect_block_name)
    use_connection(credentials)

my_flow("your_prefect_block_name")

This method leverages the ability to pass serializable data (like credentials) between tasks. You can read more about handling credentials and connections in Prefect here.