Module refinery.units.formats.archive.xtsql

Classes

class xtsql (*paths, exclude=None, list=False, join_path=False, drop_path=False, fuzzy=0, exact=False, regex=False, path=b'path')

Extract files from SQLite3 databases.

Expand source code Browse git

class xtsql(PathExtractorUnit):
    """
    Extract files from SQLite3 databases.
    """
    def unpack(self, data: bytearray):
        def _json(object):
            return json.dumps(object, tojson=json.bytes_as_string)

        if sys.version_info[:2] < (3, 11):
            raise NotImplementedError(F'python 3.11 is required to use {self.__class__.__name__}.')

        database = sqlite3.connect(':memory:')
        database.text_factory = bytes
        database.deserialize(data)
        database.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, len(data))
        cursor = database.cursor()
        result: dict[str, list[dict[str, int | float | str | bytes]]] = {}

        listing: list[tuple[bytes]] = cursor.execute(
            'SELECT name FROM sqlite_master WHERE type = ?', ('table',)).fetchall()

        for (name,) in listing:
            table = name.decode('utf8')
            quoted = table.replace('"', '""')
            select = F'SELECT * FROM "{quoted}"'
            try:
                if any(step[1] == B'VOpen' for step in cursor.execute(F'EXPLAIN {select}')):
                    self.log_info(F'skipping {table}, which does not store its own rows')
                    continue
                rows = cursor.execute(select).fetchall()
            except sqlite3.Error as error:
                self.log_warn(F'failed to read table {table}: {error!s}')
                continue
            names = [column[0] for column in cursor.description]
            result[table] = [dict(zip(names, row)) for row in rows]

        yield UnpackResult('db', functools.partial(_json, result))

        for table, rows in result.items():

            yield UnpackResult(F'db/{table}', functools.partial(_json, rows))

            for k, row in enumerate(rows):

                root = F'db/{table}/{k}'
                yield UnpackResult(root, functools.partial(_json, row))

                for name, value in row.items():
                    path = F'{root}/{name}'
                    if value is None:
                        continue
                    if isinstance(value, (int, float)):
                        value = str(value)
                    if isinstance(value, str):
                        value = value.encode(self.codec)
                    if isinstance(value, bytes):
                        yield UnpackResult(path, value)

    @classmethod
    def handles(cls, data):
        return memoryview(data)[:15] == B'SQLite format 3'

Ancestors

• PathExtractorUnit
• Unit
• UnitBase
• Entry

Subclasses

• xtsql

Class variables

var reverse

The type of the None singleton.

Methods

def unpack(self, data)

Expand source code Browse git

def unpack(self, data: bytearray):
    def _json(object):
        return json.dumps(object, tojson=json.bytes_as_string)

    if sys.version_info[:2] < (3, 11):
        raise NotImplementedError(F'python 3.11 is required to use {self.__class__.__name__}.')

    database = sqlite3.connect(':memory:')
    database.text_factory = bytes
    database.deserialize(data)
    database.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, len(data))
    cursor = database.cursor()
    result: dict[str, list[dict[str, int | float | str | bytes]]] = {}

    listing: list[tuple[bytes]] = cursor.execute(
        'SELECT name FROM sqlite_master WHERE type = ?', ('table',)).fetchall()

    for (name,) in listing:
        table = name.decode('utf8')
        quoted = table.replace('"', '""')
        select = F'SELECT * FROM "{quoted}"'
        try:
            if any(step[1] == B'VOpen' for step in cursor.execute(F'EXPLAIN {select}')):
                self.log_info(F'skipping {table}, which does not store its own rows')
                continue
            rows = cursor.execute(select).fetchall()
        except sqlite3.Error as error:
            self.log_warn(F'failed to read table {table}: {error!s}')
            continue
        names = [column[0] for column in cursor.description]
        result[table] = [dict(zip(names, row)) for row in rows]

    yield UnpackResult('db', functools.partial(_json, result))

    for table, rows in result.items():

        yield UnpackResult(F'db/{table}', functools.partial(_json, rows))

        for k, row in enumerate(rows):

            root = F'db/{table}/{k}'
            yield UnpackResult(root, functools.partial(_json, row))

            for name, value in row.items():
                path = F'{root}/{name}'
                if value is None:
                    continue
                if isinstance(value, (int, float)):
                    value = str(value)
                if isinstance(value, str):
                    value = value.encode(self.codec)
                if isinstance(value, bytes):
                    yield UnpackResult(path, value)

Inherited members

• PathExtractorUnit:
  • CustomJoinBehaviour
  • CustomPathSeparator
  • FilterEverything
  • Requires
  • act
  • assemble
  • carved
  • close
  • codec
  • console
  • filter
  • finish
  • handles
  • is_quiet
  • is_reversible
  • isatty
  • labelled
  • leniency
  • log_always
  • log_debug
  • log_detach
  • log_fail
  • log_info
  • log_level
  • log_warn
  • logger
  • name
  • nozzle
  • optional_dependencies
  • read
  • read1
  • required_dependencies
  • reset
  • run
  • source
  • superinit
• UnitBase:
  • process