Compare commits
No commits in common. "personal" and "main" have entirely different histories.
|
@ -1,7 +1,7 @@
|
||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "jo3util"
|
name = "jo3util"
|
||||||
version = "0.0.20"
|
version = "0.0.18"
|
||||||
description = ""
|
description = ""
|
||||||
dependencies = []
|
dependencies = []
|
||||||
dynamic = ["readme"]
|
dynamic = ["readme"]
|
||||||
|
|
146
src/hollow.py
146
src/hollow.py
|
@ -1,146 +0,0 @@
|
||||||
from typing import TypeVar, Generic, Callable, Type, Union
|
|
||||||
|
|
||||||
T = TypeVar("T")
|
|
||||||
|
|
||||||
|
|
||||||
def HollowObject(init_fn: Union[T, Callable[..., T]], *args, **kwargs) -> T:
|
|
||||||
""" Return an object that only gets initialized when any of its attributes are accessed.
|
|
||||||
|
|
||||||
:param init_fn: Function that initializes the real object.
|
|
||||||
:param args: Positional arguments for the init_fn.
|
|
||||||
:param kwargs: Keyword arguments for the init_fn.
|
|
||||||
"""
|
|
||||||
|
|
||||||
# Case for types.
|
|
||||||
if isinstance(init_fn, type):
|
|
||||||
base_class = init_fn
|
|
||||||
# Case for functions.
|
|
||||||
elif hasattr(init_fn, "__annotations__") and "return" in init_fn.__annotations__:
|
|
||||||
base_class = init_fn.__annotations__["return"]
|
|
||||||
# Case for callable class instances.
|
|
||||||
elif hasattr(init_fn.__call__, "__annotations__") and "return" in init_fn.__call__.__annotations__:
|
|
||||||
base_class = init_fn.__call__.__annotations__["return"]
|
|
||||||
# Case for unnannotated non-type callables.
|
|
||||||
else:
|
|
||||||
base_class = object
|
|
||||||
|
|
||||||
def initialize(hollow_object):
|
|
||||||
real_obj = init_fn(*args, **kwargs)
|
|
||||||
|
|
||||||
try:
|
|
||||||
# Change the type of `self` to match the real object
|
|
||||||
hollow_object.__class__ = real_obj.__class__
|
|
||||||
hollow_object.__dict__ = real_obj.__dict__ # Copy the real object's attributes
|
|
||||||
return False
|
|
||||||
|
|
||||||
# For certain builtin immutable types, we can't change the __class__ or __dict__.
|
|
||||||
# This means the instance will always stay a HollowObject subclass and the below
|
|
||||||
# wrapper will keep firing for all calls.
|
|
||||||
except TypeError:
|
|
||||||
hollow_object._real_obj = real_obj
|
|
||||||
return True
|
|
||||||
|
|
||||||
def wrap(fn):
|
|
||||||
def inner(self, *args, **kwargs):
|
|
||||||
if hasattr(self, "_real_obj") or initialize(self):
|
|
||||||
return object.__getattribute__(self._real_obj, fn.__name__)(*args, **kwargs)
|
|
||||||
else:
|
|
||||||
return object.__getattribute__(self, fn.__name__)(*args, **kwargs)
|
|
||||||
return inner
|
|
||||||
|
|
||||||
class HollowObject(base_class):
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
def __getattribute__(self, name):
|
|
||||||
# Avoid infinite recursions.
|
|
||||||
if name in ["__dict__", "__class__", "_real_obj", "_get_call"]:
|
|
||||||
return super().__getattribute__(name)
|
|
||||||
|
|
||||||
if hasattr(self, "_real_obj") or initialize(self):
|
|
||||||
return getattr(self._real_obj, name)
|
|
||||||
else:
|
|
||||||
return getattr(self, name)
|
|
||||||
|
|
||||||
def __setattr__(self, name, value):
|
|
||||||
# Avoid infinite recursions.
|
|
||||||
if name in ["__dict__", "__class__", "_real_obj", "_get_call"]:
|
|
||||||
return super().__setattr__(name, value)
|
|
||||||
|
|
||||||
if hasattr(self, "_real_obj") or initialize(self):
|
|
||||||
setattr(self._real_obj, name, value)
|
|
||||||
else:
|
|
||||||
setattr(self, name, value)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _get_call():
|
|
||||||
return init_fn, args, kwargs
|
|
||||||
|
|
||||||
@wrap
|
|
||||||
def __str__(self):
|
|
||||||
return ""
|
|
||||||
|
|
||||||
@wrap
|
|
||||||
def __repr__(self):
|
|
||||||
return ""
|
|
||||||
|
|
||||||
@wrap
|
|
||||||
def __add__(self, _):
|
|
||||||
pass
|
|
||||||
|
|
||||||
@wrap
|
|
||||||
def __iter__(self, _):
|
|
||||||
pass
|
|
||||||
|
|
||||||
# todo: add all the magic functions
|
|
||||||
|
|
||||||
return HollowObject()
|
|
||||||
|
|
||||||
|
|
||||||
def lazy(fn):
|
|
||||||
""" Replace fn with a lazy function.
|
|
||||||
|
|
||||||
A lazy function does not immediately fire, but instead returns an object
|
|
||||||
that contains the function call. If anything of the object is accessed,
|
|
||||||
it is created in-place and the accessed thing is returned."""
|
|
||||||
def lazy_(*args, **kwargs):
|
|
||||||
return HollowObject(fn, *args, **kwargs)
|
|
||||||
return lazy_
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
|
|
||||||
# Example usage of an object with type hints
|
|
||||||
class FullObject:
|
|
||||||
def __init__(self, value: int):
|
|
||||||
self.value = value
|
|
||||||
print("initialized")
|
|
||||||
|
|
||||||
def __str__(self):
|
|
||||||
return f"Value is {self.value}"
|
|
||||||
|
|
||||||
hollow = HollowObject(FullObject, value=42)
|
|
||||||
print(type(hollow))
|
|
||||||
# Linters will know that `hollow` behaves like `FullObject`.
|
|
||||||
#
|
|
||||||
# Only now does hollow get initialized.
|
|
||||||
print(hollow)
|
|
||||||
# The type of hollow is now FullObject, so no future calls will be made through the wrapper.
|
|
||||||
|
|
||||||
# Example usage of the lazy decorator.
|
|
||||||
|
|
||||||
print("\nstarting string tests")
|
|
||||||
|
|
||||||
@lazy
|
|
||||||
def lazy_greeting(name) -> str:
|
|
||||||
print("preparing to greet")
|
|
||||||
return "hello " + name
|
|
||||||
|
|
||||||
message = lazy_greeting("john")
|
|
||||||
print(type(message))
|
|
||||||
print([char for char in message])
|
|
||||||
# because message is a subclass of str, we can't change the class's type.
|
|
||||||
# this means that all future calls on it, will be through the wrapper
|
|
||||||
# and message stays an HollowObject type.
|
|
||||||
|
|
76
src/store.py
76
src/store.py
|
@ -6,8 +6,6 @@ from pathlib import Path
|
||||||
import pickle
|
import pickle
|
||||||
from typing import Tuple
|
from typing import Tuple
|
||||||
|
|
||||||
from .hollow import HollowObject
|
|
||||||
|
|
||||||
|
|
||||||
def with_filtered_args(fn):
|
def with_filtered_args(fn):
|
||||||
"""
|
"""
|
||||||
|
@ -60,26 +58,47 @@ class load_or_create:
|
||||||
hash_len=8,
|
hash_len=8,
|
||||||
save_args=False,
|
save_args=False,
|
||||||
save_json=False,
|
save_json=False,
|
||||||
plain_text=False,
|
plain_text=False
|
||||||
path=None,
|
|
||||||
):
|
):
|
||||||
self.load=with_filtered_args(load)[0]
|
self.load=with_filtered_args(load)[0]
|
||||||
self.load_arg_names = {p.name for p in inspect.signature(load).parameters.values()}
|
self.load_arg_names = {p.name for p in inspect.signature(load).parameters.values()}
|
||||||
self.save=with_filtered_args(save)[0]
|
self.save=with_filtered_args(save)[0]
|
||||||
self.save_arg_names = {p.name for p in inspect.signature(save).parameters.values()}
|
self.save_arg_names = {p.name for p in inspect.signature(save).parameters.values()}
|
||||||
self.path_fn=path if path is not None else path_fn
|
self.path_fn=path_fn
|
||||||
self.hash_len=hash_len
|
self.hash_len=hash_len
|
||||||
self.save_args=save_args
|
self.save_args=save_args
|
||||||
self.save_json=save_json
|
self.save_json=save_json
|
||||||
self.plain_text=plain_text
|
self.plain_text=plain_text
|
||||||
self.obj_to_args = {}
|
|
||||||
|
|
||||||
# To be initialied by Loc.
|
|
||||||
self.fn = lambda x: x
|
|
||||||
self.arg_names = []
|
|
||||||
|
|
||||||
def __call__(self, fn):
|
def __call__(self, fn):
|
||||||
return Loc(self, fn)
|
return inner(self, fn)
|
||||||
|
|
||||||
|
|
||||||
|
class inner(load_or_create):
|
||||||
|
def __init__(self, parent, fn):
|
||||||
|
self.__dict__.update(parent.__dict__)
|
||||||
|
self.fn = fn
|
||||||
|
self.arg_names = [p.name for p in inspect.signature(fn).parameters.values()]
|
||||||
|
self.obj_to_args = dict()
|
||||||
|
|
||||||
|
def __call__(self, *args, **kwargs):
|
||||||
|
# Store the keyword arguments into json and hash it to get the storage path.
|
||||||
|
path = self.path(*args, **kwargs)
|
||||||
|
merged_args = self.args_to_kwargs(args, kwargs, path=path)
|
||||||
|
|
||||||
|
obj = self.load_wrapper(**merged_args)
|
||||||
|
if obj is not None:
|
||||||
|
if "file" in self.save_arg_names: self.hash_obj({"path": path} | kwargs)
|
||||||
|
return obj
|
||||||
|
|
||||||
|
obj = self.fn(*args, **kwargs)
|
||||||
|
if obj is None: return obj
|
||||||
|
|
||||||
|
self.save_wrapper(obj, *args, **{"path": path} | kwargs)
|
||||||
|
if "file" in self.save_arg_names: self.hash_obj({"path": path} | kwargs)
|
||||||
|
if self.save_json: path.with_suffix(".kwargs.json").write_bytes(self.to_json(**kwargs))
|
||||||
|
|
||||||
|
return obj
|
||||||
|
|
||||||
def args_to_kwargs(self, args, kwargs, **extra):
|
def args_to_kwargs(self, args, kwargs, **extra):
|
||||||
return extra | kwargs | {self.arg_names[i]: a for i, a in enumerate(args)}
|
return extra | kwargs | {self.arg_names[i]: a for i, a in enumerate(args)}
|
||||||
|
@ -208,11 +227,8 @@ class load_or_create:
|
||||||
return self.obj_to_args[hash]
|
return self.obj_to_args[hash]
|
||||||
|
|
||||||
def path_from_obj(self, obj, *args, **kwargs):
|
def path_from_obj(self, obj, *args, **kwargs):
|
||||||
# If the object passed is a HollowObject, we can infer the args without
|
if isinstance(obj, LoadOrCreateCFG):
|
||||||
# calling it.
|
return self.path(*obj.args, **obj.kwargs)
|
||||||
if hasattr(obj, "_get_call"):
|
|
||||||
_, args, kwargs = obj._get_call()
|
|
||||||
return self.path(*args, **kwargs)
|
|
||||||
return self.args_from_obj(obj, *args, **kwargs)["path"]
|
return self.args_from_obj(obj, *args, **kwargs)["path"]
|
||||||
|
|
||||||
def dir_from_obj(self, obj, *args, **kwargs):
|
def dir_from_obj(self, obj, *args, **kwargs):
|
||||||
|
@ -224,32 +240,6 @@ class load_or_create:
|
||||||
to get it's path, you can pass a LoadOrCreateCFG instead, saving you
|
to get it's path, you can pass a LoadOrCreateCFG instead, saving you
|
||||||
from loading or creating that object..
|
from loading or creating that object..
|
||||||
"""
|
"""
|
||||||
return HollowObject(lambda x: x, *args, **kwargs)
|
return LoadOrCreateCFG(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
class Loc(load_or_create):
|
|
||||||
def __init__(self, parent, fn):
|
|
||||||
self.__dict__.update(parent.__dict__)
|
|
||||||
self.fn = fn
|
|
||||||
self.arg_names = [p.name for p in inspect.signature(fn).parameters.values()]
|
|
||||||
|
|
||||||
def __call__(self, *args, **kwargs):
|
|
||||||
# Store the keyword arguments into json and hash it to get the storage path.
|
|
||||||
path = self.path(*args, **kwargs)
|
|
||||||
merged_args = self.args_to_kwargs(args, kwargs, path=path)
|
|
||||||
|
|
||||||
obj = self.load_wrapper(**merged_args)
|
|
||||||
if obj is not None:
|
|
||||||
if "file" in self.save_arg_names: self.hash_obj({"path": path} | kwargs)
|
|
||||||
return obj
|
|
||||||
|
|
||||||
obj = self.fn(*args, **kwargs)
|
|
||||||
if obj is None: return obj
|
|
||||||
|
|
||||||
self.save_wrapper(obj, *args, **{"path": path} | kwargs)
|
|
||||||
if "file" in self.save_arg_names: self.hash_obj({"path": path} | kwargs)
|
|
||||||
if self.save_json: path.with_suffix(".kwargs.json").write_bytes(self.to_json(**kwargs))
|
|
||||||
|
|
||||||
return obj
|
|
||||||
|
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue
Block a user