Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 18 additions & 28 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -217,57 +217,47 @@ The `kognic.auth.serde` module provides utilities for serializing request bodies

### Serialization

`serialize_body()` converts objects to JSON-compatible dicts. Supports:
- Pydantic v2 models (`model_dump()`)
- Objects with `to_json()` or `to_dict()` methods
- Nested objects in dicts/lists are recursively serialized
`serialize_body()` converts dicts, lists, and primitives to JSON-compatible format.

```python
from pydantic import BaseModel
from kognic.auth.serde import serialize_body

serialize_body({"name": "test", "value": 42}) # {"name": "test", "value": 42}
serialize_body([1, 2, 3]) # [1, 2, 3]

# For Pydantic models, convert to dict first
from pydantic import BaseModel

class CreateRequest(BaseModel):
name: str
value: int

# Pydantic models
request = CreateRequest(name="test", value=42)
serialize_body(request) # {"name": "test", "value": 42}

# Nested in containers
serialize_body({"items": [request]}) # {"items": [{"name": "test", "value": 42}]}

# Custom classes with to_dict()
class MyModel:
def to_dict(self):
return {"key": "value"}

serialize_body(MyModel()) # {"key": "value"}
serialize_body(request.model_dump()) # {"name": "test", "value": 42}
```

### Deserialization

`deserialize()` extracts and converts API responses. Supports:
- Pydantic v2 models (`model_validate()`)
- Classes with `from_dict()` or `from_json()` methods
- Automatic envelope extraction (default key: `"data"`)
`deserialize()` extracts raw data from API responses with automatic envelope extraction (default key: `"data"`). Object conversion is done outside of the call.

```python
from kognic.auth.serde import deserialize

# Deserialize to Pydantic model
# Returns raw dict
response = client.session.get("https://api.app.kognic.com/v1/resource/123")
resource = deserialize(response, cls=ResourceModel)
data = deserialize(response)

# For Pydantic models, convert after
resource = ResourceModel.model_validate(data)

# Deserialize list of models
response = client.session.get("https://api.app.kognic.com/v1/resources")
resources = deserialize(response, cls=list[ResourceModel])
# For classes with from_dict()
resource = ResourceModel.from_dict(data)

# Custom envelope key
data = deserialize(response, cls=MyModel, enveloped_key="result")
data = deserialize(response, enveloped_key="result")

# No envelope
data = deserialize(response, cls=MyModel, enveloped_key=None)
data = deserialize(response, enveloped_key=None)
```

## Changelog
Expand Down
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,6 @@ full = [
dev = [
"httpx>=0.20,<1",
"requests>=2.20,<3",
"pydantic>=2",
"pytest",
"keyring>=23.0",
]
Expand Down
93 changes: 5 additions & 88 deletions src/kognic/auth/serde.py
Original file line number Diff line number Diff line change
@@ -1,40 +1,15 @@
"""Serialization and deserialization utilities for HTTP request/response bodies."""

from collections.abc import MutableMapping, MutableSequence
from typing import Any, Dict, List, Optional, Type, Union
from typing import Any, Dict, List, Optional, Union

ENVELOPED_KEY = "data"


def _is_list_type(cls: Type) -> bool:
"""Check if cls is a list-like type (list, List, List[T], or MutableSequence subclass)."""
try:
return issubclass(cls.__origin__, MutableSequence)
except (AttributeError, TypeError):
try:
return issubclass(cls, MutableSequence)
except TypeError:
return False


def _is_dict_type(cls: Type) -> bool:
"""Check if cls is a dict-like type (dict, Dict, Dict[K,V], or MutableMapping subclass)."""
try:
return issubclass(cls.__origin__, MutableMapping)
except (AttributeError, TypeError):
try:
return issubclass(cls, MutableMapping)
except TypeError:
return False


def serialize_body(body: Any) -> Any:
"""Serialize request body to JSON-compatible format.

Supports:
- None, dict, list, primitives (passed through)
- Objects with to_json() or to_dict method (duck typing)
- Nested objects inside containers are recursively serialized

Raises:
ValueError: If body is str or bytes at top level (not supported as request body)
Expand All @@ -59,20 +34,11 @@ def _serialize_value(value: Any) -> Any:
return {k: _serialize_value(v) for k, v in value.items()}
if isinstance(value, list):
return [_serialize_value(item) for item in value]
if hasattr(value, "model_dump") and callable(value.model_dump): # Pydantic v2
return _serialize_value(value.model_dump())
if hasattr(value, "to_json") and callable(value.to_json):
return _serialize_value(value.to_json())
if hasattr(value, "to_dict") and callable(value.to_dict):
return _serialize_value(value.to_dict())
raise TypeError(
f"Cannot serialize value of type {type(value).__name__}. Expected dict, list, primitive, or Serializable."
)
raise TypeError(f"Cannot serialize value of type {type(value).__name__}. Expected dict, list, or primitive.")


def deserialize(
response: Union[Any, Dict[str, Any], List],
cls: Optional[Type] = None,
enveloped_key: Optional[str] = ENVELOPED_KEY,
) -> Any:
"""Deserialize a response from the API.
Expand All @@ -81,14 +47,11 @@ def deserialize(

Args:
response: Response object (with .json() method) or dict/list
cls: Optional type hint for the expected return type. For basic types
(dict, list) the data is returned as-is. For model classes,
kognic-common must be installed.
enveloped_key: By Kognic convention, data is enveloped in a key.
Default is 'data'. Set to None to skip envelope extraction.

Returns:
Deserialized data
Deserialized data as raw dict/list

Raises:
ValueError: If enveloped_key is specified but not found in response
Expand All @@ -106,52 +69,6 @@ def deserialize(
f"Expected enveloped key '{enveloped_key}' not found in response json. "
f"Found keys: {response_json.keys()}"
)
data = response_json[enveloped_key]
else:
data = response_json

# Return raw data if no class specified
if cls is None:
return data

# For dict-like types, return the data as-is
if _is_dict_type(cls):
return data

# Handle list-like types
if _is_list_type(cls):
args = getattr(cls, "__args__", ())
if not args:
return data
inner_cls = args[0]
# If inner type is a basic type or generic, return as-is
if inner_cls in (dict, list, str, int, float, bool) or _is_dict_type(inner_cls) or _is_list_type(inner_cls):
return data
# Deserialize each item using inner class
return [_deserialize_object(item, inner_cls) for item in data]

# Single object deserialization
return _deserialize_object(data, cls)


def _deserialize_object(data: Any, cls: Type) -> Any:
"""Deserialize a single object using duck-typed methods.

Supports:
- Pydantic v2 models (model_validate)
- Classes with from_dict() class method
- Classes with from_json() class method
"""
if hasattr(cls, "model_validate") and callable(cls.model_validate): # Pydantic v2
return cls.model_validate(data)

if hasattr(cls, "from_dict") and callable(cls.from_dict):
return cls.from_dict(data)

if hasattr(cls, "from_json") and callable(cls.from_json):
return cls.from_json(data)
return response_json[enveloped_key]

raise TypeError(
f"Cannot deserialize to {cls.__name__}. "
f"Class must have model_validate(), from_dict(), or from_json() class method."
)
return response_json
112 changes: 10 additions & 102 deletions tests/test_deserialization.py
Original file line number Diff line number Diff line change
@@ -1,146 +1,54 @@
"""Unit tests for deserialization utilities."""

import unittest
from typing import Dict

from httpx import Response

from kognic.auth.serde import deserialize


class TestDeserialize(unittest.TestCase):
def test_deserialize_to_Dict(self):
resp = Response(200, json={"data": {"key": "value"}})
val = deserialize(resp, cls=Dict[str, str])
self.assertEqual(val, {"key": "value"})

def test_deserialize_to_dict(self):
resp = Response(200, json={"data": {"key": "value"}})
val = deserialize(resp, cls=dict[str, str])
self.assertEqual(val, {"key": "value"})

def test_deserialize_to_raw(self):
resp = Response(200, json={"data": {"key": "value"}})
val = deserialize(resp, cls=None)
val = deserialize(resp)
self.assertEqual(val, {"key": "value"})

def test_deserialize_with_custom_envelope_key(self):
resp = {"custom_key": {"key": "value"}}
val = deserialize(resp, cls=Dict[str, str], enveloped_key="custom_key")
val = deserialize(resp, enveloped_key="custom_key")
self.assertEqual(val, {"key": "value"})

def test_deserialize_from_dict(self):
resp = {"data": {"key": "value"}}
val = deserialize(resp, cls=Dict[str, str])
val = deserialize(resp)
self.assertEqual(val, {"key": "value"})

def test_deserialize_to_list(self):
def test_deserialize_list(self):
resp = Response(200, json={"data": [1, 2, 3]})
val = deserialize(resp, cls=list[int])
val = deserialize(resp)
self.assertEqual(val, [1, 2, 3])

def test_deserialize_to_list_of_dicts(self):
def test_deserialize_list_of_dicts(self):
resp = Response(200, json={"data": [{"key": "value"}]})
val = deserialize(resp, cls=list[Dict[str, str]])
val = deserialize(resp)
self.assertEqual(val, [{"key": "value"}])

def test_deserialize_empty_list(self):
resp = Response(200, json={"data": []})
val = deserialize(resp, cls=list[Dict[str, str]])
val = deserialize(resp)
self.assertEqual(val, [])

def test_deserialize_no_envelope(self):
resp = Response(200, json={"key": "value"})
val = deserialize(resp, cls=None, enveloped_key=None)
val = deserialize(resp, enveloped_key=None)
self.assertEqual(val, {"key": "value"})

def test_deserialize_missing_envelope_key_raises(self):
resp = Response(200, json={"wrong_key": {"key": "value"}})
with self.assertRaises(ValueError) as context:
deserialize(resp, cls=None)
deserialize(resp)
self.assertIn("Expected enveloped key 'data' not found", str(context.exception))

def test_deserialize_to_class_with_from_dict(self):
class MyModel:
def __init__(self, key: str):
self.key = key

@classmethod
def from_dict(cls, data: dict):
return cls(key=data["key"])

resp = Response(200, json={"data": {"key": "value"}})
result = deserialize(resp, cls=MyModel)
self.assertIsInstance(result, MyModel)
self.assertEqual(result.key, "value")

def test_deserialize_to_class_with_from_json(self):
class MyModel:
def __init__(self, key: str):
self.key = key

@classmethod
def from_json(cls, data: dict):
return cls(key=data["key"])

resp = Response(200, json={"data": {"key": "value"}})
result = deserialize(resp, cls=MyModel)
self.assertIsInstance(result, MyModel)
self.assertEqual(result.key, "value")

def test_deserialize_list_to_class_with_from_dict(self):
class MyModel:
def __init__(self, key: str):
self.key = key

@classmethod
def from_dict(cls, data: dict):
return cls(key=data["key"])

resp = Response(200, json={"data": [{"key": "a"}, {"key": "b"}]})
result = deserialize(resp, cls=list[MyModel])
self.assertIsInstance(result, list)
self.assertEqual(len(result), 2)
self.assertIsInstance(result[0], MyModel)
self.assertEqual(result[0].key, "a")
self.assertEqual(result[1].key, "b")

def test_deserialize_list_to_class_with_from_json(self):
class MyModel:
def __init__(self, key: str):
self.key = key

@classmethod
def from_json(cls, data: dict):
return cls(key=data["key"])

resp = Response(200, json={"data": [{"key": "x"}, {"key": "y"}]})
result = deserialize(resp, cls=list[MyModel])
self.assertIsInstance(result, list)
self.assertEqual(len(result), 2)
self.assertEqual(result[0].key, "x")
self.assertEqual(result[1].key, "y")

def test_deserialize_empty_list_to_class(self):
class MyModel:
@classmethod
def from_dict(cls, data: dict):
return cls()

resp = Response(200, json={"data": []})
result = deserialize(resp, cls=list[MyModel])
self.assertEqual(result, [])

def test_deserialize_unsupported_class_raises(self):
class UnsupportedModel:
pass

resp = Response(200, json={"data": {"key": "value"}})
with self.assertRaises(TypeError) as context:
deserialize(resp, cls=UnsupportedModel)
self.assertIn("Cannot deserialize to UnsupportedModel", str(context.exception))
self.assertIn("from_dict()", str(context.exception))


if __name__ == "__main__":
unittest.main()
Loading