<@ULVA73B9P> Does the prefect block schema support...
# ask-marvin
u
@Marvin Does the prefect block schema support discriminated unions? The example of the implementations as follows:
Copy code
class 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 context
m
thought for 255.5 seconds
Short answer: Yes. In Prefect 3.x, Block models are Pydantic v2 models and discriminated unions are supported. If you define a union with a discriminator, the generated JSON Schema for the block will include a oneOf plus a discriminator, and Prefect will round‑trip validate/serialize it. The UI/CLI show the raw schema; there isn’t special UI to “pick a variant,” but the schema is correct. What to change/check in your example - Make sure each variant defines the discriminator field as a Literal with a unique value. - Put the discriminator on the union type itself (Annotated or Field on the union). - Including None is fine; pydantic will add type: null in the schema. Example pattern
Copy code
from 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:
Copy code
from pprint import pprint
pprint(PrefectTransactionConfig.model_json_schema()["properties"]["result_storage"])
# Expect: {"oneOf": [...], "discriminator": {"propertyName": "type", "mapping": {...}}}
- CLI (verified command):
Copy code
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.
u
@Marvin Whether I set a valid or invalid schema, it ends up being set to null. Please tell me how to troubleshoot this.
m
thought for 342.3 seconds
Got it — if your field keeps showing up as null in prefect block type inspect and in the UI, here’s a focused way to troubleshoot and isolate the cause. First, a couple quick questions to narrow it down: - What Prefect version are you on? (please run
prefect 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.
Copy code
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:
Copy code
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.
Copy code
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
Copy code
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:
Copy code
# Temporary
result_storage: LocalFileSystemConfig | S3BucketConfig = Field(
    discriminator="type",
    description="result storage",
)
- Or replace None with an explicit “none” variant to avoid Optional:
Copy code
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)
Copy code
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 version
- The exact class definitions for LocalFileSystemConfig, S3BucketConfig, and PrefectTransactionConfig - The block type slug you’re inspecting …I can confirm if you’re hitting this known Optional-union edge case and suggest the least invasive workaround for your setup (or file a bug on your behalf).