fritter.persistent.jsonable.JSONRegistry

class documentation

class JSONRegistry(Generic[BootstrapT]): (source)

View In Hierarchy

A JSONRegistry maintains a set of functions and methods (and the classes those methods are defined on) that can scheduled against a Scheduler and then persisted into a JSON blob to reconstitute that scheduler later.

Use a JSONRegistry by instantiatint it at module scope and then using its methods as decorators on functions and methods.

Method createScheduler Create a new JSONableScheduler with the same type as if it had been loaded by this JSONRegistry.
Method function Mark the given 0-argument, None-returning, top-level function as possible to serialize within this registry. It will be serialized by its fully-qualified Python name.
Method loadScheduler Load a JSON object in the format serialized by the saver returned by JSONRegistry.createScheduler (or this method) and a runtime TimeDriver[DateTime[ZoneInfo]], returning a 2-tuple of JSONableScheduler...
Method method Mark the given method, defined at class scope in a class complying with the JSONableInstance protocol, as possible to serialize within this registry.
Method repeatedly Call work repeatedly, according to rule, with intervals computed according to the given reference time (or scheduler's current time, if not given).
Method repeatFunction Mark the given function that matches the signature of RepeatingWork, i.e. one which takes a number of steps and a Cancellable to cancel its own repetition, as looked up via its __qualname__ attribute.
Method repeatMethod Mark the given method that matches the signature of RepeatingWork, i.e. one which takes a number of steps and a Cancellable to cancel its own repetition, as looked up via its __qualname__ attribute.
Method saveScheduledCall Convert a ScheduledCall into a JSON-serializable object.
Method _loadOne Convert the given JSON-dumpable dict into an object of _JSONableType via its JSONableInstance.fromJSON classmethod registered in the given _SpecificTypeRegistration.
Method _loadRRule Undocumented
Method _registerJSONableType Mark the given class as serializable via this registry, keyed by its JSONable.typeCodeForJSON method. Used internally as supporting JSONRegistry.method and JSONRegistry.repeatMethod.
Method _registerRRule Undocumented
Method _repeaterToJSONable Convert the given JSONRepeater into a method that can be serialized.
Method _saverFor Undocumented
Method _saveRRule Undocumented
Class Variable _functions Undocumented
Class Variable _instances Undocumented
Class Variable _repeatable Undocumented
Class Variable _rules Undocumented
Class Variable _ruletype2jsonifier Undocumented
def createScheduler(self, driver: TimeDriver[DTZI]) -> tuple[JSONableScheduler[BootstrapT], Callable[[], JSONObject]]: (source)

Create a new JSONableScheduler with the same type as if it had been loaded by this JSONRegistry.

def function(self, cb: Callable[[], None]) -> JSONableCallable[BootstrapT]: (source)

Mark the given 0-argument, None-returning, top-level function as possible to serialize within this registry. It will be serialized by its fully-qualified Python name.

Use like so:

    registry = JSONRegistry[...]()

    @registry.serializedFunction
def loadScheduler(self, runtimeDriver: TimeDriver[DTZI], serializedJSON: JSONObject, bootstrap: BootstrapT) -> tuple[JSONableScheduler[BootstrapT], Callable[[], JSONObject]]: (source)

Load a JSON object in the format serialized by the saver returned by JSONRegistry.createScheduler (or this method) and a runtime TimeDriver[DateTime[ZoneInfo]], returning a 2-tuple of JSONableScheduler and a 0-argument callable which will serialize the current contents of that scheduler.

def method(self, method: Callable[[JSONableSelf], None]) -> JSONableMethodDescriptor[JSONableSelf, BootstrapT]: (source)

Mark the given method, defined at class scope in a class complying with the JSONableInstance protocol, as possible to serialize within this registry.

def repeatedly(self, scheduler: JSONableScheduler[BootstrapT], rule: RecurrenceRule[DTZI, StepsT], work: JSONableRepeatable[BootstrapT, StepsT], reference: DTZI|None = None): (source)

Call work repeatedly, according to rule, with intervals computed according to the given reference time (or scheduler's current time, if not given).

If you want to stop the repetition, work is passed a Cancellable that you can call .cancel() on to do so.

See Also
fritter.repeat
def repeatFunction(self, cb: RepeatingWork[StepsT]) -> JSONableRepeatable[BootstrapT, StepsT]: (source)

Mark the given function that matches the signature of RepeatingWork, i.e. one which takes a number of steps and a Cancellable to cancel its own repetition, as looked up via its __qualname__ attribute.

Returns
JSONableRepeatable[BootstrapT, StepsT]a function that mimics the signature of the original function, but also conforms to the JSONable protocol.
def repeatMethod(self, repeatable: _Repeater[JSONableSelf, StepsT]) -> JSONableRepeatableDescriptor[JSONableSelf, BootstrapT, StepsT]: (source)

Mark the given method that matches the signature of RepeatingWork, i.e. one which takes a number of steps and a Cancellable to cancel its own repetition, as looked up via its __qualname__ attribute.

Returns
JSONableRepeatableDescriptor[JSONableSelf, BootstrapT, StepsT]a descriptor that mimics the signature of the original method.
Note
The registration occurs when the class is defined, specifically in the __set_name__ hook in the returned descriptor, so if you are calling this in an esoteric context outside a normal class definition, be sure to invoke that hook as well.
def saveScheduledCall(self, futureCall: JSONHandle[BootstrapTInv]) -> dict[str, object]: (source)

Convert a ScheduledCall into a JSON-serializable object.

def _loadOne(self, json: JSONObject, which: _SpecificTypeRegistration[_JSONableCallableT], load: LoadProcess[BootstrapT]) -> _JSONableCallableT: (source)

Convert the given JSON-dumpable dict into an object of _JSONableType via its JSONableInstance.fromJSON classmethod registered in the given _SpecificTypeRegistration.

def _loadRRule(self, json: JSONObject) -> RecurrenceRule[DTZI, Any]: (source)

Undocumented

def _registerJSONableType(self, cls: Type[JSONableInstance[BootstrapT]]): (source)

Mark the given class as serializable via this registry, keyed by its JSONable.typeCodeForJSON method. Used internally as supporting JSONRegistry.method and JSONRegistry.repeatMethod.

def _registerRRule(self, ruleType: Type[RRuleTx], serializer: RuleJSONIfier[RRuleTx]): (source)

Undocumented

def _repeaterToJSONable(self, repeater: JSONRepeater[BootstrapT, StepsT]) -> JSONableBoundMethod[_JSONableRepeaterWrapper[BootstrapT, StepsT]]: (source)

Convert the given JSONRepeater into a method that can be serialized.

def _saverFor(self, h: Heap[Any], s: JSONableScheduler[BootstrapT]) -> Callable[[], JSONObject]: (source)

Undocumented

def _saveRRule(self, rule: RecurrenceRule[DTZI, Any]) -> JSONObject: (source)

Undocumented

_functions: _SpecificTypeRegistration[JSONableCallable[BootstrapT]] = (source)

Undocumented

_instances: _SpecificTypeRegistration[Type[JSONableInstance[BootstrapT]]] = (source)

Undocumented

_repeatable: _SpecificTypeRegistration[JSONableRepeatable[BootstrapT, Any]] = (source)

Undocumented

_rules: _SpecificTypeRegistration[RuleJSONIfier[RecurrenceRule[DTZI, Any]]] = (source)

Undocumented

_ruletype2jsonifier: dict[type[RecurrenceRule[DTZI, object]], RuleJSONIfier[RecurrenceRule[DTZI, Any]]] = (source)

Undocumented