prefect.deployments.base

Core primitives for managing Prefect projects. Projects provide a minimally opinionated build system for managing flows and deployments.

To get started, follow along with the deloyments tutorial.

configure_project_by_recipe

Given a recipe name, returns a dictionary representing base configuration options.

Parameters:

Name	Type	Description	Default
recipe	str	the name of the recipe to use	required
formatting_kwargs	dict	additional keyword arguments to format the recipe	{}

Raises:

Type	Description
ValueError	if provided recipe name does not exist.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/deployments/base.py

def configure_project_by_recipe(recipe: str, **formatting_kwargs) -> dict:
    """
    Given a recipe name, returns a dictionary representing base configuration options.

    Args:
        recipe (str): the name of the recipe to use
        formatting_kwargs (dict, optional): additional keyword arguments to format the recipe

    Raises:
        ValueError: if provided recipe name does not exist.
    """
    # load the recipe
    recipe_path = Path(__file__).parent / "recipes" / recipe / "prefect.yaml"

    if not recipe_path.exists():
        raise ValueError(f"Unknown recipe {recipe!r} provided.")

    with recipe_path.open(mode="r") as f:
        config = yaml.safe_load(f)

    config = apply_values(
        template=config, values=formatting_kwargs, remove_notset=False
    )

    return config

create_default_prefect_yaml

Creates default prefect.yaml file in the provided path if one does not already exist; returns boolean specifying whether a file was created.

Parameters:

Name	Type	Description	Default
name	str	the name of the project; if not provided, the current directory name will be used	None
contents	dict	a dictionary of contents to write to the file; if not provided, defaults will be used	None

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/deployments/base.py

def create_default_prefect_yaml(
    path: str, name: str = None, contents: dict = None
) -> bool:
    """
    Creates default `prefect.yaml` file in the provided path if one does not already exist;
    returns boolean specifying whether a file was created.

    Args:
        name (str, optional): the name of the project; if not provided, the current directory name
            will be used
        contents (dict, optional): a dictionary of contents to write to the file; if not provided,
            defaults will be used
    """
    path = Path(path)
    prefect_file = path / "prefect.yaml"
    if prefect_file.exists():
        return False
    default_file = Path(__file__).parent / "templates" / "prefect.yaml"

    with default_file.open(mode="r") as df:
        default_contents = yaml.safe_load(df)

    import prefect

    contents["prefect-version"] = prefect.__version__
    contents["name"] = name

    with prefect_file.open(mode="w") as f:
        # write header
        f.write(
            "# Welcome to your prefect.yaml file! You can use this file for storing and"
            " managing\n# configuration for deploying your flows. We recommend"
            " committing this file to source\n# control along with your flow code.\n\n"
        )

        f.write("# Generic metadata about this project\n")
        yaml.dump({"name": contents["name"]}, f, sort_keys=False)
        yaml.dump({"prefect-version": contents["prefect-version"]}, f, sort_keys=False)
        f.write("\n")

        # build
        f.write("# build section allows you to manage and build docker images\n")
        yaml.dump(
            {"build": contents.get("build", default_contents.get("build"))},
            f,
            sort_keys=False,
        )
        f.write("\n")

        # push
        f.write(
            "# push section allows you to manage if and how this project is uploaded to"
            " remote locations\n"
        )
        yaml.dump(
            {"push": contents.get("push", default_contents.get("push"))},
            f,
            sort_keys=False,
        )
        f.write("\n")

        # pull
        f.write(
            "# pull section allows you to provide instructions for cloning this project"
            " in remote locations\n"
        )
        yaml.dump(
            {"pull": contents.get("pull", default_contents.get("pull"))},
            f,
            sort_keys=False,
        )
        f.write("\n")

        # deployments
        f.write(
            "# the deployments section allows you to provide configuration for"
            " deploying flows\n"
        )
        yaml.dump(
            {
                "deployments": contents.get(
                    "deployments", default_contents.get("deployments")
                )
            },
            f,
            sort_keys=False,
        )
    return True

find_prefect_directory

Given a path, recurses upward looking for .prefect/ directories.

Once found, returns absolute path to the ./prefect directory, which is assumed to reside within the root for the current project.

If one is never found, None is returned.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/deployments/base.py

def find_prefect_directory(path: Path = None) -> Optional[Path]:
    """
    Given a path, recurses upward looking for .prefect/ directories.

    Once found, returns absolute path to the ./prefect directory, which is assumed to reside within the
    root for the current project.

    If one is never found, `None` is returned.
    """
    path = Path(path or ".").resolve()
    parent = path.parent.resolve()
    while path != parent:
        prefect_dir = path.joinpath(".prefect")
        if prefect_dir.is_dir():
            return prefect_dir

        path = parent.resolve()
        parent = path.parent.resolve()

initialize_project

Initializes a basic project structure with base files. If no name is provided, the name of the current directory is used. If no recipe is provided, one is inferred.

Parameters:

Name	Type	Description	Default
name	str	the name of the project; if not provided, the current directory name	None
recipe	str	the name of the recipe to use; if not provided, one is inferred	None
inputs	dict	a dictionary of inputs to use when formatting the recipe	None

Returns:

Type	Description
List[str]	List[str]: a list of files / directories that were created

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/deployments/base.py

def initialize_project(
    name: str = None, recipe: str = None, inputs: dict = None
) -> List[str]:
    """
    Initializes a basic project structure with base files.  If no name is provided, the name
    of the current directory is used.  If no recipe is provided, one is inferred.

    Args:
        name (str, optional): the name of the project; if not provided, the current directory name
        recipe (str, optional): the name of the recipe to use; if not provided, one is inferred
        inputs (dict, optional): a dictionary of inputs to use when formatting the recipe

    Returns:
        List[str]: a list of files / directories that were created
    """
    # determine if in git repo or use directory name as a default
    is_git_based = False
    formatting_kwargs = {"directory": str(Path(".").absolute().resolve())}
    dir_name = os.path.basename(os.getcwd())

    remote_url = _get_git_remote_origin_url()
    if remote_url:
        formatting_kwargs["repository"] = remote_url
        is_git_based = True
        branch = _get_git_branch()
        formatting_kwargs["branch"] = branch or "main"

    formatting_kwargs["name"] = dir_name

    has_dockerfile = Path("Dockerfile").exists()

    if has_dockerfile:
        formatting_kwargs["dockerfile"] = "Dockerfile"
    elif recipe is not None and "docker" in recipe:
        formatting_kwargs["dockerfile"] = "auto"

    # hand craft a pull step
    if is_git_based and recipe is None:
        if has_dockerfile:
            recipe = "docker-git"
        else:
            recipe = "git"
    elif recipe is None and has_dockerfile:
        recipe = "docker"
    elif recipe is None:
        recipe = "local"

    formatting_kwargs.update(inputs or {})
    configuration = configure_project_by_recipe(recipe=recipe, **formatting_kwargs)

    project_name = name or dir_name

    files = []
    if create_default_ignore_file("."):
        files.append(".prefectignore")
    if create_default_prefect_yaml(".", name=project_name, contents=configuration):
        files.append("prefect.yaml")
    if set_prefect_hidden_dir():
        files.append(".prefect/")

    return files

register_flow async

Register a flow with this project from an entrypoint.

Parameters:

Name	Type	Description	Default
entrypoint	str	the entrypoint to the flow to register	required
force	bool	whether or not to overwrite an existing flow with the same name	False

Raises:

Type	Description
ValueError	if force is False and registration would overwrite an existing flow

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/deployments/base.py

async def register_flow(entrypoint: str, force: bool = False):
    """
    Register a flow with this project from an entrypoint.

    Args:
        entrypoint (str): the entrypoint to the flow to register
        force (bool, optional): whether or not to overwrite an existing flow with the same name

    Raises:
        ValueError: if `force` is `False` and registration would overwrite an existing flow
    """
    try:
        fpath, obj_name = entrypoint.rsplit(":", 1)
    except ValueError as exc:
        if str(exc) == "not enough values to unpack (expected 2, got 1)":
            missing_flow_name_msg = (
                "Your flow entrypoint must include the name of the function that is"
                f" the entrypoint to your flow.\nTry {entrypoint}:<flow_name> as your"
                f" entrypoint. If you meant to specify '{entrypoint}' as the deployment"
                f" name, try `prefect deploy -n {entrypoint}`."
            )
            raise ValueError(missing_flow_name_msg)
        else:
            raise exc

    flow = await run_sync_in_worker_thread(load_flow_from_entrypoint, entrypoint)

    fpath = Path(fpath).absolute()
    prefect_dir = find_prefect_directory()
    if not prefect_dir:
        raise FileNotFoundError(
            "No .prefect directory could be found - run `prefect project"
            " init` to create one."
        )

    entrypoint = f"{fpath.relative_to(prefect_dir.parent)!s}:{obj_name}"

    flows_file = prefect_dir / "flows.json"
    if flows_file.exists():
        with flows_file.open(mode="r") as f:
            flows = json.load(f)
    else:
        flows = {}

    ## quality control
    if flow.name in flows and flows[flow.name] != entrypoint:
        if not force:
            raise ValueError(
                f"Conflicting entry found for flow with name {flow.name!r}.\nExisting"
                f" entrypoint: {flows[flow.name]}\nAttempted entrypoint:"
                f" {entrypoint}\n\nYou can try removing the existing entry for"
                f" {flow.name!r} from your [yellow]~/.prefect/flows.json[/yellow]."
            )

    flows[flow.name] = entrypoint

    with flows_file.open(mode="w") as f:
        json.dump(flows, f, sort_keys=True, indent=2)

    return flow

set_prefect_hidden_dir

Creates default .prefect/ directory if one does not already exist. Returns boolean specifying whether or not a directory was created.

If a path is provided, the directory will be created in that location.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/deployments/base.py

def set_prefect_hidden_dir(path: str = None) -> bool:
    """
    Creates default `.prefect/` directory if one does not already exist.
    Returns boolean specifying whether or not a directory was created.

    If a path is provided, the directory will be created in that location.
    """
    path = Path(path or ".") / ".prefect"

    # use exists so that we dont accidentally overwrite a file
    if path.exists():
        return False
    path.mkdir(mode=0o0700)
    return True