Update deployment defaults with project init

src/prefect/cli/project.py

@@ -87,7 +87,7 @@ async def init(name: str = None, recipe: str = None):
     )
     file_msg = (
         f"Created project in [green]{Path('.').resolve()}[/green] with the following"
-        f" new files:\n {files}"
+        f" new files:\n{files}"
     )
     app.console.print(file_msg if files else empty_msg)
 

src/prefect/cli/root.py

@@ -434,7 +434,7 @@ async def deploy(
             parameters=base_deploy["parameters"],
             description=base_deploy["description"],
             tags=base_deploy["tags"],
-            path=base_deploy["path"],
+            path=base_deploy.get("path"),
             entrypoint=base_deploy["entrypoint"],
             parameter_openapi_schema=base_deploy["parameter_openapi_schema"].dict(),
             pull_steps=project["pull"],

src/prefect/projects/base.py

@@ -33,17 +33,33 @@ def find_prefect_directory(path: Path = None) -> Optional[Path]:
         parent = path.parent.resolve()
 
 
-def create_default_deployment_yaml(path: str) -> bool:
+def create_default_deployment_yaml(path: str, field_defaults: dict = None) -> bool:
     """
     Creates default deployment.yaml file in the provided path if one does not already exist;
     returns boolean specifying whether a file was created.
     """
+    field_defaults = field_defaults or {}
+
     path = Path(path)
     if (path / "deployment.yaml").exists():
         return False
+
     default_file = Path(__file__).parent / "templates" / "deployment.yaml"
+
+    # load default file
+    with open(default_file, "r") as df:
+        default = yaml.safe_load(df)
+
+    # apply field defaults
+    for field, default_value in field_defaults.items():
+        if isinstance(default.get(field), dict):
+            default[field].update(default_value)
+        else:
+            default[field] = default_value
+
     with open(path / "deployment.yaml", "w") as f:
-        f.write(default_file.read_text())
+        yaml.dump(default, f)
+
     return True
 
 
@@ -194,10 +210,16 @@ def initialize_project(name: str = None, recipe: str = None) -> List[str]:
 
     project_name = name or dir_name
 
+    # apply deployment defaults
+    if "docker" in recipe:
+        field_defaults = {"work_pool": {"job_variables": {"image": "{{ image_name }}"}}}
+    else:
+        field_defaults = {}
+
     files = []
     if create_default_ignore_file("."):
         files.append(".prefectignore")
-    if create_default_deployment_yaml("."):
+    if create_default_deployment_yaml(".", field_defaults=field_defaults):
         files.append("deployment.yaml")
     if create_default_project_yaml(".", name=project_name, contents=configuration):
         files.append("prefect.yaml")

src/prefect/projects/templates/deployment.yaml

@@ -8,7 +8,6 @@ schedule: null
 # flow-specific fields
 flow_name: null
 entrypoint: null
-path: null
 parameters: {}
 parameter_openapi_schema: null
 

tests/projects/test_base.py

@@ -143,6 +143,35 @@ async def test_initialize_project_with_recipe(self):
         build_step = contents["build"][0]
         assert "prefect_docker.projects.steps.build_docker_image" in build_step
 
+    @pytest.mark.parametrize(
+        "recipe",
+        [
+            d.absolute().name
+            for d in Path(
+                prefect.__development_base_path__
+                / "src"
+                / "prefect"
+                / "projects"
+                / "recipes"
+            ).iterdir()
+            if d.is_dir() and "docker" in d.absolute().name
+        ],
+    )
+    async def test_initialize_project_with_docker_recipe_default_image(self, recipe):
+        files = initialize_project(recipe=recipe)
+        assert len(files) >= 3
+
+        with open("prefect.yaml", "r") as f:
+            contents = yaml.safe_load(f)
+
+        build_step = contents["build"][0]
+        assert "prefect_docker.projects.steps.build_docker_image" in build_step
+
+        with open("deployment.yaml", "r") as f:
+            contents = yaml.safe_load(f)
+
+        assert contents["work_pool"]["job_variables"]["image"] == "{{ image_name }}"
+
 
 class TestRegisterFlow:
     async def test_register_flow_works_in_root(self, project_dir):