20:53:32.166 | INFO    | prefect.agent - Submitting flow run 'adf8ae6e-675d-40dd-8a32-078aecbc248b'
20:53:32.359 | ERROR   | prefect.agent - Failed to submit flow run 'adf8ae6e-675d-40dd-8a32-078aecbc248b' to infrastructure.

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/agent.py", line 200, in submit_run
    await self.task_group.start(submit_flow_run, flow_run, infrastructure)
  File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 807, in start
    return await future
  File "/usr/local/lib/python3.9/site-packages/prefect/infrastructure/submission.py", line 47, in submit_flow_run
    infrastructure = _prepare_infrastructure(flow_run, infrastructure)
  File "/usr/local/lib/python3.9/site-packages/prefect/infrastructure/submission.py", line 35, in _prepare_infrastructure
    "env": {**base_flow_run_environment(flow_run), **infrastructure.env},
AttributeError: 'S3' object has no attribute 'env'

from prefect.tasks.notifications import SlackTask
class Alert:

    def alert_on_special_failure(task, old_state, new_state):
        if new_state.is_failed():
            if getattr(new_state.result, "flag", False) is True:
                errMsg = '--- LOADER ERROR ---'
                msg = "{}\nTask: `{}` FAILED.\nThe loader process failed: `{}`".format(
                        errMsg, task.name, new_state.result.value)
                SlackTask().run(message = msg, webhook_secret = "SLACK_PREFECT_DEV")
        return new_state

@task(name = 'send API request',
      max_retries = 3,
      retry_delay = timedelta(minutes = 5),
      state_handlers = [alert_on_special_failure])
def post_request_process(

git clone --depth 1 <https://github.com/PrefectHQ/prefect.git>
cd prefect/examples/tutorial

ofir@Ofirs-MacBook-Pro-2 prefect % ls examples/
ls: examples/: No such file or directory
ofir@Ofirs-MacBook-Pro-2 prefect %

ofir@Ofirs-MacBook-Pro-2 prefect % git checkout 1.x
Branch '1.x' set up to track remote branch '1.x' from 'origin'.
Switched to a new branch '1.x'
ofir@Ofirs-MacBook-Pro-2 prefect % ls examples
conditional.py  kafka           mapping.py      old             parameters.py   tutorial
ofir@Ofirs-MacBook-Pro-2 prefect % ls examples/tutorial
01_etl.py                       03_parameterized_etl_flow.py    05_schedules.py                 aircraftlib
02_etl_flow.py                  04_handle_failures.py           06_parallel_execution.py        requirements.txt
ofir@Ofirs-MacBook-Pro-2 prefect %

git clone --branch 1.x --depth 1 <https://github.com/PrefectHQ/prefect.git>
cd prefect/examples/tutorial

to_csv()

file-path = Path("data/fresh-data.csv”)

file-path.parent.mkdir(parents=True,exist_ok=True)

prefect<2.0

import prefect
import logging
logger = logging.getLogger(__name__)

@prefect.flow
def f():
    <http://logger.info|logger.info>("test") # doesn't appear

@flow(name="always-fail")
def entrypoint():
    raise ValueError("my error message")

if __name__ == "__main__":
    state = entrypoint(return_state=True)
    print(state)
    print(state.result)

state.result

prefect.exceptions.MissingContextError: Logger 'dystematic.pdbt.ftp' attempted to send logs to Orion without a flow run id. The Orion log handler can only send logs within flow run contexts unless the flow run id is manually provided.

RuntimeError: There is no active flow or task run context.

UserWarning: Block document has schema checksum sha256:dbeeaf09aa78947a7c576549b11e098c00a25bcbbf90b8d8b70c0c3a3fc8f4a2 which does not match the schema checksum for class 'Process'. This indicates the schema has changed and this block may not load.