鹿野孟城
02/24/2026, 6:30 AMclass PrefectTransactionConfig(BaseModel):
model_config = basemodel_config
isolation_level: TransactionIsolationLevel = Field(
default=TransactionIsolationLevel.READ_COMMITTED,
description="isolation level",
)
result_storage: LocalFileSystemConfig | S3BucketConfig | None = Field(
default=None,
description="result storage",
discriminator="type",
)
prefect block type inspect results: refers to a image context
prefect ui: refers to a image contextMarvin
02/24/2026, 6:30 AMMarvin
02/24/2026, 6:34 AMfrom typing import Literal, Annotated
from pydantic import BaseModel, Field
from prefect.blocks.core import Block
class LocalFileSystemConfig(BaseModel):
type: Literal["local"] = "local"
path: str
class S3BucketConfig(BaseModel):
type: Literal["s3"] = "s3"
bucket: str
key_prefix: str | None = None
class PrefectTransactionConfig(Block):
isolation_level: TransactionIsolationLevel = Field(
default=TransactionIsolationLevel.READ_COMMITTED,
description="isolation level",
)
# Discriminated union + Optional
result_storage: Annotated[
LocalFileSystemConfig | S3BucketConfig,
Field(discriminator="type")
] | None = Field(
default=None,
description="result storage",
)
How to verify
- In code:
from pprint import pprint
pprint(PrefectTransactionConfig.model_json_schema()["properties"]["result_storage"])
# Expect: {"oneOf": [...], "discriminator": {"propertyName": "type", "mapping": {...}}}
- CLI (verified command):
prefect block type inspect <your-block-type-slug>
Look under the properties.result_storage section for a discriminator and oneOf.
Notes and limitations
- The discriminator field must exist on every variant and be typed as a Literal (e.g., type: Literal["local"]); using type: str will not work with a discriminator.
- If you’re discriminating across Prefect Block subclasses (instead of plain Pydantic models), advanced patterns using Annotated[..., Tag(...)] + Discriminator(func) also work, but the Literal approach above is simplest.
- The Prefect UI currently renders the schema generically; it recognizes the oneOf/discriminator in the JSON Schema but does not provide a special variant picker. You’ll see a generic nested form/JSON editor for the chosen variant.
If the CLI/UI output you’re seeing does not show the discriminator:
- Please share your two variant model definitions, your block type slug, and your Prefect version. I’ll help spot what’s missing.鹿野孟城
02/24/2026, 6:51 AMMarvin
02/24/2026, 6:51 AMMarvin
02/24/2026, 6:57 AMprefect version)
- Can you share the exact Block class and the two union variant models you’re using?
- How are you registering the block (module/file path)?
Now, step-by-step troubleshooting you can run locally
1) Verify Pydantic generated the right schema
- This checks if your model definition is correct before Prefect stores it.
from pprint import pprint
schema = PrefectTransactionConfig.model_json_schema()
print("result_storage schema:")
pprint(schema["properties"].get("result_storage"))
# Helpful: confirm that it includes both "oneOf" and a "discriminator" entry.
# You should see something like:
# {
# 'oneOf': [{...}, {...}],
# 'discriminator': {'propertyName': 'type', 'mapping': {...}},
# ...
# }
If you don’t see oneOf + discriminator here, fix the model:
- Ensure each variant defines a Literal discriminator field (e.g., type: Literal["local"] = "local" and type: Literal["s3"] = "s3").
- Keep the discriminator on the union itself:
result_storage: (LocalFileSystemConfig | S3BucketConfig | None) = Field(
default=None,
description="result storage",
discriminator="type",
)
2) Sanity-check validation locally
- Make sure the union actually validates with/without None.
PrefectTransactionConfig(result_storage={"type": "s3", "bucket": "my-bucket"})
PrefectTransactionConfig(result_storage=None)
Both should succeed if the schema is correct.
3) Re-register and inspect server-side
prefect block register --file path/to/your_module.py
prefect block type inspect <your-block-type-slug>
- Inspect the fields section for result_storage. If it still shows as null while step (1) looked good, you’ve likely hit a server-side schema handling issue.
Known issue and workarounds
In Prefect 3.x there’s a known issue where Optional discriminated unions can end up being collapsed to null when the server extracts and reconstructs the “fields” section of the block schema for display. The root is how nested/annotated union types are walked while building the stored schema. You’ll often see:
- model_json_schema() looks correct locally (oneOf + discriminator)
- But prefect block type inspect (and the UI) show null for that field
Workarounds you can try right now:
- Remove Optional for now (make it required). This avoids the Optional-union edge case while registering:
# Temporary
result_storage: LocalFileSystemConfig | S3BucketConfig = Field(
discriminator="type",
description="result storage",
)
- Or replace None with an explicit “none” variant to avoid Optional:
class NoStorage(BaseModel):
type: Literal["none"] = "none"
result_storage: LocalFileSystemConfig | S3BucketConfig | NoStorage = Field(
default=NoStorage(),
discriminator="type",
description="result storage",
)
- If you currently use Annotated for the discriminator, try the simpler Field(discriminator="...") on the union instead (or vice versa), then re-register and re-check.
Confirm via API (optional, helps pinpoint server vs. model)
import asyncio
from prefect.client.orchestration import get_client
async def show_schema(slug: str):
async with get_client() as client:
bt = await client.read_block_type_by_slug(slug)
bs = await client.get_most_recent_block_schema_for_block_type(bt.id)
print("Server fields for result_storage:")
print(bs.fields.get("result_storage"))
asyncio.run(show_schema("<your-block-type-slug>"))
If this returns None/null while your local model_json_schema() had the correct oneOf + discriminator, it’s the known server-side handling.
Next steps
- If you share:
- Your Prefect versionMarvin
02/24/2026, 6:57 AM