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

add inital test suite, fix minor Mapping group bug

Browse Source
This commit is contained in:
Sam G. 2024-04-18 02:06:42 -07:00
parent 5f2be91c5d
commit b369428eb3
16 changed files with 233 additions and 1367 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
co3/co3.py
View File

@ -12,22 +12,9 @@ import logging
from collections import defaultdict from collections import defaultdict
from functools import wraps, partial from functools import wraps, partial
#from localsys.db.schema import tables
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
#def register_format(_format):
# def decorator(func):
# self.collate.format_map[_format] = func
#
# @wraps(func)
# def register(*args, **kwargs):
# return func(*args, **kwargs)
#
# return register
# return decorator
def collate(action_key, action_groups=None): def collate(action_key, action_groups=None):
def decorator(func): def decorator(func):
nonlocal action_groups nonlocal action_groups

30
co3/managers/sql.py
View File

@ -35,13 +35,10 @@ Note: Options for insert/update model
build, then single thread bulk INSERT. (**Note**: this is what the method does). build, then single thread bulk INSERT. (**Note**: this is what the method does).
''' '''
from pathlib import Path import time
from collections import defaultdict
import logging import logging
import threading import threading
import math from pathlib import Path
import time
import pprint
from concurrent.futures import wait, as_completed from concurrent.futures import wait, as_completed
import sqlalchemy as sa import sqlalchemy as sa
@ -98,7 +95,11 @@ class SQLManager(RelationalManager[SQLTable]):
def add_router(self, router): def add_router(self, router):
self.routers.append(router) self.routers.append(router)
def recreate(self, schema: Schema[SQLTable], engine: SQLEngine): def recreate(
self,
schema: Schema[SQLTable],
engine: SQLEngine
) -> None:
''' '''
Ideally this remains open, as we can't necessarily rely on a SQLAlchemy metadata Ideally this remains open, as we can't necessarily rely on a SQLAlchemy metadata
object for all kinds of SQLDatabases (would depend on the backend, for instance). object for all kinds of SQLDatabases (would depend on the backend, for instance).
@ -118,16 +119,22 @@ class SQLManager(RelationalManager[SQLTable]):
connection, connection,
component, component,
inserts: list[dict], inserts: list[dict],
commit=True
): ):
''' '''
Parameters: Parameters:
''' '''
with self._insert_lock: with self._insert_lock:
connection.execute( res = connection.execute(
sa.insert(component.obj), sa.insert(component.obj),
inserts inserts
) )
if commit:
connection.commit()
return res
def insert_many(self, connection, inserts: dict): def insert_many(self, connection, inserts: dict):
''' '''
Perform provided table inserts. Perform provided table inserts.
@ -139,9 +146,10 @@ class SQLManager(RelationalManager[SQLTable]):
if total_inserts < 1: return if total_inserts < 1: return
logger.info(f'Total of {total_inserts} sync inserts to perform') logger.info(f'Total of {total_inserts} sync inserts to perform')
start = time.time()
# TODO: add some exception handling? may be fine w default propagation # TODO: add some exception handling? may be fine w default propagation
start = time.time() res_list = []
with self._insert_lock: with self._insert_lock:
for component in inserts: for component in inserts:
comp_inserts = inserts[component] comp_inserts = inserts[component]
@ -151,10 +159,14 @@ class SQLManager(RelationalManager[SQLTable]):
f'Inserting {len(comp_inserts)} out-of-date entries into component "{component}"' f'Inserting {len(comp_inserts)} out-of-date entries into component "{component}"'
) )
self.insert(connection, component, comp_inserts) res = self.insert(connection, component, comp_inserts, commit=False)
res_list.append(res)
connection.commit() connection.commit()
logger.info(f'Insert transaction completed successfully in {time.time()-start:.2f}s') logger.info(f'Insert transaction completed successfully in {time.time()-start:.2f}s')
return res_list
def _file_sync_bools(self): def _file_sync_bools(self):
synced_bools = [] synced_bools = []
fpaths = utils.paths.iter_nested_paths(self.collector.basepath, no_dir=True) fpaths = utils.paths.iter_nested_paths(self.collector.basepath, no_dir=True)

20
co3/mapper.py
View File

@ -131,7 +131,7 @@ class Mapper[C: Component]:
type_list: list[type[CO3]], type_list: list[type[CO3]],
attr_name_map: Callable[[type[CO3]], str | C], attr_name_map: Callable[[type[CO3]], str | C],
coll_name_map: Callable[[type[CO3], str], str | C] | None = None, coll_name_map: Callable[[type[CO3], str], str | C] | None = None,
): ) -> None:
''' '''
Auto-register a set of types to the Mapper's attached Schema. Associations are Auto-register a set of types to the Mapper's attached Schema. Associations are
made from types to both attribute and collation component names, through made from types to both attribute and collation component names, through
@ -194,8 +194,10 @@ class Mapper[C: Component]:
Returns: dict with keys and values relevant for associated SQLite tables Returns: dict with keys and values relevant for associated SQLite tables
''' '''
# default is to have no actions
if action_keys is None: if action_keys is None:
action_keys = list(obj.action_map.keys()) action_keys = []
#action_keys = list(obj.action_registry.keys())
receipts = [] receipts = []
for _cls in reversed(obj.__class__.__mro__[:-2]): for _cls in reversed(obj.__class__.__mro__[:-2]):
@ -218,7 +220,7 @@ class Mapper[C: Component]:
if collation_data is None: if collation_data is None:
continue continue
_, action_groups = obj.action_registry[action_key] _, action_groups = obj.action_registry.get(action_key, (None, []))
for action_group in action_groups: for action_group in action_groups:
collation_component = self.get_collation_comp(_cls, group=action_group) collation_component = self.get_collation_comp(_cls, group=action_group)
@ -329,7 +331,7 @@ class ComposableMapper[C: ComposableComponent](Mapper[C]):
def compose( def compose(
self, self,
obj: CO3, obj: CO3 | type[CO3],
action_groups: list[str] | None = None, action_groups: list[str] | None = None,
*compose_args, *compose_args,
**compose_kwargs, **compose_kwargs,
@ -344,12 +346,14 @@ class ComposableMapper[C: ComposableComponent](Mapper[C]):
chain (for components / sa.Relationships, it's a little easier). chain (for components / sa.Relationships, it's a little easier).
Parameters: Parameters:
outer: whether to use outer joins down the chain obj: either a CO3 instance or a type reference
conversion: whether to return conversion joins or base primitives
full: whether to return fully connected primitive and conversion table
''' '''
class_ref = obj
if isinstance(obj, CO3):
class_ref = obj.__class__
attr_comp_agg = None attr_comp_agg = None
for _cls in reversed(obj.__class__.__mro__[:-2]): for _cls in reversed(class_ref.__mro__[:-2]):
attr_comp = self.get_attribute_comp(_cls) attr_comp = self.get_attribute_comp(_cls)
# require an attribute component for type consideration # require an attribute component for type consideration

204
examples/.ipynb_checkpoints/database-checkpoint.ipynb
View File

@ -1,204 +0,0 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "6f6fbc7e-4fb9-4353-b2ee-9ea819a3c896",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/home/smgr/.pyenv/versions/co4/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
}
],
"source": [
"import vegetables"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "88fd0ea8-9c94-4569-a51b-823a04f32f55",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'age': 3}"
]
},
"execution_count": 2,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"tomato = vegetables.Tomato('t1', 5)\n",
"\n",
"# test a register collation action\n",
"tomato.collate('ripe')"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "348926d9-7137-4eff-a919-508788553dd2",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"['afff61ec-e0b0-44bb-9f0d-06008a82f6a5',\n",
" '5edfa13c-0eb1-4bbc-b55e-1550ff7df3d2',\n",
" '4568a2d4-eb41-4b15-9a29-e3e22906c661']"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collect(tomato, ['ripe'])"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "4e5e7319-11bf-4051-951b-08c84e9f3874",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<Component (SQLTable)> vegetable+tomato+tomato_aging_states+tomato_cooking_states"
]
},
"execution_count": 4,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.compose(tomato, action_groups=['aging', 'cooking'])"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "aa290686-8074-4038-a3cc-ce6817844653",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"[Column('id', Integer(), table=<vegetable>, primary_key=True, nullable=False),\n",
" Column('name', String(), table=<vegetable>),\n",
" Column('color', String(), table=<vegetable>),\n",
" Column('id', Integer(), table=<tomato>, primary_key=True, nullable=False),\n",
" Column('name', String(), ForeignKey('vegetable.name'), table=<tomato>),\n",
" Column('radius', Integer(), table=<tomato>),\n",
" Column('id', Integer(), table=<tomato_aging_states>, primary_key=True, nullable=False),\n",
" Column('name', String(), ForeignKey('tomato.name'), table=<tomato_aging_states>),\n",
" Column('state', String(), table=<tomato_aging_states>),\n",
" Column('age', Integer(), table=<tomato_aging_states>)]"
]
},
"execution_count": 5,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"list(vegetables.vegetable_mapper.compose(tomato, action_groups=['aging']).obj.columns)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "f3c7e37d-ba9e-4bae-ae44-adc922bf5f4c",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"(vegetables.Tomato, vegetables.Vegetable, co3.co3.CO3, object)"
]
},
"execution_count": 6,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"tomato.__class__.__mro__"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "c21d2c54-39e2-4de3-93bc-763896ed348e",
"metadata": {},
"outputs": [],
"source": [
"from co3.databases import SQLDatabase\n",
"\n",
"db = SQLDatabase('sqlite://', echo=True)"
]
},
{
"cell_type": "code",
"execution_count": 8,
"id": "a785d202-99d3-4ae7-859e-ee22b481f8df",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<contextlib._GeneratorContextManager at 0x7dd5c619be60>"
]
},
"execution_count": 8,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"db.recreate(vegetable_schema) "
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "cda01cb0-1666-4cb1-aa64-bcdca871aff5",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "co3",
"language": "python",
"name": "co3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 5
}

145
examples/.ipynb_checkpoints/mapper-checkpoint.ipynb
View File

@ -1,145 +0,0 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "e02ccafe-e04d-4312-acba-e41cf7b1c021",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/home/smgr/.pyenv/versions/co4/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
}
],
"source": [
"from co3 import Mapper\n",
"\n",
"import vegetables"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "7d80f7b9-7458-4ad4-8c1a-3ea56e796b4e",
"metadata": {},
"outputs": [],
"source": [
"vegetable_mapper = Mapper(\n",
" vegetables.Vegetable,\n",
" vegetables.vegetable_schema\n",
")\n",
"\n",
"vegetable_mapper.attach(\n",
" vegetables.Vegetable,\n",
" vegetables.vegetable_table,\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "f9408562-bf50-4522-909c-318557f85948",
"metadata": {},
"outputs": [],
"source": [
"# manually attach component\n",
"vegetable_mapper.attach(\n",
" vegetables.Tomato,\n",
" vegetables.tomato_table,\n",
" coll_groups={\n",
" 'aging': vegetables.vegetable_schema.get_component('tomato_aging_states'),\n",
" 'cooking': vegetables.vegetable_schema.get_component('tomato_cooking_states'),\n",
" },\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "05fdd404-87ee-4187-832f-2305272758ae",
"metadata": {},
"outputs": [],
"source": [
"# attach by name in schema\n",
"vegetable_mapper.attach(\n",
" vegetables.Tomato,\n",
" vegetables.tomato_table,\n",
" coll_groups={\n",
" 'aging': 'tomato_aging_states',\n",
" 'cooking': 'tomato_cooking_states',\n",
" },\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e9b6af49-a69d-41cc-beae-1b6f171cd2f5",
"metadata": {},
"outputs": [],
"source": [
"# attach entire type hierarchy w/ type->name map\n",
"vegetable_mapper.attach_hierarchy(\n",
"# this might make more sense during init\n",
" lambda x:x.__name__.lower())\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 9,
"id": "2e4336ab-5b5f-484d-815d-164d4b6f40a0",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"defaultdict(dict,\n",
" {vegetables.Tomato: {'aging': <co3.components.SQLTable at 0x7ece94358aa0>,\n",
" 'cooking': <co3.components.SQLTable at 0x7ece94358ad0>}})"
]
},
"execution_count": 9,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetable_mapper.collation_groups"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d416f9cd-2cb6-4a6e-bab7-86ac21216b8c",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "co3",
"language": "python",
"name": "co3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 5
}

135
examples/.ipynb_checkpoints/vegetables-checkpoint.py
View File

@ -1,135 +0,0 @@
'''
just remembered tomatos aren't vegetables. whoops
'''
import random
import sqlalchemy as sa
from co3.schemas import SQLSchema
from co3 import CO3, collate, Mapper, ComposableMapper
from co3 import util
class Vegetable(CO3):
def __init__(self, name, color):
self.name = name
self.color = color
class Tomato(Vegetable):
def __init__(self, name, radius):
super().__init__(name, 'red')
self.radius = radius
@property
def attributes(self):
return vars(self)
def collation_attributes(self, action_key, action_grounp):
return {
'name': self.name,
'state': action_key,
}
@collate('ripe', action_groups=['aging'])
def ripen(self):
return {
'age': random.randint(1, 6)
}
@collate('rotten', action_groups=['aging'])
def rot(self):
return {
'age': random.randint(4, 9)
}
@collate('diced', action_groups=['cooking'])
def dice(self):
return {
'pieces': random.randint(2, 12)
}
type_list = [Vegetable, Tomato]
'''
VEGETABLE
|
TOMATO -- AGING
|
-- COOKING
Note: foreign keys need to represent values that could be known by objects _without_ first interacting
with a DB. This is slightly non-standard, given how common it is to depend on another table's integer ID
(typically a value assigned by the DB using an autoincrement, for example, and not specified explicitly
within the insertion body). As a result, SQLTable components need to be able to operate by another unique
key when expected to connect to other tables in the hierarchy. Below we use `name` with a UNIQUE constraint
for this purpose. Note that having an integer `id` is still perfectly okay so that a table can manage
uniqueness of its own rows by default.
'''
metadata = sa.MetaData()
vegetable_table = sa.Table(
'vegetable',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, unique=True),
sa.Column('color', sa.String),
)
tomato_table = sa.Table(
'tomato',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, util.db.deferred_cd_fkey('vegetable.name'), unique=True),
sa.Column('radius', sa.Integer),
)
tomato_aging_table = sa.Table(
'tomato_aging_states',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, util.db.deferred_cd_fkey('tomato.name'), unique=True),
sa.Column('state', sa.String),
sa.Column('age', sa.Integer),
)
tomato_cooking_table = sa.Table(
'tomato_cooking_states',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, util.db.deferred_cd_fkey('tomato.name'), unique=True),
sa.Column('state', sa.String),
sa.Column('pieces', sa.Integer),
)
vegetable_schema = SQLSchema.from_metadata(metadata)
def general_compose_map(c1, c2):
return c1.obj.c.name == c2.obj.c.name
vegetable_mapper = ComposableMapper(
vegetable_schema,
attr_compose_map=general_compose_map,
coll_compose_map=general_compose_map,
)
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'
vegetable_mapper.attach_many(
type_list,
attr_name_map,
coll_name_map,
)
'''
new mapping type for Mapper attachment:
Callable[ [type[CO3], str|None], tuple[str, tuple[str], tuple[str]]]
tail tuples to associate column names from central table to collation
this should complete the auto-compose horizontally
'''

429
examples/database.ipynb
View File

@ -1,429 +0,0 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "6f6fbc7e-4fb9-4353-b2ee-9ea819a3c896",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/home/smgr/.pyenv/versions/co4/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
}
],
"source": [
"import vegetables"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "88fd0ea8-9c94-4569-a51b-823a04f32f55",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'age': 5}"
]
},
"execution_count": 2,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"tomato = vegetables.Tomato('t1', 5)\n",
"\n",
"# test a register collation action\n",
"tomato.collate('ripe')"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "348926d9-7137-4eff-a919-508788553dd2",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"['9ca6772e-6621-4511-a4a6-ad451a1da91f',\n",
" '2a91b423-4e08-491c-b1d2-5ec25259191e',\n",
" '4a9edb2b-4ac5-467e-82ef-b254829ac2a2']"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collect(tomato, ['ripe'])"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "4e5e7319-11bf-4051-951b-08c84e9f3874",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<Component (SQLTable)> vegetable+tomato+tomato_aging_states+tomato_cooking_states"
]
},
"execution_count": 4,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.compose(tomato, action_groups=['aging', 'cooking'])"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "aa290686-8074-4038-a3cc-ce6817844653",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"[Column('id', Integer(), table=<vegetable>, primary_key=True, nullable=False),\n",
" Column('name', String(), table=<vegetable>),\n",
" Column('color', String(), table=<vegetable>),\n",
" Column('id', Integer(), table=<tomato>, primary_key=True, nullable=False),\n",
" Column('name', String(), ForeignKey('vegetable.name'), table=<tomato>),\n",
" Column('radius', Integer(), table=<tomato>),\n",
" Column('id', Integer(), table=<tomato_aging_states>, primary_key=True, nullable=False),\n",
" Column('name', String(), ForeignKey('tomato.name'), table=<tomato_aging_states>),\n",
" Column('state', String(), table=<tomato_aging_states>),\n",
" Column('age', Integer(), table=<tomato_aging_states>)]"
]
},
"execution_count": 5,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"list(vegetables.vegetable_mapper.compose(tomato, action_groups=['aging']).obj.columns)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "f3c7e37d-ba9e-4bae-ae44-adc922bf5f4c",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"(vegetables.Tomato, vegetables.Vegetable, co3.co3.CO3, object)"
]
},
"execution_count": 6,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"tomato.__class__.__mro__"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "c21d2c54-39e2-4de3-93bc-763896ed348e",
"metadata": {},
"outputs": [],
"source": [
"from co3.databases import SQLDatabase\n",
"\n",
"db = SQLDatabase('sqlite://') #, echo=True)"
]
},
{
"cell_type": "code",
"execution_count": 8,
"id": "a785d202-99d3-4ae7-859e-ee22b481f8df",
"metadata": {
"scrolled": true
},
"outputs": [],
"source": [
"db.recreate(vegetables.vegetable_schema)"
]
},
{
"cell_type": "code",
"execution_count": 9,
"id": "cda01cb0-1666-4cb1-aa64-bcdca871aff5",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{<Component (SQLTable)> vegetable: [{'name': 't1', 'color': 'red'}],\n",
" <Component (SQLTable)> tomato: [{'name': 't1', 'radius': 5}],\n",
" <Component (SQLTable)> tomato_aging_states: [{'name': 't1',\n",
" 'state': 'ripe',\n",
" 'age': 2}]}"
]
},
"execution_count": 9,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collector.inserts"
]
},
{
"cell_type": "code",
"execution_count": 10,
"id": "af7124ed-3031-4f28-89a6-553eb5b3cc7a",
"metadata": {},
"outputs": [],
"source": [
"with db.engine.connect() as connection:\n",
" db.manager.insert_many(\n",
" connection, \n",
" vegetables.vegetable_mapper.collector.inserts,\n",
" )"
]
},
{
"cell_type": "code",
"execution_count": 11,
"id": "0149e14e-5d07-42af-847d-af5c190f8946",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"[{'id': 1, 'name': 't1', 'radius': 5}]\n"
]
}
],
"source": [
"with db.engine.connect() as connection:\n",
" print(db.accessor.select(\n",
" connection, \n",
" vegetables.vegetable_schema.get_component('tomato')\n",
" ))"
]
},
{
"cell_type": "code",
"execution_count": 32,
"id": "668d1b8c-b47f-4a58-914d-e43402443fe6",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"[{'id': 1, 'name': 't1', 'color': 'red', 'id_1': 1, 'name_1': 't1', 'radius': 5}]\n"
]
}
],
"source": [
"agg_table = vegetables.vegetable_mapper.compose(tomato)\n",
"\n",
"with db.engine.connect() as connection:\n",
" agg_res = db.accessor.select(\n",
" connection, \n",
" agg_table,\n",
" mappings=True,\n",
" )\n",
"\n",
"print(agg_res)"
]
},
{
"cell_type": "code",
"execution_count": 33,
"id": "a051d72d-a867-46dc-bb5e-69341f39a056",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"[{'id': 1, 'name': 't1', 'color': 'red', 'id_1': 1, 'name_1': 't1', 'radius': 5, 'id_2': 1, 'name_2': 't1', 'state': 'ripe', 'age': 2}]\n"
]
}
],
"source": [
"agg_table = vegetables.vegetable_mapper.compose(tomato, action_groups=['aging'])#, outer=True)\n",
"\n",
"with db.engine.connect() as connection:\n",
" agg_res = db.accessor.select(\n",
" connection, \n",
" agg_table,\n",
" mappings=True,\n",
" )\n",
"\n",
"print(agg_res)"
]
},
{
"cell_type": "code",
"execution_count": 31,
"id": "6a80cfd7-3175-4526-96e0-374765d64a27",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"sqlalchemy.engine.row.RowMapping"
]
},
"execution_count": 31,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"type(agg_res[0])"
]
},
{
"cell_type": "code",
"execution_count": 15,
"id": "7cf05ddd-2328-4051-9cf8-4ac01352405e",
"metadata": {},
"outputs": [],
"source": [
"import sqlalchemy as sa\n",
"from co3.engines import SQLEngine\n",
"\n",
"a = SQLEngine.execute(db.engine.connect(), sa.select(agg_table.obj))"
]
},
{
"cell_type": "code",
"execution_count": 19,
"id": "c1edf68e-1fde-4a1f-8ec3-084713a8da45",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"[]"
]
},
"execution_count": 19,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"a.mappings().all()\n"
]
},
{
"cell_type": "code",
"execution_count": 37,
"id": "8b8a9e47-7f5f-4828-a99e-5d9a12697f46",
"metadata": {},
"outputs": [],
"source": [
"tomato2 = vegetables.Tomato('t2', 8)"
]
},
{
"cell_type": "code",
"execution_count": 38,
"id": "062aa4de-7aea-4fd3-b5db-82af147d023e",
"metadata": {},
"outputs": [
{
"ename": "AttributeError",
"evalue": "'Tomato' object has no attribute 'action_map'",
"output_type": "error",
"traceback": [
"\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
"\u001b[0;31mAttributeError\u001b[0m Traceback (most recent call last)",
"Cell \u001b[0;32mIn[38], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43mvegetables\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mvegetable_mapper\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcollect\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtomato2\u001b[49m\u001b[43m)\u001b[49m\n",
"File \u001b[0;32m~/Documents/projects/ontolog/co3/build/__editable__.co3-0.1.1-py3-none-any/co3/mapper.py:198\u001b[0m, in \u001b[0;36mMapper.collect\u001b[0;34m(self, obj, action_keys, action_groups)\u001b[0m\n\u001b[1;32m 179\u001b[0m \u001b[38;5;250m\u001b[39m\u001b[38;5;124;03m'''\u001b[39;00m\n\u001b[1;32m 180\u001b[0m \u001b[38;5;124;03mStages inserts up the inheritance chain, and down through components.\u001b[39;00m\n\u001b[1;32m 181\u001b[0m \n\u001b[0;32m (...)\u001b[0m\n\u001b[1;32m 195\u001b[0m \u001b[38;5;124;03mReturns: dict with keys and values relevant for associated SQLite tables\u001b[39;00m\n\u001b[1;32m 196\u001b[0m \u001b[38;5;124;03m'''\u001b[39;00m\n\u001b[1;32m 197\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m action_keys \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[0;32m--> 198\u001b[0m action_keys \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mlist\u001b[39m(\u001b[43mobj\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43maction_map\u001b[49m\u001b[38;5;241m.\u001b[39mkeys())\n\u001b[1;32m 200\u001b[0m receipts \u001b[38;5;241m=\u001b[39m []\n\u001b[1;32m 201\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m _cls \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mreversed\u001b[39m(obj\u001b[38;5;241m.\u001b[39m\u001b[38;5;18m__class__\u001b[39m\u001b[38;5;241m.\u001b[39m\u001b[38;5;18m__mro__\u001b[39m[:\u001b[38;5;241m-\u001b[39m\u001b[38;5;241m2\u001b[39m]):\n",
"\u001b[0;31mAttributeError\u001b[0m: 'Tomato' object has no attribute 'action_map'"
]
}
],
"source": [
"vegetables.vegetable_mapper.collect(tomato2)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4673ddc8-3f76-4d8c-8186-bbed4a682e0d",
"metadata": {},
"outputs": [],
"source": [
"db.insert(vegetables.vegetable_schema.get_component('tomato'), "
]
},
{
"cell_type": "code",
"execution_count": 36,
"id": "9314be4e-c1d5-4af8-ad23-0b208d24b3eb",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"[{'id': 1, 'name': 't1', 'radius': 5}]"
]
},
"execution_count": 36,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"db.select(vegetables.vegetable_schema.get_component('tomato'))"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a2efd060-f298-4ca6-8a58-7ed5acf1dd15",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "co3",
"language": "python",
"name": "co3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 5
}

405
examples/mapper.ipynb
View File

@ -1,405 +0,0 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "e02ccafe-e04d-4312-acba-e41cf7b1c021",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/home/smgr/.pyenv/versions/co4/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
}
],
"source": [
"import vegetables"
]
},
{
"cell_type": "markdown",
"id": "c0914069-7f3c-4213-8d34-f7566033e054",
"metadata": {},
"source": [
"## Development notes\n",
"- No registry actually needs to take place if there's a default type2component map or one supplied on creation. Can just collect right out of the gate\n",
"- Need connective function (type to collation) and attribute map. Do we need to this with a subclass? If a func is passed in on init, I can type it appropriately I guess `Callable[[type[CO3],str,str|None],dict]`"
]
},
{
"cell_type": "markdown",
"id": "ef733715-bb75-4263-b216-45e778a06b21",
"metadata": {},
"source": [
"## Usage\n",
"The Mapper's primary job is to associate class hierarchies with database components. This can be done in a few ways:\n",
"\n",
"1. Manually attaching a type reference to a Component\n",
"2. Attaching a type reference to a Component's name as registered in a schema\n",
"3. Automatically register the CO3 heirarchy to matching schema component names (through transformation)"
]
},
{
"cell_type": "markdown",
"id": "d2672422-3596-4eab-ac44-5da617f74b80",
"metadata": {
"jp-MarkdownHeadingCollapsed": true
},
"source": [
"## Explicit example steps"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "7d80f7b9-7458-4ad4-8c1a-3ea56e796b4e",
"metadata": {},
"outputs": [],
"source": [
"from co3 import Mapper\n",
"\n",
"vegetable_mapper = Mapper(\n",
" vegetables.Vegetable,\n",
" vegetables.vegetable_schema\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "d24d31b4-c4a6-4a1e-8bea-c44378aadfdd",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'\\nvegetable_mapper.attach(\\n vegetables.Vegetable,\\n vegetables.vegetable_table,\\n)\\n'"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"# not valid; tables need to be wrapped in CO3 Components\n",
"'''\n",
"vegetable_mapper.attach(\n",
" vegetables.Vegetable,\n",
" vegetables.vegetable_table,\n",
")\n",
"'''"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "f9408562-bf50-4522-909c-318557f85948",
"metadata": {},
"outputs": [],
"source": [
"# manually attach component\n",
"vegetable_mapper.attach(\n",
" vegetables.Tomato,\n",
" vegetables.vegetable_schema.get_component('tomato'),\n",
" coll_groups={\n",
" 'aging': vegetables.vegetable_schema.get_component('tomato_aging_states'),\n",
" 'cooking': vegetables.vegetable_schema.get_component('tomato_cooking_states'),\n",
" },\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "05fdd404-87ee-4187-832f-2305272758ae",
"metadata": {},
"outputs": [],
"source": [
"# attach by name in schema\n",
"vegetable_mapper.attach(\n",
" vegetables.Tomato,\n",
" 'tomato',\n",
" coll_groups={\n",
" 'aging': 'tomato_aging_states',\n",
" 'cooking': 'tomato_cooking_states',\n",
" },\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "e9b6af49-a69d-41cc-beae-1b6f171cd2f5",
"metadata": {},
"outputs": [],
"source": [
"# attach entire type hierarchy w/ type->name map\n",
"vegetable_mapper.attach_hierarchy(\n",
" # this might make more sense during init\n",
" vegetables.Vegetable,\n",
" lambda x:x.__name__.lower()\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 9,
"id": "0fb45a86-5c9b-41b1-a3ab-5691444f175e",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<co3.components.SQLTable at 0x7f2012b23f80>"
]
},
"execution_count": 9,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetable_mapper.get_collation_comp(vegetables.Tomato, 'cooking')"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "2e4336ab-5b5f-484d-815d-164d4b6f40a0",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'co3_root': vegetables.Vegetable,\n",
" 'schema': <co3.schemas.SQLSchema at 0x732074224aa0>,\n",
" 'collector': <co3.collector.Collector at 0x7320757da120>,\n",
" 'composer': <co3.composer.Composer at 0x7320757da9c0>,\n",
" 'attribute_comps': {vegetables.Tomato: <co3.components.SQLTable at 0x732074224cb0>},\n",
" 'collation_groups': defaultdict(dict,\n",
" {vegetables.Tomato: {'aging': <co3.components.SQLTable at 0x732074224ce0>,\n",
" 'cooking': <co3.components.SQLTable at 0x732074224d10>}})}"
]
},
"execution_count": 7,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vars(vegetable_mapper)"
]
},
{
"cell_type": "markdown",
"id": "47859e25-b803-4459-a581-f10bbcfac716",
"metadata": {},
"source": [
"## Holistic attachment"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "70c9baed-b870-4021-8949-9b713d863de6",
"metadata": {},
"outputs": [],
"source": [
"def attr_name_map(cls):\n",
" return f'{cls.__name__.lower()}'\n",
"\n",
"def coll_name_map(cls, action_group):\n",
" return f'{cls.__name__.lower()}_{action_group}_states'\n",
"\n",
"vegetables.vegetable_mapper.attach_many(\n",
" vegetables.type_list,\n",
" attr_name_map,\n",
" coll_name_map,\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "c16786d4-0b71-42d9-97f7-7893c542104e",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'age': 4}"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"# create new CO3 descendant\n",
"tomato = vegetables.Tomato('t1', 5)\n",
"\n",
"# test a register collation action\n",
"tomato.collate('ripe')"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "d7fa94ca-3ecd-4ee3-b0dc-f3b2b65ee47c",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<Component (SQLTable)> tomato"
]
},
"execution_count": 4,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.get_attribute_comp(vegetables.Tomato)"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "1adc3bc5-957f-4b5a-bc2c-2d172675826d",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'schema': <co3.schemas.SQLSchema at 0x7ab568224e60>,\n",
" 'collector': <co3.collector.Collector at 0x7ab568225190>,\n",
" 'composer': <co3.composer.Composer at 0x7ab5682251c0>,\n",
" 'attribute_comps': {vegetables.Vegetable: <Component (SQLTable)> vegetable,\n",
" vegetables.Tomato: <Component (SQLTable)> tomato},\n",
" 'collation_groups': defaultdict(dict,\n",
" {vegetables.Vegetable: {},\n",
" vegetables.Tomato: {'aging': <Component (SQLTable)> tomato_aging_states,\n",
" 'cooking': <Component (SQLTable)> tomato_cooking_states}})}"
]
},
"execution_count": 5,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vars(vegetables.vegetable_mapper)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "f32d1f65-9b1d-4600-b396-8551fbd1fcf7",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"['3bf42abc-8a12-452f-baf6-38a05fc5d420',\n",
" '271b7b84-846e-4d1d-87f6-bcabc90a7b55',\n",
" 'f9fc5d16-c5cb-47a7-9eca-7df8a3ba5d10']"
]
},
"execution_count": 6,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collect(tomato, ['ripe'])"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "380dfbea-90cc-49fc-aef1-ebb342872632",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"defaultdict(<function co3.collector.Collector.__init__.<locals>.<lambda>()>,\n",
" {'3bf42abc-8a12-452f-baf6-38a05fc5d420': (<Component (SQLTable)> vegetable,\n",
" {'name': 't1', 'color': 'red'}),\n",
" '271b7b84-846e-4d1d-87f6-bcabc90a7b55': (<Component (SQLTable)> tomato,\n",
" {'name': 't1', 'radius': 5}),\n",
" 'f9fc5d16-c5cb-47a7-9eca-7df8a3ba5d10': (<Component (SQLTable)> tomato_aging_states,\n",
" {'name': 't1', 'state': 'ripe', 'age': 1})})"
]
},
"execution_count": 7,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collector._inserts"
]
},
{
"cell_type": "code",
"execution_count": 8,
"id": "905bb2a9-9c22-4187-be15-3dd32d206e26",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{<Component (SQLTable)> vegetable: [{'name': 't1', 'color': 'red'}],\n",
" <Component (SQLTable)> tomato: [{'name': 't1', 'radius': 5}],\n",
" <Component (SQLTable)> tomato_aging_states: [{'name': 't1',\n",
" 'state': 'ripe',\n",
" 'age': 1}]}"
]
},
"execution_count": 8,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collector.inserts"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d166b9af-e3ba-4750-9dcb-d8d4e08fe4d3",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "co3",
"language": "python",
"name": "co3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 5
}

0
tests/co4_example.py → tests/co4_api_demo.py
View File

19
tests/imports.py
View File

@ -1,19 +0,0 @@
from co3 import Accessor
from co3 import CO3
from co3 import Collector
from co3 import Database
from co3 import Indexer
from co3 import Manager
from co3 import Mapper
from co3 import Relation
from co3.accessors import SQLAccessor
from co3.databases import SQLDatabase
from co3.databases import SQLiteDatabase
from co3.managers import SQLManager
from co3.relations import TabularRelation
from co3.relations import SQLTable

0
examples/vegetables.py → tests/setups/vegetables.py
View File

57
tests/test_co3.py Normal file
View File

@ -0,0 +1,57 @@
from co3.components import Relation
from setups import vegetables as veg
def test_mapper_getters():
veg_comp = veg.vegetable_schema.get_component('vegetable')
tom_comp = veg.vegetable_schema.get_component('tomato')
assert veg.vegetable_mapper.get_attribute_comp(veg.Vegetable) is veg_comp
assert veg.vegetable_mapper.get_attribute_comp(veg.Tomato) is tom_comp
tom_aging = veg.vegetable_schema.get_component('tomato_aging_states')
tom_cooking = veg.vegetable_schema.get_component('tomato_cooking_states')
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'aging') is tom_aging
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'cooking') is tom_cooking
def test_mapper_attach():
assert veg.vegetable_mapper.attach(
veg.Tomato,
'tomato',
coll_groups={
'aging': 'tomato_aging_states',
'cooking': 'tomato_cooking_states',
},
) is None
def test_mapper_attach_many():
assert veg.vegetable_mapper.attach_many(
[veg.Vegetable, veg.Tomato],
lambda t: f'{t.__name__.lower()}'
) is None
def test_mapper_collect():
tomato = veg.Tomato('t1', 10)
receipts = veg.vegetable_mapper.collect(tomato)
assert len(receipts) == 2
# attempt to retrieve receipts one at a time
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 1 # should be just one match
assert len(res1[next(iter(res1.keys()))]) == 1 # and one dict for matching comp
# try again, check no persistent match
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 0 # should be no matches for the same receipt
res2 = veg.vegetable_mapper.collector.collect_inserts([receipts[1]])
assert len(res2) == 1
assert len(res2[next(iter(res2.keys()))]) == 1

36
tests/test_database.py Normal file
View File

@ -0,0 +1,36 @@
from co3.components import Relation
from co3.databases import SQLDatabase
from setups import vegetables as veg
db = None
def test_database_init():
global db
db = SQLDatabase('sqlite://')
assert True
def test_database_recreate():
db.recreate(veg.vegetable_schema)
assert True
def test_database_insert():
tomato = veg.Tomato('t1', 5)
veg.vegetable_mapper.collect(tomato)
with db.engine.connect() as connection:
assert db.manager.insert_many(
connection,
veg.vegetable_mapper.collector.inserts,
) is not None
def test_database_access():
agg_table = veg.vegetable_mapper.compose(veg.Tomato)
with db.engine.connect() as connection:
assert db.accessor.select(
connection,
agg_table,
) is not None

25
tests/test_imports.py Normal file
View File

@ -0,0 +1,25 @@
def test_import():
from co3 import Accessor
from co3 import CO3
from co3 import Collector
from co3 import Database
from co3 import Indexer
from co3 import Manager
from co3 import Mapper
from co3 import Component
from co3.accessors import SQLAccessor
from co3.accessors import FTSAccessor
from co3.accessors import VSSAccessor
from co3.databases import SQLDatabase
from co3.databases import SQLiteDatabase
from co3.managers import SQLManager
from co3.components import ComposableComponent
from co3.components import Relation
from co3.components import SQLTable
assert True

56
tests/test_mapper.py Normal file
View File

@ -0,0 +1,56 @@
from co3.components import Relation
from setups import vegetables as veg
def test_mapper_getters():
veg_comp = veg.vegetable_schema.get_component('vegetable')
tom_comp = veg.vegetable_schema.get_component('tomato')
assert veg.vegetable_mapper.get_attribute_comp(veg.Vegetable) is veg_comp
assert veg.vegetable_mapper.get_attribute_comp(veg.Tomato) is tom_comp
tom_aging = veg.vegetable_schema.get_component('tomato_aging_states')
tom_cooking = veg.vegetable_schema.get_component('tomato_cooking_states')
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'aging') is tom_aging
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'cooking') is tom_cooking
def test_mapper_attach():
assert veg.vegetable_mapper.attach(
veg.Tomato,
'tomato',
coll_groups={
'aging': 'tomato_aging_states',
'cooking': 'tomato_cooking_states',
},
) is None
def test_mapper_attach_many():
assert veg.vegetable_mapper.attach_many(
[veg.Vegetable, veg.Tomato],
lambda t: f'{t.__name__.lower()}'
) is None
def test_mapper_collect():
tomato = veg.Tomato('t1', 10)
receipts = veg.vegetable_mapper.collect(tomato)
assert len(receipts) == 2
# attempt to retrieve receipts one at a time
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 1 # should be just one match
assert len(res1[next(iter(res1.keys()))]) == 1 # and one dict for matching comp
# try again, check no persistent match
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 0 # should be no matches for the same receipt
res2 = veg.vegetable_mapper.collector.collect_inserts([receipts[1]])
assert len(res2) == 1
assert len(res2[next(iter(res2.keys()))]) == 1

26
tests/test_schema.py Normal file
View File

@ -0,0 +1,26 @@
from co3.components import Relation
from setups import vegetables as veg
def test_schema_get():
veg_comp_raw = veg.vegetable_schema._component_map.get('vegetable')
veg_comp = veg.vegetable_schema.get_component('vegetable')
assert veg_comp_raw is veg_comp
def test_schema_contains():
vegetable_comp = veg.vegetable_schema.get_component('vegetable')
tomato_comp = veg.vegetable_schema.get_component('tomato')
tomato_aging_comp = veg.vegetable_schema.get_component('tomato_aging_states')
assert vegetable_comp in veg.vegetable_schema
assert tomato_comp in veg.vegetable_schema
assert tomato_aging_comp in veg.vegetable_schema
def test_schema_add():
veg.vegetable_schema.add_component(Relation[int]('a', 1))
veg.vegetable_schema.add_component(Relation[int]('b', 2))
assert veg.vegetable_schema.get_component('a') is not None
assert veg.vegetable_schema.get_component('b') is not None