ephemeral-prefect/api

from prefect import get_client
from prefect.server.schemas.states import State, StateType
from prefect.server.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, FlowRunFilterStateType
import asyncio

async def get_cancelling_flows():
    async with get_client() as client:
        
        state_filter = FlowRunFilter(state=FlowRunFilterState(type=FlowRunFilterStateType(any_=['CANCELLED'])))
        # Query all flow runs
        flow_runs = await client.read_flow_runs(flow_run_filter=state_filter)
        cancelled_state = State(type=StateType.CANCELLED)

        # Loop through each flow run and change its state to "CANCELLED"
        for flow in flow_runs:
            try:
                await client.set_flow_run_state(
                    flow_run_id=flow.id,
                    state=cancelled_state,
                    force=True
                )
                print(f"Flow run {flow.id} has been set to CANCELLED.")
            except Exception as e:
                print(e)

# Run the async function to get the cancelling flows
asyncio.run(get_cancelling_flows())

prefect deploy

my_flows

prefect.yaml

name: my_flows
prefect-version: 2.11.3

build:
  - prefect.deployments.steps.run_shell_script:
      id: get-commit-hash
      script: git rev-parse --short HEAD

  - prefect_docker.deployments.steps.build_docker_image:
      id: build-image
      requires: prefect-docker>=0.3.0
      image_name: my_repo/my_flows
      tag: "{{ get-commit-hash.stdout }}"

push:
  - prefect_docker.deployments.steps.push_docker_image:
      requires: prefect-docker
      image_name: "{{ build-image.image_name }}"
      tag: "{{ build-image.tag }}"
      credentials: "{{ prefect.blocks.docker-registry.docker-registry }}"

pull:
- prefect.projects.steps.set_working_directory:
    directory: /opt/prefect/my_flows

deployments:
  - name: process-lyrics-deployment
    version:
    tags: ["process-lyrics"]
    description:
    schedule: {}
    flow_name:
    entrypoint: process_lyrics/main.py:process_lyrics
    parameters:

    work_pool:
      name: pool-1
      work_queue_name:
      job_variables:
        image: "{{ image_name }}"

FROM prefecthq/prefect:2.11.3-python3.8

COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt
COPY . /opt/prefect/my_flows/

/opt/prefect/my_flows/

FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/my_flows'

prefect:2.11.3-python3.8

pullPolicy: Always

build:
      - prefect_docker.deployments.steps.build_docker_image:
          id: build_image
          requires: prefect-docker>=0.3.0
          image_name: <http://724930167716.dkr.ecr.eu-west-2.amazonaws.com/prefect-training-2023|724930167716.dkr.ecr.eu-west-2.amazonaws.com/prefect-training-2023>
          tag: 0.0.1
          dockerfile: Dockerfile
          push: true
          platform: linux/amd64

from prefect import flow, task, Parameter

ImportError                               Traceback (most recent call last)
<command-1880743847103486> in <module>
      1 from typing import Any, Awaitable, Dict, List, Optional, Union
----> 2 from prefect import flow, task
      3 from prefect import Parameter
      4 from typing import Any
      5 from datetime import timedelta

/databricks/python_shell/dbruntime/PythonPackageImportsInstrumentation/__init__.py in import_patch(name, globals, locals, fromlist, level)
    165             # Import the desired module. If you're seeing this while debugging a failed import,
    166             # look at preceding stack frames for relevant error information.
--> 167             original_result = python_builtin_import(name, globals, locals, fromlist, level)
    168 
    169             is_root_import = thread_local._nest_level == 1

/databricks/python/lib/python3.8/site-packages/prefect/__init__.py in <module>
     35 
     36 # Import user-facing API
---> 37 from prefect.states import State
     38 from prefect.logging import get_run_logger
     39 from prefect.flows import flow, Flow

/databricks/python_shell/dbruntime/PythonPackageImportsInstrumentation/__init__.py in import_patch(name, globals, locals, fromlist, level)
    165             # Import the desired module. If you're seeing this while debugging a failed import,
    166             # look at preceding stack frames for relevant error information.
--> 167             original_result = python_builtin_import(name, globals, locals, fromlist, level)
    168 
    169             is_root_import = thread_local._nest_level == 1

/databricks/python/lib/python3.8/site-packages/prefect/states.py in <module>
     10 import httpx
     11 import pendulum
---> 12 from typing_extensions import TypeGuard
     13 
     14 from prefect.client.schemas import State as State

ImportError: cannot import name 'TypeGuard' from 'typing_extensions' (/databricks/python/lib/python3.8/site-packages/typing_extensions.py)

@flow(log_prints=True)
def myflow(param: str):
    print(f'flow executed with param={param}')

CRASHED

CRASHED

RUNNING

2.8.7

CRASHED

Task run '...' already finished