__init__(__self__, *args, **kwargs)

File "C:\Users\Administrator\anaconda3\envs\prefect_test\Lib\site-packages\prefect\flows.py", line 183, in __init__

self.parameters = parameter_schema(self.fn)

^^^^^^^^^^^^^^^^^^^^^^^^^

File "C:\Users\Administrator\anaconda3\envs\prefect_test\Lib\site-packages\prefect\utilities\callables.py", line 306, in parameter_schema

model = pydantic.create_model("Parameters", __config__=ModelConfig, **model_fields)

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "pydantic\main.py", line 1026, in pydantic.main.create_model

File "pydantic\main.py", line 197, in pydantic.main.ModelMetaclass.__new__

File "pydantic\fields.py", line 506, in pydantic.fields.ModelField.infer

File "pydantic\fields.py", line 436, in pydantic.fields.ModelField.__init__

File "pydantic\fields.py", line 546, in pydantic.fields.ModelField.prepare

File "pydantic\fields.py", line 570, in pydantic.fields.ModelField._set_default_and_type

File "pydantic\fields.py", line 439, in pydantic.fields.ModelField.get_default

File "pydantic\utils.py", line 693, in pydantic.utils.smart_deepcopy

File "C:\Users\Administrator\anaconda3\envs\prefect_test\Lib\copy.py", line 172, in deepcopy

y = _reconstruct(x, memo, *rv)

^^^^^^^^^^^^^^^^^^^^^^^^^^

File "C:\Users\Administrator\anaconda3\envs\prefect_test\Lib\copy.py", line 273, in _reconstruct

y.__setstate__(state)

File "C:\Users\Administrator\anaconda3\envs\prefect_test\Lib\site-packages\prefect_dask\task_runners.py", line 330, in __setstate__

self._client = distributed.get_client()

^^^^^^^^^^^^^^^^^^^^^^^^

File "C:\Users\Administrator\anaconda3\envs\prefect_test\Lib\site-packages\distributed\worker.py", line 2782, in get_client

raise ValueError("No global client found and no address provi

PREFECT_TASKS_REFRESH_CACHE

PREFECT_TASKS_REFRESH_CACHE

python flows/flow.py

if __name__ == "__main__":
    my_flow(
        parameter_1=value_1,
        parameter_2=value_2,
        parameter_3=value_3,
    )

my_flow

@task(result=PrefectResult())
def return_results(table_name):
    return {"table_name": table_name)

trigger=all_finished:

@task(result=PrefectResult(), trigger=all_finished)
def return_results(table_name):
    return {"table_name": table_name)

/tmp

venv

@flow(name="whatever")

google.auth._default - Authentication failed using Compute Engine authentication due to unavailable metadata server.

Anonymous caller does not have storage.objects.create access to the Google Cloud Storage object. Permission 'storage.objects.create' denied on resource (or it may not exist)., 401

on_failure

Timeout context manager should be used inside a task

apply_deployment

prefect 2.8.0

TypeError: Type is not JSON serializable: <class name of the object>

run_deployment

run_deployment

@task()
def run_worker(name, flow_run_name, etc....):
    flow_run = run_deployment(
        name=name,
        flow_run_name=flow_run_name,
        timeout=timeout,
        tags=tags,
        poll_interval=poll_interval,
        parameters=parameters,
    )
    if flow_run.is_failed():
        raise FailedRun(message=f"Subflow {flow_run_name} failed to complete successfully.")

Error getting config status, workload certificates may not be configured: HTTP 404

all_finished

@flow
def flow():
   _do_something = do_something()
   _do_something2 = do_something2()
   _do_something3 = do_something3()
   _always_run = always_run(
          wait_for=[
             allow_failure(_do_something),
             allow_failure(_do_something2),
             allow_failure(_do_something3)
           ]
          )
    return _always_run

do_something

always_run

DEPLOYMENT_NAME = "local-docker-deployment"
#Import your dag definition
from dag import augmentation_pipeline
#Import prefect deployment spec library
from prefect.deployments import Deployment
from prefect.infrastructure.docker import DockerContainer
from prefect.filesystems import LocalFileSystem

docker_container_block = DockerContainer.load("example-prefect")
#fs = LocalFileSystem(basepath='C:\\Users\\ykadmin081021\\Documents\\Projects\\del2\\')
fs = LocalFileSystem.load("test-ysf")

deployment = Deployment.build_from_flow(
    flow=augmentation_pipeline,
    name=DEPLOYMENT_NAME,
    work_queue_name="test",
    infrastructure=docker_container_block,
    storage=fs
)


if __name__ == "__main__":
    deployment.apply()

FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/C:\\Users\\ykadmin081021\\Documents\\Projects\\del2\\'

/opt/prefect/

COPY