jsonapi-pydantic library

Author: isivaselvan

src/pytfe/jsonapi/__init__.py

@@ -0,0 +1,11 @@
+"""JSON:API unmarshaling library for python-tfe."""
+
+from .types import IncludedIndex, JSONAPIResponse
+from .unmarshaler import unmarshal_many_payload, unmarshal_payload
+
+__all__ = [
+    "unmarshal_payload",
+    "unmarshal_many_payload",
+    "JSONAPIResponse",
+    "IncludedIndex",
+]

src/pytfe/jsonapi/metadata.py

@@ -0,0 +1,138 @@
+"""Field metadata extractors for Pydantic models."""
+
+import inspect
+from typing import Any, get_args, get_origin, get_type_hints
+
+from pydantic import BaseModel
+from pydantic.fields import FieldInfo
+
+
+class FieldMetadata:
+    """Metadata about a Pydantic model field."""
+
+    def __init__(
+        self,
+        field_name: str,
+        field_type: type,
+        jsonapi_type: str | None = None,
+        jsonapi_name: str | None = None,
+        is_optional: bool = False,
+        is_list: bool = False,
+        inner_type: type | None = None,
+    ):
+        self.field_name = field_name
+        self.field_type = field_type
+        self.jsonapi_type = jsonapi_type or self._infer_jsonapi_type()
+        self.jsonapi_name = jsonapi_name or self._convert_to_jsonapi_name(field_name)
+        self.is_optional = is_optional
+        self.is_list = is_list
+        self.inner_type = inner_type
+
+    def _infer_jsonapi_type(self) -> str:
+        """Infer JSON:API type from field name patterns."""
+        if self.field_name == "id":
+            return "primary"
+        return "attribute"
+
+    def _convert_to_jsonapi_name(self, field_name: str) -> str:
+        """Convert Python field name to JSON:API name (snake_case to kebab-case)."""
+        return field_name.replace("_", "-")
+
+
+def get_model_metadata(model_class: type[BaseModel]) -> dict[str, FieldMetadata]:
+    """Extract metadata for all fields in a Pydantic model.
+
+    Returns:
+        Dict mapping field name to FieldMetadata
+    """
+    metadata: dict[str, FieldMetadata] = {}
+
+    # Get type hints
+    try:
+        type_hints = get_type_hints(model_class, include_extras=True)
+    except Exception:
+        type_hints = {}
+
+    # Iterate through model fields
+    for field_name, field_info in model_class.model_fields.items():
+        field_type = type_hints.get(field_name, field_info.annotation)
+
+        # Check for Field metadata
+        jsonapi_type = None
+        jsonapi_name = None
+
+        if isinstance(field_info, FieldInfo):
+            # Extract custom metadata from Field()
+            if field_info.json_schema_extra and isinstance(
+                field_info.json_schema_extra, dict
+            ):
+                jsonapi_type = field_info.json_schema_extra.get("jsonapi_type")
+                jsonapi_name = field_info.json_schema_extra.get("jsonapi_name")
+            else:
+                jsonapi_type = None
+                jsonapi_name = None
+
+            # If no explicit jsonapi_name, use the Pydantic alias if available
+            if not jsonapi_name and field_info.alias:
+                jsonapi_name = field_info.alias
+
+        # Handle Optional types
+        is_optional = False
+        is_list = False
+        inner_type = field_type
+
+        origin = get_origin(field_type)
+        args = get_args(field_type)
+
+        # Check for Optional (Union with None) - handles both Optional[X] and X | None
+        import types
+
+        if origin is types.UnionType or (
+            hasattr(types, "Union") and origin is getattr(types, "Union", None)
+        ):
+            if type(None) in args:
+                is_optional = True
+                # Get the non-None type
+                inner_type = next(
+                    (arg for arg in args if arg is not type(None)), field_type
+                )
+
+        # Check for List
+        if get_origin(inner_type) is list:
+            is_list = True
+            list_args = get_args(inner_type)
+            if list_args:
+                inner_type = list_args[0]
+
+        # Ensure proper types for FieldMetadata
+        jsonapi_type_str: str | None = None
+        if jsonapi_type is not None:
+            jsonapi_type_str = (
+                str(jsonapi_type) if not isinstance(jsonapi_type, str) else jsonapi_type
+            )
+
+        jsonapi_name_str: str | None = None
+        if jsonapi_name is not None:
+            jsonapi_name_str = (
+                str(jsonapi_name) if not isinstance(jsonapi_name, str) else jsonapi_name
+            )
+
+        metadata[field_name] = FieldMetadata(
+            field_name=field_name,
+            field_type=field_type,  # type: ignore[arg-type]
+            jsonapi_type=jsonapi_type_str,
+            jsonapi_name=jsonapi_name_str,
+            is_optional=is_optional,
+            is_list=is_list,
+            inner_type=inner_type,
+        )
+
+    return metadata
+
+
+def is_pydantic_model(obj: Any) -> bool:
+    """Check if object is a Pydantic model class."""
+    try:
+        return inspect.isclass(obj) and issubclass(obj, BaseModel)
+    except TypeError:
+        return False

src/pytfe/jsonapi/types.py

@@ -0,0 +1,81 @@
+from typing import Any, Generic, TypeVar
+
+from pydantic import BaseModel
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class JSONAPINode:
+    """Represents a JSON:API resource object node."""
+
+    def __init__(self, data: dict[str, Any]):
+        self.id: str = data.get("id", "")
+        self.type: str = data.get("type", "")
+        self.attributes: dict[str, Any] = data.get("attributes", {})
+        self.relationships: dict[str, Any] = data.get("relationships", {})
+        self.links: dict[str, Any] | None = data.get("links")
+        self.meta: dict[str, Any] | None = data.get("meta")
+        self._raw_data = data
+
+    def get_relationship_linkage(
+        self, rel_name: str
+    ) -> dict[str, Any] | list[dict[str, Any]] | None:
+        """Extract relationship linkage data (type and id)."""
+        if not self.relationships or rel_name not in self.relationships:
+            return None
+
+        rel_data = self.relationships[rel_name].get("data")
+        if rel_data is None:
+            return None
+        # Can be dict or list of dicts based on relationship type
+        return rel_data  # type: ignore[no-any-return]
+
+
+class IncludedIndex:
+    """Index for fast lookup of included resources."""
+
+    def __init__(self, included: list[dict[str, Any]] | None = None):
+        self._index: dict[tuple[str, str], JSONAPINode] = {}
+
+        if included:
+            for item in included:
+                node = JSONAPINode(item)
+                if node.type and node.id:
+                    key = (node.type, node.id)
+                    self._index[key] = node
+
+    def get(self, resource_type: str, resource_id: str) -> JSONAPINode | None:
+        """Lookup a resource by type and id."""
+        return self._index.get((resource_type, resource_id))
+
+    def resolve_relationship(
+        self, rel_data: dict[str, Any] | None
+    ) -> JSONAPINode | None:
+        """Resolve a relationship linkage to full node."""
+        if not rel_data or not isinstance(rel_data, dict):
+            return None
+
+        resource_type = rel_data.get("type")
+        resource_id = rel_data.get("id")
+
+        if not resource_type or not resource_id:
+            return None
+
+        return self.get(resource_type, resource_id)
+
+
+class JSONAPIResponse(Generic[T]):
+    """Complete JSON:API response with data and included."""
+
+    def __init__(self, response_dict: dict[str, Any]):
+        self.data: dict[str, Any] | list[dict[str, Any]] = response_dict.get("data", {})
+        self.included: list[dict[str, Any]] = response_dict.get("included", [])
+        self.links: dict[str, Any] | None = response_dict.get("links")
+        self.meta: dict[str, Any] | None = response_dict.get("meta")
+
+        # Build included index
+        self.included_index = IncludedIndex(self.included)
+
+    def is_collection(self) -> bool:
+        """Check if data is a collection (list) or single resource."""
+        return isinstance(self.data, list)