Kyle McChesney
03/31/2026, 4:30 PMMarvin
03/31/2026, 4:30 PMMarvin
03/31/2026, 4:33 PMfrom pydantic import BaseModel, Field
from prefect import flow
import base64
class FileInput(BaseModel):
filename: str = Field(description="Original filename")
file_b64: str = Field(description="Base64-encoded file content")
@flow
def process_file(inp: FileInput):
content = base64.b64decode(inp.file_b64)
# ... process content ...
2) URL to download
from pydantic import BaseModel, HttpUrl
from prefect import flow
import httpx
class FileRef(BaseModel):
file_url: HttpUrl
@flow
def process_remote_file(ref: FileRef):
resp = httpx.get(str(ref.file_url))
resp.raise_for_status()
content = resp.content
# ... process content ...
3) Storage block + path (Prefect-native)
- Create/configure a storage block (e.g., an S3 or RemoteFileSystem block) in Prefect.
- Pass the block name and a path as parameters; load and read in the flow.
from pydantic import BaseModel, Field
from prefect import flow
from prefect.filesystems import RemoteFileSystem # or use an S3/GCS-specific block
class FileLocation(BaseModel):
block_name: str = Field(description="Name of the storage block")
path: str = Field(description="Path within the storage")
@flow
def process_from_storage(loc: FileLocation):
fs = RemoteFileSystem.load(loc.block_name)
content = fs.read_path(loc.path)
# ... process content ...
Notes
- Large files are not a good fit for parameters (including base64) due to request size and UX; prefer URLs or storage blocks.
- Pydantic types like Literal/Enum render nicely (e.g., dropdowns), and Field(description=...) shows up as help text in the UI, but there’s no file picker widget.
If you can share your expected file sizes and where the files live (local, S3, GCS, HTTP), I can suggest the most ergonomic model shape and block setup.