j

    Jason

    2 months ago
    Regarding Parameters, is there any way to get the result during a flow? I'm trying to put various parameters into a Pydantic BaseModel that we're using as a config object for a task, but the BaseModel is failing the validation because the type is Parameter instead of str, for example.
    Anna Geller

    Anna Geller

    2 months ago
    Can you share more and walk us through your code? what are you trying to do?
    not sure whether this is helpful, but here is one example that shows how such model can be used to ensure only given dbt commands are passed as parameter:
    from prefect import flow, get_run_logger
    from pydantic import BaseModel, validator
    
    
    class DbtCommand(BaseModel):
        dbt_command: str
    
        @validator("dbt_command")
        def must_be_in(cls, dbt_cmd):
            valid_cmds = ["dbt run", "dbt test", "dbt docs", "dbt debug"]
            if dbt_cmd not in valid_cmds:
                raise ValueError("Invalid dbt command. Must be one of %s", valid_cmds)
            return dbt_cmd
    
    
    @flow
    def run_dbt_commands(dbt: DbtCommand):
        logger = get_run_logger()
        <http://logger.info|logger.info>("Running dbt command: %s", dbt.dbt_command)
    
    
    if __name__ == "__main__":
        run_dbt_commands(dict(dbt_command="dbt debug"))
        print("------------------------")
        run_dbt_commands(dict(dbt_command="dbt run"))
        print("------------------------")
        run_dbt_commands(dict(dbt_command="dbt clean"))