Great question! In Prefect 3.x, use

prefect.filesystems.RemoteFileSystem

inside your flow to write single files or upload whole directories to a remote backend (S3, GCS, Azure, SFTP, etc. via fsspec). Key methods you’ll use: - `write_path(path, content: bytes)`: write a single file (bytes) to the remote FS - `read_path(path) -> bytes`: read a single file back - `put_directory(local_path, to_path, ignore_file=None, overwrite=False) -> int`: upload a local directory recursively - `get_directory(from_path, local_path)`: download a remote directory Examples 1) Upload a whole directory (e.g. to S3)

from prefect import flow
from prefect.filesystems import RemoteFileSystem

@flow
def upload_dir():
    # basepath points to your remote root
    fs = RemoteFileSystem(basepath="<s3://my-bucket/my-prefix>")

    # Upload local ./data recursively under <s3://my-bucket/my-prefix/inputs/>
    uploaded = fs.put_directory(
        local_path="data",
        to_path="inputs",
        ignore_file=".prefectignore",  # optional; respected if present
        overwrite=True,                # set False to skip existing
    )
    print(f"Uploaded {uploaded} files")

if __name__ == "__main__":
    upload_dir()

2) Copy a single local file

from pathlib import Path
from prefect.filesystems import RemoteFileSystem

fs = RemoteFileSystem(basepath="<s3://my-bucket/my-prefix>")

content = Path("local/file.csv").read_bytes()
remote_full_path = fs.write_path("inputs/file.csv", content)
print(f"Wrote to {remote_full_path}")

3) Write/read text or JSON

import json
from prefect.filesystems import RemoteFileSystem

fs = RemoteFileSystem(basepath="<s3://my-bucket/my-prefix>")

# Write text
fs.write_path("notes/info.txt", "hello world".encode())

# Write JSON
payload = {"a": 1}
fs.write_path("data/payload.json", json.dumps(payload).encode())

# Read back
text = fs.read_path("notes/info.txt").decode()
obj = json.loads(fs.read_path("data/payload.json").decode())

4) Load a saved RemoteFileSystem block (if you’ve created one in Prefect)

from prefect.filesystems import RemoteFileSystem

fs = RemoteFileSystem.load("my-remote-fs")
fs.put_directory(local_path="data", to_path="inputs", overwrite=True)

5) Async usage (inside an async flow/task)

from prefect import flow
from prefect.filesystems import RemoteFileSystem

@flow
async def upload_async():
    fs = RemoteFileSystem(basepath="<s3://my-bucket/my-prefix>")
    await fs.aput_directory("data", "inputs")

Notes - `path`/`to_path` values are relative to

basepath

. For example, with

basepath="<s3://bucket/prefix>"

,

write_path("a/b.txt", ...)

writes to

<s3://bucket/prefix/a/b.txt>

. - Credentials are picked up from your environment (e.g., AWS env vars/profiles). If needed, you can pass fsspec options via

settings={...}

when constructing

RemoteFileSystem

. - There’s no

put_file()

on `RemoteFileSystem`; use

write_path

for single files,

put_directory

for folders. If you can share which backend you’re targeting (S3, GCS, Azure, SFTP, etc.) and whether you need single-file or directory uploads, I can tailor the snippet (including any

settings

you might need for auth).