Florent VanDeMoortele
08/16/2023, 4:23 PM{'$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?Nate
08/16/2023, 4:33 PM{'$ref': {'block_document_id': '779148c3-8067-496d-9c03-0ae3f81de6ca'}}
and if you don't mind, how are you passing your block to your tasks?Florent VanDeMoortele
08/16/2023, 4:34 PM@task
def task_1(bucket: GcsBucket, **kwargs):
# Here I log and I obtain the previous dict instead of block instance
# Retrieve my file from GcsBucket and some others functions
@flow
def subflow(bucket: GcsBucket, **kwargs):
# Here I log and I obtain the previous dict instead of block instance
task_1.map(arg1, unmapped(bucket))
task_2 (no need of Block)
task_3.map(arg2, unmapped(bucket))
@flow
def flow(bucket: GcsBucket, **kwargs): # Here I select block instance from the UI
subflow(bucket, **kwargs)
other_subflow
Nate
08/16/2023, 4:51 PMFlorent VanDeMoortele
08/16/2023, 4:53 PMNate
08/16/2023, 5:27 PMYou 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'
Florent VanDeMoortele
08/17/2023, 8:29 AM@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?{
"gcs_bucket": {
"$ref": {
"block_document_id": "779148c3-8067-496d-9c03-0ae3f81de6ca"
}
}
}
Nate
08/17/2023, 4:12 PMfrom 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?Florent VanDeMoortele
08/17/2023, 4:20 PMNate
08/17/2023, 4:25 PMFlorent VanDeMoortele
08/17/2023, 4:28 PMNate
08/17/2023, 4:38 PMFlorent VanDeMoortele
08/24/2023, 4:27 PM