Deploy site

This commit is contained in:
Gitea Actions
2025-06-10 03:00:57 +02:00
commit 70bff17031
2329 changed files with 367195 additions and 0 deletions

View File

@ -0,0 +1,133 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from dataproperty import LineBreakHandling
from .__version__ import __author__, __copyright__, __email__, __license__, __version__
from ._factory import TableWriterFactory
from ._function import dumps_tabledata
from ._logger import set_logger
from ._table_format import FormatAttr, TableFormat
from .error import (
EmptyTableDataError,
EmptyTableNameError,
EmptyValueError,
NotSupportedError,
WriterNotFoundError,
)
from .style import Align, Format
from .typehint import (
Bool,
DateTime,
Dictionary,
Infinity,
Integer,
IpAddress,
List,
Nan,
NoneType,
NullString,
RealNumber,
String,
)
from .writer import (
AbstractTableWriter,
AsciiDocTableWriter,
BoldUnicodeTableWriter,
BorderlessTableWriter,
CssTableWriter,
CsvTableWriter,
ElasticsearchWriter,
ExcelXlsTableWriter,
ExcelXlsxTableWriter,
HtmlTableWriter,
JavaScriptTableWriter,
JsonLinesTableWriter,
JsonTableWriter,
LatexMatrixWriter,
LatexTableWriter,
LtsvTableWriter,
MarkdownTableWriter,
MediaWikiTableWriter,
NullTableWriter,
NumpyTableWriter,
PandasDataFramePickleWriter,
PandasDataFrameWriter,
PythonCodeTableWriter,
RstCsvTableWriter,
RstGridTableWriter,
RstSimpleTableWriter,
SpaceAlignedTableWriter,
SqliteTableWriter,
TomlTableWriter,
TsvTableWriter,
UnicodeTableWriter,
YamlTableWriter,
)
__all__ = (
"__author__",
"__copyright__",
"__email__",
"__license__",
"__version__",
"LineBreakHandling",
"TableWriterFactory",
"dumps_tabledata",
"set_logger",
"FormatAttr",
"TableFormat",
"Align",
"Format",
"Bool",
"DateTime",
"Dictionary",
"Infinity",
"Integer",
"IpAddress",
"List",
"Nan",
"NoneType",
"NullString",
"RealNumber",
"String",
"EmptyTableDataError",
"EmptyTableNameError",
"EmptyValueError",
"NotSupportedError",
"WriterNotFoundError",
"AbstractTableWriter",
"AsciiDocTableWriter",
"BoldUnicodeTableWriter",
"BorderlessTableWriter",
"CssTableWriter",
"CsvTableWriter",
"ElasticsearchWriter",
"ExcelXlsTableWriter",
"ExcelXlsxTableWriter",
"HtmlTableWriter",
"JavaScriptTableWriter",
"JsonLinesTableWriter",
"JsonTableWriter",
"LatexMatrixWriter",
"LatexTableWriter",
"LtsvTableWriter",
"MarkdownTableWriter",
"MediaWikiTableWriter",
"NullTableWriter",
"NumpyTableWriter",
"PandasDataFramePickleWriter",
"PandasDataFrameWriter",
"PythonCodeTableWriter",
"RstCsvTableWriter",
"RstGridTableWriter",
"RstSimpleTableWriter",
"SpaceAlignedTableWriter",
"SqliteTableWriter",
"TomlTableWriter",
"TsvTableWriter",
"UnicodeTableWriter",
"YamlTableWriter",
)

View File

@ -0,0 +1,9 @@
from typing import Final
__author__: Final = "Tsuyoshi Hombashi"
__copyright__: Final = f"Copyright 2016-2025, {__author__}"
__license__: Final = "MIT License"
__version__ = "1.2.1"
__maintainer__: Final = __author__
__email__: Final = "tsuyoshi.hombashi@gmail.com"

View File

@ -0,0 +1,11 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import re
def strip_quote(text: str, value: str) -> str:
re_replace = re.compile(f"[\"']{value:s}[\"']", re.MULTILINE)
return re_replace.sub(value, text)

View File

@ -0,0 +1,274 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import os
from itertools import chain
from typing import Any, Final
import typepy
from ._logger import logger
from ._table_format import FormatAttr, TableFormat
from .error import WriterNotFoundError
from .writer import AbstractTableWriter
class TableWriterFactory:
"""
A factory class of table writer classes.
"""
@classmethod
def create_from_file_extension(cls, file_extension: str, **kwargs: Any) -> AbstractTableWriter:
"""
Create a table writer class instance from a file extension.
Supported file extensions are as follows:
================== ===================================
Extension Writer Class
================== ===================================
``".adoc"`` :py:class:`~.AsciiDocTableWriter`
``".asciidoc"`` :py:class:`~.AsciiDocTableWriter`
``".asc"`` :py:class:`~.AsciiDocTableWriter`
``".css"`` :py:class:`~.CssTableWriter`
``".csv"`` :py:class:`~.CsvTableWriter`
``".htm"`` :py:class:`~.HtmlTableWriter`
``".html"`` :py:class:`~.HtmlTableWriter`
``".js"`` :py:class:`~.JavaScriptTableWriter`
``".json"`` :py:class:`~.JsonTableWriter`
``".jsonl"`` :py:class:`~.JsonLinesTableWriter`
``".ltsv"`` :py:class:`~.LtsvTableWriter`
``".ldjson"`` :py:class:`~.JsonLinesTableWriter`
``".md"`` :py:class:`~.MarkdownTableWriter`
``".ndjson"`` :py:class:`~.JsonLinesTableWriter`
``".py"`` :py:class:`~.PythonCodeTableWriter`
``".rst"`` :py:class:`~.RstGridTableWriter`
``".tsv"`` :py:class:`~.TsvTableWriter`
``".xls"`` :py:class:`~.ExcelXlsTableWriter`
``".xlsx"`` :py:class:`~.ExcelXlsxTableWriter`
``".sqlite"`` :py:class:`~.SqliteTableWriter`
``".sqlite3"`` :py:class:`~.SqliteTableWriter`
``".tsv"`` :py:class:`~.TsvTableWriter`
``".toml"`` :py:class:`~.TomlTableWriter`
``".yml"`` :py:class:`~.YamlTableWriter`
================== ===================================
:param str file_extension:
File extension string (case insensitive).
:param kwargs:
Keyword arguments that pass to a writer class constructor.
:return:
Writer instance that coincides with the ``file_extension``.
:rtype:
:py:class:`~pytablewriter.writer._table_writer.TableWriterInterface`
:raises pytablewriter.WriterNotFoundError:
|WriterNotFoundError_desc| the file extension.
"""
ext: Final = os.path.splitext(file_extension)[1]
if typepy.is_null_string(ext):
file_extension = file_extension
else:
file_extension = ext
file_extension = file_extension.lstrip(".").lower()
for table_format in TableFormat:
if file_extension not in table_format.file_extensions:
continue
if table_format.format_attribute & FormatAttr.SECONDARY_EXT:
continue
logger.debug(f"create a {table_format.writer_class} instance")
return table_format.writer_class(**kwargs) # type: ignore
raise WriterNotFoundError(
"\n".join(
[
f"{file_extension:s} (unknown file extension).",
"",
"acceptable file extensions are: {}.".format(", ".join(cls.get_extensions())),
]
)
)
@classmethod
def create_from_format_name(cls, format_name: str, **kwargs: Any) -> AbstractTableWriter:
"""
Create a table writer class instance from a format name.
Supported file format names are as follows:
============================================= ===================================
Format name Writer Class
============================================= ===================================
``"adoc"`` :py:class:`~.AsciiDocTableWriter`
``"asciidoc"`` :py:class:`~.AsciiDocTableWriter`
``"css"`` :py:class:`~.CssTableWriter`
``"csv"`` :py:class:`~.CsvTableWriter`
``"elasticsearch"`` :py:class:`~.ElasticsearchWriter`
``"excel"`` :py:class:`~.ExcelXlsxTableWriter`
``"html"``/``"htm"`` :py:class:`~.HtmlTableWriter`
``"javascript"``/``"js"`` :py:class:`~.JavaScriptTableWriter`
``"json"`` :py:class:`~.JsonTableWriter`
``"json_lines"`` :py:class:`~.JsonLinesTableWriter`
``"latex_matrix"`` :py:class:`~.LatexMatrixWriter`
``"latex_table"`` :py:class:`~.LatexTableWriter`
``"ldjson"`` :py:class:`~.JsonLinesTableWriter`
``"ltsv"`` :py:class:`~.LtsvTableWriter`
``"markdown"``/``"md"`` :py:class:`~.MarkdownTableWriter`
``"mediawiki"`` :py:class:`~.MediaWikiTableWriter`
``"null"`` :py:class:`~.NullTableWriter`
``"pandas"`` :py:class:`~.PandasDataFrameWriter`
``"py"``/``"python"`` :py:class:`~.PythonCodeTableWriter`
``"rst"``/``"rst_grid"``/``"rst_grid_table"`` :py:class:`~.RstGridTableWriter`
``"rst_simple"``/``"rst_simple_table"`` :py:class:`~.RstSimpleTableWriter`
``"rst_csv"``/``"rst_csv_table"`` :py:class:`~.RstCsvTableWriter`
``"sqlite"`` :py:class:`~.SqliteTableWriter`
``"ssv"`` :py:class:`~.SpaceAlignedTableWriter`
``"tsv"`` :py:class:`~.TsvTableWriter`
``"toml"`` :py:class:`~.TomlTableWriter`
``"unicode"`` :py:class:`~.UnicodeTableWriter`
``"yaml"`` :py:class:`~.YamlTableWriter`
============================================= ===================================
:param str format_name:
Format name string (case insensitive).
:param kwargs:
Keyword arguments that pass to a writer class constructor.
:return:
Writer instance that coincides with the ``format_name``:
:rtype:
:py:class:`~pytablewriter.writer._table_writer.TableWriterInterface`
:raises pytablewriter.WriterNotFoundError:
|WriterNotFoundError_desc| for the format.
"""
format_name = format_name.casefold()
for table_format in TableFormat:
if format_name in table_format.names and not (
table_format.format_attribute & FormatAttr.SECONDARY_NAME
):
writer = table_format.writer_class(**kwargs) # type: ignore
logger.debug(f"create a {writer.FORMAT_NAME} instance")
return writer
raise WriterNotFoundError(
"\n".join(
[
f"{format_name} (unknown format name).",
"acceptable format names are: {}.".format(", ".join(cls.get_format_names())),
]
)
)
@classmethod
def get_format_names(cls) -> list[str]:
"""
:return: Available format names.
:rtype: list
:Example:
.. code:: python
>>> import pytablewriter as ptw
>>> for name in ptw.TableWriterFactory.get_format_names():
... print(name)
...
adoc
asciidoc
bold_unicode
borderless
css
csv
elasticsearch
excel
htm
html
javascript
js
json
json_lines
jsonl
latex_matrix
latex_table
ldjson
ltsv
markdown
md
mediawiki
ndjson
null
numpy
pandas
pandas_pickle
py
python
rst
rst_csv
rst_csv_table
rst_grid
rst_grid_table
rst_simple
rst_simple_table
space_aligned
sqlite
ssv
toml
tsv
unicode
yaml
"""
return sorted(list(set(chain(*(table_format.names for table_format in TableFormat)))))
@classmethod
def get_extensions(cls) -> list[str]:
"""
:return: Available file extensions.
:rtype: list
:Example:
.. code:: python
>>> import pytablewriter as ptw
>>> for name in ptw.TableWriterFactory.get_extensions():
... print(name)
...
adoc
asc
asciidoc
css
csv
htm
html
js
json
jsonl
ldjson
ltsv
md
ndjson
py
rst
sqlite
sqlite3
tex
toml
tsv
xls
xlsx
yml
"""
file_extension_set = set()
for table_format in TableFormat:
for file_extension in table_format.file_extensions:
file_extension_set.add(file_extension)
return sorted(list(file_extension_set))

View File

@ -0,0 +1,84 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from datetime import datetime
from enum import Enum
from typing import Any, Optional
import dataproperty
from pathvalidate import replace_symbol
from tabledata._core import TableData
def quote_datetime_formatter(value: datetime) -> str:
return f'"{value.strftime(dataproperty.DefaultValue.DATETIME_FORMAT):s}"'
def dateutil_datetime_formatter(value: datetime) -> str:
return 'dateutil.parser.parse("{:s}")'.format(
value.strftime(dataproperty.DefaultValue.DATETIME_FORMAT)
)
def dumps_tabledata(value: TableData, format_name: str = "rst_grid_table", **kwargs: Any) -> str:
"""
:param tabledata.TableData value: Tabular data to dump.
:param str format_name:
Dumped format name of tabular data.
Available formats are described in
:py:meth:`~pytablewriter.TableWriterFactory.create_from_format_name`
:Example:
.. code:: python
>>> dumps_tabledata(value)
.. table:: sample_data
====== ====== ======
attr_a attr_b attr_c
====== ====== ======
1 4.0 a
2 2.1 bb
3 120.9 ccc
====== ====== ======
"""
from ._factory import TableWriterFactory
if not value:
raise TypeError("value must be a tabledata.TableData instance")
writer = TableWriterFactory.create_from_format_name(format_name)
for attr_name, attr_value in kwargs.items():
setattr(writer, attr_name, attr_value)
writer.from_tabledata(value)
return writer.dumps()
def normalize_enum(
value: Any, enum_class: type[Enum], validate: bool = True, default: Optional[Enum] = None
) -> Any:
if value is None:
return default
if isinstance(value, enum_class):
return value
try:
return enum_class[replace_symbol(value.strip(), "_").upper()]
except AttributeError:
if validate:
raise TypeError(f"value must be a {enum_class} or a str: actual={type(value)}")
except KeyError:
if validate:
raise ValueError(
"invalid valid found: expected={}, actual={}".format(
"/".join(item.name for item in enum_class), value
)
)
return value

View File

@ -0,0 +1,4 @@
from ._logger import WriterLogger, logger, set_logger
__all__ = ("WriterLogger", "logger", "set_logger")

View File

@ -0,0 +1,117 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from typing import TYPE_CHECKING, Final
import dataproperty
from mbstrdecoder import MultiByteStrDecoder
from ._null_logger import NullLogger # type: ignore
if TYPE_CHECKING:
from ..writer import AbstractTableWriter
MODULE_NAME: Final = "pytablewriter"
try:
from loguru import logger
logger.disable(MODULE_NAME)
except ImportError:
logger = NullLogger()
def set_logger(is_enable: bool, propagation_depth: int = 1) -> None:
if is_enable:
logger.enable(MODULE_NAME)
else:
logger.disable(MODULE_NAME)
if propagation_depth <= 0:
return
dataproperty.set_logger(is_enable, propagation_depth - 1)
try:
import simplesqlite
simplesqlite.set_logger(is_enable, propagation_depth - 1)
except ImportError:
pass
try:
import pytablereader
pytablereader.set_logger(is_enable, propagation_depth - 1)
except ImportError:
pass
class WriterLogger:
@property
def logger(self): # type: ignore
return self.__logger
def __init__(self, writer: "AbstractTableWriter") -> None:
self.__writer = writer
self.__logger = logger
self.logger.debug(f"created WriterLogger: format={writer.format_name}")
def __enter__(self) -> "WriterLogger":
self.logging_start_write()
return self
def __exit__(self, *exc): # type: ignore
self.logging_complete_write()
return False
def logging_start_write(self) -> None:
log_entry_list = [
self.__get_format_name_message(),
self.__get_table_name_message(),
f"headers={self.__writer.headers}",
]
try:
log_entry_list.append(f"rows={len(self.__writer.value_matrix)}")
except (TypeError, AttributeError):
log_entry_list.append("rows=NaN")
log_entry_list.append(self.__get_typehint_message())
log_entry_list.extend(self.__get_extra_log_entry_list())
self.logger.debug("start write table: {}".format(", ".join(log_entry_list)))
def logging_complete_write(self) -> None:
log_entry_list = [self.__get_format_name_message(), self.__get_table_name_message()]
log_entry_list.extend(self.__get_extra_log_entry_list())
self.logger.debug("complete write table: {}".format(", ".join(log_entry_list)))
def __get_format_name_message(self) -> str:
return f"format={self.__writer.format_name:s}"
def __get_table_name_message(self) -> str:
if self.__writer.table_name:
table_name = MultiByteStrDecoder(self.__writer.table_name).unicode_str
else:
table_name = ""
return f"table-name='{table_name}'"
def __get_extra_log_entry_list(self) -> list[str]:
if self.__writer._iter_count is None:
return []
return [f"iteration={self.__writer._iter_count}/{self.__writer.iteration_length}"]
def __get_typehint_message(self) -> str:
try:
return "type-hints={}".format(
[type_hint(None, 0).typename for type_hint in self.__writer.type_hints if type_hint]
)
except (TypeError, AttributeError):
return "type-hints=[]"

View File

@ -0,0 +1,44 @@
# type: ignore
class NullLogger:
level_name = None
def remove(self, handler_id=None): # pragma: no cover
pass
def add(self, sink, **kwargs): # pragma: no cover
pass
def disable(self, name): # pragma: no cover
pass
def enable(self, name): # pragma: no cover
pass
def critical(self, __message, *args, **kwargs): # pragma: no cover
pass
def debug(self, __message, *args, **kwargs): # pragma: no cover
pass
def error(self, __message, *args, **kwargs): # pragma: no cover
pass
def exception(self, __message, *args, **kwargs): # pragma: no cover
pass
def info(self, __message, *args, **kwargs): # pragma: no cover
pass
def log(self, __level, __message, *args, **kwargs): # pragma: no cover
pass
def success(self, __message, *args, **kwargs): # pragma: no cover
pass
def trace(self, __message, *args, **kwargs): # pragma: no cover
pass
def warning(self, __message, *args, **kwargs): # pragma: no cover
pass

View File

@ -0,0 +1,354 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import enum
from collections.abc import Sequence
from typing import Optional
from .writer import (
AbstractTableWriter,
AsciiDocTableWriter,
BoldUnicodeTableWriter,
BorderlessTableWriter,
CssTableWriter,
CsvTableWriter,
ElasticsearchWriter,
ExcelXlsTableWriter,
ExcelXlsxTableWriter,
HtmlTableWriter,
JavaScriptTableWriter,
JsonLinesTableWriter,
JsonTableWriter,
LatexMatrixWriter,
LatexTableWriter,
LtsvTableWriter,
MarkdownTableWriter,
MediaWikiTableWriter,
NullTableWriter,
NumpyTableWriter,
PandasDataFramePickleWriter,
PandasDataFrameWriter,
PythonCodeTableWriter,
RstCsvTableWriter,
RstGridTableWriter,
RstSimpleTableWriter,
SpaceAlignedTableWriter,
SqliteTableWriter,
TomlTableWriter,
TsvTableWriter,
UnicodeTableWriter,
YamlTableWriter,
)
class FormatAttr:
"""
Bitmaps to represent table attributes.
"""
NONE = 1 << 1
#: Can create a file with the format.
FILE = 1 << 2
#: Table format that can represent as a text.
TEXT = 1 << 3
#: Table format that can represent as a binary file.
BIN = 1 << 4
#: Can create a source code (variables definition)
#: one of the programming language.
SOURCECODE = 1 << 5
#: Can call API for external service.
API = 1 << 6
SECONDARY_EXT = 1 << 10
SECONDARY_NAME = 1 << 11
@enum.unique
class TableFormat(enum.Enum):
"""
Enum to represent table format attributes.
"""
ASCIIDOC = (
[AsciiDocTableWriter.FORMAT_NAME, "adoc"],
AsciiDocTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["adoc", "asciidoc", "asc"],
)
CSV = ([CsvTableWriter.FORMAT_NAME], CsvTableWriter, FormatAttr.FILE | FormatAttr.TEXT, ["csv"])
CSS = (
[CssTableWriter.FORMAT_NAME],
CssTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["css"],
)
ELASTICSEARCH = (
[ElasticsearchWriter.FORMAT_NAME], # type: ignore
ElasticsearchWriter,
FormatAttr.API,
[],
)
EXCEL_XLSX = (
[ExcelXlsxTableWriter.FORMAT_NAME],
ExcelXlsxTableWriter,
FormatAttr.FILE | FormatAttr.BIN,
["xlsx"],
)
EXCEL_XLS = (
[ExcelXlsTableWriter.FORMAT_NAME],
ExcelXlsTableWriter,
FormatAttr.FILE | FormatAttr.BIN | FormatAttr.SECONDARY_NAME,
["xls"],
)
HTML = (
[HtmlTableWriter.FORMAT_NAME, "htm"],
HtmlTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["html", "htm"],
)
JAVASCRIPT = (
[JavaScriptTableWriter.FORMAT_NAME, "js"],
JavaScriptTableWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SOURCECODE,
["js"],
)
JSON = (
[JsonTableWriter.FORMAT_NAME],
JsonTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["json"],
)
JSON_LINES = (
[JsonLinesTableWriter.FORMAT_NAME, "jsonl", "ldjson", "ndjson"],
JsonLinesTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["jsonl", "ldjson", "ndjson"],
)
LATEX_MATRIX = (
[LatexMatrixWriter.FORMAT_NAME],
LatexMatrixWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["tex"],
)
LATEX_TABLE = (
[LatexTableWriter.FORMAT_NAME],
LatexTableWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SECONDARY_EXT,
["tex"],
)
LTSV = (
[LtsvTableWriter.FORMAT_NAME],
LtsvTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["ltsv"],
)
MARKDOWN = (
[MarkdownTableWriter.FORMAT_NAME, "md"],
MarkdownTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["md"],
)
MEDIAWIKI = (
[MediaWikiTableWriter.FORMAT_NAME], # type: ignore
MediaWikiTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
[],
)
NULL = (
[NullTableWriter.FORMAT_NAME], # type: ignore
NullTableWriter,
FormatAttr.NONE,
[],
)
NUMPY = (
[NumpyTableWriter.FORMAT_NAME],
NumpyTableWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SOURCECODE | FormatAttr.SECONDARY_EXT,
["py"],
)
PANDAS = (
[PandasDataFrameWriter.FORMAT_NAME],
PandasDataFrameWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SOURCECODE | FormatAttr.SECONDARY_EXT,
["py"],
)
PANDAS_PICKLE = (
[PandasDataFramePickleWriter.FORMAT_NAME], # type: ignore
PandasDataFramePickleWriter,
FormatAttr.FILE | FormatAttr.BIN,
[],
)
PYTHON = (
[PythonCodeTableWriter.FORMAT_NAME, "py"],
PythonCodeTableWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SOURCECODE,
["py"],
)
RST_CSV_TABLE = (
[RstCsvTableWriter.FORMAT_NAME, "rst_csv"],
RstCsvTableWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SECONDARY_EXT,
["rst"],
)
RST_GRID_TABLE = (
[RstGridTableWriter.FORMAT_NAME, "rst_grid", "rst"],
RstGridTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["rst"],
)
RST_SIMPLE_TABLE = (
[RstSimpleTableWriter.FORMAT_NAME, "rst_simple"],
RstSimpleTableWriter,
FormatAttr.FILE | FormatAttr.TEXT | FormatAttr.SECONDARY_EXT,
["rst"],
)
SPACE_ALIGNED = (
[SpaceAlignedTableWriter.FORMAT_NAME, "ssv"], # type: ignore
SpaceAlignedTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
[],
)
SQLITE = (
[SqliteTableWriter.FORMAT_NAME],
SqliteTableWriter,
FormatAttr.FILE | FormatAttr.BIN,
["sqlite", "sqlite3"],
)
TOML = (
[TomlTableWriter.FORMAT_NAME],
TomlTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["toml"],
)
TSV = ([TsvTableWriter.FORMAT_NAME], TsvTableWriter, FormatAttr.FILE | FormatAttr.TEXT, ["tsv"])
UNICODE = (
[UnicodeTableWriter.FORMAT_NAME], # type: ignore
UnicodeTableWriter,
FormatAttr.TEXT,
[],
)
YAML = (
[YamlTableWriter.FORMAT_NAME],
YamlTableWriter,
FormatAttr.FILE | FormatAttr.TEXT,
["yml"],
)
BOLD_UNICODE = (
[BoldUnicodeTableWriter.FORMAT_NAME], # type: ignore
BoldUnicodeTableWriter,
FormatAttr.TEXT,
[],
)
BORDERLESS = (
[BorderlessTableWriter.FORMAT_NAME], # type: ignore
BorderlessTableWriter,
FormatAttr.TEXT,
[],
)
@property
def names(self) -> list[str]:
"""
List[str]: Names associated with the table format.
"""
return self.__names
@property
def writer_class(self) -> type[AbstractTableWriter]:
"""
Type[AbstractTableWriter]: Table writer class object associated with the table format.
"""
return self.__writer_class
@property
def format_attribute(self) -> int:
"""
FormatAttr: Table attributes bitmap.
"""
return self.__format_attribute
@property
def file_extensions(self) -> list[str]:
"""
List[str]: File extensions associated with the table format.
"""
return self.__file_extensions
def __init__(
self,
names: Sequence[str],
writer_class: type[AbstractTableWriter],
format_attribute: int,
file_extensions: Sequence[str],
) -> None:
self.__names = list(names)
self.__writer_class = writer_class
self.__format_attribute = format_attribute
self.__file_extensions = list(file_extensions)
@classmethod
def find_all_attr(cls, format_attribute: int) -> list["TableFormat"]:
"""Searching table formats that have specific attributes.
Args:
format_attribute (FormatAttr):
Table format attributes to look for.
Returns:
List[TableFormat]: Table formats that matched the attribute.
"""
return [
table_format
for table_format in TableFormat
if table_format.format_attribute & format_attribute
]
@classmethod
def from_name(cls, format_name: str) -> Optional["TableFormat"]:
"""Get a table format from a format name.
Args:
format_name (str): Table format specifier.
Returns:
Optional[TableFormat]: A table format enum value corresponding to the ``format_name``.
"""
format_name = format_name.casefold().strip()
for table_format in TableFormat:
if format_name in table_format.names:
return table_format
return None
@classmethod
def from_file_extension(cls, file_extension: str) -> Optional["TableFormat"]:
"""Get a table format from a file extension.
Args:
file_extension (str): File extension.
Returns:
Optional[TableFormat]:
A table format enum value corresponding to the ``file_extension``.
"""
ext = file_extension.lower().strip().lstrip(".")
for table_format in TableFormat:
if ext in table_format.file_extensions:
return table_format
return None

View File

@ -0,0 +1,34 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
class NotSupportedError(Exception):
pass
class EmptyTableNameError(Exception):
"""
Exception raised when a table writer class of the |table_name| attribute
is null and the class is not accepted null |table_name|.
"""
class EmptyValueError(Exception):
"""
Exception raised when a table writer class of the |value_matrix| attribute
is null, and the class is not accepted null |value_matrix|.
"""
class EmptyTableDataError(Exception):
"""
Exception raised when a table writer class of the |headers| and
|value_matrix| attributes are null.
"""
class WriterNotFoundError(Exception):
"""
Exception raised when appropriate loader writer found.
"""

View File

@ -0,0 +1,21 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from ._elasticsearch import ElasticsearchIndexNameSanitizer
from ._excel import sanitize_excel_sheet_name, validate_excel_sheet_name
from ._javascript import JavaScriptVarNameSanitizer, sanitize_js_var_name, validate_js_var_name
from ._python import PythonVarNameSanitizer, sanitize_python_var_name, validate_python_var_name
__all__ = (
"ElasticsearchIndexNameSanitizer",
"JavaScriptVarNameSanitizer",
"PythonVarNameSanitizer",
"sanitize_excel_sheet_name",
"sanitize_js_var_name",
"sanitize_python_var_name",
"validate_excel_sheet_name",
"validate_js_var_name",
"validate_python_var_name",
)

View File

@ -0,0 +1,94 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import abc
import re
from re import Pattern
from typing import Final
from pathvalidate.error import ErrorReason, ValidationError
from typepy import is_null_string
from ._interface import NameSanitizer
def _preprocess(name: str) -> str:
return name.strip()
class VarNameSanitizer(NameSanitizer):
@property
@abc.abstractmethod
def _invalid_var_name_head_re(self) -> Pattern[str]: # pragma: no cover
pass
@property
@abc.abstractmethod
def _invalid_var_name_re(self) -> Pattern[str]: # pragma: no cover
pass
def validate(self) -> None:
self._validate(self._value)
def sanitize(self, replacement_text: str = "") -> str:
var_name = self._invalid_var_name_re.sub(replacement_text, self._str)
# delete invalid char(s) in the beginning of the variable name
is_require_remove_head: Final = any(
[
is_null_string(replacement_text),
self._invalid_var_name_head_re.search(replacement_text) is not None,
]
)
if is_require_remove_head:
var_name = self._invalid_var_name_head_re.sub("", var_name)
else:
match = self._invalid_var_name_head_re.search(var_name)
if match is not None:
var_name = match.end() * replacement_text + self._invalid_var_name_head_re.sub(
"", var_name
)
if not var_name:
return ""
try:
self._validate(var_name)
except ValidationError as e:
if e.reason == ErrorReason.RESERVED_NAME and e.reusable_name is False:
var_name += "_"
return var_name
def _validate(self, value: str) -> None:
self._validate_null_string(value)
unicode_var_name: Final = _preprocess(value)
if self._is_reserved_keyword(unicode_var_name):
raise ValidationError(
description=f"{unicode_var_name:s} is a reserved keyword by python",
reason=ErrorReason.RESERVED_NAME,
reusable_name=False,
reserved_name=unicode_var_name,
)
match = self._invalid_var_name_re.search(unicode_var_name)
if match is not None:
raise ValidationError(
description="invalid char found in the variable name: '{}'".format(
re.escape(match.group())
),
reason=ErrorReason.INVALID_CHARACTER,
)
match = self._invalid_var_name_head_re.search(unicode_var_name)
if match is not None:
raise ValidationError(
description="the first character of the variable name is invalid: '{}'".format(
re.escape(match.group())
),
reason=ErrorReason.INVALID_CHARACTER,
)

View File

@ -0,0 +1,28 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import re
from re import Pattern
from typing import Final
from ._base import VarNameSanitizer
class ElasticsearchIndexNameSanitizer(VarNameSanitizer):
__RE_INVALID_INDEX_NAME: Final[Pattern[str]] = re.compile(
"[" + re.escape('\\/*?"<>|,"') + r"\s]+"
)
__RE_INVALID_INDEX_NAME_HEAD: Final[Pattern[str]] = re.compile("^[_]+")
@property
def reserved_keywords(self) -> list[str]:
return []
@property
def _invalid_var_name_head_re(self) -> Pattern[str]:
return self.__RE_INVALID_INDEX_NAME_HEAD
@property
def _invalid_var_name_re(self) -> Pattern[str]:
return self.__RE_INVALID_INDEX_NAME

View File

@ -0,0 +1,78 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import re
from typing import Final
from pathvalidate import validate_pathtype
from pathvalidate.error import ErrorReason, ValidationError
from ._base import _preprocess
__MAX_SHEET_NAME_LEN: Final = 31
__INVALID_EXCEL_CHARS: Final = "[]:*?/\\"
__RE_INVALID_EXCEL_SHEET_NAME: Final = re.compile(
f"[{re.escape(__INVALID_EXCEL_CHARS):s}]", re.UNICODE
)
def validate_excel_sheet_name(sheet_name: str) -> None:
"""
:param str sheet_name: Excel sheet name to validate.
:raises pathvalidate.ValidationError (ErrorReason.INVALID_CHARACTER):
If the ``sheet_name`` includes invalid char(s):
|invalid_excel_sheet_chars|.
:raises pathvalidate.ValidationError (ErrorReason.INVALID_LENGTH):
If the ``sheet_name`` is longer than 31 characters.
"""
validate_pathtype(sheet_name)
if len(sheet_name) > __MAX_SHEET_NAME_LEN:
raise ValidationError(
description="sheet name is too long: expected<={:d}, actual={:d}".format(
__MAX_SHEET_NAME_LEN, len(sheet_name)
),
reason=ErrorReason.INVALID_LENGTH,
)
unicode_sheet_name = _preprocess(sheet_name)
match = __RE_INVALID_EXCEL_SHEET_NAME.search(unicode_sheet_name)
if match is not None:
raise ValidationError(
description="invalid char found in the sheet name: '{:s}'".format(
re.escape(match.group())
),
reason=ErrorReason.INVALID_CHARACTER,
)
def sanitize_excel_sheet_name(sheet_name: str, replacement_text: str = "") -> str:
"""
Replace invalid characters for an Excel sheet name within
the ``sheet_name`` with the ``replacement_text``.
Invalid characters are as follows:
|invalid_excel_sheet_chars|.
The ``sheet_name`` truncate to 31 characters
(max sheet name length of Excel) from the head, if the length
of the name is exceed 31 characters.
:param str sheet_name: Excel sheet name to sanitize.
:param str replacement_text: Replacement text.
:return: A replacement string.
:rtype: str
:raises ValueError: If the ``sheet_name`` is an invalid sheet name.
"""
try:
unicode_sheet_name = _preprocess(sheet_name)
except AttributeError as e:
raise ValueError(e)
modify_sheet_name = __RE_INVALID_EXCEL_SHEET_NAME.sub(replacement_text, unicode_sheet_name)
return modify_sheet_name[:__MAX_SHEET_NAME_LEN]

View File

@ -0,0 +1,38 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import abc
from pathvalidate import validate_pathtype
class NameSanitizer(metaclass=abc.ABCMeta):
@property
@abc.abstractmethod
def reserved_keywords(self) -> list[str]: # pragma: no cover
pass
@abc.abstractmethod
def validate(self) -> None: # pragma: no cover
pass
@abc.abstractmethod
def sanitize(self, replacement_text: str = "") -> str: # pragma: no cover
pass
@property
def _str(self) -> str:
return str(self._value)
def __init__(self, value: str) -> None:
self._validate_null_string(value)
self._value = value.strip()
def _is_reserved_keyword(self, value: str) -> bool:
return value in self.reserved_keywords
@staticmethod
def _validate_null_string(text: str) -> None:
validate_pathtype(text)

View File

@ -0,0 +1,144 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import re
from re import Pattern
from typing import Final
from ._base import VarNameSanitizer
class JavaScriptVarNameSanitizer(VarNameSanitizer):
__JS_RESERVED_KEYWORDS_ES6: Final = [
"break",
"case",
"catch",
"class",
"const",
"continue",
"debugger",
"default",
"delete",
"do",
"else",
"export",
"extends",
"finally",
"for",
"function",
"if",
"import",
"in",
"instanceof",
"new",
"return",
"super",
"switch",
"this",
"throw",
"try",
"typeof",
"var",
"void",
"while",
"with",
"yield",
]
__JS_RESERVED_KEYWORDS_FUTURE: Final = [
"enum",
"implements",
"interface",
"let",
"package",
"private",
"protected",
"public",
"static",
"await",
"abstract",
"boolean",
"byte",
"char",
"double",
"final",
"float",
"goto",
"int",
"long",
"native",
"short",
"synchronized",
"throws",
"transient",
"volatile",
]
__JS_BUILTIN_CONSTANTS: Final = ["null", "true", "false"]
__RE_INVALID_VAR_NAME: Final = re.compile("[^a-zA-Z0-9_$]")
__RE_INVALID_VAR_NAME_HEAD: Final = re.compile("^[^a-zA-Z$]+")
@property
def reserved_keywords(self) -> list[str]:
return (
self.__JS_RESERVED_KEYWORDS_ES6
+ self.__JS_RESERVED_KEYWORDS_FUTURE
+ self.__JS_BUILTIN_CONSTANTS
)
@property
def _invalid_var_name_head_re(self) -> Pattern[str]:
return self.__RE_INVALID_VAR_NAME_HEAD
@property
def _invalid_var_name_re(self) -> Pattern[str]:
return self.__RE_INVALID_VAR_NAME
def validate_js_var_name(var_name: str) -> None:
"""
:param str var_name: Name to validate.
:raises pathvalidate.ValidationError (ErrorReason.INVALID_CHARACTER):
If the ``var_name`` is invalid as a JavaScript identifier.
:raises pathvalidate.ValidationError (ErrorReason.RESERVED_NAME):
If the ``var_name`` is equals to
`JavaScript reserved keywords
<https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Lexical_grammar#Keywords>`__.
.. note::
Currently, not supported unicode variable names.
"""
JavaScriptVarNameSanitizer(var_name).validate()
def sanitize_js_var_name(var_name: str, replacement_text: str = "") -> str:
"""
Make a valid JavaScript variable name from ``var_name``.
To make a valid name:
- Replace invalid characters for a JavaScript variable name within
the ``var_name`` with the ``replacement_text``
- Delete invalid chars for the beginning of the variable name
- Append underscore (``"_"``) at the tail of the name if sanitized name
is one of the JavaScript reserved names
:JavaScriptstr filename: Name to sanitize.
:param str replacement_text: Replacement text.
:return: A replacement string.
:rtype: str
:raises ValueError: If ``var_name`` or ``replacement_text`` is invalid.
:Example:
:ref:`example-sanitize-var-name`
.. note::
Currently, not supported Unicode variable names.
.. seealso::
:py:func:`.validate_js_var_name`
"""
return JavaScriptVarNameSanitizer(var_name).sanitize(replacement_text)

View File

@ -0,0 +1,118 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import re
from re import Pattern
from typing import Final
from ._base import VarNameSanitizer
class PythonVarNameSanitizer(VarNameSanitizer):
__PYTHON_RESERVED_KEYWORDS: Final = [
"and",
"del",
"from",
"not",
"while",
"as",
"elif",
"global",
"or",
"with",
"assert",
"else",
"if",
"pass",
"yield",
"break",
"except",
"import",
"print",
"class",
"exec",
"in",
"raise",
"continue",
"finally",
"is",
"return",
"def",
"for",
"lambda",
"try",
]
__PYTHON_BUILTIN_CONSTANTS: Final = [
"False",
"True",
"None",
"NotImplemented",
"Ellipsis",
"__debug__",
]
__RE_INVALID_VAR_NAME: Final = re.compile("[^a-zA-Z0-9_]")
__RE_INVALID_VAR_NAME_HEAD: Final = re.compile("^[^a-zA-Z]+")
@property
def reserved_keywords(self) -> list[str]:
return self.__PYTHON_RESERVED_KEYWORDS + self.__PYTHON_BUILTIN_CONSTANTS
@property
def _invalid_var_name_head_re(self) -> Pattern[str]:
return self.__RE_INVALID_VAR_NAME_HEAD
@property
def _invalid_var_name_re(self) -> Pattern[str]:
return self.__RE_INVALID_VAR_NAME
def validate_python_var_name(var_name: str) -> None:
"""
:param str var_name: Name to validate.
:raises pathvalidate.ValidationError (ErrorReason.INVALID_CHARACTER):
If the ``var_name`` is invalid as
`Python identifier
<https://docs.python.org/3/reference/lexical_analysis.html#identifiers>`__.
:raises pathvalidate.ValidationError (ErrorReason.RESERVED_NAME):
If the ``var_name`` is equals to
`Python reserved keywords
<https://docs.python.org/3/reference/lexical_analysis.html#keywords>`__
or
`Python built-in constants
<https://docs.python.org/3/library/constants.html>`__.
:Example:
:ref:`example-validate-var-name`
"""
PythonVarNameSanitizer(var_name).validate()
def sanitize_python_var_name(var_name: str, replacement_text: str = "") -> str:
"""
Make a valid Python variable name from ``var_name``.
To make a valid name:
- Replace invalid characters for a Python variable name within
the ``var_name`` with the ``replacement_text``
- Delete invalid chars for the beginning of the variable name
- Append underscore (``"_"``) at the tail of the name if sanitized name
is one of the Python reserved names
:param str filename: Name to sanitize.
:param str replacement_text: Replacement text.
:return: A replacement string.
:rtype: str
:raises ValueError: If ``var_name`` or ``replacement_text`` is invalid.
:Example:
:ref:`example-sanitize-var-name`
.. seealso::
:py:func:`.validate_python_var_name`
"""
return PythonVarNameSanitizer(var_name).sanitize(replacement_text)

View File

@ -0,0 +1,53 @@
from dataproperty import Align, Format
from ._cell import Cell
from ._font import FontSize, FontStyle, FontWeight
from ._style import DecorationLine, Style, ThousandSeparator, VerticalAlign
from ._styler import (
GFMarkdownStyler,
HtmlStyler,
LatexStyler,
MarkdownStyler,
NullStyler,
ReStructuredTextStyler,
TextStyler,
get_align_char,
)
from ._styler_interface import StylerInterface
from ._theme import (
CheckStyleFilterKeywordArgsFunc,
ColSeparatorStyleFilterFunc,
StyleFilterFunc,
Theme,
fetch_theme,
list_themes,
)
__all__ = (
"Align",
"Format",
"Cell",
"FontSize",
"FontStyle",
"FontWeight",
"Style",
"ThousandSeparator",
"VerticalAlign",
"DecorationLine",
"GFMarkdownStyler",
"HtmlStyler",
"LatexStyler",
"MarkdownStyler",
"NullStyler",
"ReStructuredTextStyler",
"StylerInterface",
"TextStyler",
"CheckStyleFilterKeywordArgsFunc",
"ColSeparatorStyleFilterFunc",
"StyleFilterFunc",
"Theme",
"get_align_char",
"fetch_theme",
"list_themes",
)

View File

@ -0,0 +1,30 @@
from dataclasses import dataclass
from typing import Any
from ._style import Style
@dataclass(frozen=True)
class Cell:
"""
A data class representing a cell in a table.
"""
row: int
"""row index. ``-1`` means that the table header row."""
col: int
"""column index."""
value: Any
"""data for the cell."""
default_style: Style
"""default |Style| for the cell."""
def is_header_row(self) -> bool:
"""
Return |True| if the cell is a header.
"""
return self.row < 0

View File

@ -0,0 +1,23 @@
from enum import Enum, unique
@unique
class FontSize(Enum):
NONE = "none"
TINY = "tiny"
SMALL = "small"
MEDIUM = "medium"
LARGE = "large"
@unique
class FontStyle(Enum):
NORMAL = "normal"
ITALIC = "italic"
TYPEWRITER = "typewriter"
@unique
class FontWeight(Enum):
NORMAL = "normal"
BOLD = "bold"

View File

@ -0,0 +1,373 @@
import warnings
from enum import Enum, unique
from typing import Any, Final, Optional, Union
from dataproperty import Align
from tcolorpy import Color
from .._function import normalize_enum
from ._font import FontSize, FontStyle, FontWeight
@unique
class DecorationLine(Enum):
NONE = "none"
LINE_THROUGH = "line_through"
STRIKE = "strike"
UNDERLINE = "underline"
@unique
class ThousandSeparator(Enum):
NONE = "none" #: no thousands separator
COMMA = "comma" #: ``','`` as thousands separator
SPACE = "space" #: ``' '`` as thousands separator
UNDERSCORE = "underscore" #: ``'_'`` as thousands separator
@unique
class VerticalAlign(Enum):
BASELINE = (1 << 0, "baseline")
TOP = (1 << 1, "top")
MIDDLE = (1 << 2, "middle")
BOTTOM = (1 << 3, "bottom")
@property
def align_code(self) -> int:
return self.__align_code
@property
def align_str(self) -> str:
return self.__align_string
def __init__(self, code: int, string: str) -> None:
self.__align_code = code
self.__align_string = string
_s_to_ts: Final[dict[str, ThousandSeparator]] = {
"": ThousandSeparator.NONE,
",": ThousandSeparator.COMMA,
" ": ThousandSeparator.SPACE,
"_": ThousandSeparator.UNDERSCORE,
}
def _normalize_thousand_separator(value: Union[str, ThousandSeparator]) -> ThousandSeparator:
if isinstance(value, ThousandSeparator):
return value
thousand_separator = normalize_enum(
value,
ThousandSeparator,
default=ThousandSeparator.NONE,
validate=False,
)
if isinstance(thousand_separator, ThousandSeparator):
return thousand_separator
norm_value = _s_to_ts.get(value)
if norm_value is None:
raise ValueError(f"unknown thousand separator: {value}")
return norm_value
class Style:
"""Style specifier class for table elements.
Args:
color (Union[|str|, tcolorpy.Color, |None|]):
Text color for cells.
When using str, specify a color code (``"#XXXXXX"``) or a color name.
.. note::
In the current version, only applicable for part of text format writer classes.
fg_color (Union[|str|, tcolorpy.Color, |None|]):
Alias to :py:attr:`~.color`.
bg_color (Union[|str|, tcolorpy.Color, |None|]):
Background color for cells.
When using str, specify a color code (``"#XXXXXX"``) or a color name.
.. note::
In the current version, only applicable for part of text format writer classes.
align (|str| / :py:class:`~.style.Align`):
Horizontal text alignment for cells.
This can be only applied for text format writer classes.
Possible string values are:
- ``"auto"`` (default)
- Detect data type for each column and set alignment that appropriate
for the type automatically
- ``"left"``
- ``"right"``
- ``"center"``
vertical_align (|str| / :py:class:`~.style.VerticalAlign`):
Vertical text alignment for cells.
This can be only applied for HtmlTableWriter class.
Possible string values are:
- ``"baseline"`` (default)
- ``"top"``
- ``"middle"``
- ``"bottom"``
font_size (|str| / :py:class:`~.style.FontSize`):
Font size specification for cells in a column.
This can be only applied for HTML/Latex writer classes.
Possible string values are:
- ``"tiny"``
- ``"small"``
- ``"medium"``
- ``"large"``
- ``"none"`` (default: no font size specification)
font_weight (|str| / :py:class:`~.style.FontWeight`):
Font weight specification for cells in a column.
This can be only applied for HTML/Latex/Markdown writer classes.
Possible string values are:
- ``"normal"`` (default)
- ``"bold"``
font_style (|str| / :py:class:`~.style.FontStyle`):
Font style specification for cells in a column.
This can be applied only for HTML/Latex/Markdown writer classes.
Possible string values are:
- ``"normal"`` (default)
- ``"italic"``
- ``"typewriter"`` (only for Latex writer)
decoration_line (|str| / :py:class:`~.style.DecorationLine`)
Experiental.
Possible string values are:
- ``"line-through"``
- ``"strike"`` (alias for ``"line-through"``)
- ``"underline"``
- ``"none"`` (default)
thousand_separator (|str| / :py:class:`~.style.ThousandSeparator`):
Thousand separator specification for numbers in a column.
This can be only applied for text format writer classes.
Possible string values are:
- ``","``/``"comma"``
- ``" "``/``"space"``
- ``"_"``/``"underscore"``
- ``""``/``"none"`` (default)
Example:
:ref:`example-style`
"""
@property
def align(self) -> Align:
return self.__align
@align.setter
def align(self, value: Align) -> None:
self.__align = value
@property
def vertical_align(self) -> VerticalAlign:
return self.__valign
@property
def decoration_line(self) -> DecorationLine:
return self.__decoration_line
@property
def font_size(self) -> FontSize:
return self.__font_size
@property
def font_style(self) -> FontStyle:
return self.__font_style
@property
def font_weight(self) -> FontWeight:
return self.__font_weight
@property
def color(self) -> Optional[Color]:
return self.__fg_color
@property
def fg_color(self) -> Optional[Color]:
return self.__fg_color
@property
def bg_color(self) -> Optional[Color]:
return self.__bg_color
@property
def thousand_separator(self) -> ThousandSeparator:
return self.__thousand_separator
@property
def padding(self) -> Optional[int]:
return self.__padding
@padding.setter
def padding(self, value: Optional[int]) -> None:
self.__padding = value
def __init__(self, **kwargs: Any) -> None:
self.__kwargs = kwargs
self.__update_color(initialize=True)
self.__update_align(initialize=True)
self.__update_font(initialize=True)
self.__update_misc(initialize=True)
if self.__kwargs:
warnings.warn(f"unknown style attributes found: {self.__kwargs.keys()}", UserWarning)
def __repr__(self) -> str:
items = []
if self.align:
items.append(f"align={self.align.align_string}")
if self.padding is not None:
items.append(f"padding={self.padding}")
if self.vertical_align:
items.append(f"valign={self.vertical_align.align_str}")
if self.color:
items.append(f"color={self.color}")
if self.bg_color:
items.append(f"bg_color={self.bg_color}")
if self.decoration_line is not DecorationLine.NONE:
items.append(f"decoration_line={self.decoration_line.value}")
if self.font_size is not FontSize.NONE:
items.append(f"font_size={self.font_size.value}")
if self.font_style:
items.append(f"font_style={self.font_style.value}")
if self.font_weight:
items.append(f"font_weight={self.font_weight.value}")
if self.thousand_separator is not ThousandSeparator.NONE:
items.append(f"thousand_separator={self.thousand_separator.value}")
return "({})".format(", ".join(items))
def __eq__(self, other: Any) -> bool:
if self.__class__ is not other.__class__:
return False
return all(
[
self.align == other.align,
self.font_size == other.font_size,
self.font_style == other.font_style,
self.font_weight == other.font_weight,
self.thousand_separator == other.thousand_separator,
]
)
def __ne__(self, other: Any) -> bool:
if self.__class__ is not other.__class__:
return True
return not self.__eq__(other)
def update(self, **kwargs: Any) -> None:
"""Update specified style attributes."""
self.__kwargs = kwargs
self.__update_color(initialize=False)
self.__update_align(initialize=False)
self.__update_font(initialize=False)
self.__update_misc(initialize=False)
if self.__kwargs:
warnings.warn(f"unknown style attributes found: {self.__kwargs.keys()}", UserWarning)
def __update_color(self, initialize: bool) -> None:
fg_color = self.__kwargs.pop("color", None) or self.__kwargs.pop("fg_color", None)
if fg_color:
self.__fg_color: Optional[Color] = Color(fg_color)
elif initialize:
self.__fg_color = None
bg_color = self.__kwargs.pop("bg_color", None)
if bg_color:
self.__bg_color: Optional[Color] = Color(bg_color)
elif initialize:
self.__bg_color = None
def __update_font(self, initialize: bool) -> None:
font_size = self.__kwargs.pop("font_size", None)
if font_size:
self.__font_size = normalize_enum(
font_size,
FontSize,
validate=False,
default=FontSize.NONE,
)
elif initialize:
self.__font_size = FontSize.NONE
self.__validate_attr("font_size", (FontSize, str))
font_style = self.__kwargs.pop("font_style", None)
if font_style:
self.__font_style = normalize_enum(font_style, FontStyle, default=FontStyle.NORMAL)
elif initialize:
self.__font_style = FontStyle.NORMAL
self.__validate_attr("font_style", (FontStyle,))
font_weight = self.__kwargs.pop("font_weight", None)
if font_weight:
self.__font_weight = normalize_enum(font_weight, FontWeight, default=FontWeight.NORMAL)
elif initialize:
self.__font_weight = FontWeight.NORMAL
self.__validate_attr("font_weight", (FontWeight,))
def __update_align(self, initialize: bool) -> None:
align = self.__kwargs.pop("align", None)
if align:
self.__align = normalize_enum(align, Align, default=Align.AUTO)
elif initialize:
self.__align = Align.AUTO
self.__validate_attr("align", (Align,))
valign = self.__kwargs.pop("vertical_align", None)
if valign:
self.__valign = normalize_enum(valign, VerticalAlign, default=VerticalAlign.BASELINE)
elif initialize:
self.__valign = VerticalAlign.BASELINE
self.__validate_attr("vertical_align", (VerticalAlign,))
def __update_misc(self, initialize: bool) -> None:
padding = self.__kwargs.pop("padding", None)
if padding is not None:
self.__padding = padding
elif initialize:
self.__padding = None
decoration_line = self.__kwargs.pop("decoration_line", None)
if decoration_line:
self.__decoration_line = normalize_enum(
decoration_line, DecorationLine, default=DecorationLine.NONE
)
elif initialize:
self.__decoration_line = DecorationLine.NONE
self.__validate_attr("decoration_line", (DecorationLine,))
thousand_separator = self.__kwargs.pop("thousand_separator", None)
if thousand_separator:
self.__thousand_separator = _normalize_thousand_separator(thousand_separator)
elif initialize:
self.__thousand_separator = ThousandSeparator.NONE
self.__validate_attr("thousand_separator", (ThousandSeparator,))
def __validate_attr(self, attr_name: str, expected_types: tuple[type, ...]) -> None:
value = getattr(self, attr_name)
expected = " or ".join(c.__name__ for c in expected_types)
if not isinstance(value, expected_types):
raise TypeError(f"{attr_name} must be instance of {expected}: actual={type(value)}")

View File

@ -0,0 +1,331 @@
import re
from typing import TYPE_CHECKING, Any, Final, Optional
from dataproperty import Align
from tcolorpy import Color, tcolor
from ._font import FontSize, FontStyle, FontWeight
from ._style import DecorationLine, Style, ThousandSeparator
from ._styler_interface import StylerInterface
if TYPE_CHECKING:
from ..writer._table_writer import AbstractTableWriter
_align_char_mapping: Final[dict[Align, str]] = {
Align.AUTO: "<",
Align.LEFT: "<",
Align.RIGHT: ">",
Align.CENTER: "^",
}
def get_align_char(align: Align) -> str:
return _align_char_mapping[align]
def _to_latex_rgb(color: Color, value: str) -> str:
return r"\textcolor{" + color.color_code + "}{" + value + "}"
class AbstractStyler(StylerInterface):
def __init__(self, writer: "AbstractTableWriter") -> None:
self._writer = writer
self._font_size_map = self._get_font_size_map()
def get_font_size(self, style: Style) -> Optional[str]:
return self._font_size_map.get(style.font_size)
def get_additional_char_width(self, style: Style) -> int:
return 0
def apply(self, value: Any, style: Style) -> str:
return value
def apply_align(self, value: str, style: Style) -> str:
return value
def apply_terminal_style(self, value: str, style: Style) -> str:
return value
def _get_font_size_map(self) -> dict[FontSize, str]:
return {}
class NullStyler(AbstractStyler):
def get_font_size(self, style: Style) -> Optional[str]:
return ""
class TextStyler(AbstractStyler):
def apply_terminal_style(self, value: str, style: Style) -> str:
if not self._writer.enable_ansi_escape:
return value
ansi_styles = []
if style.decoration_line in (DecorationLine.STRIKE, DecorationLine.LINE_THROUGH):
ansi_styles.append("strike")
if style.decoration_line == DecorationLine.UNDERLINE:
ansi_styles.append("underline")
if style.font_weight == FontWeight.BOLD:
ansi_styles.append("bold")
if self._writer.colorize_terminal:
return tcolor(value, color=style.color, bg_color=style.bg_color, styles=ansi_styles)
return tcolor(value, styles=ansi_styles)
def __get_align_format(self, style: Style) -> str:
align_char = get_align_char(style.align)
format_items = ["{:" + align_char]
if style.padding is not None and style.padding > 0:
format_items.append(str(style.padding))
format_items.append("s}")
return "".join(format_items)
def apply_align(self, value: str, style: Style) -> str:
return self.__get_align_format(style).format(value)
def apply(self, value: str, style: Style) -> str:
if value:
if style.thousand_separator == ThousandSeparator.SPACE:
value = value.replace(",", " ")
elif style.thousand_separator == ThousandSeparator.UNDERSCORE:
value = value.replace(",", "_")
return value
class HtmlStyler(TextStyler):
def _get_font_size_map(self) -> dict[FontSize, str]:
return {
FontSize.TINY: "font-size:x-small",
FontSize.SMALL: "font-size:small",
FontSize.MEDIUM: "font-size:medium",
FontSize.LARGE: "font-size:large",
}
class LatexStyler(TextStyler):
class Command:
BOLD: Final = r"\bf"
ITALIC: Final = r"\it"
TYPEWRITER: Final = r"\tt"
UNDERLINE: Final = r"\underline"
STRIKEOUT: Final = r"\sout"
def get_additional_char_width(self, style: Style) -> int:
dummy_value = "d"
applied_value = self.apply(dummy_value, style)
return len(applied_value) - len(dummy_value)
def apply(self, value: Any, style: Style) -> str:
value = super().apply(value, style)
if not value:
return value
font_size = self.get_font_size(style)
commands = []
if font_size:
commands.append(font_size)
if style.font_weight == FontWeight.BOLD:
commands.append(self.Command.BOLD)
if style.font_style == FontStyle.ITALIC:
commands.append(self.Command.ITALIC)
elif style.font_style == FontStyle.TYPEWRITER:
commands.append(self.Command.TYPEWRITER)
if style.decoration_line in (DecorationLine.STRIKE, DecorationLine.LINE_THROUGH):
commands.append(self.Command.STRIKEOUT)
elif style.decoration_line == DecorationLine.UNDERLINE:
commands.append(self.Command.UNDERLINE)
for cmd in commands:
value = cmd + "{" + value + "}"
value = self.__apply_color(value, style)
return value
def __apply_color(self, value: str, style: Style) -> str:
if not style.fg_color:
return value
value = _to_latex_rgb(style.fg_color, value)
return value
def _get_font_size_map(self) -> dict[FontSize, str]:
return {
FontSize.TINY: r"\tiny",
FontSize.SMALL: r"\small",
FontSize.MEDIUM: r"\normalsize",
FontSize.LARGE: r"\large",
}
class MarkdownStyler(TextStyler):
def get_additional_char_width(self, style: Style) -> int:
width = 0
if style.font_weight == FontWeight.BOLD:
width += 4
if style.font_style == FontStyle.ITALIC:
width += 2
return width
def apply(self, value: Any, style: Style) -> str:
value = super().apply(value, style)
if not value:
return value
value = self._apply_font_weight(value, style)
value = self._apply_font_style(value, style)
return value
def _apply_font_weight(self, value: Any, style: Style) -> str:
if style.font_weight == FontWeight.BOLD:
value = f"**{value}**"
return value
def _apply_font_style(self, value: Any, style: Style) -> str:
if style.font_style == FontStyle.ITALIC:
value = f"_{value}_"
return value
class GFMarkdownStyler(MarkdownStyler):
"""
A styler class for GitHub Flavored Markdown
"""
def get_additional_char_width(self, style: Style) -> int:
width = super().get_additional_char_width(style)
if style.decoration_line in (DecorationLine.STRIKE, DecorationLine.LINE_THROUGH):
width += 4
if self.__use_latex(style):
dummy_value = "d"
value = self.apply(dummy_value, style)
width += len(value) - len(dummy_value)
return width
def apply(self, value: Any, style: Style) -> str:
value = super().apply(value, style)
if not value:
return value
use_latex = self.__use_latex(style)
if use_latex:
value = self.__escape_for_latex(value)
value = LatexStyler.Command.TYPEWRITER + "{" + value + "}"
value = self.__apply_decoration_line(value, style)
if use_latex:
value = r"$$" + self.__apply_color(value, style) + r"$$"
return value
def __use_latex(self, style: Style) -> bool:
return style.fg_color is not None
def __escape_for_latex(self, value: str) -> str:
value = re.sub(r"[\s_]", r"\\\\\g<0>", value)
return value.replace("-", r"\text{-}")
def __apply_decoration_line(self, value: str, style: Style) -> str:
use_latex = self.__use_latex(style)
if style.decoration_line in (DecorationLine.STRIKE, DecorationLine.LINE_THROUGH):
if use_latex:
value = r"\enclose{horizontalstrike}{" + value + "}"
else:
value = f"~~{value}~~"
elif style.decoration_line == DecorationLine.UNDERLINE:
if use_latex:
value = r"\underline{" + value + "}"
return value
def __apply_color(self, value: str, style: Style) -> str:
if not style.fg_color:
return value
return _to_latex_rgb(style.fg_color, value)
def _apply_font_weight(self, value: Any, style: Style) -> str:
if not self.__use_latex(style):
return super()._apply_font_weight(value, style)
if style.font_weight == FontWeight.BOLD:
value = LatexStyler.Command.BOLD + "{" + value + "}"
return value
def _apply_font_style(self, value: Any, style: Style) -> str:
if not self.__use_latex(style):
return super()._apply_font_style(value, style)
if style.font_style == FontStyle.ITALIC:
value = LatexStyler.Command.ITALIC + "{" + value + "}"
return value
class ReStructuredTextStyler(TextStyler):
def get_additional_char_width(self, style: Style) -> int:
from ..writer import RstCsvTableWriter
width = 0
if style.font_weight == FontWeight.BOLD:
width += 4
elif style.font_style == FontStyle.ITALIC:
width += 2
if (
style.thousand_separator == ThousandSeparator.COMMA
and self._writer.format_name == RstCsvTableWriter.FORMAT_NAME
):
width += 2
return width
def apply(self, value: Any, style: Style) -> str:
from ..writer import RstCsvTableWriter
value = super().apply(value, style)
if not value:
return value
if style.font_weight == FontWeight.BOLD:
value = f"**{value}**"
elif style.font_style == FontStyle.ITALIC:
# in reStructuredText, some custom style definition will be required to
# set for both bold and italic (currently not supported)
value = f"*{value}*"
if (
style.thousand_separator == ThousandSeparator.COMMA
and self._writer.format_name == RstCsvTableWriter.FORMAT_NAME
):
value = f'"{value}"'
return value

View File

@ -0,0 +1,26 @@
import abc
from typing import Any, Optional
from ._style import Style
class StylerInterface(metaclass=abc.ABCMeta):
@abc.abstractmethod
def apply(self, value: Any, style: Style) -> str: # pragma: no cover
raise NotImplementedError()
@abc.abstractmethod
def apply_align(self, value: str, style: Style) -> str: # pragma: no cover
raise NotImplementedError()
@abc.abstractmethod
def apply_terminal_style(self, value: str, style: Style) -> str: # pragma: no cover
raise NotImplementedError()
@abc.abstractmethod
def get_font_size(self, style: Style) -> Optional[str]: # pragma: no cover
raise NotImplementedError()
@abc.abstractmethod
def get_additional_char_width(self, style: Style) -> int: # pragma: no cover
raise NotImplementedError()

View File

@ -0,0 +1,93 @@
import importlib
import pkgutil
import re
from collections.abc import Sequence
from typing import Any, Final, NamedTuple, Optional, Protocol
from .._logger import logger
from ..style import Cell, Style
PLUGIN_NAME_PEFIX: Final = "pytablewriter"
PLUGIN_NAME_SUFFIX: Final = "theme"
KNOWN_PLUGINS: Final = (
f"{PLUGIN_NAME_PEFIX}_altrow_{PLUGIN_NAME_SUFFIX}",
f"{PLUGIN_NAME_PEFIX}_altcol_{PLUGIN_NAME_SUFFIX}",
)
class StyleFilterFunc(Protocol):
def __call__(self, cell: Cell, **kwargs: Any) -> Optional[Style]: ...
class ColSeparatorStyleFilterFunc(Protocol):
def __call__(
self, left_cell: Optional[Cell], right_cell: Optional[Cell], **kwargs: Any
) -> Optional[Style]: ...
class CheckStyleFilterKeywordArgsFunc(Protocol):
def __call__(self, **kwargs: Any) -> None: ...
class Theme(NamedTuple):
style_filter: Optional[StyleFilterFunc]
col_separator_style_filter: Optional[ColSeparatorStyleFilterFunc]
check_style_filter_kwargs: Optional[CheckStyleFilterKeywordArgsFunc]
def list_themes() -> Sequence[str]:
return list(load_ptw_plugins())
def load_ptw_plugins() -> dict[str, Theme]:
plugin_regexp: Final = re.compile(
rf"^{PLUGIN_NAME_PEFIX}[_-].+[_-]{PLUGIN_NAME_SUFFIX}", re.IGNORECASE
)
discovered_plugins: Final = {
name: importlib.import_module(name)
for _finder, name, _ispkg in pkgutil.iter_modules()
if plugin_regexp.search(name) is not None
}
logger.debug(f"discovered_plugins: {list(discovered_plugins)}")
themes: dict[str, Theme] = {}
for theme, plugin in discovered_plugins.items():
style_filter = plugin.style_filter if hasattr(plugin, "style_filter") else None
col_sep_style_filter = (
plugin.col_separator_style_filter
if hasattr(plugin, "col_separator_style_filter")
else None
)
check_kwargs_func = (
plugin.check_style_filter_kwargs
if hasattr(plugin, "check_style_filter_kwargs")
else None
)
themes[theme] = Theme(style_filter, col_sep_style_filter, check_kwargs_func)
return themes
def fetch_theme(plugin_name: str) -> Theme:
loaded_themes: Final = load_ptw_plugins()
theme_regexp: Final = re.compile(
rf"^{PLUGIN_NAME_PEFIX}[_-]{plugin_name}[_-]{PLUGIN_NAME_SUFFIX}", re.IGNORECASE
)
matched_theme = None
for loaded_theme in loaded_themes:
if theme_regexp.search(loaded_theme):
matched_theme = loaded_theme
break
else:
err_msgs = [f"{plugin_name} theme is not installed."]
if plugin_name in KNOWN_PLUGINS:
err_msgs.append(f"try 'pip install {plugin_name}' to install the theme.")
raise RuntimeError(" ".join(err_msgs))
return loaded_themes[matched_theme]

View File

@ -0,0 +1,38 @@
from dataproperty.typing import TypeHint
from typepy import (
Binary,
Bool,
Bytes,
DateTime,
Dictionary,
Infinity,
Integer,
IpAddress,
List,
Nan,
NoneType,
NullString,
RealNumber,
String,
)
from typepy.type import AbstractType
__all__ = (
"Binary",
"Bool",
"Bytes",
"DateTime",
"Dictionary",
"Infinity",
"Integer",
"IpAddress",
"List",
"Nan",
"NoneType",
"NullString",
"RealNumber",
"String",
"TypeHint",
"AbstractType",
)

View File

@ -0,0 +1,74 @@
from ._elasticsearch import ElasticsearchWriter
from ._null import NullTableWriter
from ._table_writer import AbstractTableWriter
from .binary import (
ExcelXlsTableWriter,
ExcelXlsxTableWriter,
PandasDataFramePickleWriter,
SqliteTableWriter,
)
from .text import (
AsciiDocTableWriter,
BoldUnicodeTableWriter,
BorderlessTableWriter,
CssTableWriter,
CsvTableWriter,
HtmlTableWriter,
JsonLinesTableWriter,
JsonTableWriter,
LatexMatrixWriter,
LatexTableWriter,
LtsvTableWriter,
MarkdownTableWriter,
MediaWikiTableWriter,
RstCsvTableWriter,
RstGridTableWriter,
RstSimpleTableWriter,
SpaceAlignedTableWriter,
TomlTableWriter,
TsvTableWriter,
UnicodeTableWriter,
YamlTableWriter,
)
from .text.sourcecode import (
JavaScriptTableWriter,
NumpyTableWriter,
PandasDataFrameWriter,
PythonCodeTableWriter,
)
__all__ = (
"AbstractTableWriter",
"AsciiDocTableWriter",
"BoldUnicodeTableWriter",
"BorderlessTableWriter",
"CssTableWriter",
"CsvTableWriter",
"ElasticsearchWriter",
"ExcelXlsTableWriter",
"ExcelXlsxTableWriter",
"HtmlTableWriter",
"JavaScriptTableWriter",
"JsonLinesTableWriter",
"JsonTableWriter",
"LatexMatrixWriter",
"LatexTableWriter",
"LtsvTableWriter",
"MarkdownTableWriter",
"MediaWikiTableWriter",
"NullTableWriter",
"NumpyTableWriter",
"PandasDataFramePickleWriter",
"PandasDataFrameWriter",
"PythonCodeTableWriter",
"RstCsvTableWriter",
"RstGridTableWriter",
"RstSimpleTableWriter",
"SpaceAlignedTableWriter",
"SqliteTableWriter",
"TomlTableWriter",
"TsvTableWriter",
"UnicodeTableWriter",
"YamlTableWriter",
)

View File

@ -0,0 +1,12 @@
from textwrap import dedent
HEADER_ROW = -1
import_error_msg_template = dedent(
"""\
dependency packages for {0} not found.
you can install the dependencies with 'pip install pytablewriter[{0}]'
"""
)

View File

@ -0,0 +1,205 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import copy
from collections.abc import Generator
from typing import Any
import dataproperty
from dataproperty import ColumnDataProperty
from typepy import Typecode
from ..error import EmptyValueError
from ._msgfy import to_error_message
from ._table_writer import AbstractTableWriter
DataType = dict[str, str]
Properties = dict[str, DataType]
def _get_es_datatype(column_dp: ColumnDataProperty) -> DataType:
if column_dp.typecode in (
Typecode.NONE,
Typecode.NULL_STRING,
Typecode.INFINITY,
Typecode.NAN,
):
return {"type": "keyword"}
if column_dp.typecode == Typecode.STRING:
return {"type": "text"}
if column_dp.typecode == Typecode.DATETIME:
return {"type": "date", "format": "date_optional_time"}
if column_dp.typecode == Typecode.REAL_NUMBER:
return {"type": "double"}
if column_dp.typecode == Typecode.BOOL:
return {"type": "boolean"}
if column_dp.typecode == Typecode.IP_ADDRESS:
return {"type": "ip"}
if column_dp.typecode == Typecode.INTEGER:
assert column_dp.bit_length is not None
if column_dp.bit_length <= 8:
return {"type": "byte"}
elif column_dp.bit_length <= 16:
return {"type": "short"}
elif column_dp.bit_length <= 32:
return {"type": "integer"}
elif column_dp.bit_length <= 64:
return {"type": "long"}
raise ValueError(
f"too large integer bits: expected<=64bits, actual={column_dp.bit_length:d}bits"
)
raise ValueError(f"unknown typecode: {column_dp.typecode}")
class ElasticsearchWriter(AbstractTableWriter):
"""
A table writer class for Elasticsearch.
:Dependency Packages:
- `elasticsearch-py <https://github.com/elastic/elasticsearch-py>`__
.. py:attribute:: index_name
:type: str
Alias attribute for |table_name|.
.. py:attribute:: document_type
:type: str
:value: "table"
Specify document type for indices.
.. py:method:: write_table()
Create an index and put documents for each row to Elasticsearch.
You need to pass an
`elasticsearch.Elasticsearch <https://elasticsearch-py.rtfd.io/en/master/api.html#elasticsearch>`__
instance to |stream| before calling this method.
|table_name|/:py:attr:`~pytablewriter.ElasticsearchWriter.index_name`
used as the creating index name,
invalid characters in the name are replaced with underscores (``'_'``).
Document data types for documents are automatically detected from the data.
:raises ValueError:
If the |stream| has not elasticsearch.Elasticsearch instance.
:Example:
:ref:`example-elasticsearch-table-writer`
"""
FORMAT_NAME = "elasticsearch"
@property
def format_name(self) -> str:
return self.FORMAT_NAME
@property
def support_split_write(self) -> bool:
return True
@property
def table_name(self) -> str:
return super().table_name
@table_name.setter
def table_name(self, value: str) -> None:
from pathvalidate import ErrorReason, ValidationError
from ..sanitizer import ElasticsearchIndexNameSanitizer
try:
self._table_name = ElasticsearchIndexNameSanitizer(value).sanitize(replacement_text="_")
except ValidationError as e:
if e.reason is ErrorReason.NULL_NAME:
self._table_name = ""
else:
raise
@property
def index_name(self) -> str:
return self.table_name
@index_name.setter
def index_name(self, value: str) -> None:
self.table_name = value
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.stream = None
self.is_padding = False
self.is_formatting_float = False
self._is_require_table_name = True
self._quoting_flags = copy.deepcopy(dataproperty.NOT_QUOTING_FLAGS)
self._dp_extractor.type_value_map = copy.deepcopy(dataproperty.DefaultValue.TYPE_VALUE_MAP)
self.document_type = "table"
def write_null_line(self) -> None:
pass
def _get_mappings(self) -> dict[str, dict[str, dict[str, Properties]]]:
properties: Properties = {}
for header, column_dp in zip(self.headers, self._column_dp_list):
properties[header] = _get_es_datatype(column_dp)
return {"mappings": {self.document_type: {"properties": properties}}}
def _get_body(self) -> Generator:
str_datatype = (Typecode.DATETIME, Typecode.IP_ADDRESS, Typecode.INFINITY, Typecode.NAN)
for value_dp_list in self._table_value_dp_matrix:
values = [
value_dp.data if value_dp.typecode not in str_datatype else value_dp.to_str()
for value_dp in value_dp_list
]
yield dict(zip(self.headers, values))
def _write_table(self, **kwargs: Any) -> None:
import elasticsearch as es
if not isinstance(self.stream, es.Elasticsearch):
raise ValueError("stream must be an elasticsearch.Elasticsearch instance")
try:
self._verify_value_matrix()
except EmptyValueError:
self._logger.logger.debug("no tabular data found")
return
self._preprocess()
mappings = self._get_mappings()
try:
result = self.stream.indices.create(index=self.index_name, body=mappings)
self._logger.logger.debug(result)
except es.TransportError as e:
for error in e.errors:
if error == "index_already_exists_exception":
# ignore already existing index
self._logger.logger.debug(to_error_message(e))
else:
raise
for body in self._get_body():
try:
self.stream.index(index=self.index_name, body=body)
except es.exceptions.RequestError as e:
self._logger.logger.error(f"{to_error_message(e)}, body={body}")
def _write_value_row_separator(self) -> None:
pass

View File

@ -0,0 +1,86 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
import abc
from typing import IO, Any, Union
class TableWriterInterface(metaclass=abc.ABCMeta):
"""
Interface class for writing a table.
"""
@property
@abc.abstractmethod
def format_name(self) -> str: # pragma: no cover
"""Format name for the writer.
Returns:
|str|
"""
@property
@abc.abstractmethod
def support_split_write(self) -> bool: # pragma: no cover
"""Indicates whether the writer class supports iterative table writing (``write_table_iter``) method.
Returns:
bool: |True| if the writer supported iterative table writing.
"""
@abc.abstractmethod
def write_table(self, **kwargs: Any) -> None: # pragma: no cover
"""
|write_table|.
"""
def dump(
self, output: Union[str, IO], close_after_write: bool, **kwargs: Any
) -> None: # pragma: no cover
raise NotImplementedError(f"{self.format_name} writer did not support dump method")
def dumps(self) -> str: # pragma: no cover
raise NotImplementedError(f"{self.format_name} writer did not support dumps method")
def write_table_iter(self, **kwargs: Any) -> None: # pragma: no cover
"""
Write a table with iteration.
"Iteration" means that divide the table writing into multiple writes.
This method is helpful, especially for extensive data.
The following are the premises to execute this method:
- set iterator to the |value_matrix|
- set the number of iterations to the |iteration_length| attribute
Call back function (Optional):
A callback function is called when each iteration of writing a table is completed.
You can set a callback function via the |write_callback| attribute.
Raises:
pytablewriter.NotSupportedError: If the writer class does not support this method.
.. note::
The following classes do not support this method:
- |HtmlTableWriter|
- |RstGridTableWriter|
- |RstSimpleTableWriter|
``support_split_write`` attribute return |True| if the class
is supporting this method.
"""
self._write_table_iter(**kwargs)
@abc.abstractmethod
def _write_table_iter(self, **kwargs: Any) -> None: # pragma: no cover
pass
@abc.abstractmethod
def close(self) -> None: # pragma: no cover
pass
@abc.abstractmethod
def _write_value_row_separator(self) -> None: # pragma: no cover
pass

View File

@ -0,0 +1,56 @@
"""
Import from https://github.com/thombashi/msgfy
"""
import inspect
import os.path
from types import FrameType
from typing import Optional
DEFAULT_ERROR_MESSAGE_FORMAT = "{exception}: {error_msg}"
DEFAULT_DEBUG_MESSAGE_FORMAT = "{exception} {file_name}({line_no}) {func_name}: {error_msg}"
error_message_format = DEFAULT_ERROR_MESSAGE_FORMAT
debug_message_format = DEFAULT_DEBUG_MESSAGE_FORMAT
def _to_message(exception_obj: Exception, format_str: str, frame: Optional[FrameType]) -> str:
if not isinstance(exception_obj, Exception):
raise ValueError("exception_obj must be an instance of a subclass of the Exception class")
if frame is None:
return str(exception_obj)
try:
return (
format_str.replace("{exception}", exception_obj.__class__.__name__)
.replace("{file_name}", os.path.basename(frame.f_code.co_filename))
.replace("{line_no}", str(frame.f_lineno))
.replace("{func_name}", frame.f_code.co_name)
.replace("{error_msg}", str(exception_obj))
)
except AttributeError:
raise ValueError("format_str must be a string")
def to_error_message(exception_obj: Exception, format_str: Optional[str] = None) -> str:
if not format_str:
format_str = error_message_format
frame = inspect.currentframe()
if frame is None:
return str(exception_obj)
return _to_message(exception_obj, format_str, frame.f_back)
def to_debug_message(exception_obj: Exception, format_str: Optional[str] = None) -> str:
if not format_str:
format_str = debug_message_format
frame = inspect.currentframe()
if frame is None:
return str(exception_obj)
return _to_message(exception_obj, format_str, frame.f_back)

View File

@ -0,0 +1,61 @@
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from typing import IO, Any, Union
from ._interface import TableWriterInterface
from .text._interface import IndentationInterface, TextWriterInterface
class NullTableWriter(IndentationInterface, TextWriterInterface, TableWriterInterface):
FORMAT_NAME = "null"
def __init__(self, **kwargs: Any) -> None:
self.table_name = kwargs.get("table_name", "")
self.value_matrix = kwargs.get("value_matrix", [])
self.is_formatting_float = kwargs.get("is_formatting_float", True)
self.headers = kwargs.get("headers", [])
self.type_hints = kwargs.get("type_hints", [])
self.max_workers = kwargs.get("max_workers", 1)
def __repr__(self) -> str:
return self.dumps()
@property
def format_name(self) -> str:
return self.FORMAT_NAME
@property
def support_split_write(self) -> bool:
return True
def set_indent_level(self, indent_level: int) -> None:
pass
def inc_indent_level(self) -> None:
pass
def dec_indent_level(self) -> None:
pass
def write_null_line(self) -> None:
pass
def write_table(self, **kwargs: Any) -> None:
pass
def dump(self, output: Union[str, IO], close_after_write: bool = True, **kwargs: Any) -> None:
pass
def dumps(self) -> str:
return ""
def _write_table_iter(self, **kwargs: Any) -> None:
pass
def close(self) -> None:
pass
def _write_value_row_separator(self) -> None:
pass

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,11 @@
from ._excel import ExcelXlsTableWriter, ExcelXlsxTableWriter
from ._pandas import PandasDataFramePickleWriter
from ._sqlite import SqliteTableWriter
__all__ = (
"ExcelXlsTableWriter",
"ExcelXlsxTableWriter",
"PandasDataFramePickleWriter",
"SqliteTableWriter",
)

View File

@ -0,0 +1,501 @@
import abc
import copy
import warnings
from typing import IO, TYPE_CHECKING, Any, Final, Optional, Union, cast
import dataproperty
import typepy
from dataproperty import DataProperty
from tabledata import TableData
from typepy import Integer
from .._common import import_error_msg_template
from ._excel_workbook import ExcelWorkbookInterface, ExcelWorkbookXls, ExcelWorkbookXlsx
from ._interface import AbstractBinaryTableWriter
if TYPE_CHECKING:
from xlwt import XFStyle
class ExcelTableWriter(AbstractBinaryTableWriter, metaclass=abc.ABCMeta):
"""
An abstract class of a table writer for Excel file format.
"""
FORMAT_NAME = "excel"
@property
def format_name(self) -> str:
return self.FORMAT_NAME
@property
def workbook(self) -> Optional[ExcelWorkbookInterface]:
return self._workbook
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._workbook: Optional[ExcelWorkbookInterface] = None
self._dp_extractor.type_value_map = {
typepy.Typecode.INFINITY: "Inf",
typepy.Typecode.NAN: "NaN",
}
self._first_header_row = 0
self._last_header_row = self.first_header_row
self._first_data_row = self.last_header_row + 1
self._first_data_col = 0
self._last_data_row: Optional[int] = None
self._last_data_col: Optional[int] = None
self._current_data_row = self._first_data_row
self._quoting_flags = copy.deepcopy(dataproperty.NOT_QUOTING_FLAGS)
self._quoting_flags[typepy.Typecode.DATETIME] = True
@property
def first_header_row(self) -> int:
"""int: Index of the first row of the header.
.. note:: |excel_attr|
"""
return self._first_header_row
@property
def last_header_row(self) -> int:
"""int: Index of the last row of the header.
.. note:: |excel_attr|
"""
return self._last_header_row
@property
def first_data_row(self) -> int:
"""int: Index of the first row of the data (table body).
.. note:: |excel_attr|
"""
return self._first_data_row
@property
def last_data_row(self) -> Optional[int]:
"""int: Index of the last row of the data (table body).
.. note:: |excel_attr|
"""
return self._last_data_row
@property
def first_data_col(self) -> int:
"""int: Index of the first column of the table.
.. note:: |excel_attr|
"""
return self._first_data_col
@property
def last_data_col(self) -> Optional[int]:
"""int: Index of the last column of the table.
.. note:: |excel_attr|
"""
return self._last_data_col
def is_opened(self) -> bool:
return self.workbook is not None
def open(self, file_path: str) -> None:
"""
Open an Excel workbook file.
:param str file_path: Excel workbook file path to open.
"""
if self.workbook and self.workbook.file_path == file_path:
self._logger.logger.debug(f"workbook already opened: {self.workbook.file_path}")
return
self.close()
self._open(file_path)
@abc.abstractmethod
def _open(self, workbook_path: str) -> None: # pragma: no cover
pass
def close(self) -> None:
"""
Close the current workbook.
"""
if self.is_opened():
self.workbook.close() # type: ignore
self._workbook = None
def from_tabledata(self, value: TableData, is_overwrite_table_name: bool = True) -> None:
"""
Set following attributes from |TableData|
- :py:attr:`~.table_name`.
- :py:attr:`~.headers`.
- :py:attr:`~.value_matrix`.
And create worksheet named from :py:attr:`~.table_name` ABC
if not existed yet.
:param tabledata.TableData value: Input table data.
"""
super().from_tabledata(value)
if self.is_opened():
self.make_worksheet(self.table_name)
def make_worksheet(self, sheet_name: Optional[str] = None) -> None:
"""Make a worksheet to the current workbook.
Args:
sheet_name (str):
Name of the worksheet to create. The name will be automatically generated
(like ``"Sheet1"``) if the ``sheet_name`` is empty.
"""
if sheet_name is None:
sheet_name = self.table_name
if not sheet_name:
sheet_name = ""
self._stream = self.workbook.add_worksheet(sheet_name) # type: ignore
self._current_data_row = self._first_data_row
def dump(self, output: Union[str, IO], close_after_write: bool = True, **kwargs: Any) -> None:
"""Write a worksheet to the current workbook.
Args:
output (str):
Path to the workbook file to write.
close_after_write (bool, optional):
Close the workbook after write.
Defaults to |True|.
"""
if not isinstance(output, str):
raise TypeError(f"output must be a str: actual={type(output)}")
self.open(output)
try:
self.make_worksheet(self.table_name)
self.write_table(**kwargs)
finally:
if close_after_write:
self.close()
@abc.abstractmethod
def _write_header(self) -> None:
pass
@abc.abstractmethod
def _write_cell(self, row: int, col: int, value_dp: DataProperty) -> None:
pass
def _write_table(self, **kwargs: Any) -> None:
self._preprocess_table_dp()
self._preprocess_table_property()
self._write_header()
self._write_value_matrix()
self._postprocess()
def _write_value_matrix(self) -> None:
for value_dp_list in self._table_value_dp_matrix:
for col_idx, value_dp in enumerate(value_dp_list):
self._write_cell(self._current_data_row, col_idx, value_dp)
self._current_data_row += 1
def _get_last_column(self) -> int:
if typepy.is_not_empty_sequence(self.headers):
return len(self.headers) - 1
if typepy.is_not_empty_sequence(self.value_matrix):
return len(self.value_matrix[0]) - 1
raise ValueError("data not found")
def _postprocess(self) -> None:
self._last_data_row = self._current_data_row
self._last_data_col = self._get_last_column()
class ExcelXlsTableWriter(ExcelTableWriter):
"""
A table writer class for Excel file format: ``.xls`` (older or equal to Office 2003).
``xlwt`` package required to use this class.
.. py:method:: write_table()
Write a table to the current opened worksheet.
:raises IOError: If failed to write data to the worksheet.
.. note::
Specific values in the tabular data are converted when writing:
- |None|: written as an empty string
- |inf|: written as ``Inf``
- |nan|: written as ``NaN``
"""
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__col_style_table: dict[int, Any] = {}
def _open(self, workbook_path: str) -> None:
self._workbook = ExcelWorkbookXls(workbook_path)
def _write_header(self) -> None:
if not self.is_write_header or typepy.is_empty_sequence(self.headers):
return
for col, value in enumerate(self.headers):
self.stream.write(self.first_header_row, col, value)
def _write_cell(self, row: int, col: int, value_dp: DataProperty) -> None:
if value_dp.typecode in [typepy.Typecode.REAL_NUMBER]:
try:
cell_style = self.__get_cell_style(col)
except ValueError:
pass
else:
self.stream.write(row, col, value_dp.data, cell_style)
return
self.stream.write(row, col, value_dp.data)
def _postprocess(self) -> None:
super()._postprocess()
self.__col_style_table = {}
def __get_cell_style(self, col: int) -> "XFStyle":
try:
import xlwt
except ImportError:
warnings.warn(import_error_msg_template.format("excel"))
raise
if col in self.__col_style_table:
return self.__col_style_table.get(col) # type: ignore
try:
col_dp = self._column_dp_list[col]
except KeyError:
return {} # type: ignore
if col_dp.typecode not in [typepy.Typecode.REAL_NUMBER]:
raise ValueError()
if not Integer(col_dp.minmax_decimal_places.max_value).is_type():
raise ValueError()
float_digit = col_dp.minmax_decimal_places.max_value
if float_digit is None or float_digit <= 0:
raise ValueError()
num_format_str = "#,{:s}0.{:s}".format("#" * int(float_digit), "0" * int(float_digit))
cell_style = xlwt.easyxf(num_format_str=num_format_str)
self.__col_style_table[col] = cell_style
return cell_style
class ExcelXlsxTableWriter(ExcelTableWriter):
"""
A table writer class for Excel file format: ``.xlsx`` (newer or equal to Office 2007).
.. py:method:: write_table()
Write a table to the current opened worksheet.
:raises IOError: If failed to write data to the worksheet.
Examples:
:ref:`example-excel-table-writer`
.. note::
Specific values in the tabular data are converted when writing:
- |None|: written as an empty string
- |inf|: written as ``Inf``
- |nan|: written as ``NaN``
"""
MAX_CELL_WIDTH: Final[int] = 60
class TableFormat:
HEADER: Final = "header"
CELL: Final = "cell"
NAN: Final = "nan"
class Default:
FONT_NAME: Final[str] = "MS Gothic"
FONT_SIZE: Final[int] = 9
CELL_FORMAT: Final[dict[str, Union[int, str, bool]]] = {
"font_name": FONT_NAME,
"font_size": FONT_SIZE,
"align": "top",
"text_wrap": True,
"top": 1,
"left": 1,
"bottom": 1,
"right": 1,
}
HEADER_FORMAT: Final[dict[str, Union[int, str, bool]]] = {
"font_name": FONT_NAME,
"font_size": FONT_SIZE,
"bg_color": "#DFDFFF",
"bold": True,
"left": 1,
"right": 1,
}
NAN_FORMAT: Final[dict[str, Union[int, str, bool]]] = {
"font_name": FONT_NAME,
"font_size": FONT_SIZE,
"font_color": "silver",
"top": 1,
"left": 1,
"bottom": 1,
"right": 1,
}
@property
def __nan_format_property(self) -> dict[str, Union[int, str, bool]]:
return self.format_table.get(self.TableFormat.NAN, self.default_format)
@property
def __cell_format_property(self) -> dict[str, Union[int, str, bool]]:
return self.format_table.get(self.TableFormat.CELL, self.default_format)
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.default_format = self.Default.CELL_FORMAT
self.format_table = {
self.TableFormat.CELL: self.Default.CELL_FORMAT,
self.TableFormat.HEADER: self.Default.HEADER_FORMAT,
self.TableFormat.NAN: self.Default.NAN_FORMAT,
}
self.__col_cell_format_cache: dict[int, Any] = {}
self.__col_numprops_table: dict[int, dict[str, str]] = {}
def _open(self, workbook_path: str) -> None:
self._workbook = ExcelWorkbookXlsx(workbook_path)
def _write_header(self) -> None:
if not self.is_write_header or typepy.is_empty_sequence(self.headers):
return
header_format_props = self.format_table.get(self.TableFormat.HEADER, self.default_format)
header_format = self.__add_format(header_format_props)
self.stream.write_row(
row=self.first_header_row, col=0, data=self.headers, cell_format=header_format
)
for row in range(self.first_header_row, self.last_header_row):
self.stream.write_row(
row=row, col=0, data=[""] * len(self.headers), cell_format=header_format
)
def _write_cell(self, row: int, col: int, value_dp: DataProperty) -> None:
base_props = dict(self.__cell_format_property)
format_key = f"{col:d}_{value_dp.typecode.name:s}"
if value_dp.typecode in [typepy.Typecode.INTEGER, typepy.Typecode.REAL_NUMBER]:
num_props = self.__get_number_property(col)
base_props.update(num_props)
cell_format = self.__get_cell_format(format_key, base_props)
try:
self.stream.write_number(row, col, float(value_dp.data), cell_format)
return
except TypeError:
pass
if value_dp.typecode is typepy.Typecode.NAN:
base_props = dict(self.__nan_format_property)
cell_format = self.__get_cell_format(format_key, base_props)
self.stream.write(row, col, value_dp.data, cell_format)
def __get_number_property(self, col: int) -> dict[str, str]:
if col in self.__col_numprops_table:
return self.__col_numprops_table[col]
try:
col_dp = self._column_dp_list[col]
except KeyError:
return {}
if col_dp.typecode not in [typepy.Typecode.INTEGER, typepy.Typecode.REAL_NUMBER]:
return {}
num_props = {}
if Integer(col_dp.minmax_decimal_places.max_value).is_type():
float_digit = col_dp.minmax_decimal_places.max_value
if float_digit is not None and float_digit > 0:
num_props = {"num_format": "0.{:s}".format("0" * int(float_digit))}
self.__col_numprops_table[col] = num_props
return num_props
def __get_cell_format(self, format_key, cell_props) -> dict: # type: ignore
cell_format = self.__col_cell_format_cache.get(format_key)
if cell_format is not None:
return cell_format
# cache miss
cell_format = self.__add_format(cell_props)
self.__col_cell_format_cache[format_key] = cell_format
return cell_format
def __add_format(self, dict_property): # type: ignore
assert self.workbook
return self.workbook.workbook.add_format(dict_property)
def __set_cell_width(self) -> None:
font_size = cast(int, self.__cell_format_property.get("font_size"))
if not Integer(font_size).is_type():
return
for col_idx, col_dp in enumerate(self._column_dp_list):
width = min(col_dp.ascii_char_width, self.MAX_CELL_WIDTH) * (font_size / 10.0) + 2
self.stream.set_column(col_idx, col_idx, width=width)
def _preprocess_table_property(self) -> None:
super()._preprocess_table_property()
self.__set_cell_width()
def _postprocess(self) -> None:
super()._postprocess()
self.stream.autofilter(
self.last_header_row, self.first_data_col, self.last_data_row, self.last_data_col
)
self.stream.freeze_panes(self.first_data_row, self.first_data_col)
self.__col_cell_format_cache = {}
self.__col_numprops_table = {}

View File

@ -0,0 +1,141 @@
import abc
import warnings
from typing import Any, Optional
import typepy
from ..._logger import logger
from ...sanitizer import sanitize_excel_sheet_name
from .._common import import_error_msg_template
from .._msgfy import to_error_message
class ExcelWorkbookInterface(metaclass=abc.ABCMeta):
@property
@abc.abstractmethod
def workbook(self) -> Any: # pragma: no cover
pass
@property
@abc.abstractmethod
def file_path(self) -> Optional[str]: # pragma: no cover
pass
@abc.abstractmethod
def open(self, file_path: str) -> None: # pragma: no cover
pass
@abc.abstractmethod
def close(self) -> None: # pragma: no cover
pass
@abc.abstractmethod
def add_worksheet(self, worksheet_name: Optional[str]) -> Any: # pragma: no cover
pass
class ExcelWorkbook(ExcelWorkbookInterface):
@property
def workbook(self) -> Any:
return self._workbook
@property
def file_path(self) -> Optional[str]:
return self._file_path
def _clear(self) -> None:
self._workbook = None
self._file_path: Optional[str] = None
self._worksheet_table: dict[str, Any] = {}
def __init__(self, file_path: str) -> None:
self._clear()
self._file_path = file_path
def __del__(self) -> None:
self.close()
class ExcelWorkbookXls(ExcelWorkbook):
def __init__(self, file_path: str) -> None:
super().__init__(file_path)
self.open(file_path)
def open(self, file_path: str) -> None:
try:
import xlwt
except ImportError:
warnings.warn(import_error_msg_template.format("excel"))
raise
self._workbook = xlwt.Workbook()
def close(self) -> None:
if self.workbook is None:
return
try:
self.workbook.save(self._file_path)
except IndexError as e:
logger.debug(to_error_message(e))
self._clear()
def add_worksheet(self, worksheet_name: Optional[str]) -> Any:
if typepy.is_not_null_string(worksheet_name):
assert worksheet_name
worksheet_name = sanitize_excel_sheet_name(worksheet_name)
if worksheet_name in self._worksheet_table:
# the work sheet is already exists
return self._worksheet_table.get(worksheet_name)
else:
sheet_id = 1
while True:
worksheet_name = f"Sheet{sheet_id:d}"
if worksheet_name not in self._worksheet_table:
break
sheet_id += 1
worksheet = self.workbook.add_sheet(worksheet_name)
self._worksheet_table[worksheet.get_name()] = worksheet
return worksheet
class ExcelWorkbookXlsx(ExcelWorkbook):
def __init__(self, file_path: str) -> None:
super().__init__(file_path)
self.open(file_path)
def open(self, file_path: str) -> None:
try:
import xlsxwriter
except ImportError:
warnings.warn(import_error_msg_template.format("excel"))
raise
self._workbook = xlsxwriter.Workbook(file_path)
def close(self) -> None:
if self.workbook is None:
return
self._workbook.close() # type: ignore
self._clear()
def add_worksheet(self, worksheet_name: Optional[str]) -> Any:
if typepy.is_not_null_string(worksheet_name):
assert worksheet_name
worksheet_name = sanitize_excel_sheet_name(worksheet_name)
if worksheet_name in self._worksheet_table:
# the work sheet is already exists
return self._worksheet_table.get(worksheet_name)
else:
worksheet_name = None
worksheet = self.workbook.add_worksheet(worksheet_name)
self._worksheet_table[worksheet.get_name()] = worksheet
return worksheet

View File

@ -0,0 +1,58 @@
import abc
from typing import Any
from .._table_writer import AbstractTableWriter
class BinaryWriterInterface(metaclass=abc.ABCMeta):
@abc.abstractmethod
def is_opened(self) -> bool: # pragma: no cover
pass
@abc.abstractmethod
def open(self, file_path: str) -> None: # pragma: no cover
"""
Open a file for output stream.
Args:
file_path (str): path to the file.
"""
class AbstractBinaryTableWriter(AbstractTableWriter, BinaryWriterInterface):
@property
def stream(self) -> Any:
return self._stream
@stream.setter
def stream(self, value: Any) -> None:
raise RuntimeError(
"cannot assign a stream to binary format writers. use open method instead."
)
@property
def support_split_write(self) -> bool:
return True
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.table_name = kwargs.get("table_name", "")
self._stream = None
def __del__(self) -> None:
self.close()
def is_opened(self) -> bool:
return self.stream is not None
def dumps(self) -> str:
raise NotImplementedError("binary format writers did not support dumps method")
def _verify_stream(self) -> None:
if self.stream is None:
raise OSError("null output stream. required to open(file_path) first.")
def _write_value_row_separator(self) -> None:
pass

View File

@ -0,0 +1,99 @@
from typing import IO, Any, Optional, Union
import tabledata
from ...error import EmptyValueError
from ._interface import AbstractBinaryTableWriter
class PandasDataFramePickleWriter(AbstractBinaryTableWriter):
"""
A table writer class for pandas DataFrame pickle.
.. py:method:: write_table()
Write a table to a pandas DataFrame pickle file.
"""
FORMAT_NAME = "pandas_pickle"
@property
def format_name(self) -> str:
return self.FORMAT_NAME
@property
def support_split_write(self) -> bool:
return False
def __init__(self, **kwargs: Any) -> None:
import copy
import dataproperty
super().__init__(**kwargs)
self.is_padding = False
self.is_formatting_float = False
self._use_default_header = True
self._quoting_flags = copy.deepcopy(dataproperty.NOT_QUOTING_FLAGS)
self.__filepath: Optional[str] = None
def is_opened(self) -> bool:
return self.__filepath is not None
def open(self, file_path: str) -> None:
self.__filepath = file_path
def close(self) -> None:
super().close()
self.__filepath = None
def dump(self, output: Union[str, IO], close_after_write: bool = True, **kwargs: Any) -> None:
"""Write data to a DataFrame pickle file.
Args:
output (str): Path to an output DataFrame pickle file.
"""
if not isinstance(output, str):
raise TypeError(f"output must be a str: actual={type(output)}")
self.open(output)
try:
self.write_table(**kwargs)
finally:
if close_after_write:
self.close()
def _verify_stream(self) -> None:
pass
def _write_table(self, **kwargs: Any) -> None:
if self.__filepath is None or not self.is_opened():
self._logger.logger.error("required to open(file_path) first.")
return
try:
self._verify_value_matrix()
except EmptyValueError:
self._logger.logger.debug("no tabular data found")
return
self._preprocess()
table_data = tabledata.TableData(
self.table_name,
self.headers,
[
[value_dp.data for value_dp in value_dp_list]
for value_dp_list in self._table_value_dp_matrix
],
type_hints=self.type_hints,
max_workers=self.max_workers,
)
table_data.as_dataframe().to_pickle(self.__filepath)
def _write_table_iter(self, **kwargs: Any) -> None:
self._write_table(**kwargs)

View File

@ -0,0 +1,104 @@
from os.path import abspath
from typing import IO, Any, Union
import tabledata
from ...error import EmptyValueError
from ._interface import AbstractBinaryTableWriter
class SqliteTableWriter(AbstractBinaryTableWriter):
"""
A table writer class for `SQLite <https://www.sqlite.org/index.html>`__ database.
.. py:method:: write_table()
Write a table to a `SQLite <https://www.sqlite.org/index.html>`__ database.
:raises pytablewriter.EmptyTableNameError:
If the |table_name| is empty.
:Example:
:ref:`example-sqlite-table-writer`
"""
FORMAT_NAME = "sqlite"
@property
def format_name(self) -> str:
return self.FORMAT_NAME
def __init__(self, **kwargs: Any) -> None:
import copy
import dataproperty
super().__init__(**kwargs)
self.is_padding = False
self.is_formatting_float = False
self._use_default_header = True
self._is_require_table_name = True
self._is_require_header = True
self._quoting_flags = copy.deepcopy(dataproperty.NOT_QUOTING_FLAGS)
def open(self, file_path: str) -> None:
"""
Open a SQLite database file.
:param str file_path: SQLite database file path to open.
"""
from simplesqlite import SimpleSQLite
if self.is_opened():
if self.stream.database_path == abspath(file_path):
self._logger.logger.debug(f"database already opened: {self.stream.database_path}")
return
self.close()
self._stream = SimpleSQLite(file_path, "w", max_workers=self.max_workers)
def dump(self, output: Union[str, IO], close_after_write: bool = True, **kwargs: Any) -> None:
"""Write data to the SQLite database file.
Args:
output (str):
path to the output SQLite database file.
close_after_write (bool, optional):
Close the output after write.
Defaults to |True|.
"""
if not isinstance(output, str):
raise TypeError(f"output must be a str: actual={type(output)}")
self.open(output)
try:
self.write_table(**kwargs)
finally:
if close_after_write:
self.close()
def _write_table(self, **kwargs: Any) -> None:
try:
self._verify_value_matrix()
except EmptyValueError:
self._logger.logger.debug("no tabular data found")
return
self._preprocess()
table_data = tabledata.TableData(
self.table_name,
self.headers,
[
[value_dp.data for value_dp in value_dp_list]
for value_dp_list in self._table_value_dp_matrix
],
type_hints=self.type_hints,
max_workers=self.max_workers,
)
self.stream.create_table_from_tabledata(table_data)

View File

@ -0,0 +1,44 @@
from ._asciidoc import AsciiDocTableWriter
from ._borderless import BorderlessTableWriter
from ._css import CssTableWriter
from ._csv import CsvTableWriter
from ._html import HtmlTableWriter
from ._json import JsonTableWriter
from ._jsonlines import JsonLinesTableWriter
from ._latex import LatexMatrixWriter, LatexTableWriter
from ._ltsv import LtsvTableWriter
from ._markdown import MarkdownFlavor, MarkdownTableWriter, normalize_md_flavor
from ._mediawiki import MediaWikiTableWriter
from ._rst import RstCsvTableWriter, RstGridTableWriter, RstSimpleTableWriter
from ._spacealigned import SpaceAlignedTableWriter
from ._toml import TomlTableWriter
from ._tsv import TsvTableWriter
from ._unicode import BoldUnicodeTableWriter, UnicodeTableWriter
from ._yaml import YamlTableWriter
__all__ = (
"AsciiDocTableWriter",
"BoldUnicodeTableWriter",
"BorderlessTableWriter",
"CssTableWriter",
"CsvTableWriter",
"HtmlTableWriter",
"JsonTableWriter",
"JsonLinesTableWriter",
"LatexMatrixWriter",
"LatexTableWriter",
"LtsvTableWriter",
"MarkdownFlavor",
"MarkdownTableWriter",
"normalize_md_flavor",
"MediaWikiTableWriter",
"RstCsvTableWriter",
"RstGridTableWriter",
"RstSimpleTableWriter",
"SpaceAlignedTableWriter",
"TomlTableWriter",
"TsvTableWriter",
"UnicodeTableWriter",
"YamlTableWriter",
)

View File

@ -0,0 +1,147 @@
import copy
from collections.abc import Sequence
from typing import Any, Final
import dataproperty as dp
import typepy
from dataproperty import ColumnDataProperty, DataProperty, LineBreakHandling
from mbstrdecoder import MultiByteStrDecoder
from ...style import (
Align,
FontStyle,
FontWeight,
Style,
StylerInterface,
TextStyler,
get_align_char,
)
from .._table_writer import AbstractTableWriter
from ._text_writer import TextTableWriter
class AsciiDocStyler(TextStyler):
def apply(self, value: str, style: Style) -> str:
value = super().apply(value, style)
if not value:
return value
try:
fg_color = style.fg_color.name.lower() # type: ignore
except AttributeError:
fg_color = None
try:
bg_color = style.bg_color.name.lower() # type: ignore
except AttributeError:
bg_color = None
if fg_color and bg_color:
value = f"[{fg_color} {bg_color}-background]##{value}##"
elif fg_color:
value = f"[{fg_color}]##{value}##"
elif bg_color:
value = f"[{bg_color}-background]##{value}##"
if style.font_weight == FontWeight.BOLD:
value = f"*{value}*"
if style.font_style == FontStyle.ITALIC:
value = f"_{value}_"
return value
class AsciiDocTableWriter(TextTableWriter):
"""
A table writer class for `AsciiDoc <https://asciidoc.org/>`__ format.
"""
FORMAT_NAME = "asciidoc"
@property
def format_name(self) -> str:
return self.FORMAT_NAME
@property
def support_split_write(self) -> bool:
return True
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.column_delimiter = "\n"
self.is_padding = False
self.is_write_header_separator_row = True
self.is_write_value_separator_row = True
self.is_write_opening_row = True
self.is_write_closing_row = True
self.update_preprocessor(line_break_handling=LineBreakHandling.NOP)
self._quoting_flags = copy.deepcopy(dp.NOT_QUOTING_FLAGS)
def _create_styler(self, writer: AbstractTableWriter) -> StylerInterface:
return AsciiDocStyler(writer)
def _write_value_row(
self, row: int, values: Sequence[str], value_dp_list: Sequence[DataProperty]
) -> None:
self._write_row(
row,
[
self.__modify_row_element(row, col_idx, value, value_dp)
for col_idx, (value, value_dp) in enumerate(zip(values, value_dp_list))
],
)
def _get_opening_row_items(self) -> list[str]:
cols: Final = ", ".join(
f"{get_align_char(col_dp.align)}{col_dp.ascii_char_width}"
for col_dp in self._column_dp_list
)
rows = [f'[cols="{cols}", options="header"]']
if typepy.is_not_null_string(self.table_name):
rows.append("." + MultiByteStrDecoder(self.table_name).unicode_str)
rows.append("|===")
return ["\n".join(rows)]
def _get_header_row_separator_items(self) -> list[str]:
return [""]
def _get_value_row_separator_items(self) -> list[str]:
return self._get_header_row_separator_items()
def _get_closing_row_items(self) -> list[str]:
return ["|==="]
def __apply_align(self, value: str, style: Style) -> str:
return f"{get_align_char(Align.CENTER)}|{value}"
def _apply_style_to_header_item(
self, col_dp: ColumnDataProperty, value_dp: DataProperty, style: Style
) -> str:
value = self._styler.apply(col_dp.dp_to_str(value_dp), style=style)
return self.__apply_align(value, style)
def __modify_row_element(
self, row_idx: int, col_idx: int, value: str, value_dp: DataProperty
) -> str:
col_dp: Final = self._column_dp_list[col_idx]
style: Final = self._fetch_style(row_idx, col_dp, value_dp)
align = col_dp.align
if style and style.align and style.align != align:
forma_stirng = "{0:s}|{1:s}"
align = style.align
elif value_dp.align != align:
forma_stirng = "{0:s}|{1:s}"
align = value_dp.align
else:
forma_stirng = "|{1:s}"
return forma_stirng.format(get_align_char(align), value)

Some files were not shown because too many files have changed in this diff Show More