Jeremy Knickerbocker
01/28/2025, 1:29 AMfrom typing import Optional, Union
from prefect.blocks.core import Block
from prefect import flow, get_run_logger, task
from pydantic import Field
class Car(Block):
make: Optional[str] = Field(default=None, description="Make")
model: Optional[str] = Field(default=None, description="Model")
year: Optional[int] = Field(default=None, description="Year")
class Bike(Block):
brand: Optional[str] = Field(default=None, description="Brand")
tire_size: Optional[int] = Field(default=None, description="Tire Size")
class Person(Block):
name: Optional[str] = Field(default=None, description="Name")
age: Optional[int] = Field(default=None, description="Age")
transportion: Optional[Union[Car, Bike]] = Field(
default=None, description="Mode of transportation used by this person"
)
@task(name="Create Blocks")
def create_blocks():
bike = Bike(brand="Giant", tire_size=26)
person = Person(name="John", age=30, transportion=bike)
bike.save("bike1", overwrite=True)
person.save("person1", overwrite=True)
@task(name="Check Nested Blocks")
def check_nested_blocks():
person = Person.load("person1")
logger = get_run_logger()
logger.info(f"Person: {person}")
logger.info(f"{person.__dict__=}")
logger.info(f"{person.transportion}")
logger.info(f"{person.transportion.__dict__=}")
@flow(name="TestFlow")
def test_blocks_flow():
create_blocks()
check_nested_blocks()
if __name__ == "__main__":
test_blocks_flow()
The output shows their transportation as a Car
when it should be a Bike
python block_test.py
20:27:40.822 | INFO | Flow run 'melodic-zebra' - Beginning flow run 'melodic-zebra' for flow 'TestFlow'
20:27:40.824 | INFO | Flow run 'melodic-zebra' - View at <http://127.0.0.1:4203/runs/flow-run/d99d392a-319c-40ad-9c9f-4e6c4c63b0ca>
20:27:41.111 | INFO | Task run 'Create Blocks-4c4' - Finished in state Completed()
20:27:41.152 | INFO | Task run 'Check Nested Blocks-8cb' - Person: Person(name='John', age=30, transportion=Car(make=None, model=None, year=None))
20:27:41.153 | INFO | Task run 'Check Nested Blocks-8cb' - person.__dict__={'name': 'John', 'age': 30, 'transportion': Car(make=None, model=None, year=None)}
20:27:41.154 | INFO | Task run 'Check Nested Blocks-8cb' - Car(make=None, model=None, year=None)
20:27:41.154 | INFO | Task run 'Check Nested Blocks-8cb' - person.transportion.__dict__={'make': None, 'model': None, 'year': None}
20:27:41.156 | INFO | Task run 'Check Nested Blocks-8cb' - Finished in state Completed()
20:27:41.184 | INFO | Flow run 'melodic-zebra' - Finished in state Completed()
When I switch the order of the transportion: Optional[Union[Car, Bike]]
to transportion: Optional[Union[Bike, Car]]
then the code shows Bike
Marvin
01/28/2025, 1:29 AMMarvin
01/28/2025, 1:29 AMJeremy Knickerbocker
01/28/2025, 1:30 AMMarvin
01/28/2025, 1:30 AMBaseModel
features, as Prefect blocks are built on top of it. Here's a concise guide on how you might approach this:
1. Subclass the Block: Start by creating a class that subclasses Block
. This class will represent your custom block type.
2. Define Custom Fields: Use Pydantic's field definitions to specify the types of fields your block will accept. You can use Pydantic's Union
to allow a field to accept multiple types.
3. Register the Block: Use Prefect's registration methods to make your custom block available for use.
Here's a simple example:
python
from pydantic import BaseModel, Field
from typing import Union
from prefect.blocks.core import Block
class MyCustomBlock(Block):
my_field: Union[int, str] = Field(..., description="This field can be an int or a str")
# Register the block
MyCustomBlock.register_type_and_schema()
This example defines a block with a field my_field
that can be either an integer or a string. You can read more about creating custom blocks in the Prefect documentation.
If you have specific needs or additional complexities, feel free to ask!