<commit_before># -*- test-case-name: ranger-ims-server.ext.test.test_sqlite -*-
"""
SQLite utilities
"""

from pathlib import Path
from sqlite3 import (
    Connection as BaseConnection, Cursor as BaseCursor, Error as SQLiteError,
    Row as BaseRow, connect as sqliteConnect,
)
from typing import (
    Any, Callable, Iterable, Mapping, Optional, Tuple, TypeVar, Union, cast
)
from typing.io import TextIO

from attr import attrib, attrs
from attr.validators import instance_of, optional

from twisted.logger import Logger


__all__ = (
    "Connection",
    "Parameter",
    "QueryPlanExplanation",
    "Row",
    "SQLiteError",
    "createDB",
    "explainQueryPlans",
    "openDB",
    "printSchema",
)


TConnection = TypeVar("TConnection", bound="Connection")
TCursor     = TypeVar("TCursor", bound="Cursor")
TBaseCursor = TypeVar("TBaseCursor", bound="BaseCursor")

CursorFactory = Callable[..., TCursor]

Parameter = Optional[Union[bytes, str, int, float]]


class Row(BaseRow):
    """
    Subclass of :class:`sqlite3.Row` that has a :class:`dict`-like ``get``
    method.
    """

    def get(self, key: str, default: Any = None) -> Any:
        if key in self.keys():
            return self[key]
        else:
            return default



class Cursor(BaseCursor):
    """
    Subclass of `sqlite3.Cursor` that adds logging of SQL statements for
    debugging purposes.
    """

    _log = Logger()


    def executescript(self, sql_script: str) -> TCursor:
        """
        See :meth:`sqlite3.Cursor.executescript`.
        """
        self._log.debug("EXECUTE SCRIPT:\n{script}", script=sql_script)
        return cast(TCursor, super().executescript(sql_script))


    def execute(self, sql: str, parameters: Mapping = None) -> TCursor:
        """
        See :meth:`sqlite3.Cursor.execute`.
        """
        if parameters is None:
            parameters = {}
        self._log.debug(
            "EXECUTE: {sql} <- {parameters}", sql=sql, parameters=parameters
        )
        return cast(TCursor, super().execute(sql, parameters))



class Connection(BaseConnection):
    """
    Subclass of `sqlite3.Connection` that adds logging of SQL statements for
    debugging purposes and an improved row type.
    """

    _log = Logger()


    def cursor(
        self, factory: CursorFactory = cast(CursorFactory, Cursor)
    ) -> TCursor:
        return cast(TCursor, super().cursor(factory=factory))
<commit_msg>Add Parameters type, clean up docstrings.<commit_after># -*- test-case-name: ranger-ims-server.ext.test.test_sqlite -*-
"""
SQLite utilities
"""

from pathlib import Path
from sqlite3 import (
    Connection as BaseConnection, Cursor as BaseCursor, Error as SQLiteError,
    Row as BaseRow, connect as sqliteConnect,
)
from typing import (
    Any, Callable, Iterable, Mapping, Optional, Tuple, TypeVar, Union, cast
)
from typing.io import TextIO

from attr import attrib, attrs
from attr.validators import instance_of, optional

from twisted.logger import Logger


__all__ = (
    "Connection",
    "Parameter",
    "QueryPlanExplanation",
    "Row",
    "SQLiteError",
    "createDB",
    "explainQueryPlans",
    "openDB",
    "printSchema",
)


TConnection = TypeVar("TConnection", bound="Connection")
TCursor     = TypeVar("TCursor", bound="Cursor")
TBaseCursor = TypeVar("TBaseCursor", bound="BaseCursor")

CursorFactory = Callable[..., TCursor]

Parameter = Optional[Union[bytes, str, int, float]]
Parameters = Mapping[str, Parameter]


class Row(BaseRow):
    """
    Subclass of :class:`sqlite3.Row` that has a :class:`dict`-like ``get``
    method.
    """

    def get(self, key: str, default: Any = None) -> Any:
        if key in self.keys():
            return self[key]
        else:
            return default



class Cursor(BaseCursor):
    """
    Subclass of :class:`sqlite3.Cursor` that adds logging of SQL statements for
    debugging purposes.
    """

    _log = Logger()


    def executescript(self, sql_script: str) -> TCursor:
        """
        See :meth:`sqlite3.Cursor.executescript`.
        """
        self._log.debug("EXECUTE SCRIPT:\n{script}", script=sql_script)
        return cast(TCursor, super().executescript(sql_script))


    def execute(self, sql: str, parameters: Parameters = None) -> TCursor:
        """
        See :meth:`sqlite3.Cursor.execute`.
        """
        if parameters is None:
            parameters = {}
        self._log.debug(
            "EXECUTE: {sql} <- {parameters}", sql=sql, parameters=parameters
        )
        return cast(TCursor, super().execute(sql, parameters))



class Connection(BaseConnection):
    """
    Subclass of :class:`sqlite3.Connection` that adds logging of SQL statements
    for debugging purposes and an improved row type.
    """

    _log = Logger()


    def cursor(
        self, factory: CursorFactory = cast(CursorFactory, Cursor)
    ) -> TCursor:
        """
        See :meth:`sqlite3.Cursor.cursor`.
        """
        return cast(TCursor, super().cursor(factory=factory))

{'fraction_non_alphanumeric': 0.12577237392864263, 'fraction_numerical': 0.00259118995415587, 'mean_word_length': 2.963665086887836, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1319307', 'n_tokens_mistral': 1516, 'n_tokens_neox': 1463, 'n_words': 406}