olog/co3
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

add "implicit groups" to CO3 registry for dynamic key support

Browse Source
This commit is contained in:
Sam G. 2024-05-01 16:46:25 -07:00
parent b05fdda61a
commit a7c355d6ed
4 changed files with 260 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

200
co3/co3.py
View File

@ -1,12 +1,25 @@
'''
CO3
CO3 is an abstract base class for scaffolding object hierarchies and managing operations
with associated database schemas. It facilitates something like a "lightweight ORM" for
classes/tables/states with fixed transformations of interest. The canonical use case is
managing hierarchical document relations, format conversions, and syntactical components.
'''
Generic collation syntax:
.. code-block:: python
class Type(CO3):
@collate
def group(self, key):
# disambiguate key
...
@collate('key', groups=['group1', 'group2'])
def key(self):
# key-specific logic
...
'''
import inspect
import logging
from collections import defaultdict
@ -15,30 +28,149 @@ from functools import wraps, partial
logger = logging.getLogger(__name__)
def collate(action_key, action_groups=None):
def decorator(func):
nonlocal action_groups
def collate(key, groups=None):
'''
Collation decorator for CO3 subtype action registry.
Dynamic decorator; can be used as ``collate`` without any arguments, or with all. In
the former case, ``key`` will be a function, so we check for this.
.. admonition:: Usage
Collation registration is the process of exposing various actions for use in
**hierarchical collection** (see ``Mapper.collect``). Collation *keys* are unique
identifiers of a particular action that emits data. Keys can belong to an arbitrary
number of *groups*, which serve as semantically meaningful collections of similar
actions. Group assignment also determines the associated *collation component*
to be used as a storage target; the results of actions $K_G$ belonging to group
$G$ will all be stored in the attached $G$-component. Specification of key-group
relations can be done in a few ways:
- Explicit key-group specification: a specific key and associated groups can be
provided as arguments to the decorator:
.. code-block:: python
@collate('key', groups=['group1', 'group2'])
def _key(self):
# key-specific logic
...
The registry dictionaries will then have the following items:
.. code-block:: python
key_registry = {
...,
'key': (_key, ['group1', 'group2']),
...
}
group_registry = {
...,
'group1': [..., 'key', ...],
'group2': [..., 'key', ...],
...
}
If ``groups`` is left unspecified, the key will be attached to the default
``None`` group.
- Implicit key-group association: in some cases, you may want to support an entire
"action class," and associate any operations under the class to the same storage
component. Here we still use the notion of connecting groups to components, but
allow the key to be dynamically specified and passed through to the collation
method:
.. code-block:: python
@collate
def group(self, key):
# disambiguate key
...
and in the registries:
.. code-block:: python
key_registry = {
...,
None: {..., 'group': group, ...},
...
}
group_registry = {
...,
'group': [..., None, ...],
...
}
A few important notes:
- Implicit key-group specifications attach the *group* to a single method,
whereas in the explicit case, groups can be affiliated with many keys. When
explicitly provided, only those exact key values are supported. But in the
implicit case, *any* key is allowed; the group still remains a proxy for the
entire action class, but without needing to map from specifically stored key
values. That is, the utility of the group remains consistent across implicit
and explicit cases, but stores the associations differently.
- The ``None`` key, rather than point to a ``(<method>, <group-list>)`` tuple,
instead points to a dictionary of ``group``-``method`` pairs. When attempting
execute a key under a particular group, the group registry indicates
whether the key is explicitly supported. If ``None`` is present for the group,
then ``key_registry[None][<group-name>]`` can be used to recover the method
implicitly affiliated with the key (along with any other key under the group).
- When any method has been implicitly registered, *any* key (even when
attempting to specify an explicit key) will match that group. This can
effectively mean keys are not unique when an implicit group has been
registered. There is a protection in place here, however; in methods like
``CO3.collate`` and ``Mapper.collect``, an implicit group must be directly
named in order for a given key to be considered. That is, when attempting
collation outside specific group context, provided keys will only be
considered against explicitly registered keys.
'''
func = None
if inspect.isfunction(key):
func = key
key = None
groups = [func.__name__]
if groups is None:
groups = [None]
def decorator(f):
f._collation_data = (key, groups)
return f
if func is not None:
return decorator(func)
if action_groups is None:
action_groups = [None]
func._action_data = (action_key, action_groups)
return func
return decorator
class FormatRegistryMeta(type):
'''
Metaclass handling collation registry at the class level.
'''
def __new__(cls, name, bases, attrs):
action_registry = {}
group_registry = defaultdict(list)
key_registry = {}
group_registry = defaultdict(list)
def register_action(method):
nonlocal action_registry, group_registry
nonlocal key_registry, group_registry
if hasattr(method, '_action_data'):
action_key, action_groups = method._action_data
action_registry[action_key] = (method, action_groups)
if hasattr(method, '_collation_data'):
key, groups = method._collation_data
for action_group in action_groups:
group_registry[action_group].append(action_key)
if key is None:
# only add a "None" entry if there is _some_ implicit group
if None not in key_registry:
key_registry[None] = {}
# only a single group possible here
key_registry[None][groups[0]] = method
else:
key_registry[key] = (method, groups)
for group in groups:
group_registry[group].append(key)
# add registered superclass methods; iterate over bases (usually just one), then
# that base's chain down (reversed), then methods from each subclass
@ -53,8 +185,8 @@ class FormatRegistryMeta(type):
for attr_name, attr_value in attrs.items():
register_action(attr_value)
attrs['action_registry'] = action_registry
attrs['group_registry'] = group_registry
attrs['key_registry'] = key_registry
attrs['group_registry'] = group_registry
return super().__new__(cls, name, bases, attrs)
@ -97,7 +229,7 @@ class CO3(metaclass=FormatRegistryMeta):
'''
return []
def collation_attributes(self, action_key, action_group):
def collation_attributes(self, key, group):
'''
Return "connective" collation component data, possibly dependent on
instance-specific attributes and the action arguments. This is typically the
@ -111,12 +243,28 @@ class CO3(metaclass=FormatRegistryMeta):
'''
return {}
def collate(self, action_key, *action_args, **action_kwargs):
if action_key not in self.action_registry:
logger.debug(f'Collation for {action_key} not supported')
def collate(self, key, group=None, *args, **kwargs):
if key is None:
return None
if key not in self.key_registry:
# keys can't match implicit group if that group isn't explicitly provided
if group is None:
logger.debug(
f'Collation for "{key}" not supported, or implicit group not specified'
)
return None
method = self.key_registry[None].get(group)
if method is None:
logger.debug(
f'Collation key "{key}" not registered and group {group} not implicit'
)
return None
return method(self, key, *args, **kwargs)
else:
action_method = self.action_registry[action_key][0]
return action_method(self, *action_args, **action_kwargs)
method = self.key_registry[key][0]
return method(self, *args, **kwargs)

79
co3/mapper.py
View File

@ -1,8 +1,6 @@
'''
Mapper
Used to house useful objects for storage schemas (e.g., SQLAlchemy table definitions).
Provides a general interface for mapping from CO4 class names to storage structures for
Provides a general interface for mapping from CO3 class names to storage structures for
auto-collection and composition.
Example:
@ -32,6 +30,7 @@ Example:
hierarchy). As such, to fully collect from a type, the Mapper needs to leave
registration open to various types, not just those part of the same hierarchy.
'''
import logging
from typing import Callable, Any
from collections import defaultdict
@ -42,6 +41,8 @@ from co3.component import Component
from co3.components import ComposableComponent
logger = logging.getLogger(__name__)
class Mapper[C: Component]:
'''
Mapper base class for housing schema components and managing relationships between CO3
@ -60,7 +61,7 @@ class Mapper[C: Component]:
.. admonition:: Dev note
the Composer needs reconsideration, or at least its positioning directly in this
The Composer needs reconsideration, or at least its positioning directly in this
class. It may be more appropriate to have at the Schema level, or even just
dissolved altogether if arbitrary named Components can be attached to schemas.
@ -82,19 +83,27 @@ class Mapper[C: Component]:
self.attribute_comps: dict[type[CO3], C] = {}
self.collation_groups: dict[type[CO3], dict[str|None, C]] = defaultdict(dict)
def _check_component(self, comp: str | C):
def _check_component(self, comp: str | C, strict=True):
if type(comp) is str:
comp_key = comp
comp = self.schema.get_component(comp_key)
if comp is None:
raise ValueError(
f'Component key {comp_key} not available in attached schema'
)
err_msg = f'Component key "{comp_key}" not available in attached schema'
if strict:
raise ValueError(err_msg)
else:
logger.info(err_msg)
return None
else:
if comp not in self.schema:
raise TypeError(
f'Component {comp} not registered to Mapper schema {self.schema}'
)
err_msg = f'Component "{comp}" not registered to Mapper schema {self.schema}'
if strict:
raise TypeError(err_msg)
else:
logger.info(err_msg)
return None
return comp
@ -104,6 +113,7 @@ class Mapper[C: Component]:
attr_comp : str | C,
coll_comp : str | C | None = None,
coll_groups : dict[str | None, str | C] | None = None,
strict = True,
) -> None:
'''
Parameters:
@ -115,18 +125,18 @@ class Mapper[C: Component]:
names to components
'''
# check attribute component in registered schema
attr_comp = self._check_component(attr_comp)
attr_comp = self._check_component(attr_comp, strict=strict)
self.attribute_comps[type_ref] = attr_comp
# check default component in registered schema
if coll_comp is not None:
coll_comp = self._check_component(coll_comp)
coll_comp = self._check_component(coll_comp, strict=strict)
self.collation_groups[type_ref][None] = coll_comp
# check if any component in group dict not in registered schema
if coll_groups is not None:
for coll_key in coll_groups:
coll_groups[coll_key] = self._check_component(coll_groups[coll_key])
coll_groups[coll_key] = self._check_component(coll_groups[coll_key], strict=strict)
self.collation_groups[type_ref].update(coll_groups)
@ -135,6 +145,7 @@ class Mapper[C: Component]:
type_list: list[type[CO3]],
attr_name_map: Callable[[type[CO3]], str | C],
coll_name_map: Callable[[type[CO3], str], str | C] | None = None,
strict = False,
) -> None:
'''
Auto-register a set of types to the Mapper's attached Schema. Associations are
@ -156,10 +167,10 @@ class Mapper[C: Component]:
coll_groups = {}
if coll_name_map:
for action_group in _type.group_registry:
coll_groups[action_group] = coll_name_map(_type, action_group)
for group in _type.group_registry:
coll_groups[group] = coll_name_map(_type, group)
self.attach(_type, attr_comp, coll_groups=coll_groups)
self.attach(_type, attr_comp, coll_groups=coll_groups, strict=strict)
def get_attr_comp(
self,
@ -185,8 +196,8 @@ class Mapper[C: Component]:
def collect(
self,
obj : CO3,
action_keys : list[str] = None,
action_groups : list[str] = None,
keys : list[str] = None,
groups : list[str] = None,
) -> list:
'''
Stages inserts up the inheritance chain, and down through components.
@ -201,15 +212,15 @@ class Mapper[C: Component]:
Parameters:
obj: CO3 instance to collect from
action_keys: keys for actions to collect from
action_group: action group names to run all actions for
keys: keys for actions to collect from
group: action group names to run all actions for
Returns: dict with keys and values relevant for associated SQLite tables
'''
# default is to have no actions
if action_keys is None:
action_keys = []
#action_keys = list(obj.action_registry.keys())
if keys is None:
keys = []
#keys = list(obj.key_registry.keys())
receipts = []
for _cls in reversed(obj.__class__.__mro__[:-2]):
@ -225,22 +236,22 @@ class Mapper[C: Component]:
receipts=receipts,
)
for action_key in action_keys:
collation_data = obj.collate(action_key)
for key in keys:
collation_data = obj.collate(key)
# if method either returned no data or isn't registered, ignore
if collation_data is None:
continue
_, action_groups = obj.action_registry.get(action_key, (None, []))
for action_group in action_groups:
collation_component = self.get_coll_comp(_cls, group=action_group)
_, groups = obj.key_registry.get(key, (None, []))
for group in groups:
collation_component = self.get_coll_comp(_cls, group=group)
if collation_component is None:
continue
# gather connective data for collation components
connective_data = obj.collation_attributes(action_key, action_group)
connective_data = obj.collation_attributes(key, group)
self.collector.add_insert(
collation_component,
@ -345,7 +356,7 @@ class ComposableMapper[C: ComposableComponent](Mapper[C]):
def compose(
self,
co3_ref: CO3 | type[CO3],
action_groups: list[str] | None = None,
groups: list[str] | None = None,
*compose_args,
**compose_kwargs,
):
@ -375,9 +386,9 @@ class ComposableMapper[C: ComposableComponent](Mapper[C]):
# compose horizontally with components from provided action groups
coll_comp_agg = attr_comp
if action_groups is not None:
for action_group in action_groups:
coll_comp = self.get_coll_comp(_cls, group=action_group)
if groups is not None:
for group in groups:
coll_comp = self.get_coll_comp(_cls, group=group)
if coll_comp is None:
continue

29
tests/setups/vegetables.py
View File

@ -15,6 +15,11 @@ class Vegetable(CO3):
self.name = name
self.color = color
#@abstractmethod
@collate
def cut(self, method):
raise NotImplementedError
class Tomato(Vegetable):
def __init__(self, name, radius):
super().__init__(name, 'red')
@ -24,30 +29,40 @@ class Tomato(Vegetable):
def attributes(self):
return vars(self)
def collation_attributes(self, action_key, action_group):
def collation_attributes(self, key, group):
return {
'name': self.name,
'state': action_key,
'state': key,
}
@collate('ripe', action_groups=['aging'])
@collate('ripe', groups=['aging'])
def ripen(self):
return {
'age': random.randint(1, 6)
}
@collate('rotten', action_groups=['aging'])
@collate('rotten', groups=['aging'])
def rot(self):
return {
'age': random.randint(4, 9)
}
@collate('diced', action_groups=['cooking'])
@collate('diced', groups=['cooking'])
def dice(self):
return {
'pieces': random.randint(2, 12)
}
@collate
def cut(self, method):
if method == 'slice':
return {
'pieces': random.randint(2, 5)
}
elif method == 'dice':
return self.dice()
type_list = [Vegetable, Tomato]
'''
@ -114,8 +129,8 @@ vegetable_mapper = ComposableMapper(
def attr_name_map(cls):
return f'{cls.__name__.lower()}'
def coll_name_map(cls, action_group):
return f'{cls.__name__.lower()}_{action_group}_states'
def coll_name_map(cls, group):
return f'{cls.__name__.lower()}_{group}_states'
vegetable_mapper.attach_many(
type_list,

30
tests/test_co3.py
View File

@ -11,13 +11,18 @@ def test_co3_registry():
keys_to_groups = defaultdict(list)
# collect groups each key is associated
for action_group, action_keys in tomato.group_registry.items():
for action_key in action_keys:
keys_to_groups[action_key].append(action_group)
for group, keys in tomato.group_registry.items():
for key in keys:
keys_to_groups[key].append(group)
# check against `action_registry`, should map keys to all groups
for action_key, (_, action_groups) in tomato.action_registry.items():
assert keys_to_groups.get(action_key) == action_groups
assert set(tomato.key_registry.get(None,{}).keys()) == set(keys_to_groups.get(None,[]))
# check against `registry`, should map keys to all groups
for key, group_obj in tomato.key_registry.items():
if key is None: continue
_, groups = group_obj
assert keys_to_groups.get(key) == groups
def test_co3_attributes():
assert tomato.attributes is not None
@ -26,11 +31,12 @@ def test_co3_components():
assert tomato.components is not None
def test_co3_collation_attributes():
for action_group, action_keys in tomato.group_registry.items():
for action_key in action_keys:
assert tomato.collation_attributes(action_key, action_group) is not None
for group, keys in tomato.group_registry.items():
for key in keys:
assert tomato.collation_attributes(key, group) is not None
def test_co3_collate():
for action_group, action_keys in tomato.group_registry.items():
for action_key in action_keys:
assert tomato.collate(action_key) is not None
for group, keys in tomato.group_registry.items():
for key in keys:
if key is None: continue
assert tomato.collate(key) is not None