jason810496 commented on code in PR #71536:
URL: https://github.com/apache/airflow/pull/71536#discussion_r3810130882
##########
airflow-core/src/airflow/serialization/stub_arg_bindings.py:
##########
@@ -274,6 +319,8 @@ def build_arg_bindings(op: DecoratedOperator) ->
list[dict[str, Any]] | None:
xcom_entry["value_schema"] = value_schema
spec.append(xcom_entry)
continue
+ _reject_nested_xcom(value, task_id, name)
+ value = _to_json_value(value, annotations[name])
Review Comment:
I had not considered Enum or Decimal initially. The table in the PR
description now shows the behavior before and after this PR. The behavior of
Enum and Decimal remains unchanged after this PR.
##########
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/task_arg_binding.py:
##########
@@ -37,7 +37,33 @@
"ArgValueSchema", Annotated[dict[str, JsonValue],
Field(title="ArgValueSchema")]
)
"""JSON-schema fragment constraining the value a stub-task argument binds to;
generated
-by pydantic from the stub annotation, carried verbatim, unknown keywords
ignored."""
+by pydantic from the stub annotation, carried verbatim, unknown keywords
ignored.
+
+``format`` carries the part of the contract ``type`` alone cannot: which
native type a
+lang SDK should decode the value into. Every SDK is expected to follow the
same table,
+so a Dag author sees one behaviour regardless of the task's language:
+
+=================== ========== ====================================
=========================
+Python annotation ``type`` ``format`` / wire spelling Native
target
+=================== ========== ====================================
=========================
+``datetime`` string ``date-time`` ``2024-01-02T03:04:05Z``
timestamp
+``date`` string ``date`` ``2024-01-02`` date
+``time`` string ``time`` ``03:04:05`` time
of day
+``timedelta`` string ``duration`` ``P1DT2H3M4S`` ``-PT1M30S``
duration
+``UUID`` string ``uuid`` ``6ba7b810-9dad-...-...`` UUID
+``bytes`` string ``binary`` (raw text, **not** base64) byte
string
Review Comment:
I overlooked bytes in my review. We do not need to support bytes at all. I
also added the Enum, Decimal, and Path cases to the docstring table.
##########
airflow-core/src/airflow/serialization/stub_arg_bindings.py:
##########
@@ -108,36 +109,76 @@ def _infer_value_schema(annotation: Any) -> dict[str,
Any] | None:
# get_type_hints normalizes a bare ``None`` annotation to NoneType; a
parameter
# that can only ever be None constrains nothing worth shipping.
return None
+ wire_form = _get_wire_form(annotation)
+ # Deep-copy so callers embedding the fragment never alias the cached dict.
+ return copy.deepcopy(wire_form.schema) if wire_form else None
+
+
+class _ValueWireForm(NamedTuple):
+ """The schema describing an annotation's JSON form, and the adapter that
renders values into it."""
+
+ adapter: TypeAdapter
+ schema: dict[str, Any]
+
+
+def _get_wire_form(annotation: Any) -> _ValueWireForm | None:
try:
- schema = _generate_value_schema(annotation)
+ return _build_wire_form(annotation)
except TypeError:
- # Unhashable annotations cannot key the cache; generate directly. Any
pydantic
+ # Unhashable annotations cannot key the cache; build directly. Any
pydantic
# failure inside the body degrades to None there, so this retry never
re-raises.
- schema = _generate_value_schema.__wrapped__(annotation)
- # Deep-copy so callers embedding the fragment never alias the cached dict.
- return copy.deepcopy(schema) if schema else None
+ return _build_wire_form.__wrapped__(annotation)
@cache
-def _generate_value_schema(annotation: Any) -> dict[str, Any] | None:
+def _build_wire_form(annotation: Any) -> _ValueWireForm | None:
"""
- Generate the schema for one annotation, cached for the process lifetime.
+ Build the adapter and schema for one annotation together, cached for the
process lifetime.
TypeAdapter construction is one of pydantic's most expensive operations and
annotations are static, so re-serializations of the same Dag must not
re-pay it.
+
+ Pairing them is what keeps a literal from being rendered in a spelling its
own
+ ``value_schema`` does not describe: both come from the same adapter,
including when
+ the temporal-normalization retry below settles on a different annotation.
"""
# PydanticUserError/TypeError cover annotations pydantic can't schema;
either way,
# that degrades to no schema rather than failing Dag serialization.
- try:
- return
TypeAdapter(annotation).json_schema(schema_generator=_ValueSchemaGenerator)
- except (PydanticUserError, TypeError):
- normalized = _normalize_temporal_annotation(annotation)
- if normalized is annotation:
- return None
+ for candidate in (annotation, _normalize_temporal_annotation(annotation)):
try:
- return
TypeAdapter(normalized).json_schema(schema_generator=_ValueSchemaGenerator)
+ adapter = TypeAdapter(candidate)
+ return _ValueWireForm(adapter,
adapter.json_schema(schema_generator=_ValueSchemaGenerator))
except (PydanticUserError, TypeError):
- return None
+ continue
+ return None
+
+
+def _to_json_value(value: Any, annotation: Any) -> Any:
+ """
+ Render a native value in the JSON form its ``value_schema`` advertises.
+
+ A ``datetime``/``timedelta``/``UUID`` is not JSON-serializable, so without
this it
+ could not cross the language boundary at all. Dumping it through the same
adapter
+ that produced the schema gives every lang SDK exactly one spelling per
format --
+ RFC 3339 timestamps, ISO-8601 durations, canonical UUIDs -- instead of
each Dag
+ author picking their own.
+
+ Values pydantic cannot render for this annotation pass through untouched,
leaving
+ the JSON-serializability check to reject them.
+ """
+ if annotation is Parameter.empty or annotation is None or annotation is
Any:
+ return value
+ if isinstance(value, datetime.datetime):
+ # A naive timestamp is ambiguous once it leaves Python: Go would read
it as UTC,
+ # JavaScript as the worker's local time, and Java would refuse to
parse it. Pin
+ # the offset here, using the same default timezone the rest of Airflow
applies.
+ value = coerce_datetime(value)
Review Comment:
Good point. I had not considered collection types. Addressed in 108dbf5f8b,
thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
