diff --git a/pyatlan_v9/model/string_coercion.py b/pyatlan_v9/model/string_coercion.py new file mode 100644 index 000000000..48d929042 --- /dev/null +++ b/pyatlan_v9/model/string_coercion.py @@ -0,0 +1,197 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. +"""Lenient coercion of values destined for ``str``-typed model slots. + +Hand-maintained: unlike its neighbours, this module is **not** emitted by +``PythonMsgspecRenderer.pkl``. + +Why this exists +--------------- +Atlas's ``string`` type accepts anything and stores its stringification. From +``AtlasBuiltInTypes.AtlasStringType`` in ``atlas-metastore``:: + + public boolean isValidValue(Object obj) { return true; } + public String getNormalizedValue(Object obj) { + if (obj != null) { return obj.toString(); } + return null; + } + +The generated models mirror the *declared* Atlas type — a typedef attribute of +type ``map`` becomes ``Dict[str, str]``, and the entity-header +``customAttributes`` (``Map`` on ``AtlasEntity``) becomes +``Dict[str, str]`` — but they cannot mirror that leniency, because ``msgspec`` +enforces ``str`` exactly. ``strict=False`` does not help: lax mode coerces +``str`` -> ``int``, never ``int`` -> ``str``. A ``dec_hook`` does not help +either; it fires only for types ``msgspec`` cannot natively handle, so a +``str``/``int`` mismatch raises before any hook runs. + +The result is a decoder that is stricter than the server it models. A producer +that emits ``{"ordinal_position": 1}`` in ``customAttributes`` has its **whole +record** rejected here, while Atlas accepts it and stores ``"1"``. + +So this module restores the server's leniency, opt-in, for callers decoding +payloads that have not yet been through Atlas's normalisation. + +What it does +------------ +The coercion plan is derived from the model itself via :mod:`msgspec.inspect`, +so it covers every ``str``-typed slot — plain ``str`` fields, ``Dict[str, str]``, +``List[Dict[str, str]]``, ``List[str]``, and the same shapes on nested related +structs — with no hand-maintained list of field names to fall behind. + +Values are coerced to their **compact JSON text**, which matches +``getNormalizedValue`` for scalars (``1`` -> ``"1"``, ``true`` -> ``"true"``). +For a container in a string slot it does not: Atlas would store Java's +``Object.toString()`` (``{a=b}``); this emits ``{"a":"b"}``. JSON is chosen +deliberately — it is the honest serialisation of a JSON value, and it round +trips. + +``None`` is dropped from a slot that cannot hold it (``Dict[str, str]`` values, +and non-optional fields), and preserved where the model declares +``Optional``. Dropping rather than stringifying is deliberate: a literal +``"None"``/``"null"`` is a value the producer never wrote. +""" + +from __future__ import annotations + +from typing import Any, Callable, Dict, Optional + +import msgspec +import msgspec.inspect as _inspect + +__all__ = ["coerce_string_slots"] + +_Coercer = Callable[[Any], Any] + + +class _Drop: + """Sentinel: remove the key rather than decode a value it cannot hold.""" + + __slots__ = () + + def __repr__(self) -> str: # pragma: no cover - debugging aid + return "" + + +_DROP = _Drop() + +# One plan per model class, keyed by encoded (camelCase) field name. Built +# lazily; entries are inserted *before* the class is walked so a cyclic type +# graph (``Related*`` structs reference each other) terminates. +_plans: Dict[type, Dict[str, _Coercer]] = {} + + +def _to_str(value: Any) -> Any: + """Stringify one value the way Atlas's ``string`` type would.""" + if isinstance(value, str): + return value + return msgspec.json.encode(value).decode() + + +def _coerce_str_map(value: Any) -> Any: + """Coerce a ``Dict[str, str]`` payload, dropping null values. + + A non-mapping payload is returned untouched: there is no sane coercion, and + letting ``msgspec`` reject it keeps a genuine producer-side format change + visible instead of inventing a value. + """ + if not isinstance(value, dict): + return value + return {key: _to_str(item) for key, item in value.items() if item is not None} + + +def _coerce_sequence(inner: _Coercer) -> _Coercer: + def coerce(value: Any) -> Any: + if not isinstance(value, list): + return value + return [inner(item) for item in value] + + return coerce + + +def _coerce_struct(cls: type) -> _Coercer: + def coerce(value: Any) -> Any: + if isinstance(value, dict): + coerce_string_slots(value, cls) + return value + + return coerce + + +def _value_coercer(field_type: _inspect.Type) -> Optional[_Coercer]: + """Build a coercer for one type, or ``None`` if it holds no ``str`` slot.""" + if isinstance(field_type, _inspect.StrType): + return _to_str + if isinstance(field_type, _inspect.DictType): + if isinstance(field_type.value_type, _inspect.StrType): + return _coerce_str_map + return None + if isinstance(field_type, (_inspect.ListType, _inspect.SetType)): + inner = _value_coercer(field_type.item_type) + return _coerce_sequence(inner) if inner is not None else None + if isinstance(field_type, _inspect.StructType): + # Resolved at call time, not build time, so a cyclic type graph is fine. + return _coerce_struct(field_type.cls) + if isinstance(field_type, _inspect.UnionType): + for member in field_type.types: + coercer = _value_coercer(member) + if coercer is not None: + return coercer + return None + + +def _field_coercer(field_type: _inspect.Type) -> Optional[_Coercer]: + inner = _value_coercer(field_type) + if inner is None: + return None + nullable = isinstance(field_type, _inspect.UnionType) and any( + isinstance(member, _inspect.NoneType) for member in field_type.types + ) + + def coerce(value: Any) -> Any: + if value is None: + return None if nullable else _DROP + return inner(value) + + return coerce + + +def _plan_for(cls: type) -> Dict[str, _Coercer]: + plan = _plans.get(cls) + if plan is not None: + return plan + plan = {} + # Publish before walking: a self-referential type graph resolves to this + # same (eventually populated) dict instead of recursing forever. + _plans[cls] = plan + info = _inspect.type_info(cls) + for field in getattr(info, "fields", ()): + coercer = _field_coercer(field.type) + if coercer is not None: + plan[field.encode_name] = coercer + return plan + + +def coerce_string_slots(data: Dict[str, Any], cls: type) -> Dict[str, Any]: + """Coerce, in place, every value in ``data`` bound for a ``str`` slot of ``cls``. + + Args: + data: A flattened entity dict, keyed by Atlas (camelCase) field names. + cls: The model class ``data`` is about to be converted into. + + Returns: + The same dict, mutated. + """ + plan = _plan_for(cls) + if not plan: + return data + for key in list(data): + coercer = plan.get(key) + if coercer is None: + continue + coerced = coercer(data[key]) + if coerced is _DROP: + del data[key] + else: + data[key] = coerced + return data diff --git a/pyatlan_v9/model/transform.py b/pyatlan_v9/model/transform.py index 36806cfee..f2366f6df 100644 --- a/pyatlan_v9/model/transform.py +++ b/pyatlan_v9/model/transform.py @@ -19,6 +19,7 @@ import msgspec from pyatlan_v9.model.assets.asset import Asset +from pyatlan_v9.model.string_coercion import coerce_string_slots _T = TypeVar("_T") @@ -248,7 +249,7 @@ def _flatten_entity_dict(data: dict[str, Any]) -> dict[str, Any]: return flattened -def from_atlas_format(data: dict[str, Any]) -> Asset: +def from_atlas_format(data: dict[str, Any], *, lenient_strings: bool = False) -> Asset: """Convert Atlas API format to a flattened msgspec Struct. Takes an entity from the Atlas API and converts it to the SDK's @@ -256,6 +257,14 @@ def from_atlas_format(data: dict[str, Any]) -> Asset: Args: data: A dictionary in Atlas API format. + lenient_strings: Coerce values bound for ``str``-typed slots the way + Atlas's own ``string`` type does, instead of rejecting the record. + Off by default, so decoding an Atlas API response is unchanged. + Turn it on when decoding a payload that has **not** been through + Atlas's normalisation - connector output, fixtures, a local + validator - where an ``int`` in a ``Dict[str, str]`` is a producer + quirk Atlas would have stringified, not a contract breach. + See :mod:`pyatlan_v9.model.string_coercion`. Returns: The appropriate Asset subclass instance with flattened attributes. @@ -265,10 +274,18 @@ def from_atlas_format(data: dict[str, Any]) -> Asset: flattened = _flatten_entity_dict(data) + if lenient_strings: + coerce_string_slots(flattened, cls) + return msgspec.convert(flattened, cls, strict=False) -def from_atlas_json(json_bytes: bytes, type_name: str | None = None) -> Asset: # noqa: ARG001 +def from_atlas_json( + json_bytes: bytes, + type_name: str | None = None, # noqa: ARG001 + *, + lenient_strings: bool = False, +) -> Asset: """Convert Atlas API JSON bytes directly to a flattened msgspec Struct. This is the fastest path - parses JSON and converts in minimal steps. @@ -276,6 +293,7 @@ def from_atlas_json(json_bytes: bytes, type_name: str | None = None) -> Asset: Args: json_bytes: Raw JSON bytes from the API response. type_name: Optional type name hint (if known ahead of time). + lenient_strings: Passed through to :func:`from_atlas_format`. Returns: The appropriate Asset subclass instance with flattened attributes. @@ -287,7 +305,7 @@ def from_atlas_json(json_bytes: bytes, type_name: str | None = None) -> Asset: if "entity" in data: data = data["entity"] - return from_atlas_format(data) + return from_atlas_format(data, lenient_strings=lenient_strings) def to_bulk_payload(entities: list[Asset]) -> dict[str, Any]: diff --git a/tests_v9/unit/test_string_coercion.py b/tests_v9/unit/test_string_coercion.py new file mode 100644 index 000000000..d74bdd397 --- /dev/null +++ b/tests_v9/unit/test_string_coercion.py @@ -0,0 +1,276 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. +"""Tests for ``lenient_strings`` decoding. + +Atlas's ``string`` type accepts any value and stores its stringification +(``AtlasBuiltInTypes.AtlasStringType``: ``isValidValue`` returns ``true``, +``getNormalizedValue`` returns ``obj.toString()``). The generated models mirror +the declared Atlas type but not that leniency, so a producer that has not been +through Atlas's normalisation gets whole records rejected over one value. + +Every payload shape below was observed in connector output that Atlas itself +accepted. +""" + +from __future__ import annotations + +import json + +import msgspec +import pytest + +from pyatlan_v9.model.transform import ( + from_atlas_format, + from_atlas_json, + get_type, +) + + +def _column(custom: object, *, omit: bool = False) -> dict: + entity = { + "typeName": "Column", + "attributes": {"name": "c1", "qualifiedName": "default/db/sch/tbl/c1"}, + } + if not omit: + entity["customAttributes"] = custom + return entity + + +# --------------------------------------------------------------------------- +# Dict[str, str] - the entity-header ``customAttributes`` slot +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("custom", "expected"), + [ + pytest.param({"type_name": "int4"}, {"type_name": "int4"}, id="already-str"), + pytest.param({"ordinal_position": 1}, {"ordinal_position": "1"}, id="int"), + pytest.param({"len": -1.0}, {"len": "-1.0"}, id="float"), + # JSON text, so "true" - matching what Atlas stores, not Python's repr. + pytest.param({"is_secure": True}, {"is_secure": "true"}, id="bool"), + # Dropped, not stringified: "null" is a value the producer never wrote. + pytest.param({"numeric_precision": None}, {}, id="null-value"), + pytest.param({"a": 1, "b": None, "c": "x"}, {"a": "1", "c": "x"}, id="mixed"), + pytest.param({}, {}, id="empty"), + # A container in a string slot: Atlas would store Java's toString(); + # this stores compact JSON. + pytest.param({"k": {"n": "o"}}, {"k": '{"n":"o"}'}, id="nested-object"), + pytest.param({"k": [1, 2]}, {"k": "[1,2]"}, id="nested-array"), + ], +) +def test_string_map_values_are_coerced(custom, expected): + asset = from_atlas_format(_column(custom), lenient_strings=True) + + assert asset.custom_attributes == expected + + +def test_null_map_is_dropped_because_the_slot_is_not_optional(): + """``Dict[str, str]`` with no ``None`` member cannot hold an explicit null.""" + asset = from_atlas_format(_column(None), lenient_strings=True) + + assert asset.custom_attributes is msgspec.UNSET + + +def test_absent_map_is_untouched(): + asset = from_atlas_format(_column(None, omit=True), lenient_strings=True) + + assert asset.custom_attributes is msgspec.UNSET + + +def test_non_mapping_payload_still_raises(): + """No sane coercion exists, and inventing one would hide producer drift.""" + with pytest.raises(msgspec.ValidationError): + from_atlas_format(_column("not-a-map"), lenient_strings=True) + + +# --------------------------------------------------------------------------- +# List[Dict[str, str]] - a typedef attribute (``array>``) +# --------------------------------------------------------------------------- + + +def _datasource_field(upstream: object) -> dict: + return { + "typeName": "TableauDatasourceField", + "attributes": { + "name": "f", + "qualifiedName": "default/tableau/1/p/ds/f", + "upstreamColumns": upstream, + }, + } + + +@pytest.mark.parametrize( + ("upstream", "expected"), + [ + pytest.param([{"id": "7"}], [{"id": "7"}], id="already-str"), + pytest.param([{"id": 7}], [{"id": "7"}], id="int"), + pytest.param( + [{"col": {"nested": "obj"}}], [{"col": '{"nested":"obj"}'}], id="object" + ), + pytest.param([{"a": 1}, {"b": None}], [{"a": "1"}, {}], id="ragged"), + pytest.param([], [], id="empty"), + ], +) +def test_list_of_string_maps_is_coerced(upstream, expected): + asset = from_atlas_format(_datasource_field(upstream), lenient_strings=True) + + assert asset.upstream_columns == expected + + +# --------------------------------------------------------------------------- +# Plain ``str`` slots, and nullability +# --------------------------------------------------------------------------- + + +def test_scalar_in_a_plain_string_slot_is_coerced(): + entity = { + "typeName": "Column", + "attributes": {"name": 12345, "qualifiedName": "default/db/sch/tbl/c1"}, + } + + asset = from_atlas_format(entity, lenient_strings=True) + + assert asset.name == "12345" + + +def test_null_is_preserved_in_an_optional_string_slot(): + """``name`` is ``Union[str, None]``; a declared null is a real value there.""" + entity = { + "typeName": "Column", + "attributes": {"name": None, "qualifiedName": "default/db/sch/tbl/c1"}, + } + + asset = from_atlas_format(entity, lenient_strings=True) + + assert asset.name is None + + +# --------------------------------------------------------------------------- +# Nested related structs - the shape a name-keyed fixup would miss +# --------------------------------------------------------------------------- + + +def test_nested_related_struct_is_coerced(): + """``RelatedTableauDatasourceField.upstream_columns`` is a typed slot too. + + It is reached through a relationship attribute, not the top level, so the + coercion has to follow the model's own nesting. + """ + entity = { + "typeName": "TableauWorksheet", + "attributes": {"name": "w", "qualifiedName": "default/tableau/1/p/wb/w"}, + "relationshipAttributes": { + "datasourceFields": [ + { + "typeName": "TableauDatasourceField", + "guid": "abc", + "attributes": { + "name": "f", + "upstreamColumns": [{"id": 7}], + }, + } + ] + }, + } + + asset = from_atlas_format(entity, lenient_strings=True) + + assert asset.datasource_fields[0].upstream_columns == [{"id": "7"}] + + +# --------------------------------------------------------------------------- +# The flag is off by default, and genuine breakage still raises +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "entity", + [ + pytest.param(_column({"ordinal_position": 1}), id="int-in-map"), + pytest.param(_datasource_field([{"col": {"n": "o"}}]), id="object-in-list-map"), + ], +) +def test_default_is_unchanged(entity): + """Decoding an Atlas API response must behave exactly as before.""" + with pytest.raises(msgspec.ValidationError): + from_atlas_format(entity) + + with pytest.raises(msgspec.ValidationError): + from_atlas_json(json.dumps(entity).encode()) + + +@pytest.mark.parametrize( + "entity", + [ + pytest.param( + { + "typeName": "Column", + "attributes": {"name": "c1", "order": {"not": "an int"}}, + }, + id="wrong-typed-non-string-field", + ), + pytest.param( + { + "typeName": "Column", + "attributes": {"name": "c1"}, + "createTime": {"not": "an int"}, + }, + id="wrong-typed-top-level-field", + ), + ], +) +def test_non_string_slots_are_left_strict(entity): + """Leniency is confined to ``str`` slots; everything else still raises.""" + with pytest.raises(msgspec.ValidationError): + from_atlas_format(entity, lenient_strings=True) + + +def test_malformed_json_still_raises(): + with pytest.raises(msgspec.DecodeError): + from_atlas_json(b"{not json", lenient_strings=True) + + +def test_json_entry_point_applies_the_flag(): + asset = from_atlas_json( + json.dumps(_column({"ordinal_position": 1})).encode(), lenient_strings=True + ) + + assert asset.custom_attributes == {"ordinal_position": "1"} + + +def test_wrapped_entity_shape_applies_the_flag(): + payload = {"entity": _column({"ordinal_position": 1})} + + asset = from_atlas_json(json.dumps(payload).encode(), lenient_strings=True) + + assert asset.custom_attributes == {"ordinal_position": "1"} + + +# --------------------------------------------------------------------------- +# The plan is derived from the model, not a hand-maintained field list +# --------------------------------------------------------------------------- + + +def test_plan_covers_every_string_map_field_of_a_type(): + """A new ``map`` typedef attribute is covered on regeneration. + + Guards the property that matters: the plan comes from the model, so nothing + has to be added here when the models are regenerated. + """ + from pyatlan_v9.model.string_coercion import _plan_for + + plan = _plan_for(get_type("TableauDatasourceField")) + + assert "upstreamColumns" in plan + assert "upstreamFields" in plan + assert "customAttributes" in plan + assert "qualifiedName" in plan + + +def test_plan_is_cached_per_class(): + from pyatlan_v9.model.string_coercion import _plan_for + + cls = get_type("Column") + + assert _plan_for(cls) is _plan_for(cls)