I don't understand what's the good practice to work with block during the flow. I have a block that I want to pass in some tasks of my flow. When I run it locally, it works. But when I run a deployment on the cloud, each task get a dict with this format:

{'$ref': {'block_document_id': '779148c3-8067-496d-9c03-0ae3f81de6ca'}}

instead of the block instance. How can I pass a block instance in a task of my flow?

hi @Florent VanDeMoortele - where are you seeing this?

{'$ref': {'block_document_id': '779148c3-8067-496d-9c03-0ae3f81de6ca'}}

and if you don't mind, how are you passing your block to your tasks?

I log it

hmm im having a hard time reproducing that. you can share what your remote infra is like? or if you’re encountering this with specific block types?

I have this problem with custom blocks and also from collections (like prefect_gcs here with GcsBucket block)

You see, when you pass a block instance to a task, Prefect serializes the block instance into a dictionary.

Marvin is incorrect here - you're free to pass types like Blocks to tasks (and thats not how blocks look when JSON serialized) for example, this works fine both locally and remotely

from prefect import flow, task, unmapped
from prefect.blocks.core import Block

class MyBlock(Block):
    custom_field: str = "default value"
    
# MyBlock().save("test")

@task
def accept_foo_and_block(foo: str, block: Block):
    print(f"Received foo {foo!r} of type {type(foo)}")
    print(f"Received block {block!r} of type {type(block)}")
    
@flow(log_prints=True)
def some_flow():
    accept_foo_and_block.map(
        foo=["a", "b"], block=unmapped(MyBlock.load("test"))
    )

if __name__ == "__main__":
    some_flow()

it seems there must be something going on with the block types that are giving you the

'$ref'

Thank you @Nate. I try with your example. When I load my block into the flow: it's okay. But, here block is an argument of my flow (to allow to chose into the Prefect UI). If I try with:

@flow(log_prints=True)
def some_flow(block: MyBlock):
    accept_foo_and_block.map(
        foo=["a", "b"], block=unmapped(block)
    )

It doesn't work. I think the problem append when Prefect UI transmits arguments to flow. Do you have any idea to resolve this problem?

hmm im not sure i understand your problem. if i have a flow like this

from prefect import flow
from prefect_gcp import GcsBucket

@flow(log_prints=True)
def foo(bucket: GcsBucket):
    print(bucket)

if __name__ == "__main__":
    foo()

and deploy it, I can use the form view to select / create a block, or go to the JSON tab to specify a JSON representation of the block are you not seeing this behaviour?

Yes, exactly. When I use the form, I select the block and when my flow run I obtain a dict representation with $ref instead of block instance

interesting - I see the behaviour you're describing now. certainly seems like a bug

This morning, I found this PR: https://github.com/PrefectHQ/prefect-ui-library/pull/1557 It's possibly a similar issue

sounds good!

Hi @Nate , I can't retrieve the Github issue about the block bug in UI. Could you please send me the link? So I can subscribe to it and follow this issue. Thank you !