def METHOD_NAME(self) -> Flow:
    df = self.df_task.bind(
        flow=self,
        credentials=self.epicor_credentials,
        config_key=self.epicor_config_key,
        validate_date_filter=self.validate_date_filter,
        end_date_field=self.end_date_field,
        start_date_field=self.start_date_field,
    )
    df_mapped = cast_df_to_str.bind(df, flow=self)
    df_with_metadata = add_ingestion_metadata_task.bind(df_mapped, flow=self)
    parquet = df_to_parquet.bind(
        df=df_with_metadata,
        path=self.local_file_path,
        if_exists=self.if_exists,
        flow=self,
    )
    create_duckdb_table = self.create_duckdb_table_task.bind(
        path=self.local_file_path,
        schema=self.duckdb_schema,
        table=self.duckdb_table,
        if_exists=self.if_exists,
        if_empty=self.if_empty,
        flow=self,
    )
    create_duckdb_table.set_upstream(parquet, flow=self)