Compare commits

..

No commits in common. "personal" and "main" have entirely different histories.

3 changed files with 34 additions and 190 deletions

View File

@ -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"]

View File

@ -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.

View File

@ -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