our_filename.py

our_filename.py

class SomeClass:
    @staticmethod
    def create_params_file(base_filename: str, **kwargs) -> str:
        return something

@task
def some_other_task():
    return something

.run()

@task def my_task

import prefect
from prefect import task
from prefect.tasks.prefect import create_flow_run


def create_params_file(base_filename: str, **kwargs) -> str:
    """Create a parameter file in a well-known location."""


@task
def dbt(
    k1,
    k2
    run_name: str = "__internal__",
    **kwargs,  # noqa: U100 used to set dependencies with other tasks
):
    run_name = f"__{prefect.context.flow_name}__.{run_name}"

    create_flow_run.run(
        flow_name="a_flow_name",
        project_name=get_project_name(prefect.context.flow_id),
        run_name=run_name,
        parameters={
            "parameters_filename": create_params_file(
                k1=v1, k2=v2, ...
            ),
        },
    )

from thatfile import dbt

with Flow(...) as flow:
  ...
  dbt(...)

stored_as_script

with Flow(
    "my_flow",
    run_config=LocalRun(labels=["input"]),
) as flow:

FlowStorageError

KubernetesRun

LocalRun