from project_name.tasks import …

prefect register

def cancel(obj: Union[Task, Flow], _: State, new_state: State) -> State:
    if isinstance(new_state, (Cancelling, Cancelled)):
        obj.some_arg.cancel() # let them know we're no longer waiting

    return new_state


class RemoteTask(Task):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        state_handlers = kwargs.pop('state_handlers', [])
        state_handlers.append(cancel)
        kwargs['state_handlers'] = state_handlers
        super().__init__(*args, **kwargs)

server

server

slack_notifier

PREFECT__SERVER__UI__ENDPOINT

[server]
  [server.ui]
  endpoint = "..."

"unauthorized. invalid prefect cloud api key"

The following error messages were provided by the GraphQL server:
          
            INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
                 "input.tenant_id"; Expected non-nullable type UUID! not to be null.

[server]
host = "<http://mypublicIP>"
port = "4200"
host_port = "4200"
endpoint = "${server.host}:${server.port}"
    [server.ui]
    host = "<http://mypublicIP>"
    port = "8080"
    host_port = "8080"
    host_ip = "mypublicIP"
    endpoint = "${server.ui.host}:${server.ui.port}"
    apollo_url = "<http://mypublicIP:4200/graphql>"

ERROR: for tmp_ui_1  Cannot start service ui: driver failed programming external connectivity on endpoint tmp_ui_1 (1f380ce2f089ba70a8c422d2302bb447fef87c48bdd81d2dd93898b661c661a0): Error starting userland proxy: listen tcp4 xx.xx.xx.xx:8080: bind: cannot assign requested address

docker build my_processor .
docker run -v ${PWD}/data:/app/data my_processor

FROM python:3.7-slim
WORKDIR /app
COPY . .
RUN pip install .
CMD ["python3", "preprocess.py"]

flow.py

from prefect import Flow, task
...

@task(name="Get data")
def get_data_from_s3():
    """Retrieve csv from S3 and saves as /data/raw/file.csv"""
    pass


@task(name="Process data")
def process_through_docker_image():
    """Start docker container which
        - reads the csv-file
        - processes the data
        - writes as json
    """
    pass

@task(name="Push data")
def upload_to_s3():
    """Push data to S3"""
    pass

with Flow("foo") as f:
    # First get data: get_data_from_s3()
    ...

    # Then process the data: process_through_docker_image()
    ...

    # Finally, upload the processed data to S3: upload_to_s3()
    ...

f.register()

Hey everyone, I have an issue. We're getting this error: "Unexpected error: ClientError(\'An error occurred (ThrottlingException) when calling the DescribeTasks operation (reached max retries: 4): Rate exceeded\')"

ImportError: cannot import name 'apply' from 'dask.compatibility' (/usr/local/lib/python3.9/site-packages/dask/compatibility.py)

prefect server start --postgres-url postgres://<username>:<password>@hostname:<port>/<dbname>

prefect-server database upgrade

prefect server start --external-postgres --no-upgrade

prefect-server database upgrade

Error: HTTPConnectionPool(host='localhost', port=3000): Max retries exceeded with url: /v1/query (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f6aaf865730>: Failed to establish a new connection: [Errno 111] Connection refused'))

PREFECT__ENGINE__EXECUTOR__DEFAULT_CLASS

dask-gateway==0.9.0
dask==2020.12.0
distributed==2020.12.0
prefect==0.14.1
click==7.1.2

gevent

gevent

ssl

import gevent.monkey
gevent.monkey.patch_all()

AZURE_STORAGE_CONNECTION_STRING

prefectConfig

PREFECT__CONTEXT__SECRETS__AZURE_STORAGE_CONNECTION_STRING

github_repo/
├── flow.py
├── flow_dependencies/sub_task.py

import flow_dependencies.saubtask

flow.py

flow.storage = Git(...)