--gpus

task

pytest

result=

/opt/prefect/healthcheck.py:151: UserWarning: Flow uses module which is not importable. Refer to documentation on how to import custom modules <https://docs.prefect.io/api/latest/storage.html#docker>
  flows = cloudpickle_deserialization_check(flow_file_paths)
Traceback (most recent call last):
  File "/opt/prefect/healthcheck.py", line 151, in <module>
    flows = cloudpickle_deserialization_check(flow_file_paths)
  File "/opt/prefect/healthcheck.py", line 44, in cloudpickle_deserialization_check
    flows.append(cloudpickle.loads(flow_bytes))
ModuleNotFoundError: No module named 'snowflake.sqlalchemy'

flow.storage = Docker(registry_url="xxx", image_name="xxx",
                          python_dependencies=["pandas", "oauthlib ", "requests", "requests_oauthlib", "oauth2client",
                                               "snowflake-connector-python", "pyarrow", "fastparquet"],
                          secrets=["GCP_CREDENTIALS"], files={"C:/Users/laura.vaida.000/anaconda3/envs/prefect/Lib/site-packages/sqlalchemy": "/modules/sqlachemy.py"})

KeyError: 'Task slug []-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did you change the flow without re-registering it?\n- Did you register the flow without updating it in your storage location (if applicable)?'

subscription {
  flow_run_state_by_pk(id: "848a9771-4962-405b-86f1-2d4561ffffff") {
    state
  }
}

result:

{
  "errors": [
    {
      "path": [
        "flow_run_state_by_pk"
      ],
      "message": "Cannot read property 'flow_run_state_by_pk' of undefined",
      "extensions": {
        "code": "INTERNAL_SERVER_ERROR"
      }
    }
  ],
  "data": {
    "flow_run_state_by_pk": null
  }
}

import prefect
from prefect import Flow, task
from prefect.tasks.kubernetes.job import RunNamespacedJob


start_body = {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
        "name": "start"
    },
    "spec": {
        "template": {
            "spec": {
                "containers": [
                    {
                        "name": "start",
                        "image": "alpine",
                        "command": [
                            "echo",
                            "start"
                        ]
                    }
                ],
                "restartPolicy": "Never"
            }
        },
        "backoffLimit": 4
    }
}

@task
def print_kube_res(x):
    print(str(x))

with Flow("Kubernetes Job") as flow:
    start = RunNamespacedJob(body=start_body, kubernetes_api_key_secret=None)
    p = print_kube_res(start)
    start.set_downstream(p)

registry_url

ok, i tried with the following:

@task(log_stdout=True)
def create_engine(snowflake_salesforce):
    config = configparser.ConfigParser()
    engine=create_engine(URL(**config[snowflake_salesforce]))
with Flow('UWG-Mail') as flow:
    snowflake_credentials=PrefectSecret("Snowflake_Salesforce")
    connection=create_engine(snowflake_salesforce=snowflake_credentials)

Unexpected error: TypeError("unhashable type: 'dict'")
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 863, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 298, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "<input>", line 36, in create_engine
  File "/usr/local/lib/python3.8/configparser.py", line 959, in __getitem__
    if key != self.default_section and not self.has_section(key):
  File "/usr/local/lib/python3.8/configparser.py", line 668, in has_section
    return section in self._sections
TypeError: unhashable type: 'dict'

Parameter

DatabricksRunNow

Task

notebook_param

Github Storage

KubernetesRun

pip install -e .

/tasks

my-happy-project/
├── flows/
│   ├── flow_1.py
│   └── flow_2.py
│
└── tasks/
    ├── shared_stuff_1.py
    └── shared_stuff_2.py

get_etl_flow

register_flow

run_flow

get_etl_flow

register_flow

register_flow

run_flow

register_flow

with case(results, None):
    return # End the Flow

with case(cond, "a"):
        run_if_cond_is_a()

    with case(cond, "b"):
        run_if_cond_is_b()