Jamie Place
06/30/2025, 10:20 AMMarvin
06/30/2025, 10:20 AMMarvin
06/30/2025, 10:22 AMpython
from typing import Annotated
from your_module import YourPydanticModel
@flow
def your_flow(params: Annotated[dict, YourPydanticModel]):
validated_params = YourPydanticModel(**params)
# Your flow logic here
2. Create Local Model Wrappers
Create a local Pydantic model that inherits from your imported models:
python
from pydantic import BaseModel
from your_module import ImportedModel
class LocalWrapper(BaseModel):
imported_data: ImportedModel
@flow
def your_flow(params: LocalWrapper):
# Your flow logic here
3. Use Model Config
If you're using Pydantic v2, ensure your models have the proper configuration:
python
from pydantic import BaseModel, ConfigDict
class YourModel(BaseModel):
model_config = ConfigDict(from_attributes=True)
# Your model fields here
4. Explicit Schema Definition
Define the schema explicitly in your flow using Pydantic's schema generation:
python
from your_module import YourPydanticModel
from prefect import flow
@flow(flow_run_name="Flow with Schema")
def your_flow(params: YourPydanticModel):
"""
Flow with explicit schema definition
Args:
params (YourPydanticModel): Parameters matching schema:
{YourPydanticModel.schema_json(indent=4)}
"""
# Your flow logic here
A few important notes:
- Make sure you're using consistent Pydantic versions across your codebase (either v1 or v2)
- If possible, keep model definitions close to where they're used
- Consider using simpler parameter types for UI visibility if the full Pydantic model complexity isn't necessary for the UI
If none of these approaches work for your specific use case, you might want to consider breaking down complex Pydantic models into simpler parameters for the flow interface while handling the model validation inside the flow function.
The Prefect UI parameter visibility is still an evolving feature, and while these workarounds should help, you might encounter edge cases with very complex nested models.Jamie Place
06/30/2025, 10:39 AMJamie Place
06/30/2025, 10:40 AMMarvin
06/30/2025, 10:40 AMMarvin
06/30/2025, 10:41 AMpython
from typing import Optional, List
from pydantic import BaseModel, Field
from prefect import flow
class TrainingConfig(BaseModel):
learning_rate: float = Field(
default=0.001,
description="The learning rate for the model",
gt=0,
le=1
)
batch_size: int = Field(
default=32,
description="Number of samples per batch",
gt=0
)
epochs: int = Field(
default=10,
description="Number of training epochs",
gt=0
)
@flow
def train_model(config: TrainingConfig) -> None:
"""
Train a machine learning model with the given configuration.
Parameters
----------
config : TrainingConfig
The training configuration with the following schema:
{
"learning_rate": float (0 < x ≤ 1)
The learning rate for the model training
"batch_size": int (> 0)
Number of samples per training batch
"epochs": int (> 0)
Number of training epochs
}
Returns
-------
None
This flow doesn't return any values
Notes
-----
The schema for the config parameter is:
{schema}
""".format(schema=TrainingConfig.schema_json(indent=4))
# Your training logic here
print(f"Training with learning rate: {config.learning_rate}")
print(f"Batch size: {config.batch_size}")
print(f"Epochs: {config.epochs}")
For more complex nested models, you can expand this pattern:
```python
from typing import List, Optional
from pydantic import BaseModel, Field
from prefect import flow
class DataConfig(BaseModel):
input_path: str = Field(
..., # ... means required
description="Path to input data"
)
features: List[str] = Field(
default_factory=list,
description="List of feature columns"
)
class ModelConfig(BaseModel):
architecture: str = Field(
default="mlp",
description="Neural network architecture type"
)
hidden_layers: List[int] = Field(
default=[64, 32],
description="Number of neurons in hidden layers"
)
class ExperimentConfig(BaseModel):
data: DataConfig = Field(
...,
description="Data configuration"
)
model: ModelConfig = Field(
default_factory=ModelConfig,
description="Model configuration"
)
training: TrainingConfig = Field(
default_factory=TrainingConfig,
description="Training configuration"
)
@flow
def run_experiment(config: ExperimentConfig) -> None:
"""
Run a machine learning experiment with the given configuration.
Parameters
----------
config : ExperimentConfig
The complete experiment configuration.
Schema details:
{schema}
Returns
-------
None
This flow doesn't return any values
Examples
--------
>>> config = ExperimentConfig(
... data=DataConfig(
... input_path="data/train.csv",
... features=["feature1", "feature2"]
... ),
... model=ModelConfig(
... architecture="mlp",
... hidden_layers=[128, 64]
... ),
... training=TrainingConfig(
... learning_rate=0.001,
... batch_size=32,
... epochs=10
... )
... )
>>> run_experiment(config)
""".format(schema=ExperimentConfig.schema_json(indent=4))
# Access nested configurations
print(f"Loading data from: {config.data.input_path}")
print(f"Using features: {config.data.features}")
print(f"Model architecture: {config.model.architecture}")
print(f"Training for {Jamie Place
06/30/2025, 10:47 AMfrom prefect import flow, task
from project.config import ModelConfig
@flow
def main(input_obj: ModelConfig):
return
Here is the config
class ModelStructure(str, Enum):
PhishNetMLP = "PhishNet"
PhishNetResidual = "PhishNetLarge"
class ModelConfig(BaseModel):
structure: ModelStructure
n_layers: int
dropout: float
This, however, results in this view in the uiMarvin
06/30/2025, 10:48 AMMarvin
06/30/2025, 10:48 AMpython
from prefect import flow
from pydantic import BaseModel
from project.config import ModelConfig, ModelStructure
class LocalModelConfig(BaseModel):
"""Wrapper for ModelConfig to ensure UI visibility"""
structure: ModelStructure = Field(
description="Model structure type",
# Explicitly list the enum values
json_schema_extra={
"choices": [e.value for e in ModelStructure]
}
)
n_layers: int = Field(description="Number of layers in the model")
dropout: float = Field(description="Dropout rate")
@flow
def main(input_obj: LocalModelConfig):
# Convert to original ModelConfig if needed
config = ModelConfig(**input_obj.model_dump())
return
2. Option 2: Modify Original ModelConfig
If you can modify the original config file, update it like this:
python
# project/config.py
from enum import Enum
from pydantic import BaseModel, Field
class ModelStructure(str, Enum):
PhishNetMLP = "PhishNet"
PhishNetResidual = "PhishNetLarge"
class ModelConfig(BaseModel):
structure: ModelStructure = Field(
description="Model structure type",
json_schema_extra={
"choices": [e.value for e in ModelStructure]
}
)
n_layers: int = Field(
description="Number of layers in the model",
gt=0
)
dropout: float = Field(
description="Dropout rate",
ge=0.0,
le=1.0
)
class Config:
# This helps with serialization
json_schema_extra = {
"title": "Model Configuration",
"description": "Configuration for the phishing detection model"
}
3. Option 3: Use Type Annotations with Dictionary
python
from typing import Annotated
from prefect import flow
from project.config import ModelConfig
@flow
def main(input_obj: Annotated[dict, ModelConfig]):
config = ModelConfig(**input_obj)
return
The key points to make this work:
1. Always add Field
descriptions and constraints to your Pydantic models
2. For enums, explicitly specify the choices using json_schema_extra
3. Use proper type hints and annotations
4. Consider adding validation ranges for numeric fields (like gt=0
for n_layers)
The most robust solution is Option 1 or 2, as they provide the best UI experience with proper type hints and descriptions. Option 3 is a fallback if you can't modify the original code.
Here's how the complete flow would look with Option 1:
python
from prefect import flow
from pydantic import BaseModel, Field
from project.config import ModelConfig, ModelStructure
class LocalModelConfig(BaseModel):
"""Wrapper for ModelConfig to ensure UI visibility"""
structure: ModelStructure = Field(
description="Model structure type",
json_schema_extra={
"choices": [e.value for e in ModelStructure]
}
)
n_layers: int = Field(
description="Number of layers in the model",
gt=0
)
dropout: float = Field(
description="Dropout rate",
ge=0.0,
le=1.0
)
class Config:
json_schema_extra = {
"title": "Phishing Model Configuration",
"description": "Configuration parameters for the phishing detection model"
}
@flow(name="Train Phishing Model")
def main(input_obj: LocalModelConfig):
"""
Train a phishing detection model with the given configuration.
Args:
input_obj: Model configuration parameters including structure,
number of layers, and dropout rate.
"""
# Convert to original ModelConfig if needed
config = ModelConfig(**input_obj.model_dump())
return
This should give you a clean UI with proper parameter inputs, including