Source code for pytablewriter.writer.binary._sqlite

from os.path import abspath
from typing import IO, Any, Union

import tabledata

from ...error import EmptyValueError
from ._interface import AbstractBinaryTableWriter

[docs] class SqliteTableWriter(AbstractBinaryTableWriter): """ A table writer class for `SQLite <>`__ database. .. py:method:: write_table() Write a table to a `SQLite <>`__ 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)
[docs] 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 == abspath(file_path): self._logger.logger.debug(f"database already opened: {}") return self.close() self._stream = SimpleSQLite(file_path, "w", max_workers=self.max_workers)
[docs] 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)}") 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, [ [ 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, )