dump-sqlmodel 1.3.0

A simple library that help you dump and load SQLModel object

dump-sqlmodel

A simple library that helps you dump and load your SQLModel objects

Git Repo

I. Introduction

This is a simple library that helps you dump and load your SQLModel objects. It also preserve complex relationship object.

So why not model_dump?

Because when you have a cyclic Relationship reference, model_dump will just call recursively and reach recursion limit.

In this library, it uses seen mechanism and reference mechanism. In simple term, each time it sees a Relationship, it will check if the relationship is in a seen list. If so, create a reference that links to the origin Relationship, or else, dump the Relationship and push a reference link to the seen list

II. So are there any disadvantages?

I think there are a lots ;-;

• I just implement a few obj type that can't be JSON serializable, which includes Enum, datetime and pydantic BaseModel.
• It requires manually garbage collector implement to get rid of the reference list or you will evntually get memory leak eror (I'm sorry ;-;)
• And more...

III. Installation

If you still choose me, then this is installation guide.

You can install it with pip using this command:

pip install dump-sqlmodel

Or if you want to use Redis, use this command:

pip install 'dump-sqlmodel[redis]'

IV. How to use this?

First, import the class:

from sqlmodel_dump import SQLModelDump

Or import the async or redis version:

from sqlmodel_dump.ext.io import SQLModelDump
from sqlmodel_dump.ext.redis import SQLModelDump
from sqlmodel_dump.ext.aio_redis import SQLModelDump

Second, init it:

sqlmodel_dump = SQLModelDump()

Or with the redis version

sqlmodel_dump = SQLModelDump(redis=<your redis client>)
sqlmodel_dump = SQLModelDump(redis=<your async redis client>)

Third, use it:

enrollment = <some SQLModel>
dump_data = sqlmodel_dump.dumps(enrollment)
print(dump_data)

{
    "__type": "model",
    "__class": "Enrollment",
    "id": "f2915205-831c-4941-b116-7c76b27ebd2f",
    "classroom": {
        "__type": "relationship",
        "__class": "Class",
        "__ref": "Class:b607a4a3-4a10-453b-aed4-cd575ebb6c36" # An address to the reference in `_ref` object
    },
    ...
}

But there is an option called self_ref, what is that?

With self_ref set to False, the Relationship reference will be stored in SQLModelDump._ref dict. This is the default mode and need a garbage collector in order not to raise the memory leak error..

Example: Return object:

{
    "__type": "model",
    "__class": "Enrollment",
    "id": "f2915205-831c-4941-b116-7c76b27ebd2f",
    "classroom": {
        "__type": "relationship",
        "__class": "Class",
        "__ref": "Class:b607a4a3-4a10-453b-aed4-cd575ebb6c36" # An address to the reference in `_ref` object
    },
    ...
}

The _ref object

{
    "Class:b607a4a3-4a10-453b-aed4-cd575ebb6c36": { # The relationship value
        "__type": "model",
        "__class": "Class",
        "id": "b607a4a3-4a10-453b-aed4-cd575ebb6c36",
        "name": "Test Class",
        "subject": "Test",
        "teacher_id": "3135756c-be25-41a5-a6a8-be2c6f667c99",
        "teacher": {
            "__type": "relationship",
            "__class": "Teacher",
            "__ref": "Teacher:3135756c-be25-41a5-a6a8-be2c6f667c99" # Reference to another object
        }
    },
    ...
}

With self_ref set to True, the reference will be stored in the result object itself. This can help eliminate the garbage collector but the result object size will be significantly increased

Example:

{
    "__type": "model",
    "__class": "Enrollment",
    "id": "ae4efb62-cd94-47bf-a6e7-3f9c93703c00",
    "class_id": "bdd2df1c-5b33-467b-9c0b-a1e69f4324b2",
    "classroom": {
        "__type": "relationship",
        "__class": "Class",
        "__ref": "Class:bdd2df1c-5b33-467b-9c0b-a1e69f4324b2",
        "__val": { # The relationship will be store in itself
            "__type": "model",
            "__class": "Class",
            "id": "bdd2df1c-5b33-467b-9c0b-a1e69f4324b2",
            "created_at": {
                "__type": "datetime",
                "__val": 1772292084.449564
            },
            "updated_at": {
                "__type": "datetime",
                "__val": 1772292084.449624
            },
            "name": "Test Class",
            "subject": "Test",
            ...
        }
    },
    ...
}

V. API

1. SQLModelDump

SQLModelDump(
    max_depth: int = 3,
    self_ref: bool = False,
    key_factory: Callable[[Any], str] | None = None,
    attr_to_key: dict[str | type[Any], list[str]] | None = None,
    ref_alternative: Callable[[str], Any] | None = None
)

Create a SQLModelDump instance

Args:

• max_depth int control how deep the serializer and deserializer will go
• self_ref bool Use self_ref mode
• attr_to_key dict A map of class or class name and keys that will be used to create the reference key. For example, if you want class Enrollment use its class_id and student_id for reference key, it should be { Enrollment: ["class_id", "student_id"] }. This is the alternative way to modify the reference key.
• key_factory Callable A custom function to produce the key for reference, should accept only one args is the reference object
• ref_alternative Callable A function to call when a reference is missing, could be an async function

dumps(obj: SQLModel) -> dict[str, Any]

Dump the SQLModel object

Args:

• obj SQLModel The object to dumps

Return dict[str, Any] The dumped object

loads(obj: Any) -> Any

Load the SQLModel object

Args:

• obj SQLModel The object to loads

Return Any The loaded object

2. Async SQLModelDump

SQLModelDump(
    max_depth: int = 3,
    self_ref: bool = False,
    key_factory: Callable[[Any], str] | None = None,
    attr_to_key: dict[str | type[Any], list[str]] | None = None,
    ref_alternative: Callable[[str], Coroutine[Any, Any, Any]] | None = None
)

Create an async SQLModelDump instance, which accept ref_alternative as an async function.

Args:

• max_depth int control how deep the serializer and deserializer will go
• self_ref bool Use self_ref mode
• attr_to_key dict A map of class or class name and keys that will be used to create the reference key. For example, if you want class Enrollment use its class_id and student_id for reference key, it should be { Enrollment: ["class_id", "student_id"] }. This is the alternative way to modify the reference key.
• key_factory Callable A custom function to produce the key for reference, should accept only one args is the reference object
• ref_alternative Callable A function to call when a reference is missing, could be an async function

3. Redis SQLModelDump

ext.redis.SQLModelDump(
    redis: Redis,
    max_depth: int = 3,
    self_ref: bool = False,
    key_factory: Callable[[Any], str] | None = None,
    attr_to_key: dict[str | type[Any], list[str]] | None = None,
    ref_alternative: Callable[[str], Any] | None = None,
    running_loop: AbstractEventLoop | None = None,
    store_obj_to_redis: bool = False,
    set_kwargs: dict[str, Any] = {},
    get_kwargs: dict[str, Any] = {},
)

Create a SQLModel with Redis as the reference manager

Args:

• redis redis.Redis Redis client
• max_depth int control how deep the serializer and deserializer will go
• self_ref bool Use self_ref mode
• attr_to_key dict A map of class or class name and keys that will be used to create the reference key. For example, if you want class Enrollment use its class_id and student_id for reference key, it should be { Enrollment: ["class_id", "student_id"] }. This is the alternative way to modify the reference key.
• key_factory Callable A custom function to produce the key for reference, should accept only one args is the reference object
• ref_alternative Callable A function to call when a reference is missing, could be an async function
• set_kwargs dict[str, Any] The kwargs to pass to set function
• get_kwargs dict[str, Any] The kwargs to pass to get function

loads_from_redis(key: str) -> Any

Loads the object from Redis

Args

• key str The Redis key

Return Any The loaded object

dumps(obj: SQLModel, store_obj_to_redis: bool) -> dict[str, Any]

Dump the SQLModel object

Args:

• obj SQLModel The object to dumps
• store_obj_to_redis bool Will the result object be stored in the Redis before returning

Return dict[str, Any] The dumped object

The rest Like SQLModelDump

4. Async Redis SQLModelDump

ext.aio_redis.SQLModelDump(
    redis: Redis,
    max_depth: int = 3,
    self_ref: bool = False,
    key_factory: Callable[[Any], str] | None = None,
    attr_to_key: dict[str | type[Any], list[str]] | None = None,
    ref_alternative: Callable[[str], Any] | None = None,
    store_obj_to_redis: bool = False,
    set_kwargs: dict[str, Any] = {},
    get_kwargs: dict[str, Any] = {},
)

Create a SQLModel with Redis as the reference manager

Args:

• redis redis.asyncio.Redis Async Redis client
• max_depth int control how deep the serializer and deserializer will go
• self_ref bool Use self_ref mode
• attr_to_key dict A map of class or class name and keys that will be used to create the reference key. For example, if you want class Enrollment use its class_id and student_id for reference key, it should be { Enrollment: ["class_id", "student_id"] }. This is the alternative way to modify the reference key.
• key_factory Callable A custom function to produce the key for reference, should accept only one args is the reference object
• ref_alternative Callable A function to call when a reference is missing, could be an async function
• store_obj_to_redis bool Will the result object be stored in the Redis before returning
• set_kwargs dict[str, Any] The kwargs to pass to set function
• get_kwargs dict[str, Any] The kwargs to pass to get function

The rest Like Redis SQLModelDump but use async :D

5. Serializer

Serializer(
    self_ref: bool = False,
    max_depth: int = 3,
    key_factory: Callable[[Any], str] | None = None,
    attr_to_key: dict[str | type[Any], list[str]] | None = None,
)

Create a Serializer

Args:

• self_ref bool Use self_ref mode. The output will be a self_ref object
• max_depth int Controll how deep the serializer will dig into the object.
• attr_to_key dict A map of class or class name and keys that will be used to create the reference key. For example, if you want class Enrollment use its class_id and student_id for reference key, it should be { Enrollment: ["class_id", "student_id"] }. This is the alternative way to modify the reference key.
• key_factory Callable[[Any], str] Modify how a reference key will be created. Accept the refering object as the input and the output must be a string as the reference key. The default key pattern is obj_class_name:obj_id

ref: dict[str, Any]

Store references. Will be an empty object if self_ref is set to True

base_class: dict[str, type[Any]]

Store reference class to recontruct them when come to deserialization. Should be passed to Deserializer.

serialize(obj: Any, current_depth: int = 1) -> Any

Serialize object into a json object.

Args:

• obj Any Object to serialize
• current_depth int To control the object depth

Return Any The dumped object

Note: This function will be call recusively while running.

6. Deserializer

Deserializer(
    get_class: Callable[[str], type[Any] | None],
    get_ref: Callable[[str], Any] | None = None,
    self_ref: bool = False,
    max_depth: int = 3,
    ref_alternative: Callable[[str], Any] | None = None
)

Create a Deserializer

Args:

• get_class Callable[[str], type[Any] | None] The function to get the reference class.
• get_ref Callable[[str], Any] | None The function to get the reference.
• self_ref bool Use self_ref mode. Will return a self_ref object.
• max_depth int Controll how deep the deserializer will dig into the object.
• ref_alternative Callable[[str], Any] | None The function to call when the ref is missing. Could be a function to call to the DB. Should return the obj, not a dict, for example, a SQLModel object.

deserialize(obj: Any, current_depth: int = 1) -> Any

Deserialize json object into a object.

Args:

• obj Any Object to deserialize
• current_depth int To control the object depth

Return Any The loaded object

7. AsyncDeserializer

ext.aio.AsyncDeserializer(
    get_class: Callable[[str], type[Any] | None],
    get_ref: Callable[[str], Any] | None = None,
    self_ref: bool = False,
    max_depth: int = 3,
    ref_alternative: Callable[[str], Coroutine[Any, Any, Any]] | None = None
)

Create a AsyncDeserializer, which accept ref_alternative as an async function.

Args:

• get_class Callable[[str], type[Any] | None] The function to get the reference class.
• get_ref Callable[[str], Any] | None The function to get the reference.
• self_ref bool Use self_ref mode. Will return a self_ref object.
• max_depth int Controll how deep the deserializer will dig into the object.
• ref_alternative Callable[[str], Any] | None The function to call when the ref is missing. Could be a function to call to the DB. Should return the obj, not a dict, for example, a SQLModel object.

async deserialize(obj: Any, current_depth: int = 1) -> Any

Deserialize json object into a object.

Args:

• obj Any Object to deserialize
• current_depth int To control the object depth

Return Any The loaded object