prefect agent docker start --token <token>--name "EMG-DCK-01" --volume company:/mnt/company --label production --log-level DEBUG --show-flow-logs

docker run -it -v company:/mnt/company node:lts bash

cd /mnt/company
ls

**shows all files**

PS C:\Users\azure.automation> docker volume inspect company
[
    {
        "CreatedAt": "2021-07-17T22:43:31Z",
        "Driver": "local",
        "Labels": {},
        "Mountpoint": "/var/lib/docker/volumes/company/_data",
        "Name": "company",
        "Options": {
            "device": "//192.168.1.2/Company",
            "o": "user=myusername,domain=mydomain,password=mypassword",
            "type": "cifs"
        },
        "Scope": "local"
    }
]

Traceback (most recent call last):
  File "runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "\executors.py", line 537, in prepare_upstream_states_for_
mapping
    value = upstream_state.result[i]
KeyError: 0
[2021-07-19 18:23:21+0200] ERROR - prefect.ETL | Unexpected error occured in FlowRunner: KeyError(0)

DaskExecutor

dask_kubernetes.KubeCluster

./test_prefect

## test_prefect.py
result = LocalResult(dir="./test_prefect")

@task(target="func_task_target.txt", checkpoint=True, result = result)
def test(val):
    return val

with Flow("simple_test",result=result) as test_flow:
    result = test(99)
    

test_flow.run()

PREFECT__FLOWS__CHECKPOINTING=true

python test_prefect.py

Error grabbing logs: unexpected EOF

{map_index}

first_map_fn[0]

second_map_fn[0]

first_map_fn[1]

run_always

prefect agent docker start --key "SERVICE_ACCOUNT_KEY" --name "AGENT_NAME"

docker.errors.APIError: 400 Client Error for <http+docker://localhost/v1.40/containers/create?name=nickel-okapi>: Bad Request ("invalid IP address in add-host: "host-gateway"")

wait=True

StartFlowRun

StartFlowRun

wait=False

Cloud Hook

success

Psst! We have recently added Automations which offer more functionality than Cloud Hooks and will eventually replace them.

def post_to_slack(task, old_state, new_state):
    # <https://github.com/PrefectHQ/prefect/blob/d61fa6aac9330c5817cc8e8b8f8cca2d634ea7e1/src/prefect/engine/state.py>
    if new_state.is_retrying():
        msg = f"<@XXXX> {os.path.basename(__file__)} | {task} | {new_state}"
        <http://requests.post|requests.post>("<https://hooks.slack.com/services/>", json={"text": msg})

    if new_state.is_failed():
        msg = f"<@XXXX> {os.path.basename(__file__)} | {task} | {new_state}"
        <http://requests.post|requests.post>("<https://hooks.slack.com/services/>", json={"text": msg})

    # if new_state.is_successful():
    #     msg = f"<@XXXX> {os.path.basename(__file__)} | {task} | {new_state}"
    #     <http://requests.post|requests.post>("<https://hooks.slack.com/services/>", json={"text": msg})

    return new_state

Failed to load and execute Flow's environment: ValidationError({'type': ['Unsupported value: Git']})

import prefect
from prefect.storage import Git
from prefect.run_configs import ECSRun, LocalRun
from prefect import task, Flow, Parameter
from prefect.client import Secret

RUN_CONFIG = ECSRun(image='image/image',
                    cpu='1 vcpu', memory='2 GB')
                  
STORAGE = Git(repo='name/repo', flow_path='path_to_this_file', git_token_secret_name='token')

@task
def say_hello():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hi")
    
with Flow("git-storage", storage=STORAGE, run_config=RUN_CONFIG) as flow:
    say_hello()