msgspec RuntimeStruct

runtime_struct.py

from collections import deque
from threading import RLock
from typing import Any, ClassVar

import msgspec


def cached_call(func):
    def wrapper(*args, **kwargs):
        if not (len(args) < 2 or args[1] is None):
            key = (args, func)

            with wrapper.lock:
                if key in wrapper.previous_args:
                    return

                wrapper.previous_args.append(key)

        return func(*args, **kwargs)

    wrapper.lock = RLock()
    wrapper.previous_args = deque(maxlen=3)  # ?
    return wrapper


class RuntimeStruct(msgspec.Struct):
    _validate_init: ClassVar[bool] = True
    _validate_assignment: ClassVar[bool] = True

    def validate(self) -> None:
        msgspec.convert(msgspec.to_builtins(self), type(self))

    @cached_call
    def _validate(self, caller_hash: int | None = None) -> None:
        self.validate()

    def __post_init__(self) -> None:
        if not self.__struct_config__.eq:  # ?
            raise ValueError(
                f"Subclasses of {type(self).mro()[-4]!r} are required to use eq=True."
            )

        if self._validate_init:
            self._validate(3528134149580498495)

    def __setattr__(self, name: str, value: Any) -> None:
        old_value = getattr(self, name)

        super().__setattr__(name, value)

        if self._validate_assignment:
            try:
                self._validate()
            except msgspec.ValidationError:
                super().__setattr__(name, old_value)
                raise


class InitAndAssignment(RuntimeStruct):
    foo: str


class AssignmentOnly(RuntimeStruct):
    _validate_init = False

    foo: str


unknown_str_variable: Any = "42"
unknown_int_variable: Any = 42


obj1 = InitAndAssignment(foo=unknown_int_variable)
obj1.foo = unknown_int_variable

obj2 = AssignmentOnly(foo=unknown_int_variable)
obj2.foo = unknown_int_variable