<@ULVA73B9P> What is the correct way to specify a ...
# marvin-ai
m
@Marvin What is the correct way to specify a union parameter to a Prefect flow? I'd like to be able to effectively choose between two input forms, based on an initial selection. I've tried this, which nearly works, but I'm having trouble hiding the discriminator field in the Union type:
Copy code
class DateRange(BaseModel):
    model_config = {'title': 'Range'}
    mode: Literal['range'] = Field('range', exclude=True)

    start_date: date = Field(title='Start Date (inclusive)')
    end_date: date = Field(title='End Date (inclusive)')


class DateList(BaseModel):
    model_config = {'title': 'List'}
    mode: Literal['list'] = Field('list', exclude=True)

    dates: list[date] = Field(default_factory=list, title='Dates')

# This is parsed properly in the UI, but shows me a const, uneditable "mode" field.
DateParam1 = DateList | DateRange

# This is not parsed properly, and forces the UI to require JSON input
DateParam2 = Annotated[DateList | DateRange, Field(discriminator='mode')]
m
thought for 68.3 seconds
Based on the research and GitHub issues, I can provide a solution for your Union parameter issue. The key is to use
Annotated
with
Tag
and
Discriminator
to properly handle the Union type while controlling the discriminator field visibility. Here's how you can modify your code:
Copy code
python
from typing import Annotated, Union, Any
from datetime import date
from pydantic import BaseModel, Field, Tag, Discriminator
from typing_extensions import Literal

def date_param_discriminator(v: Any) -> str:
    if isinstance(v, dict):
        return v.get("mode", "")
    return getattr(v, "mode", "")

class DateRange(BaseModel):
    model_config = {'title': 'Range'}
    mode: Literal['range'] = Field('range', exclude=True)
    start_date: date = Field(title='Start Date (inclusive)')
    end_date: date = Field(title='End Date (inclusive)')

class DateList(BaseModel):
    model_config = {'title': 'List'}
    mode: Literal['list'] = Field('list', exclude=True)
    dates: list[date] = Field(default_factory=list, title='Dates')

# This version will properly handle the discriminator while hiding the mode field
DateParam = Annotated[
    Union[
        Annotated[DateRange, Tag("range")],
        Annotated[DateList, Tag("list")]
    ],
    Discriminator(date_param_discriminator)
]
This solution: 1. Uses
Annotated
with
Tag
to mark each type in the Union with its corresponding mode 2. Uses a
Discriminator
with a custom discriminator function to determine which type to use 3. Keeps the
mode
field excluded from the UI while still using it for type discrimination 4. Maintains the proper titles and field descriptions in the UI You can then use this in your flow like this:
Copy code
python
from prefect import flow

@flow
def process_dates(date_param: DateParam):
    if isinstance(date_param, DateRange):
        print(f"Processing date range: {date_param.start_date} to {date_param.end_date}")
    else:
        print(f"Processing date list: {date_param.dates}")
The UI will now show the proper form fields without exposing the discriminator field, and the type discrimination will work correctly when the flow is called. This approach is based on Pydantic's recommended way of handling discriminated unions and is compatible with Prefect 3.x's parameter handling. The discriminator function helps determine which type to use based on the
mode
field, while keeping that field hidden from the UI. Note: If you're experiencing any issues with Pydantic versions, make sure you're using a compatible version as there have been some recent compatibility issues with newer Pydantic releases (as evidenced in the GitHub issues).
m
@Marvin Tried this, and got an error on deployment of this form:
Copy code
PydanticUserError("`Tag` not provided for choice {'type': 'tagged-union', 'choices': {'range': {'type': 'definition-ref', 'schema_ref': '...', 'metadata': {'pydantic.internal.tagged_union_tag': 'range'}}, 'list': {'type': 'definition-ref', 'schema_ref': '...', 'metadata': {'pydantic.internal.tagged_union_tag': 'list'}}}, 'discriminator': <function date_param_discriminator at 0xffff746e6680>} used with `Discriminator`")
m
thought for 140.6 seconds
Based on the research and the error message, it appears there's an issue with how the discriminator is being handled in newer versions of Pydantic. Let's try a different approach using Pydantic's newer discriminated union pattern. Here's the modified solution:
Copy code
python
from typing import Annotated, Union, Literal
from datetime import date
from pydantic import BaseModel, Field, ConfigDict

class DateRange(BaseModel):
    model_config = ConfigDict(title='Range')
    type: Literal["range"] = Field("range")
    start_date: date = Field(title='Start Date (inclusive)')
    end_date: date = Field(title='End Date (inclusive)')

class DateList(BaseModel):
    model_config = ConfigDict(title='List')
    type: Literal["list"] = Field("list")
    dates: list[date] = Field(default_factory=list, title='Dates')

# Use a simpler Union approach with a discriminator field
class DateParam(BaseModel):
    model_config = ConfigDict(
        json_schema_extra={
            "discriminator": {"propertyName": "type"},
            "oneOf": [
                {"$ref": f"#/definitions/{DateRange.__name__}"},
                {"$ref": f"#/definitions/{DateList.__name__}"},
            ],
        }
    )
    __root__: Union[DateRange, DateList]

    def __init__(self, **data):
        super().__init__(__root__=data)

    @property
    def value(self) -> Union[DateRange, DateList]:
        return self.__root__
You can then use it in your flow like this:
Copy code
python
from prefect import flow

@flow
def process_dates(date_param: DateParam):
    param = date_param.value
    if isinstance(param, DateRange):
        print(f"Processing date range: {param.start_date} to {param.end_date}")
    else:
        print(f"Processing date list: {param.dates}")
This approach: 1. Uses a more straightforward discriminator pattern with a
type
field 2. Wraps the Union in a container model that properly handles the discriminator 3. Avoids the complexity of the
Tag
and
Discriminator
annotations that were causing issues 4. Still provides a clean interface in the UI with proper type discrimination The UI should now show the proper form fields and handle the type discrimination correctly. The
type
field will be visible but will be automatically set based on which form variant is selected. Note: If you're still experiencing issues, you might want to check your Pydantic version. The recent Pydantic 2.11.1 release has caused some compatibility issues with Prefect. You might need to pin to a specific version of Pydantic that's known to work well with your version of Prefect. Try this version and let me know if you encounter any other issues. We can further refine the solution based on your specific needs or any errors you encounter.
m
Darn. This doesn't work. I would love to see a "recipe" for this pattern in the Prefect docs. I've found other places this question was asked, but the provided solutions never seem to work for me - the UI falls back to JSON input. It looks like I'm using pydantic version 2.9.2 if that matters. I did get a working version that shows the discriminator in the UI, which is unfortunate, but I've spent a bit too much time on this.
n
the discriminator is not necessary for a simple union
sorry i didnt read the whole thread but what's wrong with a
Literal
or
Union
directly?
m
🤦 thank you so much, I'd seen the discriminator at some point and got overly hung up on it, it seems. I went back to the simple union and it does exactly what I want!