mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 09:53:32 +01:00
109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Optional, Iterable
|
|
|
|
import abc
|
|
|
|
from ...log import *
|
|
|
|
from .Table import Table
|
|
from .Column import Column
|
|
from .DataType import DataType
|
|
from .CompositeForeignKey import CompositeForeignKey
|
|
|
|
class Schema(abc.ABC): # export
|
|
|
|
def __init__(self) -> None:
|
|
self.___tables: Optional[list[Table]] = None
|
|
self.__foreign_keys: Optional[list[CompositeForeignKey]] = None
|
|
self.__access_defining_columns: Optional[list[str]] = None
|
|
|
|
@property
|
|
def __tables(self):
|
|
if self.___tables is None:
|
|
ret = dict()
|
|
for name in self._table_names():
|
|
slog(DEBUG, f'Caching metadata for table "{name}"')
|
|
assert(isinstance(name, str))
|
|
ret[name] = self._table(name)
|
|
self.___tables = ret
|
|
return self.___tables
|
|
|
|
# ------ API to be implemented
|
|
|
|
@abc.abstractmethod
|
|
def _table_names(self) -> Iterable[str]:
|
|
throw(ERR, "Called pure virtual base class method")
|
|
return []
|
|
|
|
@abc.abstractmethod
|
|
def _table(self, name: str) -> Table:
|
|
throw(ERR, "Called pure virtual base class method")
|
|
return None # type: ignore
|
|
|
|
@abc.abstractmethod
|
|
def _foreign_keys(self) -> list[CompositeForeignKey]:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def _access_defining_columns(self):
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def _model_module_search_paths(self) -> list[tuple[str, type]]:
|
|
pass
|
|
|
|
# ------ API to be called
|
|
|
|
def __len__(self):
|
|
return len(self.__tables)
|
|
|
|
def __iter__(self):
|
|
yield from self.__tables.values()
|
|
|
|
def __repr__(self):
|
|
return '|'.join([table.name for table in self.__tables])
|
|
|
|
def __getitem__(self, index):
|
|
return self.__tables[index]
|
|
|
|
@property
|
|
def table_names(self) -> Iterable[str]:
|
|
return self.__tables.keys()
|
|
|
|
@property
|
|
def tables(self) -> Iterable[Table]:
|
|
return self.__tables.values()
|
|
|
|
@property
|
|
def access_defining_columns(self):
|
|
if self.__access_defining_columns is None:
|
|
self.__access_defining_columns = self._access_defining_columns()
|
|
return self.__access_defining_columns
|
|
|
|
@property
|
|
def foreign_key_constraints(self) -> list[CompositeForeignKey]:
|
|
if self.__foreign_keys is None:
|
|
self.__foreign_keys = self._foreign_keys()
|
|
return self.__foreign_keys
|
|
|
|
def table(self, name: str) -> Table:
|
|
return self.__tables[name]
|
|
|
|
def table_by_model_name(self, name: str, throw=False) -> Table:
|
|
for table in self.__tables.values():
|
|
if table.model_name == name:
|
|
return table
|
|
if throw:
|
|
raise Exception(f'Table "{name}" not found in database metadata')
|
|
return None # type: ignore
|
|
|
|
def primary_keys(self, table_name: str) -> Iterable[str]:
|
|
return self.__tables[table_name].primary_keys
|
|
|
|
def columns(self, table_name: str) -> Iterable[Column]:
|
|
return self.__tables[table_name].columns
|
|
|
|
@property
|
|
def model_module_search_paths(self) -> list[tuple[str, type]]:
|
|
return self._model_module_search_paths()
|