<@ULVA73B9P> When creating flows I always add a Py...
# ask-marvin
j
@Marvin When creating flows I always add a Pydantic Model as the type of the flow input to surface the input parameters conveniently in the UI. When the Pydantic Model is in the same script as the flow function everything works fine. When I import the Pydantic Model, or some parameters are of an imported pedantic model type the UI fails to show the parameters at all. Is there anything I can do to solve this issue? I am using prefect 3.x
m
thought for 99.5 seconds
Based on the research, there are a few approaches you can take to resolve this issue: 1. Use Type Annotations Instead of Direct Models Instead of using the imported Pydantic model directly as the parameter type, you can create a local type annotation that references the imported model:
Copy code
python
   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:
Copy code
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:
Copy code
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:
Copy code
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.
j
Expand more on the last point, using explicit schema definitions? Is it possible to do this with the numpy doctoring format?
@Marvin Expand more on the last point, using explicit schema definitions? Is it possible to do this with the numpy doctoring format?
m
thought for 54.6 seconds
Here's a comprehensive example of how to use explicit schema definitions with NumPy-style docstrings in your Prefect flows:
Copy code
python
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}
Copy code
""".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 {
j
@Marvin This is an example flow. I expect the parameters in the UI to show up easily here. The ModelConfig class is a pydantic class Here is the flow
Copy code
from prefect import flow, task

from project.config import ModelConfig


@flow
def main(input_obj: ModelConfig):
    return
Here is the config
Copy code
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 ui
m
thought for 45.6 seconds
Based on your specific case, here's how you can modify your code to make the parameters visible in the UI: 1. Option 1: Use a Local Model Wrapper
Copy code
python
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:
Copy code
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
Copy code
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:
Copy code
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