cast None -> {} in KeyValueLabels

src/prefect/client/cloud.py

@@ -13,7 +13,6 @@
 from prefect.client.schemas.objects import (
     IPAllowlist,
     IPAllowlistMyAccessResponse,
-    KeyValueLabels,
     Workspace,
 )
 from prefect.exceptions import ObjectNotFound, PrefectException
@@ -23,6 +22,7 @@
     PREFECT_CLOUD_API_URL,
     PREFECT_TESTING_UNIT_TEST_MODE,
 )
+from prefect.types import KeyValueLabels
 
 PARSE_API_URL_REGEX = re.compile(r"accounts/(.{36})/workspaces/(.{36})")
 

src/prefect/client/orchestration.py

@@ -3925,13 +3925,13 @@ def read_flow_run(self, flow_run_id: UUID) -> FlowRun:
     def read_flow_runs(
         self,
         *,
-        flow_filter: FlowFilter = None,
-        flow_run_filter: FlowRunFilter = None,
-        task_run_filter: TaskRunFilter = None,
-        deployment_filter: DeploymentFilter = None,
-        work_pool_filter: WorkPoolFilter = None,
-        work_queue_filter: WorkQueueFilter = None,
-        sort: FlowRunSort = None,
+        flow_filter: Optional[FlowFilter] = None,
+        flow_run_filter: Optional[FlowRunFilter] = None,
+        task_run_filter: Optional[TaskRunFilter] = None,
+        deployment_filter: Optional[DeploymentFilter] = None,
+        work_pool_filter: Optional[WorkPoolFilter] = None,
+        work_queue_filter: Optional[WorkQueueFilter] = None,
+        sort: Optional[FlowRunSort] = None,
         limit: Optional[int] = None,
         offset: int = 0,
     ) -> List[FlowRun]:

src/prefect/client/schemas/objects.py

@@ -23,9 +23,6 @@
     HttpUrl,
     IPvAnyNetwork,
     SerializationInfo,
-    StrictBool,
-    StrictFloat,
-    StrictInt,
     Tag,
     field_validator,
     model_serializer,
@@ -56,6 +53,7 @@
 from prefect.settings import PREFECT_CLOUD_API_URL, PREFECT_CLOUD_UI_URL
 from prefect.types import (
     MAX_VARIABLE_NAME_LENGTH,
+    KeyValueLabels,
     Name,
     NonNegativeInteger,
     PositiveInteger,
@@ -71,8 +69,6 @@
 
 R = TypeVar("R", default=Any)
 
-KeyValueLabels = dict[str, Union[StrictBool, StrictInt, StrictFloat, str]]
-
 
 DEFAULT_BLOCK_SCHEMA_VERSION = "non-versioned"
 DEFAULT_AGENT_WORK_POOL_NAME = "default-agent-pool"
@@ -1188,27 +1184,6 @@ class ConcurrencyLimit(ObjectBaseModel):
     )
 
 
-class BlockSchema(ObjectBaseModel):
-    """An ORM representation of a block schema."""
-
-    checksum: str = Field(default=..., description="The block schema's unique checksum")
-    fields: Dict[str, Any] = Field(
-        default_factory=dict, description="The block schema's field schema"
-    )
-    block_type_id: Optional[UUID] = Field(default=..., description="A block type ID")
-    block_type: Optional[BlockType] = Field(
-        default=None, description="The associated block type"
-    )
-    capabilities: List[str] = Field(
-        default_factory=list,
-        description="A list of Block capabilities",
-    )
-    version: str = Field(
-        default=DEFAULT_BLOCK_SCHEMA_VERSION,
-        description="Human readable identifier for the block schema",
-    )
-
-
 class BlockSchemaReference(ObjectBaseModel):
     """An ORM representation of a block schema reference."""
 

src/prefect/types/__init__.py

@@ -82,13 +82,23 @@ def check_variable_value(value: object) -> object:
     return value
 
 
+def cast_none_to_empty_dict(value: Any) -> dict[str, Any]:
+    if value is None:
+        return {}
+    return value
+
+
 StrictVariableValue = Annotated[VariableValue, BeforeValidator(check_variable_value)]
 
 LaxUrl = Annotated[str, BeforeValidator(lambda x: str(x).strip())]
 
-
 StatusCode = Annotated[int, Field(ge=100, le=599)]
 
+KeyValueLabels = Annotated[
+    dict[str, Union[StrictBool, StrictInt, StrictFloat, str]],
+    BeforeValidator(cast_none_to_empty_dict),
+]
+
 
 class SecretDict(pydantic.Secret[Dict[str, Any]]):
     pass

tests/test_flow_engine.py

@@ -16,7 +16,7 @@
 from prefect import Flow, __development_base_path__, flow, task
 from prefect.client.orchestration import PrefectClient, SyncPrefectClient
 from prefect.client.schemas.filters import FlowFilter, FlowRunFilter
-from prefect.client.schemas.objects import FlowRun, KeyValueLabels, StateType
+from prefect.client.schemas.objects import FlowRun, StateType
 from prefect.client.schemas.sorting import FlowRunSort
 from prefect.concurrency.asyncio import concurrency as aconcurrency
 from prefect.concurrency.sync import concurrency
@@ -45,6 +45,7 @@
 from prefect.server.schemas.core import ConcurrencyLimitV2
 from prefect.server.schemas.core import FlowRun as ServerFlowRun
 from prefect.testing.utilities import AsyncMock
+from prefect.types import KeyValueLabels
 from prefect.utilities.callables import get_call_parameters
 from prefect.utilities.filesystem import tmpchdir
 

tests/test_types.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+from prefect.types import KeyValueLabels
+
+
+def test_allow_none_as_empty_dict():
+    class Model(BaseModel):
+        labels: KeyValueLabels
+
+    assert Model(labels=None).labels == {}  # type: ignore