Hello. When I call a task and pass a Block instance, it looks as if the task receives a copy of the block instance, not the actual reference:

@task
def mytask(obj):
  get_run_logger().info(f"in mytask: {id(obj)}")  # task shows different id than flow
  with obj.get_connection() as connection:
    connection.execute_string("SELECT 1;")

@flow
def myflow():
  obj = SnowflakeConnector.load(block_name)
  get_run_logger().info(f"in flow: {id(obj)}")
  mytask(obj)
  mytask.submit(obj)

Is this behaviour by design that I can rely upon? Are there cases where I would get the actual reference instead of a copy? _After having used the (SnowflakeConnector) object in the task, its underlying connection might have been closed. If subsequent tasks would receive the same instance, this would result in a "Connection already closed" exception. Getting a copy ensures that I can safely call get_connection in every task._

Maybe I’m not quite understanding your question, but you are creating a local object variable in

myflow()

then passing it as a reference to

mytask

. I’m fairly sure this is expected / consistent behavior unless you made it a global and were passing in a global reference?

Yes, it is a local object. If fact, the object might be the result of another task. Since it is an object, I was expecting it to be passed by reference. If I would have created a "regular" (not a BaseModel) object it would behave differently, i.e. it would get passed by reference and the id would be the same in both the flow and the task. But the object being a pydantic BaseModel causes a copy of it to get passed to the task. That is confusing so I want to make sure the behaviour I'm seeing is expected.

I can’t say that I know the answer here unfortunately , I’ll have to defer to someone else on this behavior

I see in prefect utilities

collections.py

that it explicitly makes copies if the object is a BaseModel instance. Would be great to know the thought behind this and whether I can rely on it in my code.