from prefect import flow, get_run_logger
from prefect.deployments import Deployment
from prefect.filesystems import RemoteFileSystem
from prefect.infrastructure import DockerContainer


@flow
def my_docker_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello from Docker!")


minio_block = RemoteFileSystem(
    basepath="<s3://prefect-flows/flows>",
    key_type="hash",
    settings=dict(
        use_ssl = False,
        key = "blablabla",
        secret = "blablabla",
        client_kwargs = dict(endpoint_url = "<http://minio:9000>")
    ),
)
minio_block.save("minio", overwrite=True)


deployment = Deployment.build_from_flow(
    name="docker-example",
    flow=my_docker_flow,
    storage=RemoteFileSystem.load('minio'),
    infrastructure=DockerContainer(
        image = 'prefect-orion:2.1.1',
        image_pull_policy = 'IF_NOT_PRESENT',
        networks = ['prefect'],
        env = {
            "USE_SSL": False,
            "AWS_ACCESS_KEY_ID": "blablabla",
            "AWS_SECRET_ACCESS_KEY": "blablabla",
            "ENDPOINT_URL": '<http://minio:9000>',
        }
    ),
)
deployment.apply()

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 839, in exec_module
  File "<frozen importlib._bootstrap_external>", line 975, in get_code
  File "<frozen importlib._bootstrap_external>", line 1032, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/flows/test_flow.py'

packager

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 839, in exec_module
  File "<frozen importlib._bootstrap_external>", line 975, in get_code
  File "<frozen importlib._bootstrap_external>", line 1032, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/flows/test_flow.py'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 246, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.8/site-packages/prefect/client.py", line 105, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/deployments.py", line 73, in load_flow_from_flow_run
    flow = await run_sync_in_worker_thread(import_object, str(import_path))
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 56, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/usr/local/lib/python3.8/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/importtools.py", line 193, in import_object
    module = load_script_as_module(script_path)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/importtools.py", line 156, in load_script_as_module
    raise ScriptError(user_exc=exc, path=path) from exc
prefect.exceptions.ScriptError: Script at '/flows/test_flow.py' encountered an exception

botocore.exceptions.EndpointConnectionError: Could not connect to the endpoint URL: "<http://minio:9000/prefect-flows?list-type=2&prefix=flows%2F&delimiter=&encoding-type=url>"