Last active
March 4, 2024 19:27
-
-
Save rohanrhu/8119a461b736e695607e65704a39cd25 to your computer and use it in GitHub Desktop.
Meowing Cat's Marshaling Utilities
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- coding: utf-8 -*- | |
# Meowing Cat's Marshaling Utilities | |
# Copyright (C) 2024 Oğuzhan Eroğlu <[email protected]> (https://github.com/rohanrhu) | |
# Licensed under the MIT License. | |
# You may obtain a copy of the License at: https://opensource.org/licenses/MIT | |
""" | |
* Marshaling utilities for database models and things. | |
# ! IMPORTANT: | |
# * @serializable_class is a decorator to make a class serializable. | |
# * It avoids infinite recursions by using the `recursion_alternative` parameter if it is provided | |
# * or if it is not provided, it avoids infinite recursions by using circular references. | |
# * For JSON serialization, you must avoid circular references too by using the `recursion_alternative` parameter. | |
* Usage: | |
```python | |
# -*- coding: utf-8 -*- | |
import json | |
import sys | |
sys.path.append(".") | |
from lib.marshal import serializable_class | |
@serializable_class | |
class Kind: | |
id: int = 0 | |
name: str = "" | |
is_active: bool = False | |
cats: list = [] | |
def __init__(self, p_id: int, p_name: str = "", p_is_active: bool = False, p_cats: list = []) -> None: | |
self.id = p_id | |
self.name = p_name | |
self.is_active = p_is_active | |
self.cats = p_cats | |
self.serializable_property("id") | |
self.serializable_property("name") | |
self.serializable_property("is_active") | |
self.serializable_property("cats") | |
@serializable_class | |
class Cat: | |
id: int = 0 | |
name: str = "" | |
is_active: bool = False | |
kind: Kind | |
def __init__(self, p_id: int, p_name: str = "", p_is_active: bool = False, p_kind: Kind = None) -> None: | |
self.id = p_id | |
self.name = p_name | |
self.is_active = p_is_active | |
self.kind = p_kind | |
self.serializable_property("id") | |
self.serializable_property("name") | |
self.serializable_property("is_active") | |
self.serializable_property("kind", p_recursion_alternative="id") | |
# self.serializable_property("kind") # .. or use for circular references. | |
# ... or | |
# self.serializable_property(self.kind, p_recursion_alternative=lambda kind: kind.id) | |
# ... or | |
# self.serializable_property(self.kind, p_serializable_prop="...") | |
# ... or | |
# self.serializable_property(self.kind, p_serializable_prop=lambda kind: kind.id) | |
# ... or | |
# self.serializable_property(self.kind, p_serializable_prop="...") | |
# if `p_recursion_alternative` is not provided, serialized data still will be recursive | |
# The `p_recursion_alternative` is called like this: | |
# ```python | |
# recursion_alternative(obj, level, serializeds) | |
# ``` | |
kind = Kind( | |
p_id = 1, | |
p_name = "Tabby Cat", | |
p_is_active = True | |
) | |
instance = Cat( | |
p_id = 1, | |
p_name = "Meowing Cat", | |
p_is_active = True, | |
p_kind = kind | |
) | |
kind.cats.append(instance) | |
serialized = kind.serializable | |
print(json.dumps(serialized, indent=4)) | |
``` | |
""" | |
from typing import Any, Callable, AnyStr, Hashable | |
def is_mutable(obj): | |
immutable_types = (int, float, bool, str, tuple, frozenset, bytes, type(None), range) | |
return not isinstance(obj, immutable_types) | |
class SerializableProperty: | |
name: AnyStr | |
value: AnyStr | Callable | |
recursion_alternative: AnyStr | Callable | None | |
def __init__(self, p_name: AnyStr, p_value: AnyStr | Callable, p_recursion_alternative: AnyStr | Callable | None = None): | |
self.name = p_name | |
self.value = p_value | |
self.recursion_alternative = p_recursion_alternative | |
def get_serializables(p_cls) -> dict: | |
if not hasattr(p_cls, "__serializables__"): | |
raise TypeError(f"{p_cls} is not a serializable class.\n\ | |
Hint: Did you forget to add the @serializable_class decorator?") | |
return getattr(p_cls, "__serializables__", {}) | |
def get_serializable_property(p_obj: Any, p_prop: AnyStr | None) -> SerializableProperty | None: | |
if not hasattr(p_obj, "__serializables__"): | |
return None | |
if p_prop is None: | |
return None | |
serializables = get_serializables(p_obj) | |
serializable_property = None | |
if p_prop in serializables.keys(): | |
serializable_property = serializables[p_prop] | |
return serializable_property | |
def serializable_property( | |
p_obj: Any, | |
p_prop: AnyStr, | |
p_serializable_prop: AnyStr | Callable | None = None, | |
p_recursion_alternative: AnyStr | Callable | None = None | |
): | |
""" | |
Makes a property serializable. | |
:param cls: The class to add the property to. | |
:param prop: The property to make serializable. | |
:param serializable_prop: The name of the serializable property. | |
If string, the provided property will be used. | |
If function, the provided function will be | |
called with the property's value. | |
If not provided, the property's itself will be used. | |
:param recursion_alternative: An alternative function to use for recursion. | |
If not provided, the default serialization | |
function will be used. | |
""" | |
if not hasattr(p_obj, "__serializables__"): | |
raise TypeError(f"{p_obj} is not a serializable class.\n\ | |
Hint: Did you forget to add the @serializable_class decorator?") | |
if not hasattr(p_obj, p_prop): | |
return None | |
serializable_property = SerializableProperty( | |
p_prop, | |
p_serializable_prop or getattr(p_obj, p_prop), | |
p_recursion_alternative | |
) | |
get_serializables(p_obj)[p_prop] = serializable_property | |
def hash_or_id(p_obj: Any) -> int: | |
if isinstance(p_obj, Hashable): | |
return hash(p_obj) | |
return id(p_obj) | |
def serialize(p_obj: Any, p_name: AnyStr = None, p_level=0, p_serializeds=None, p_parent=None): | |
""" | |
Serialize an object to a dictionary. | |
:param obj: The object to serialize. | |
:param level: The current level of recursion. | |
""" | |
if p_serializeds is None: | |
p_serializeds = {} | |
if not hasattr(p_obj, "__serializables__") and p_level == 0: | |
raise TypeError(f"{p_obj} is not a serializable property.\n\ | |
Hint: Did you forget to add self.serializable_property(self.prop)?") | |
serialize_with_level = lambda obj, prop=None, parent=None: serialize(obj, prop, p_level + 1, p_serializeds, parent) | |
if hash_or_id(p_obj) in p_serializeds.keys(): | |
obj_serializable = p_serializeds[hash_or_id(p_obj)] | |
serializable_property = get_serializable_property(p_parent, p_name) | |
if serializable_property: | |
if serializable_property.recursion_alternative is not None: | |
recursion_alternative = serializable_property.recursion_alternative | |
if isinstance(recursion_alternative, str): | |
return serialize_with_level(getattr(p_obj, recursion_alternative)) | |
if callable(recursion_alternative): | |
return serialize_with_level(recursion_alternative(p_obj, p_level, p_serializeds)) | |
if recursion_alternative is None: | |
return None | |
raise TypeError(f"Recursion alternative must be a string\ | |
or a function, not {type(recursion_alternative)}") | |
return obj_serializable | |
if hasattr(p_obj, "__serializables__"): | |
serializables = get_serializables(p_obj) | |
serialized = {} | |
p_serializeds[hash_or_id(p_obj)] = serialized | |
for prop_name, prop in serializables.items(): | |
serialized[prop.name] = serialize_with_level(getattr(p_obj, prop.name), prop.name, p_obj) | |
return serialized | |
if hasattr(p_obj, "serializable"): | |
return serialize_with_level(p_obj.serializable, p_name, p_parent) | |
if hasattr(p_obj, "__dict__"): | |
serialized = {} | |
for prop in p_obj.__dict__.keys(): | |
if not prop.startswith("_") and not callable(getattr(p_obj, prop)): | |
serialized[prop] = serialize_with_level(getattr(p_obj, prop)) | |
return serialized | |
if isinstance(p_obj, str): | |
return p_obj | |
if isinstance(p_obj, int): | |
return p_obj | |
if isinstance(p_obj, float): | |
return p_obj | |
if isinstance(p_obj, bool): | |
return p_obj | |
if isinstance(p_obj, list): | |
return [serialize_with_level(item) for item in p_obj] | |
if isinstance(p_obj, dict): | |
return {key: serialize_with_level(value) for key, value in p_obj.items()} | |
if isinstance(p_obj, tuple): | |
return tuple(serialize_with_level(item) for item in p_obj) | |
if isinstance(p_obj, set): | |
return {serialize_with_level(item) for item in p_obj} | |
if isinstance(p_obj, bytes): | |
return p_obj.decode("utf-8") | |
if p_obj is None: | |
return None | |
return str(p_obj) | |
def serializable_class(p_cls): | |
""" | |
@serializable_class | |
Decorator to make a class serializable. | |
:param cls: The class to make serializable. | |
""" | |
setattr(p_cls, "__serializables__", {}) | |
setattr(p_cls, "serializable", property(serialize)) | |
setattr(p_cls, "serializable_property", serializable_property) | |
return p_cls |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment