jw-python/tools/python/jwutils/db/schema/Column.py

# -*- 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)