mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 01:52:56 +01:00
schema.Table: Add property relationships
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
parent
9685f07eb8
commit
ba1b7eda19
1 changed files with 25 additions and 1 deletions
|
|
@ -1,6 +1,6 @@
|
||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
from typing import Optional, Iterable, Any # TODO: Need any for many things, as I can't figure out how to avoid circular imports from here
|
from typing import Optional, Union, Iterable, Self, Any # TODO: Need any for many things, as I can't figure out how to avoid circular imports from here
|
||||||
|
|
||||||
import abc
|
import abc
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
|
|
@ -38,7 +38,9 @@ class Table(abc.ABC): # export
|
||||||
self.__location_rule: Optional[Iterable[str]] = None
|
self.__location_rule: Optional[Iterable[str]] = None
|
||||||
self.__row_location_rule: Optional[Iterable[str]] = None
|
self.__row_location_rule: Optional[Iterable[str]] = None
|
||||||
self.__foreign_keys_to_parent_table: Optional[OrderedDict[str, Any]] = None
|
self.__foreign_keys_to_parent_table: Optional[OrderedDict[str, Any]] = None
|
||||||
|
self.__relationships: Optional[list[tuple[str, Self]]] = None
|
||||||
self.__model_class: Optional[Any] = None
|
self.__model_class: Optional[Any] = None
|
||||||
|
self.__relationship_by_foreign_table: Optional[dict[str, Self]] = None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def __columns(self) -> OrderedDict[str, Any]:
|
def __columns(self) -> OrderedDict[str, Any]:
|
||||||
|
|
@ -134,6 +136,9 @@ class Table(abc.ABC): # export
|
||||||
def _query_name(self) -> str:
|
def _query_name(self) -> str:
|
||||||
return 'tbl/' + self.__name
|
return 'tbl/' + self.__name
|
||||||
|
|
||||||
|
def _relationships(self) -> list[Self]:
|
||||||
|
return []
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def _row_query_name(self) -> str:
|
def _row_query_name(self) -> str:
|
||||||
return 'row/' + self.__name
|
return 'row/' + self.__name
|
||||||
|
|
@ -336,3 +341,22 @@ class Table(abc.ABC): # export
|
||||||
self.__foreign_keys_to_parent_table[pt].append(cfk)
|
self.__foreign_keys_to_parent_table[pt].append(cfk)
|
||||||
parent_table_name = parent_table if isinstance(parent_table, str) else parent_table.name
|
parent_table_name = parent_table if isinstance(parent_table, str) else parent_table.name
|
||||||
return self.__foreign_keys_to_parent_table[parent_table_name]
|
return self.__foreign_keys_to_parent_table[parent_table_name]
|
||||||
|
|
||||||
|
@property
|
||||||
|
def relationships(self) -> list[tuple[str, Self]]:
|
||||||
|
if self.__relationships is None:
|
||||||
|
ret = []
|
||||||
|
for member_name, table_name in self._relationships():
|
||||||
|
ret.append((member_name, self.schema[table_name]))
|
||||||
|
self.__relationships = ret
|
||||||
|
return self.__relationships
|
||||||
|
|
||||||
|
def relationship(self, table: Union[Self, str]) -> bool:
|
||||||
|
if isinstance(table, Table):
|
||||||
|
table = table.name
|
||||||
|
if self.__relationship_by_foreign_table is None:
|
||||||
|
ret: dict[str, Self] = dict()
|
||||||
|
for member_name, table_name in self.relationships:
|
||||||
|
ret[member_name] = self.schema[table_name]
|
||||||
|
self.__relationship_by_foreign_table = ret
|
||||||
|
return self.__relationship_by_foreign_table.get(table)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue