Short answer: - Dynamic asset keys are supported in Prefect 3.x. - Your snippet will create a materialized task inside a task and will likely run, but it’s not recommended. Define the materialized task at module scope and set the key dynamically with

.with_options(...)

. - Also, your example key uses angle brackets

< >

, which are not allowed in asset keys. Notes on asset keys: - Must be valid URIs like

<s3://bucket/path/file.parquet>

- Cannot include

<

or

>

(or spaces,

#

,

?

,

%

, etc.) - Max length 512 chars Recommended pattern for dynamic keys

from prefect import flow
from prefect.assets import Asset, AssetProperties, materialize, add_asset_metadata

# Define once, at module scope
@materialize("<s3://bucket/placeholder>")
def emit(asset_key: str, meta: dict):
    # Only valid when this task is running, and asset_key matches one of this task's assets
    add_asset_metadata(asset_key, meta)

@flow
def process(file_id: str):
    key = f"<s3://bucket/{file_id}/data.parquet>"  # no angle brackets
    asset = Asset(key=key, properties=AssetProperties(
        name=f"file {file_id}",
        description="processed file",
        owners=["<mailto:data@company.com|data@company.com>"],
    ))

    # Override the asset(s) for this run
    emit_dynamic = emit.with_options(assets=[asset])
    emit_dynamic(key, {"rows": 123, "source_id": file_id})

Why not define

@materialize

inside a task? - It creates a new task object at runtime, which can lead to confusing run graphs and makes configuration/caching harder to reason about. - You lose clarity and reusability; the recommended approach is to configure assets dynamically via

.with_options(...)

on a single, top-level materialized task. If you prefer avoiding string keys in

add_asset_metadata

, you can use the instance method:

from prefect.assets import Asset, AssetProperties, materialize

@materialize("<s3://bucket/placeholder>")
def emit(asset: Asset, meta: dict):
    asset.add_metadata(meta)

@flow
def process(file_id: str):
    asset = Asset(
        key=f"<s3://bucket/{file_id}/data.parquet>",
        properties=AssetProperties(name=f"file {file_id}")
    )
    emit.with_options(assets=[asset])(asset, {"rows": 123})

Docs for reference: - Assets overview: https://docs.prefect.io/v3/concepts/assets/ - Assets API: https://docs.prefect.io/v3/api-ref/prefect/assets/ If you share a bit more about how

process(file_id)

is called (from a flow vs. ad-hoc) and whether you have upstream inputs to model as

asset_deps

, I can tailor the example further.

Great question. In Prefect 3.x, assets for a materializing task must be declared before the task starts. You can absolutely handle “N assets” where N is only known after some discovery/processing — the pattern is: - Do discovery first (in a separate task or inline in your flow) - Build the full list of asset keys (or Asset objects) - Call your materialized task via

.with_options(assets=[...])

using that list - Inside the materialized task, add per-asset metadata as needed Example pattern

from prefect import flow, task
from prefect.assets import Asset, AssetProperties, materialize

@task
def discover_keys() -> list[str]:
    # Do whatever processing/discovery you need
    # Return concrete keys you’ll materialize
    return [f"<s3://bucket/data/{i}.parquet>" for i in range(5)]

# Define once at module scope
@materialize("<s3://bucket/placeholder>")
def create_assets(keys: list[str]):
    # Do your create/write/update work here
    # Optionally add metadata for each asset key
    for k in keys:
        # Only valid because k is one of the declared assets via with_options
        # You can use either function or instance method:
        # add_asset_metadata(k, {...})
        Asset(key=k).add_metadata({"status": "created"})

@flow
def pipeline():
    keys = discover_keys()
    # Declare the N assets dynamically just before execution
    task_run = create_assets.with_options(assets=keys)
    task_run(keys)

Key points - You cannot add new assets to the task after it has started. Attempts to add metadata for undeclared assets will raise an error. - Use strings or Asset objects in

assets=[...]

. Strings are auto-wrapped as

Asset(key=...)

. - If you prefer instance methods, construct

Asset(key=...)

and call

asset.add_metadata(...)

inside the materialized task. It’s valid as long as that asset is in the task’s declared

assets

. If your discovery requires a first pass over inputs and then a second pass to materialize, split into two tasks: one to discover and return the keys, then call the materializing task with

.with_options(assets=discovered_keys)

.

It does not have to be a single file. In Prefect 3.x, an asset key is just a URI-style identifier for any logical data object: a file, a directory/prefix, a table, a topic, etc. Prefect doesn’t introspect the location; it tracks what you declare. - Directory/prefix assets are fine, e.g.

<s3://bucket/dataset/2025-12-12/>

- Keys must be valid URIs and avoid restricted characters; a trailing slash for directories is okay - If you want to model both a directory and its child files, declare each as its own asset and relate them via

asset_deps

(discovered first, then passed with

.with_options(...)

) Example

from prefect.assets import Asset, AssetProperties, materialize

dir_asset = Asset(
    key="<s3://bucket/dataset/2025-12-12/>",
    properties=AssetProperties(name="Daily dataset prefix")
)

@materialize(dir_asset)
def write_partitioned_dataset(parts: list[bytes]):
    # write multiple files under <s3://bucket/dataset/2025-12-12/>
    # ...
    dir_asset.add_metadata({"partitions": len(parts), "bytes": 123456})

If you also want per-file assets and lineage to the directory:

file_assets = [f"<s3://bucket/dataset/2025-12-12/part-{i}.parquet>" for i in range(n)]

# Declare both the directory and the files up front (compute n first)
task = write_partitioned_dataset.with_options(
    assets=["<s3://bucket/dataset/2025-12-12/>"],
    asset_deps=file_assets
)
task(parts)

Docs: Assets overview and Assets API

Short answer: not inherently bad, but be intentional. There’s no hard limit on the number of assets, but very high cardinality can impact event volume, UI/search performance, and Cloud API rate limits. Guidelines - Model identity carefully: - Use a single asset for a logical dataset (e.g., a partition directory/prefix) and attach per-run or per-partition details as metadata, rather than creating one asset per tiny file. - Create discrete assets when they truly represent different artifacts (different location, schema, SLA/owner). - Prefer partitioned/dynamic assets: - For daily/hourly data, set the asset key per run via

.with_options(assets=[...])

. This scales better than defining thousands of static assets. - Watch event volume: - Each materialized asset emits an event; upstream

asset_deps

emit reference events. Thousands of assets per run can create a large burst of events/logs. - Keep UI usability in mind: - Many unique keys can make search and lineage views heavier. Organize with sensible prefixes and naming. - Rate limits (Prefect Cloud): - The SDK retries on 429s, but avoid massive simultaneous bursts of asset events. Batch work or stagger runs if needed. - Single source of truth: - Only one workflow should set

AssetProperties

for a given key; others should reference by string key and add runtime metadata only. Practical patterns - Directory-level asset with metadata counts: - One asset like

<s3://bucket/dataset/2025-12-12/>

, add metadata:

{"files": 1243, "rows": 8_321_554}

- Partitioned assets via `.with_options(...)`: -

assets=[f"<s3://bucket/dataset/{date}/>"]

in a daily flow - If you must emit many assets, consider: - Grouping them into logical batches - Emitting only key assets and recording the rest as metadata (e.g., list of file names, counts, checksums) Docs - Assets overview: Assets overview - Assets API: Assets API

Got it. For a structure like:

document-lake/
└── raw/
    └── gdrive/
        └── <file_id>/
            └── file.pdf

you have a few modeling options. The best choice depends on how much per-file lineage you need and your expected file volume. Option A: Per-file assets (fine-grained lineage) - Model each file as its own asset, e.g.

<dl://raw/gdrive/{file_id}/file.pdf>

(replace

dl://

with your actual store scheme like

<s3://bucket/>...

or `gcs://bucket/...`). - Use discovery to build the list of file asset keys, then call the materialized task with

.with_options(assets=[...])

. - Pros: precise lineage per file, per-file metadata - Cons: high cardinality and event volume for large N Example

from prefect import flow, task
from prefect.assets import materialize, add_asset_metadata

@task
def discover_file_ids() -> list[str]:
    # query GDrive and return a list of file_ids to ingest
    return ["id1", "id2", "id3"]

@materialize("<dl://raw/gdrive/placeholder>")  # dummy; will be overridden
def ingest_files(file_ids: list[str], base_prefix: str):
    # write to document-lake/raw/gdrive/{file_id}/file.pdf
    for fid in file_ids:
        # ... download/write ...
        add_asset_metadata(f"{base_prefix}/{fid}/file.pdf", {"source": "gdrive", "file_id": fid})

@flow
def etl_run():
    base_prefix = "<dl://raw/gdrive>"
    file_ids = discover_file_ids()
    asset_keys = [f"{base_prefix}/{fid}/file.pdf" for fid in file_ids]

    ingest = ingest_files.with_options(assets=asset_keys)
    ingest(file_ids, base_prefix)

Option B: Directory/batch assets (coarse-grained, scalable) - Model a folder or run partition as the asset, e.g.

<dl://raw/gdrive/{run_id}/>

or just

<dl://raw/gdrive/>

. - Store per-file details as metadata on the folder asset (counts, lists of file_ids, checksums). - Pros: fewer assets, scalable; great for very large N - Cons: you lose per-file lineage as separate assets Example

from prefect import flow, task
from prefect.assets import Asset, AssetProperties, materialize

@task
def discover_file_ids() -> list[str]:
    return ["id1", "id2", "id3"]

@materialize("<dl://raw/gdrive/placeholder/>")  # will be overridden
def ingest_batch(file_ids: list[str], folder_asset: Asset):
    processed = []
    for fid in file_ids:
        # ... download/write to folder_asset.key/{fid}/file.pdf ...
        processed.append({"file_id": fid, "bytes": 12345})
    # summarize on the folder asset
    folder_asset.add_metadata({
        "count": len(processed),
        "files": processed,   # keep this list reasonable in size
    })

@flow
def etl_run(run_id: str):
    file_ids = discover_file_ids()
    folder = Asset(
        key=f"<dl://raw/gdrive/{run_id}/>",
        properties=AssetProperties(name=f"gdrive raw {run_id}")
    )
    ingest = ingest_batch.with_options(assets=[folder])
    ingest(file_ids, folder)

Option C: Hybrid - Use a folder asset as the primary record and “promote” only important files to discrete assets. - Keep everything else as metadata on the folder asset. This balances lineage detail with scalability. Additional tips - Use dynamic keys via

.with_options(assets=[...])

after discovery; assets must be declared before the materialized task starts. - Keys must be valid URIs with allowed characters; avoid spaces and

< >

. A custom scheme like

dl://

is fine, or use your real store like

<s3://bucket/>...

or

<gcs://bucket/>...

. - If you have downstream processing (e.g., clean/parsed), you can link to raw assets via `asset_deps`: - Build

asset_deps=[f"<dl://raw/gdrive/{fid}/file.pdf>" ...]

for the processed asset(s). - For very large N per run, consider: - Batch by subdirectory (e.g.,

.../gdrive/{run_id}/batch-{k}/

) and materialize one asset per batch - Keep per-file metadata lightweight (counts, a sample, or a separate manifest file in storage rather than embedding huge lists)

If you share expected file counts per run and whether you need per-file lineage in the UI, I can recommend a concrete modeling choice and tailor the code to your storage backend (S3/GCS/local). Links: Assets overview • Assets API

Short answer: include the user identifier in the asset key when it’s part of the asset’s identity (uniqueness, storage segregation, lineage, or lifecycle). Otherwise, keep it as metadata. When to include the user in the key - Uniqueness: file IDs may collide across users - dl://raw/gdrive/{user_id}/{file_id}/file.pdf avoids clashes - Isolation/retention: per-user directories, quotas, or deletion policies - Authorization boundaries: downstream jobs or storage ACLs operate per user - Idempotency: deterministic keys prevent duplicates on retries When to keep it as metadata instead - You want to limit asset cardinality (many users × many files) - You don’t need per-user lineage in the UI - Privacy/PII concerns: keys are widely visible; better to store

user_id

as metadata or a hashed/surrogate value Privacy and governance - Don’t put secrets or sensitive PII (e.g., emails) in keys; consider a hashed ID if you need it in the key - Keys must be valid URIs and under 512 chars; avoid spaces and special characters Patterns - Per-file, per-user asset (fine-grained lineage)

from prefect import flow
from prefect.assets import materialize, add_asset_metadata

@materialize("<dl://raw/gdrive/placeholder>")  # overridden per run
def ingest(user_id: str, file_ids: list[str], base: str):
    for fid in file_ids:
        # write to <dl://raw/gdrive/{user_id}/{fid}/file.pdf> ...
        add_asset_metadata(f"{base}/{user_id}/{fid}/file.pdf",
                           {"user_id": user_id, "source": "gdrive"})

@flow
def run_for_user(user_id: str, file_ids: list[str]):
    base = "<dl://raw/gdrive>"
    keys = [f"{base}/{user_id}/{fid}/file.pdf" for fid in file_ids]
    ingest.with_options(assets=keys)(user_id, file_ids, base)

- Directory-per-user asset (scalable; per-file details as metadata)

from prefect.assets import Asset, AssetProperties, materialize

@materialize("<dl://raw/gdrive/placeholder/>")
def ingest_batch(user_dir: Asset, file_ids: list[str]):
    # write files under user_dir.key ...
    user_dir.add_metadata({"user_id": "...", "count": len(file_ids)})

# in your flow:
user_dir = Asset(key=f"<dl://raw/gdrive/{user_id}/>",
                 properties=AssetProperties(name=f"gdrive raw {user_id}"))
ingest_batch.with_options(assets=[user_dir])(user_dir, file_ids)

Decision checklist - Could two different users produce the “same” file_id/path? Include user in key. - Do you need per-user lineage/visibility in the UI? Include user in key or use per-user directories. - Are you concerned about asset explosion? Prefer directory-level assets and store

user_id

as metadata. - Any PII constraints? Use metadata or a hashed/surrogate in the key. Happy to tailor this to your expected user count and file volume. Docs: Assets overview • Assets API

Fun question. I’d keep the graph, but make it one of several “perspectives” optimized for different jobs and scales. If I were rebuilding it, I’d design around these views: - Catalog (default) - Virtualized table with instant search, prefix filters (scheme/bucket/path), tags, owners, freshness, status - Left “folder” tree by URI, middle results, right details panel (metadata, last materialization, deps, recent runs) - Saved filters/views with shareable URLs - Lineage (scoped graph) - Start from a seed selection; expand upstream/downstream by radius with clustering (by path, owner, flow, or “bucket”) - Mini-map, lasso, on-demand expansion, and “collapse to group” for high-degree nodes - Edge filters (only materialized deps, only references, failed-only) - Time slider to show lineage “as of” or highlight changes since last run - Partitions/Time - Calendar or heatmap per asset showing daily/hourly partitions and their state - Aggregate rollups (success rate, latency, recency), drill-down to a specific partition materialization - Health & SLAs - Freshness targets per asset; badges for stale/late; propagate “at risk” through lineage with simple rules - Bulk actions (acknowledge, open issue, trigger upstream flow) - Compare & Diff - Compare two materializations of the same asset: metadata diff, size/count deltas, schema/sample changes when available - Batch/Directory assets - Treat directories/prefixes as first-class: show contained children counts, link to a manifest, summarize metrics without listing every file - Promote “notable” children to discrete assets while others stay as metadata - Run overlays - Overlay a specific flow run onto the graph: highlight which assets were materialized/referenced, durations, errors UX/scale features I’d prioritize: - Progressive rendering and server-side clustering for large graphs; never auto-render the whole universe - Keyboard-first navigation, quick-open by key prefix, copyable deep links to any filtered state - Clear empty states and guard rails: “You’ve selected 12,431 assets — consider narrowing to a prefix or owner” - Privacy controls: hide/scrub sensitive key segments; support hashed identifiers in display mapping Why this mix: - Table/Tree is best for discovery, search, and scale - Graph shines for understanding impact and relationships but must be scoped and clusterable - Time/Partitions captures an asset’s lifecycle, which a static graph can’t - Health/SLAs makes it operationally useful, not just informative If you share what you do most (discovery vs. debugging vs. monitoring) and your typical/peak asset counts, I can sketch a concrete layout and defaults tuned to your scale. Refs: Assets overview • Assets API