Hi all, How would I go about running a pipline that has to run different but dependent steps in different images? So for example I might have Step 1(task or flow) run in image 1, and then if successful Step 2 run in image 2? Here's a yaml I have so far:

pull:
- prefect.projects.steps.git_clone_project:
    repository: myscmprovider/myrepo.git
    branch: main
    access_token: '{{ prefect.blocks.secret.token }}'
- prefect.deployments.steps.set_working_directory:
    id: chdir
    directory: /opt/prefect/myrepo/k8-multi-container/

deployments:
  - name: deployment-1
    entrypoint: .\workflow\flow_demo.py:flow_1
    work_pool:
        name: aks-worker
        job_variables:
            image: myimage_1
            cpu_request: "4"
            cpu_limit: "6"
            memory_request: "4Gi"
            memory_limit: "6Gi"

  - name: deployment-2
    entrypoint: .\workflow\another_flow.py:flo_2
    work_pool:
        name: aks-worker
        job_variables:
            image: myimage_2
            cpu_request: "4"
            cpu_limit: "6"
            memory_request: "4Gi"
            memory_limit: "6Gi"

But I don't understand completely how these two deployments are linked? How does the dependency between flow 1 and flow 2 get established? They don't need to pass data, but flow 2 shouldn't run if flow 1 failed. And how would I kick it off all at once? Right now I do prefect deployments run to run things.

hi @YSF - you wouldnt establish the dependency in the definition of the deployment, you would do that in the implementation of a flow that conditionally invokes those deployments with

run_deployment

for example

from prefect import flow
from prefect.client.schemas.core import FlowRun
from prefect.deployments import run_deployment

@flow
def parent_flow():
   flow_run_1: FlowRun = run_deployment("flow-1/deployment-1")

   if flow_run_1.state.is_completed():
      flow_run_2: FlowRun = run_deployment("flow-2/deployment-2")

Hey @Nate thank you! This makes sense, it threw me a no module found named prefect.client.schemas.core, and I went to the api reference and the page is just empty back until 2.10.8? Is this being deprecated? Or does it just live somewhere else now? https://docs.prefect.io/2.10.21/api-ref/prefect/client/schemas/

whoops, i was just typing that in slack so there was no LSP to catch me I would use the client's schema

from prefect.client.schemas.objects import FlowRun