janware/jw-python
1
mirror of ssh://git.janware.com/srv/git/janware/proj/jw-python synced 2026-01-15 09:53:32 +01:00
Code Releases Activity

schema: Continue

Browse source 
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2025-01-30 20:34:44 +01:00
commit c50c614f13
7 changed files with 223 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

137
tools/python/jwutils/db/schema/Table.py
View file

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from typing import Optional, Iterable, List, Any # TODO: Need any for many things, as I can't figure out how to avoid circular imports from here
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
import abc
from collections import OrderedDict
@ -18,26 +18,27 @@ class Table(abc.ABC): # export
assert(isinstance(name, str))
self.__schema = schema
self.__name = name
self.__primary_keys: Optional[Iterable[str]] = None
self.__unique_constraints: Optional[List[ColumnSet]] = None
self.__foreign_key_constraints: Optional[List[CompositeForeignKey]] = None
self.___columns: Optional[OrderedDict[str, Any]] = None
self.___foreign_key_parent_tables: Optional[OrderedDict[str, Any]] = None
self.__primary_keys: Optional[Iterable[str]] = None
self.__unique_constraints: Optional[list[ColumnSet]] = None
self.__foreign_key_constraints: Optional[list[CompositeForeignKey]] = None
self.__nullable_columns: Optional[Iterable[str]] = None
self.__non_nullable_columns: Optional[Iterable[str]] = None
self.__null_insertible_columns: Optional[Iterable[str]] = None
self.__not_null_insertible_columns: Optional[Iterable[str]] = None
self.__log_columns: Optional[Iterable[str]] = None
self.__display_columns: Optional[Iterable[str]] = None
self.__column_default: Optional[dict[str, Any]] = None
self.__base_location: Optional[Iterable[str]] = None
self.__location: Optional[Iterable[str]] = None
self.__row_location: Optional[Iterable[str]] = None
def __repr__(self) -> str:
return self.__name
self.__foreign_keys_to_parent_table: Optional[OrderedDict[str, Any]] = None
@property
def __columns(self):
def __columns(self) -> OrderedDict[str, Any]:
if self.___columns is None:
ret: OrderedDict[str, Any] = OrderedDict()
for name in self._column_names():
@ -45,16 +46,47 @@ class Table(abc.ABC): # export
self.___columns = ret
return self.___columns
@property
def __foreign_key_parent_tables(self) -> OrderedDict[str, Any]:
if self.___foreign_key_parent_tables is None:
self.___foreign_key_parent_tables = OrderedDict()
for cfk in self.foreign_key_constraints:
self.___foreign_key_parent_tables[cfk.parent_table.name] = cfk.parent_table
return self.___foreign_key_parent_tables
def __len__(self):
return len(self.__columns)
def __iter__(self):
yield from self.__columns.values()
def __repr__(self) -> str:
return self.__name
def __getitem__(self, index):
return self.__columns[index]
def __eq__(self, rhs) -> bool:
if isinstance(rhs, Table):
if self.__name != rhs.__name:
return False
return True
elif isinstance(rhs, str):
if self.__name != rhs:
return False
return True
throw(ERR, f'Tried to compare table {self} to type {type(rhs)}: {rhs}')
return False # Unreachable but requested by mypy
def __hash__(self) -> int:
return hash(self.name)
# -- To be reimplemented
@abc.abstractmethod
def _column_names(self) -> Iterable[str]:
pass
@abc.abstractmethod
def _log_columns(self) -> Iterable[str]:
pass
@abc.abstractmethod
def _column_data_type(self, name) -> DataType:
pass
@ -63,6 +95,18 @@ class Table(abc.ABC): # export
def _primary_keys(self) -> Iterable[str]:
pass
@abc.abstractmethod
def _log_columns(self) -> Iterable[str]:
pass
@abc.abstractmethod
def _display_columns(self) -> Iterable[str]:
return self._primary_keys()
@abc.abstractmethod
def _nullable_columns(self) -> Iterable[str]:
pass
@abc.abstractmethod
def _auto_increment_columns(self) -> Iterable[str]:
pass
@ -72,7 +116,7 @@ class Table(abc.ABC): # export
pass
@abc.abstractmethod
def _unique_constraints(self) -> List[List[str]]:
def _unique_constraints(self) -> list[list[str]]:
pass
def _model_name(self) -> Optional[str]:
@ -84,9 +128,13 @@ class Table(abc.ABC): # export
throw(ERR, "Not implemented")
return None
@abc.abstractmethod
def _query_name(self) -> str:
return self.__name
# -- common URL schema for all data
def _base_location(self) -> Optional[str]:
return f'/self.name'
return f'/{self.name}'
def _location(self) -> Optional[str]:
ret = ''
@ -98,9 +146,10 @@ class Table(abc.ABC): # export
def _row_location(self) -> Optional[str]:
ret = self._location()
for col in self.primary_keys:
if col not in self.__schema.access_defining_columns:
ret += f'/<{col}>'
if ret is not None:
for col in self.primary_keys:
if col not in self.__schema.access_defining_columns:
ret += f'/<{col}>'
return ret
# -- To be used
@ -136,6 +185,9 @@ class Table(abc.ABC): # export
def model_name(self) -> Optional[str]:
return self._model_name()
def query_name(self) -> str:
return self._query_name()
@property
def base_location(self):
if self.__base_location is None:
@ -160,6 +212,22 @@ class Table(abc.ABC): # export
self.__primary_keys = self._primary_keys()
return self.__primary_keys
@property
def log_columns(self):
if self.__log_columns is None:
self.__log_columns = self._log_columns()
return self.__log_columns
@property
def display_columns(self):
if self.__display_columns is None:
self.__display_columns = self._display_columns()
return self.__display_columns
@property
def auto_increment_columns(self) -> Iterable[str]:
return self._auto_increment_columns()
@property
def nullable_columns(self) -> Iterable[str]:
if self.__nullable_columns is None:
@ -199,19 +267,9 @@ class Table(abc.ABC): # export
return self.__not_null_insertible_columns
@property
def log_columns(self):
if self.__log_columns is None:
self.__log_columns = self._log_columns()
return self.__log_columns
@property
def auto_increment_columns(self) -> Iterable[str]:
return self._auto_increment_columns()
@property
def unique_constraints(self) -> List[ColumnSet]:
def unique_constraints(self) -> list[ColumnSet]:
if self.__unique_constraints is None:
ret: List[ColumnSet] = []
ret: list[ColumnSet] = []
impl = self._unique_constraints()
if impl is not None:
for columns in impl:
@ -220,11 +278,26 @@ class Table(abc.ABC): # export
return self.__unique_constraints
@property
def foreign_key_constraints(self) -> List[CompositeForeignKey]:
def foreign_key_constraints(self) -> list[CompositeForeignKey]:
if self.__foreign_key_constraints is None:
ret: List[Any] = []
ret: list[Any] = []
for composite_key in self.__schema.foreign_key_constraints:
if composite_key.child_table == self:
ret.append(composite_key)
self.__foreign_key_constraints = ret
return self.__foreign_key_constraints
@property
def foreign_key_parent_tables(self):
return self.__foreign_key_parent_tables.values()
def foreign_keys_to_parent_table(self, parent_table) -> Iterable[CompositeForeignKey]:
if self.__foreign_keys_to_parent_table is None:
self.__foreign_keys_to_parent_table = OrderedDict()
for cfk in self.foreign_key_constraints:
pt = cfk.parent_table.name
if pt not in self.__foreign_keys_to_parent_table:
self.__foreign_keys_to_parent_table[pt] = []
self.__foreign_keys_to_parent_table[pt].append(cfk)
parent_table_name = parent_table if isinstance(parent_table, str) else parent_table.name
return self.__foreign_keys_to_parent_table[parent_table_name]