Prefect.Deployment doesn't have a property pull_steps

[davzucky]

First check

• I added a descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the Prefect documentation for this issue.
• I checked that this issue is related to Prefect and not one of its dependencies.

Bug summary

The new deployment using Project was released with Prefect 2.10.x. This adds the ability not to have to pull flow definition from external storage like it was the case before.

However, to enable that, when we do deployment using code, prefect.Deployment doesn't have the property pull_steps that we could set.

Reproduction

Prefect Deployment model does not have an argument. However, the function in the same module load_flow_from_flow_run is using this property.
This is working because this property exists on the ORM layer.
The call to client.create_deployment will need to be updated as well.

If you are happy with adding that, I will create a pull request

Error

I cannot set the property in the deployment object

Versions

(venv) dzucker@x1clinux ~/git $ prefect version
Version:             2.10.4
API version:         0.8.4
Python version:      3.10.10
Git commit:          b6d0433a
Built:               Thu, Apr 13, 2023 5:34 PM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         ephemeral
Server:
  Database:          sqlite
  SQLite version:    3.41.2

Additional context

We need that because we control and run all our deployments using code.

[serinamarie]

Hi @davzucky, thanks for the issue! What I'm hearing is that prior to running the prefect deploy command, you want to edit the pull step? When running prefect project init, the prefect.yaml file is created in your directory with an automatically populated pull step specific to your project. This is then used when creating a deployment:

prefect/src/prefect/cli/root.py

Lines 505 to 520 in b6d0433

	deployment_id = await client.create_deployment(
	flow_id=flow_id,
	name=base_deploy["name"],
	work_queue_name=base_deploy["work_pool"]["work_queue_name"],
	work_pool_name=base_deploy["work_pool"]["name"],
	version=base_deploy["version"],
	schedule=base_deploy["schedule"],
	parameters=base_deploy["parameters"],
	description=base_deploy["description"],
	tags=base_deploy["tags"],
	path=base_deploy.get("path"),
	entrypoint=base_deploy["entrypoint"],
	parameter_openapi_schema=base_deploy["parameter_openapi_schema"].dict(),
	pull_steps=project["pull"],
	infra_overrides=base_deploy["work_pool"]["job_variables"],
	)

Can you explain more on why the prefect.yaml file pull step isn't sufficient to your use case?

[github-actions]

This issue is stale because it has been open 30 days with no activity. To keep this issue open remove stale label or comment.

[davzucky]

Hi @serinamarie
The main reasons we need this feature are the following.

1. We cannot push to the central Docker registry. However, we can only push to a dev registry and promote to a prod later.
2. Our CI and CD are split between repositories. All our applications are in one repository, and we have our IaaC in another. This manages state, compliance, and deployment
3. We like doing deployment using code as it allows us to do deep testing. This allows us as well consistency between Block requirement and deployment setup
4. We have different Kubernetes clusters which run the same app (Docker image); however, we don't deploy the same flow on each cluster.

Hope this point help you to understand our current use case and how we arrive at using code for deployment.

[github-actions]

This issue is stale because it has been open 30 days with no activity. To keep this issue open remove stale label or comment.

[milesgranger]

This is similar but slightly different, but I for one would love be able to modify pull_steps during an implementation of BaseWorker.run method. 🤞

[tekumara]

Hitting this too, eg:

param_s3: Deployment = Deployment.build_from_flow(
    name="s3",
    flow=flows.param_flow.param,
    work_pool_name="kubes-pool",
    pull_steps=[
        {
            "prefect_aws.deployments.steps.pull_from_s3": {
                "requires": "prefect-aws>=0.3.0",
                "bucket": bucket,
                "folder": folder,
            }
        }
    ],
)

Errors with

...
  File "/Users/tekumara/code/prefect-demo/.venv/lib/python3.10/site-packages/prefect/deployments/deployments.py", line 787, in build_from_flow
    deployment = cls(name=name, **kwargs)
  File "/Users/tekumara/code/prefect-demo/.venv/lib/python3.10/site-packages/prefect/_internal/compatibility/experimental.py", line 230, in __init__
    cls_init(__pydantic_self__, **data)
  File "pydantic/main.py", line 341, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for Deployment
pull_steps
  extra fields not permitted (type=value_error.extra)

[mmartsen]

we need this too. our use case is slightly different, we deploy only one deployment, which is kind-of "template", and the rest of deployments are done by the users as copies of that template with schedule and input parameters modifications. So we need to be able to programmatically create/update pull steps thru the api

[ionicsolutions]

We'd also like to add/change pull_steps programmatically. We're creating Prefect deployments through a Python script that hides complexity from our users.

[marcm-ml]

I think this is resolved by using the RunnerDeployment class in prefect.deployments.runner. You could achieve the desired functionality and use apply() to register the Deployment with UI including pull_steps. The question is, if this is a stable Api and the intended "prefect" way. Since this is also used in the Flow.serve() and Flow.deploy() methods, I would assume that you can safely use this atm

[carlosjourdan]

In my case, I'm building deployments programaticaly using Deployment.build_from_flow, and would only seem logical that this allows for a pull_steps parameter, but it doesn't.

I don't see how I could use @marcm-ml's suggestion of using RunnerDeployment given that RunnerDeployment.from_flow also does not accept pull_steps as a parameter.

[marcm-ml]

You must provide the storage keyword in the RunnerDeployment.from_* methods. There are RemoteStorage (prefect.runner.storage.RemoteStorage) and Git Repository (prefect.runner.storage.GitRepository) classes available that implement the correct storage API the runner deployment expects. These storage classes implement the pull_steps and set them in the runner deployment

[carlosjourdan]

Thanks @marcm-ml. But what I'm trying to accomplish is to run a shell script before the flow startup, using the prefect.deployment.steps.utility module. My flows run on local storage, and from the docs I assumed I would be able to insert this action on the pull_steps list without needing to tinker with the storage. Am I wrong?

[marcm-ml]

Mh without too much context I can't suggest to much but maybe you can get away with implementing the Protocol in prefect.runner.storage via a custom class where you implement the methods of the protocol with the things you want it to do, e.g. executing a shell script. I hope this helps