mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 18:03:31 +01:00
121 lines
4.1 KiB
Python
121 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Optional, Any
|
|
|
|
import abc
|
|
|
|
from ...log import *
|
|
|
|
from .DataType import DataType
|
|
|
|
class Column(abc.ABC): # export
|
|
|
|
def __init__(self, table, name, data_type: DataType):
|
|
self.__name: str = name
|
|
self.__table: Any = table
|
|
self.__is_nullable: Optional[bool] = None
|
|
self.__is_null_insertible: Optional[bool] = None
|
|
self.__is_primary_key: Optional[bool] = None
|
|
self.__default_value: Optional[Any] = None
|
|
self.__default_value_cached: bool = False
|
|
self.__is_auto_increment: Optional[bool] = None
|
|
self.__translate: Optional[bool] = None
|
|
self.__data_type: DataType = data_type
|
|
self.__foreign_keys: Optional[Any] = None
|
|
self.__foreign_keys_cached: bool = False
|
|
self.__foreign_keys_by_table: Optional[dict[str, Any]] = None
|
|
|
|
def __repr__(self):
|
|
return f'{self.__table.name}.{self.__name}: {self.__data_type}'
|
|
|
|
def __eq__(self, rhs) -> bool:
|
|
if isinstance(rhs, Column):
|
|
if self.__table != rhs.__table:
|
|
return False
|
|
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 column {self} to type {type(rhs)}: {rhs}')
|
|
return False # Unreachable but requested by mypy
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.__name
|
|
|
|
@property
|
|
def data_type(self):
|
|
return self.__data_type
|
|
|
|
@property
|
|
def table(self) -> str:
|
|
return self.__table
|
|
|
|
@property
|
|
def is_nullable(self) -> bool:
|
|
if self.__is_nullable is None:
|
|
self.__is_nullable = self.__name in self.__table.nullable_columns
|
|
return self.__is_nullable
|
|
|
|
@property
|
|
def is_null_insertible(self):
|
|
if self.__is_null_insertible is None:
|
|
ret = False
|
|
if self.is_nullable:
|
|
ret = True
|
|
elif self.is_auto_increment:
|
|
ret = True
|
|
elif self.default_value is not None:
|
|
ret = True
|
|
self.__is_null_insertible = ret
|
|
return self.__is_null_insertible
|
|
|
|
@property
|
|
def is_primary_key(self) -> bool:
|
|
if self.__is_primary_key is None:
|
|
self.__is_primary_key = self.__name in self.__table.primary_keys
|
|
return self.__is_primary_key
|
|
|
|
@property
|
|
def is_auto_increment(self) -> bool:
|
|
if self.__is_auto_increment is None:
|
|
self.__is_auto_increment = self.__name in self.__table.auto_increment_columns
|
|
return self.__is_auto_increment
|
|
|
|
@property
|
|
def translate(self) -> bool:
|
|
if self.__translate is None:
|
|
self.__translate = self.__name in self.__table.translate_columns
|
|
return self.__translate
|
|
|
|
@property
|
|
def default_value(self) -> Optional[Any]:
|
|
if self.__default_value_cached is False:
|
|
self.__default_value = self.__table.column_default(self.name)
|
|
self.__default_value_cached = True
|
|
return self.__default_value
|
|
|
|
# Returns Column object on parent table
|
|
@property
|
|
def foreign_keys(self) -> Optional[Any]:
|
|
if not self.__foreign_keys_cached:
|
|
fks: list[Any] = []
|
|
for cfk in self.__table.foreign_key_constraints:
|
|
for fk in cfk:
|
|
if fk.child_column == self:
|
|
fks.append(fk.parent_column)
|
|
self.__foreign_keys_cached = True
|
|
self.__foreign_keys = fks if fks else None
|
|
return self.__foreign_keys
|
|
|
|
def foreign_key(self, table) -> Optional[Any]:
|
|
if self.__foreign_keys_by_table is None:
|
|
self.__foreign_keys_by_table = dict()
|
|
for col in self.foreign_keys: # type: ignore # Any not iterable
|
|
assert(col.table.name not in self.__foreign_keys_by_table)
|
|
self.__foreign_keys_by_table[col.table.name] = col
|
|
table_name = table if isinstance(table, str) else table.name
|
|
return self.__foreign_keys_by_table.get(table_name)
|