ÖmerFİ
12/15/2025, 2:30 PMMarvin
12/15/2025, 2:30 PMMarvin
12/15/2025, 2:34 PM{"bucket_block": "my-s3", "key": "incoming/file.csv"} and read it in the flow.
- Base64 string for small files
- Encode the file client-side and pass a base64 string parameter, then decode in the flow. Only suitable for small inputs because of request size limits.
Examples
1) Pass a file path/URL as a parameter (CSV or Excel)
from prefect import flow
import pandas as pd
@flow
def process_file(source: str):
if source.lower().endswith(".csv"):
df = pd.read_csv(source) # s3://..., gs://..., https://..., or local path
else:
import io, requests
if source.startswith(("http://", "https://")):
content = requests.get(source).content
df = pd.read_excel(io.BytesIO(content))
else:
df = pd.read_excel(source) # local or fsspec-supported URLs
return df.head()
2) Use an S3 block for robust access and credentials handling
from prefect import flow
from prefect_aws.s3 import S3Bucket
import pandas as pd, io
@flow
def process_s3_csv(bucket_block: str, key: str):
s3 = S3Bucket.load(bucket_block)
data = s3.read_path(key) # bytes
if key.lower().endswith(".csv"):
import pandas as pd, io
df = pd.read_csv(io.BytesIO(data))
else:
df = pd.read_excel(io.BytesIO(data))
return df.head()
3) Base64 for small files
from prefect import flow
import base64, io, pandas as pd
@flow
def process_b64(file_b64: str):
data = base64.b64decode(file_b64)
try:
df = pd.read_csv(io.BytesIO(data))
except Exception:
df = pd.read_excel(io.BytesIO(data))
return df.head()
Triggering with parameters
- UI: open the deployment, click Run, and paste JSON parameters (e.g., {"source": "<s3://my-bucket/file.csv>"} or {"bucket_block": "my-s3", "key": "folder/file.xlsx"}).
- CLI:
prefect deployment run my-flow/my-deployment \
--param source="<s3://my-bucket/data.csv>" \
--watch
- Python:
from prefect.deployments import run_deployment
run_deployment(
name="my-flow/my-deployment",
parameters={"source": "<s3://my-bucket/data.csv>"}
)
Docs
- Trigger ad‑hoc deployment runs
- Blocks concept
- Blob storage patterns (applies to S3/GCS/Azure blocks)
- Variables (OK for small, non‑sensitive values; not for large files)
If you can share where your flows run (e.g., local process/Docker/Kubernetes) and where the files live (local disk, S3, GCS, etc.), I can suggest the simplest wiring for your setup.ÖmerFİ
12/15/2025, 3:01 PMMarvin
12/15/2025, 3:01 PMMarvin
12/15/2025, 3:04 PM{"s3_path": "s3://.../file.csv"}.
- In the flow, load via a storage block (S3/GCS/Azure) and read the file.
2) Base64 for small files
- Frontend reads the file, base64‑encodes it, and sends as a parameter. Flow decodes and processes. Only for small inputs due to request size limits.
3) Custom proxy API
- Build a small service that accepts multipart/form-data, writes to storage, then calls Prefect’s API to start the run with a reference to the stored file.
4) Interactive flows with RunInput
- For pauseable workflows, implement a UI that sends base64/path inputs mid‑run using Prefect’s RunInput API. Best for “review/approve/upload” steps.
Calling the API to create a run (Cloud example)
- Endpoint:
- POST https://api.prefect.cloud/api/accounts/{ACCOUNT_ID}/workspaces/{WORKSPACE_ID}/deployments/{DEPLOYMENT_ID}/create_flow_run
- Headers:
- Authorization: Bearer YOUR_PREFECT_API_KEY
- Content-Type: application/json
- X-PREFECT-API-VERSION: optional for compatibility
- Body:
- JSON with parameters (your file path/base64 and any other params), plus optional tags, name, etc.
- Docs:
- Trigger ad‑hoc deployment runs
- Blocks concept
- Blob storage patterns (S3/GCS/Azure)
- Variables (small, non‑sensitive values only)
If you share where your workers run and where you want to store files (local, S3, GCS, Azure), I can sketch a minimal custom UI + flow pattern tailored to your setup.