[KV][global state storage] Key/Vale store implemented with SQLAlchemy + PostgreSQL

gstate.py

from typing import Any
import copy

import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB

from my_proj.db import db


class GState(db.Model):
    """
    Global system state storage.
    """
    __tablename__ = 'g_state'

    key = sa.Column(sa.String(256), primary_key=True)
    value = sa.Column(JSONB)

    def __repr__(self):
        return f'GState({self.key!r})'

    @classmethod
    def get(cls, key: str, default=None) -> Any:
        """
        Get state by key

        Returns: the state value by key, or None if not exists.
        """
        st = db.s.get(cls, key)
        return st.value if st else default

    @classmethod
    def set(cls, key: str, value: Any, commit=False) -> None:
        """
        Set or overwrite state by key

        Args:
            key: the key of the state.
            value: the value of the state, supports any json-serializable object.
            commit: whether to commit the current database session.
        """
        st = db.s.get(cls, key) or cls(key=key)
        st.value = copy.deepcopy(value)

        db.s.add(st)

        if commit:
            db.s.commit()

    @classmethod
    def delete(cls, key: str, commit=False) -> None:
        """
        Delete state by key
        """
        db.s.query(cls).filter(cls.key == key).delete()
        if commit:
            db.s.commit()