dump-sqlmodel 1.0.1

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.
• And so many 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 redis, async redis version:

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 call 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] = lambda x: (
        f"{x.__class__.__name__}:{getattr(x, 'id', 'None')}"
    ),
    ref_alternative: Callable[[str], Any] | None = None,
    running_loop: AbstractEventLoop | 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
• 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
• running_loop AbstractEventLoop The current event loop to pass to deserializer to run ref_alternative if it is an async function

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

Dump the SQLModel object

Args:

• obj SQLModel The object to dumps

Return dict[str, Any] The dumped object

SQLModelDump.loads(self, obj: Any) -> Any

Load the SQLModel object

Args:

• obj SQLModel The object to loads

Return Any The loaded object

2. Redis SQLModelDump

ext.redis.RedisSQLModelDump(
    redis: Redis,
    max_depth: int = 3,
    self_ref: bool = False,
    key_factory: Callable[[Any], str] = lambda x: (
        f"{x.__class__.__name__}:{getattr(x, 'id', '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
• 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
• running_loop AbstractEventLoop The current event loop to pass to deserializer to run ref_alternative if it is 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

RedisSQLModelDump.loads_from_redis(key: str) -> Any

Loads the object from Redis

Args

• key str The Redis key

Return Any The loaded object

The rest Like SQLModelDump

3. Async Redis SQLModelDump

ext.aio_redis.SQLModelDump(
    redis: Redis,
    max_depth: int = 3,
    self_ref: bool = False,
    key_factory: Callable[[Any], str] = lambda x: (
        f"{x.__class__.__name__}:{getattr(x, 'id', '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
• 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

4. Serializer

Serializer(
    self_ref: bool = False,
    max_depth: int = 3,
    key_factory: Callable[[Any], str] = lambda x: (
        f"{x.__class__.__name__}:{getattr(x, 'id', '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.
• 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

Serializer.ref: dict[str, Any]

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

Serializer.base_class: dict[str, type[Any]]

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

Serializer.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.

5. 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,
    running_loop: AbstractEventLoop | 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.
• running_loop AbstractEventLoop | None The current running event loop. Will be used to call the async_deserialize method.

Deserializer.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

Note: This is the sync wrapper for async_deserialize method

Deserializer.async_deserialize(obj: Any, current_depth: int = 1) -> Awaitable[Any]

Deserialize json object into a object but with async :D.

Args:

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

Return Any The loaded object