prefect
prefect copied to clipboard
Allow blocks to reference nested blocks via capabilities
First check
- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.
Prefect Version
2.x
Describe the current behavior
Blocks can be nested inside of other blocks, but it has to be a specific block or a union of specific blocks. This means that if a block from the core library needs to reference a block in a collection library, that library collection would need to be added as a dependency to the core library. This rose as a problem during the development of a block for working with an AWS ECR registry that could be used with the existing DockerContainer
block.
Describe the proposed behavior
The proposed behavior is to allow blocks to reference other blocks via the capabilities that they offer. This information would be included in the block schema stored with the Orion server so that the available options for a nested block could be dynamically shown to the user in the UI based on the blocks with the specified capabilities that are available on the Orion server.
In the Python client, a block author could specify an abstract base class that represents a set of block capabilities. That block could then be registered as any other block and used for saving and loading block documents like a normal block. On loading, the client should be able to marshal the block document data into the correct class.
This would allow blocks to nest blocks that could be contained in other libraries without needing to declare an explicit dependency.
Example Use
from abc import ABC, abstractmethod
from prefect.blocks.core import Block
class InterfaceBlock(Block, ABC):
_block_schema_capabilities = ["do-thing"]
@abstractmethod
def do_thing():
pass
class ConcreteBlock(InterfaceBlock):
def do_thing():
print("Did that thing!")
class AnotherConcreteBlock(InterfaceBlock):
def do_thing():
print("Didn't do that thing.")
class ParentBlock(Block):
thing_doer: InterfaceBlock
def a_method():
self.thing_doer.do_thing()
block = ParentBlock(thing_doer=ConcreteBlock())
block.save("example")
example_block = ParentBlock.load("example")
example_block.a_method() # Did that thing!
Additional context
No response