Short answer: yes — but not via dedicated attributes like

.region

. After you

rfs = RemoteFileSystem.load("name")

, you can read anything you stored on the block via: -

rfs.basepath

(e.g., `s3://my-bucket/prefix`) -

rfs.settings

(a dict passed straight through to fsspec/s3fs) For S3, region/DNS/endpoint-style are set in

settings

(there’s no separate

.region

field). Typical S3 options go under

client_kwargs

and

config_kwargs

for s3fs:

from prefect.filesystems import RemoteFileSystem

# Create and save
rfs = RemoteFileSystem(
    basepath="<s3://my-bucket/prefix>",
    settings={
        # Passed to boto3 client via s3fs
        "client_kwargs": {
            "region_name": "us-west-2",
            "endpoint_url": "<https://s3.my-company.internal>",  # custom DNS/endpoint
        },
        "use_ssl": True,
        "requester_pays": False,
        # Control path vs virtual-hosted–style addressing
        "config_kwargs": {"s3": {"addressing_style": "virtual"}},  # or "path"
    },
)
rfs.save("my-s3")

# Load and access later
rfs = RemoteFileSystem.load("my-s3")
region = rfs.settings.get("client_kwargs", {}).get("region_name")
endpoint = rfs.settings.get("client_kwargs", {}).get("endpoint_url")
addressing_style = (
    rfs.settings.get("config_kwargs", {}).get("s3", {}).get("addressing_style")
)

Notes: - You can also access the underlying fsspec filesystem via

rfs.filesystem

(for S3 this is an

s3fs.S3FileSystem

), but the most reliable place to read/write these options is

rfs.settings

. - AWS creds are typically picked up from environment variables; you can also pass

key

,

secret

,

token

in

settings

if needed. If you’re using something other than S3, you’ll still use

rfs.settings

— just pass the kwargs supported by that fsspec implementation. Want me to help you set the exact settings for your S3 endpoint? If so, which region and what kind of DNS/endpoint are you targeting (AWS S3 vs S3-compatible like MinIO)?

Short answer: not directly.

RemoteFileSystem

doesn’t define

exists()

. Use the underlying fsspec filesystem: -

rfs.filesystem.exists(path)

- Other utilities are also on `rfs.filesystem`:

glob

,

ls

,

isfile

,

isdir

, etc. Example:

rfs = RemoteFileSystem.load("my-s3")
# Check existence (use a fully-qualified path or resolve relative to basepath)
rfs.filesystem.exists("<s3://my-bucket/prefix/file.txt>")
# or relative to the basepath:
rfs.filesystem.exists(rfs._resolve_path("file.txt"))

Note:

read_path

,

write_path

,

get_directory

, and

put_directory

are provided on

RemoteFileSystem

itself, but exploration helpers like

exists()

are on the fsspec object at

rfs.filesystem

.

Yes, with one caveat. You can do:

s3 = rfs.filesystem
s3.exists("my/path/to/file")  # works only if this is a fully-qualified path for the FS

For S3 specifically, fsspec/s3fs expects a bucket in the path. So either: - Pass a full path:

s3.exists("<s3://my-bucket/prefix/my/path/to/file>")
# or "my-bucket/prefix/my/path/to/file"

- Or resolve relative to your block’s basepath:

s3.exists(rfs._resolve_path("my/path/to/file"))  # resolves against rfs.basepath

Tip:

rfs._resolve_path(...)

is an internal helper but handy for making relative paths work with your

basepath

.

Yes —

rfs.basepath

is the configured base path string. Example:

rfs = RemoteFileSystem.load("my-s3")
print(rfs.basepath)  # e.g., "<s3://my-bucket/prefix>"

If you want to build a full path for fsspec calls:

full_path = f"{rfs.basepath.rstrip('/')}/my/path/to/file"
rfs.filesystem.exists(full_path)

You can also use the helper that resolves relative paths against the basepath:

rfs.filesystem.exists(rfs._resolve_path("my/path/to/file"))

Short answer: RemoteFileSystem (RFS) can create folders/prefixes, but it cannot create the bucket/container itself. In Prefect 3.x, you can absolutely create the bucket from within a flow—use the cloud SDK (e.g., boto3 for S3) in a task, then continue using RFS or a Prefect storage block. Here’s a concise S3 example using prefect-aws to check/create, then use RFS:

from prefect import flow, task
from prefect_aws import AwsCredentials
from prefect.filesystems import RemoteFileSystem
from botocore.exceptions import ClientError

@task
def ensure_s3_bucket(bucket_name: str, region: str = "us-east-1", aws_creds: AwsCredentials | None = None) -> bool:
    """
    Returns True if created, False if already existed.
    """
    session = aws_creds.get_boto3_session() if aws_creds else None
    s3 = (session.client("s3", region_name=region) if session else __import__("boto3").client("s3", region_name=region))

    # Check existence
    try:
        s3.head_bucket(Bucket=bucket_name)
        return False  # exists
    except ClientError as e:
        code = e.response.get("Error", {}).get("Code", "")
        if code not in ("404", "NoSuchBucket"):
            # 403 usually means it exists but not owned by you—surface the error
            if code == "403":
                raise
            # otherwise proceed to creation
            pass

    # Create if missing
    if region == "us-east-1":
        s3.create_bucket(Bucket=bucket_name)
    else:
        s3.create_bucket(
            Bucket=bucket_name,
            CreateBucketConfiguration={"LocationConstraint": region},
        )
    return True

@flow
def ensure_bucket_then_use_rfs(bucket: str, region: str = "us-east-1", prefix: str = "data/"):
    # Load AWS credentials block if you have one saved
    aws_creds = AwsCredentials.load("my-aws-creds")  # or construct AwsCredentials(...)
    created = ensure_s3_bucket(bucket, region, aws_creds)
    print(f"Bucket created? {created}")

    # Now use RFS normally (RFS will create folders/prefixes as needed)
    rfs = RemoteFileSystem(basepath=f"s3://{bucket}/{prefix}", settings=aws_creds.dict())
    rfs.write_path("hello.txt", b"hello world")

Notes - RFS auto-creates prefixes (directories), not buckets. - For sustained S3 I/O, consider the

prefect-aws

S3Bucket block for convenience. It does not create buckets either, but it simplifies reads/writes once the bucket exists. If you’re on GCP or Azure, do the same pattern with their SDKs inside a Prefect task: - GCS (using google-cloud-storage): check

client.get_bucket(...)

, create with

client.create_bucket(...)

- Azure Blob (using azure-storage-blob): call

create_container

, catch

ResourceExistsError

Links - RemoteFileSystem and result storage overview: Prefect 3 docs - prefect-aws S3 utilities and blocks: prefect-aws S3 docs - prefect-aws credentials block: prefect-aws credentials If you share which cloud you’re on (S3/GCS/Azure), I can drop in a tailored “ensure bucket/container” task snippet for that provider.

Got it — MinIO + RemoteFileSystem. You can absolutely create a MinIO bucket from within a Prefect flow; RFS will not create the bucket for you, but it will create prefixes/folders. The simplest way is to use the MinIO Python SDK (or boto3 pointed at your MinIO endpoint) in a small task, then continue using your RFS block. Option A: Use the MinIO Python SDK (recommended for MinIO)

from prefect import flow, task
from minio import Minio
from minio.error import S3Error
from prefect.filesystems import RemoteFileSystem

@task
def ensure_minio_bucket(endpoint: str, access_key: str, secret_key: str, bucket: str, secure: bool = False) -> bool:
    """
    Ensure a MinIO bucket exists. Returns True if created, False if it already existed.
    """
    client = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure)
    if client.bucket_exists(bucket):
        return False
    client.make_bucket(bucket)
    return True

@flow
def ensure_bucket_then_use_rfs_minio():
    # Your MinIO connection details
    endpoint = "minio:9000"           # e.g. "localhost:9000" or "<http://minio.my-domain.com|minio.my-domain.com>"
    access_key = "MINIO_ACCESS_KEY"
    secret_key = "MINIO_SECRET_KEY"
    bucket = "my-bucket"
    prefix = "data/"
    secure = False                    # True if using https

    created = ensure_minio_bucket(endpoint, access_key, secret_key, bucket, secure)
    print(f"Bucket created? {created}")

    # Now use your RemoteFileSystem (RFS) with the MinIO endpoint
    rfs = RemoteFileSystem(
        basepath=f"s3://{bucket}/{prefix}",
        settings={
            # s3fs credentials
            "key": access_key,
            "secret": secret_key,
            "use_ssl": secure,
            # point s3fs at MinIO
            "client_kwargs": {"endpoint_url": f"{'https' if secure else 'http'}://{endpoint}"},
        },
    )

    # RFS can now read/write as usual (it will create prefixes but not buckets)
    rfs.write_path("healthcheck.txt", b"ok")

Option B: Use boto3 pointed at MinIO (also works)

from prefect import flow, task
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
from prefect.filesystems import RemoteFileSystem

@task
def ensure_minio_bucket_boto3(endpoint: str, access_key: str, secret_key: str, bucket: str, secure: bool = False) -> bool:
    s3 = boto3.client(
        "s3",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        endpoint_url=f"{'https' if secure else 'http'}://{endpoint}",
        config=Config(signature_version="s3v4"),
        region_name="us-east-1",
    )
    try:
        s3.head_bucket(Bucket=bucket)
        return False
    except ClientError as e:
        code = e.response.get("Error", {}).get("Code", "")
        if code not in ("404", "NoSuchBucket"):
            # 403 can mean the bucket exists but isn’t owned by your creds
            if code == "403":
                raise
        # Create if missing
        s3.create_bucket(Bucket=bucket)
        return True

@flow
def ensure_bucket_then_use_rfs_minio_boto3():
    endpoint = "minio:9000"
    access_key = "MINIO_ACCESS_KEY"
    secret_key = "MINIO_SECRET_KEY"
    bucket = "my-bucket"
    prefix = "data/"
    secure = False

    created = ensure_minio_bucket_boto3(endpoint, access_key, secret_key, bucket, secure)
    print(f"Bucket created? {created}")

    rfs = RemoteFileSystem(
        basepath=f"s3://{bucket}/{prefix}",
        settings={
            "key": access_key,
            "secret": secret_key,
            "use_ssl": secure,
            "client_kwargs": {"endpoint_url": f"{'https' if secure else 'http'}://{endpoint}"},
        },
    )
    rfs.write_path("healthcheck.txt", b"ok")

Notes - RFS: creates prefixes automatically but not buckets. - MinIO SDK is very straightforward for bucket lifecycle:

bucket_exists

+

make_bucket

. - For boto3, the

endpoint_url

must point to MinIO, and you may need

signature_version="s3v4"

. Helpful links

- MinIO Python SDK: minio-py on GitHub - s3fs kwargs (used by RemoteFileSystem settings): s3fs docs - prefect-aws credentials (includes MinIOCredentials, if you prefer that route): prefect-aws credentials docs If you share your MinIO endpoint scheme (http/https) and how you’re configuring your existing RFS block, I can tailor the snippet exactly to your setup.