Hi, In previous versions (tried 2.0b4) it was possible to create an s3 storage block backed by minio, using the

/api/blocks

endpoint. In the 2.0b8 api I see endpoints for block schemas, block_types and block_document, but as far as I can tell none of these create a block? Through the UI, it is not possible to supply the necessary

client_kwargs

and

config_kwargs

. Is this no longer possible, or am I misunderstanding something? For reference I used a request like this:

PAYLOAD=$(cat <<EOF
{
    "name": "minio",
    "block_spec_id": ${FILE_STORAGE_ID},
    "data": {
        "base_path": "<s3://prefect-flows>",
        "key_type": "hash",
        "options": {
            "use_ssl": false,
            "key": "blablabla",
            "secret": "blablabla",
            "client_kwargs": {"endpoint_url": "<http://minio:9000>"},
            "config_kwargs":{"signature_version": "s3v4"},
        }
    }
}
EOF
)
BLOCK_ID=$(echo -n "$PAYLOAD" | curl -vs -XPOST -H "Content-Type: application/json" <http://localhost:4200/api/blocks/> -d@- | jq -r '.id')

Storage blocks have been simplified and will likely still undergo some changes until the final GA release - check out the file systems docs https://orion-docs.prefect.io/concepts/filesystems/; specifically, RemoteFileSystem is what you're looking for to work with S3/minio

Ok, thank you! But is it possible to use a RemoteFileSystem for the flow code? From the documentation, it appears they let you interact with, say minio, from inside the flow. E.g for saving/loading data needed in a flow That is not what I am after. I would like to use minio as a storage backend, for storing flow code, which is what used to be possible. I do not see anything in either the Storage or FileSystems sections about using RemoteFileSystem as a storage backend in this manner.

here is an example:

from prefect import task, flow
from prefect import get_run_logger
from prefect.deployments import Deployment
from prefect.packaging import FilePackager
from prefect.filesystems import RemoteFileSystem


@task
def say_hi(user_name: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello %s!", user_name)


@flow
def hello(user: str = "world"):
    say_hi(user)


Deployment(
    flow=hello,
    name="s3_file_packager_with_remote_s3fs",
    packager=FilePackager(
        filesystem=RemoteFileSystem(basepath="<s3://prefect-orion/flows/>")
    ),
)


if __name__ == "__main__":
    hello(user="from Slack")

Thank you I will try it out!