Custom Types

Compressed String

A unicode string, but compressed. See example.

Compressed Binary

A big binary blob, but compressed. See example.

Compressed JSON

A json serializable object, but compressed. See example.

JSON Serializable

Any JSON serializable object, if implemented to_json(self): and from_json(cls, json_str): method.

declare your JSON serializable object, which will be the value of a column in Database.

import jsonpickle


# a custom python class
class ComputerDetails:
    def __init__(self, os: str, cpu: int, memory: int, disk: int):
        self.os = os
        self.cpu = cpu
        self.memory = memory
        self.disk = disk

    def to_json(self) -> str:
        return jsonpickle.encode(self)

    @classmethod
    def from_json(cls, json_str: str) -> "Computer":
        return jsonpickle.decode(json_str)

then declare your ORM model

import sqlalchemy as sa
import sqlalchemy.orm as orm
import sqlalchemy_mate.api as sam

Base = orm.declarative_base()


class Computer(Base):
    __tablename__ = "computer"

    id: orm.Mapped[int] = orm.mapped_column(sa.Integer, primary_key=True)
    # make sure you set the type hint and factory_class right
    details: orm.Mapped[ComputerDetails] = orm.mapped_column(
        sam.types.JSONSerializableType(factory_class=ComputerDetails),
        nullable=True,
    )

Use ORM to insert a row and get the Computer object.

engine = sam.engine_creator.EngineCreator().create_sqlite(
    "/tmp/sqlalchemy_mate_json_serializable.sqlite"
)
Base.metadata.create_all(engine)
sam.deleting.delete_all(engine, Computer.__table__)


with orm.Session(engine) as ses:
    computer = Computer(
        id=1,
        details=ComputerDetails(
            os="Linux",
            cpu=4,
            memory=8,
            disk=256,
        ),
    )

    ses.add(computer)
    ses.commit()

    computer = ses.get(Computer, 1)
    print(f"{computer.details.os = }")
    print(f"{computer.details.cpu = }")
    print(f"{computer.details.memory = }")
    print(f"{computer.details.disk = }")

computer.details.os = 'Linux'
computer.details.cpu = 4
computer.details.memory = 8
computer.details.disk = 256

If you query the raw table definition (The ComputerDetail column is a str (JSON encoded)), you will see the encoded JSON.

t_computer = sa.Table(
    "computer",
    sa.MetaData(),
    sa.Column("id", sa.Integer, primary_key=True),
    sa.Column("details", sa.String),
)


with engine.connect() as conn:
    stmt = sa.select(t_computer)
    for row in conn.execute(stmt).all():
        print(row)

(1, '{"py/object": "__main__.ComputerDetails", "os": "Linux", "cpu": 4, "memory": 8, "disk": 256}')