Source code for pytablereader.interface

"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""

import abc
import threading

import path
import typepy
from dataproperty import DataPropertyExtractor

from pytablereader import InvalidTableNameError

from ._constant import SourceType
from ._constant import TableNameTemplate as tnt


class TableLoaderInterface(metaclass=abc.ABCMeta):
    """
    Interface class of table loader class.
    """

    @abc.abstractproperty
    def format_name(self):  # pragma: no cover
        pass

    @abc.abstractproperty
    def source_type(self):  # pragma: no cover
        pass

    @abc.abstractmethod
    def load(self):  # pragma: no cover
        pass

    @abc.abstractmethod
    def inc_table_count(self):  # pragma: no cover
        pass


[docs]class AbstractTableReader(TableLoaderInterface, metaclass=abc.ABCMeta): """ The abstract class of table data file loader. .. py:attribute:: table_name Table name string. .. py:attribute:: source Table data source to load. """ __table_count_lock = threading.Lock() __global_table_count = 0 __format_table_count = {} @property def source_type(self): return self._validator.source_type @property def quoting_flags(self): return self.__quoting_flags @property def dp_extractor(self): return self.__dp_extractor def __init__(self, source, quoting_flags, type_hints, type_hint_rules=None): self.table_name = tnt.DEFAULT self.source = source self.__quoting_flags = quoting_flags self.type_hints = type_hints self.type_hint_rules = type_hint_rules self._validator = None self._logger = None self.__dp_extractor = DataPropertyExtractor() self.__dp_extractor.quoting_flags = self.quoting_flags self.__dp_extractor.update_strict_level_map({typepy.Typecode.BOOL: 1}) def get_format_key(self): return f"{self.format_name:s}{self.__get_format_table_count():d}" def make_table_name(self): return self._make_table_name() def inc_table_count(self): with self.__table_count_lock: self.__global_table_count += 1 self.__format_table_count[self.format_name] = self.__get_format_table_count() + 1 @abc.abstractmethod def _get_default_table_name_template(self): # pragma: no cover pass def _validate(self): self._validate_table_name() self._validate_source() def _validate_table_name(self): try: if typepy.is_null_string(self.table_name): raise ValueError("table name is empty") except (TypeError, AttributeError): raise TypeError("table_name must be a string") def _validate_source(self): self._validator.validate() def __get_format_table_count(self): return self.__format_table_count.get(self.format_name, 0) def _get_filename_tablename_mapping(self): filename = "" if all([self.source_type == SourceType.FILE, typepy.is_not_null_string(self.source)]): filename = path.Path(self.source).stem return (tnt.FILENAME, filename) def _get_basic_tablename_keyvalue_mapping(self): from collections import OrderedDict return OrderedDict( [ (tnt.DEFAULT, self._get_default_table_name_template()), (tnt.FORMAT_NAME, self.format_name), (tnt.FORMAT_ID, str(self.__get_format_table_count())), (tnt.GLOBAL_ID, str(self.__global_table_count)), self._get_filename_tablename_mapping(), ] ) def _expand_table_name_format(self, table_name_kv_mapping): self._validate_table_name() table_name = self.table_name for template, value in table_name_kv_mapping.items(): table_name = table_name.replace(template, value) return self._sanitize_table_name(table_name) def _make_table_name(self): self._validate_table_name() return self._expand_table_name_format(self._get_basic_tablename_keyvalue_mapping()) @staticmethod def _sanitize_table_name(table_name): if typepy.is_null_string(table_name): raise InvalidTableNameError("table name is empty after the template replacement") return table_name.strip("_") @classmethod def clear_table_count(cls): with cls.__table_count_lock: cls.__global_table_count = 0 cls.__format_table_count = {}