Library Reference

Database

A convenient frontend for a sqlite3 database, with a few bells and whistles to save code.

If you have a Database called 'db', you can generally access any table in it via 'db.TABLE_NAME'. If the table name conflicts with an existing property or method, you can instead use 'db[TABLE_NAME]' as a fallback.

Source code in hissdb/db.py

class Database:
    """
    A convenient frontend for a sqlite3 database, with a few bells
    and whistles to save code.

    If you have a Database called 'db', you can generally access any
    table in it via 'db.TABLE_NAME'. If the table name conflicts with
    an existing property or method, you can instead use 'db[TABLE_NAME]'
    as a fallback.
    """
    def __init__(
        self,
        path: str,
        autoconnect: bool = True,
        autocommit: bool = True,
        verbose: bool = False,
    ):
        """
        Database constructor.

        Arguments:
            path: the file path for a new or existing database, or
                ':memory:' for an in-memory database with no
                corresponding file
            autoconnect: whether to autoconnect to the database whenever
                a query requires it. If this is False, you must manually
                run the connect() method or keep all database operations
                inside a 'with' statement.
            autocommit: whether to write all changes to the database at
                the close of a 'with' statement. Note that even if this
                is True, you must manually run the commit() method if
                you make changes *outside* of a context manager.
            verbose: whether to print each SQL statement to the console
                as it is executed
        """
        self._path = Path(path)
        self._verbose = verbose
        self._autoconnect = autoconnect
        self._autocommit = autocommit
        self._connection = None
        self._tables = {}

        if self._path.exists():
            self.connect()
            cur = self.execute('SELECT sql FROM sqlite_schema')
            schema_rows = cur.fetchall()
            for schema_row in schema_rows:
                if not schema_row[0]:
                    continue
                table = Table._from_schema(schema_row[0])
                self._tables[table._name] = table
                table._db = self
            if not self._autoconnect:
                self.disconnect()


    def __setattr__(self, attr: str, value):
        """
        If value is a Table object, do CREATE TABLE. Otherwise,
        do normal __setattr__ behavior.
        """
        if type(value) is not Table:
            super().__setattr__(attr, value)
            return

        self.__setitem__(attr, value)


    def __setitem__(self, item, value):
        assert type(value) is Table
        if item in self._tables:
            del self[item]
        value._name = item
        self.execute(value._schema)
        self._tables[item] = value
        value._db = self


    def __delattr__(self, attr):
        if type(getattr(self, attr)) is Table:
            self.execute(f'DROP TABLE {attr}')
            self._tables.pop(attr)
        else:
            super().__delattr__(attr)


    def __delitem__(self, item):
        "Remove a Table from the database"
        self.execute(f'DROP TABLE {item}')
        self._tables.pop(item)


    def __getattr__(self, attr: str):
        if '_tables' in self.__dict__ and attr in self._tables:
            return self[attr]
        else:
            raise AttributeError(
                f'{self} does not have any property or '
                f'table with the name "{attr}"'
            )


    def __getitem__(self, item) -> Table:
        if '.' in item:
            table, col = item.split('.')
            return self[table][col]
        elif '(' in item:
            table, col = item.split('(')
            return self[table][col[:-1]]
        else:
            return self._tables[item]


    def __contains__(self, item) -> bool:
        return item in self._tables


    def __repr__(self):
        return f"{__class__.__name__}('{self._path}')"


    def __enter__(self):
        """
        Context manager to handle connections to the database. If
        self._autocommit is True, then any changes are automatically
        committed at the close of the the 'with' statement.
        """
        self.connect(reuse_existing = True)
        return self


    def __exit__(self, exception_type, exception_value, traceback):
        if exception_type:
            self.disconnect(commit = False)
        else:
            self.disconnect()


    def create_table(self,
        name: str,
        columns: dict[str, str] = {},
        foreign_keys: dict[str, str] = {},
        primary_key: tuple[str] = (),
        if_not_exist: bool = False,
        **kwargs,
    ) -> Table:
        """
        Add a table to the database.

        Arguments:
            name: the name of the new table
            columns: a dict where each key is the name of a column, and
                each value contains the column constraints, e.g.
                {'id': 'INTEGER PRIMARY KEY', 'title': 'TEXT NOT NULL'}.
            foreign_keys: a dict where each key is the name of a column
                in this table, and each value represents a column in
                another table. Values can be specified as Column objects
                or as strings like 'users(id)' or 'users.id'.
            primary_key: optional tuple with the names of multiple
                columns that should be combined to form the primary key.
        """
        if if_not_exist and name in self._tables:
            return self._tables[name]
        table = Table(columns, foreign_keys, primary_key, **kwargs)
        self[name] = table
        return table


    def drop_table(self, name: str):
        """
        Delete the given table and its contents.
        Same as 'del self[name]'
        """
        del self[name]


    @property
    def _columns(self) -> dict[str, Column]:
        cols = {}
        for table in self._tables:
            cols.update(table._columns)
        return cols


    @property
    def connection(self):
        if self._connection:
            return self._connection
        elif self._autoconnect:
            return self.connect()
        else:
            raise AttributeError(
                f'{self} is not connected, and autoconnect is disabled,'
                f' so you must first run {self}.connect() before you '
                f'can perform this action'
            )


    def connect(self, reuse_existing: bool = True):
        """
        Connect to the database. Returns a sqlite3 connection object,
        but you should not need to use it.
        """
        if self._connection and reuse_existing:
            return self._connection
        else:
            self._connection = sqlite3.connect(self._path)
            return self._connection


    def commit(self):
        "Save recent changes to the database"
        self.connection.commit()


    def rollback(self):
        "Undo all changes since the last commit"
        self.connection.rollback()    


    def disconnect(self, commit: bool = 'AUTO'):
        """
        Close the connection to the database. If commit is False, roll
        back any changes. If commit is True, commit them.

        If commit is 'AUTO', only commit if self._autocommit is True, but
        don't rollback either way.
        """
        if commit == True or (commit == 'AUTO' and self._autocommit):
            self.commit()
        elif commit == False:
            self.rollback()
        self.connection.close()
        self._connection = None


    def execute(self,
        statement: str,
        placeholders: dict = {},
        many: bool = False,
    ):
        """
        Feed the given statement and placeholders to the execute() or
        executemany() method of this database's SQLite3 connection. If
        self._verbose is True, also print the executed statement.
        """
        if hasattr(statement, 'placeholders'):
            placeholders = copy(placeholders)
            placeholders.update(statement.placeholders)
        if many:
            func = self.connection.executemany
        else:
            func = self.connection.execute
        if self._verbose:
            render = statement
            if type(placeholders) is dict:
                for k, v in placeholders.items():
                    render = sub(rf':{k}(\b|$)', repr(v), render)
            print(f'{render};\n')

        return func(statement, placeholders)

__delitem__(item)

Remove a Table from the database

Source code in hissdb/db.py

def __delitem__(self, item):
    "Remove a Table from the database"
    self.execute(f'DROP TABLE {item}')
    self._tables.pop(item)

__enter__()

Context manager to handle connections to the database. If self._autocommit is True, then any changes are automatically committed at the close of the the 'with' statement.

Source code in hissdb/db.py

def __enter__(self):
    """
    Context manager to handle connections to the database. If
    self._autocommit is True, then any changes are automatically
    committed at the close of the the 'with' statement.
    """
    self.connect(reuse_existing = True)
    return self

__init__(path, autoconnect=True, autocommit=True, verbose=False)

Database constructor.

Parameters:

Name	Type	Description	Default
path	str	the file path for a new or existing database, or ':memory:' for an in-memory database with no corresponding file	required
autoconnect	bool	whether to autoconnect to the database whenever a query requires it. If this is False, you must manually run the connect() method or keep all database operations inside a 'with' statement.	True
autocommit	bool	whether to write all changes to the database at the close of a 'with' statement. Note that even if this is True, you must manually run the commit() method if you make changes outside of a context manager.	True
verbose	bool	whether to print each SQL statement to the console as it is executed	False

Source code in hissdb/db.py

def __init__(
    self,
    path: str,
    autoconnect: bool = True,
    autocommit: bool = True,
    verbose: bool = False,
):
    """
    Database constructor.

    Arguments:
        path: the file path for a new or existing database, or
            ':memory:' for an in-memory database with no
            corresponding file
        autoconnect: whether to autoconnect to the database whenever
            a query requires it. If this is False, you must manually
            run the connect() method or keep all database operations
            inside a 'with' statement.
        autocommit: whether to write all changes to the database at
            the close of a 'with' statement. Note that even if this
            is True, you must manually run the commit() method if
            you make changes *outside* of a context manager.
        verbose: whether to print each SQL statement to the console
            as it is executed
    """
    self._path = Path(path)
    self._verbose = verbose
    self._autoconnect = autoconnect
    self._autocommit = autocommit
    self._connection = None
    self._tables = {}

    if self._path.exists():
        self.connect()
        cur = self.execute('SELECT sql FROM sqlite_schema')
        schema_rows = cur.fetchall()
        for schema_row in schema_rows:
            if not schema_row[0]:
                continue
            table = Table._from_schema(schema_row[0])
            self._tables[table._name] = table
            table._db = self
        if not self._autoconnect:
            self.disconnect()

__setattr__(attr, value)

If value is a Table object, do CREATE TABLE. Otherwise, do normal setattr behavior.

Source code in hissdb/db.py

def __setattr__(self, attr: str, value):
    """
    If value is a Table object, do CREATE TABLE. Otherwise,
    do normal __setattr__ behavior.
    """
    if type(value) is not Table:
        super().__setattr__(attr, value)
        return

    self.__setitem__(attr, value)

commit()

Save recent changes to the database

Source code in hissdb/db.py

def commit(self):
    "Save recent changes to the database"
    self.connection.commit()

connect(reuse_existing=True)

Connect to the database. Returns a sqlite3 connection object, but you should not need to use it.

Source code in hissdb/db.py

def connect(self, reuse_existing: bool = True):
    """
    Connect to the database. Returns a sqlite3 connection object,
    but you should not need to use it.
    """
    if self._connection and reuse_existing:
        return self._connection
    else:
        self._connection = sqlite3.connect(self._path)
        return self._connection

create_table(name, columns={}, foreign_keys={}, primary_key=(), if_not_exist=False, **kwargs)

Add a table to the database.

Parameters:

Name	Type	Description	Default
name	str	the name of the new table	required
columns	dict[str, str]	a dict where each key is the name of a column, and each value contains the column constraints, e.g. {'id': 'INTEGER PRIMARY KEY', 'title': 'TEXT NOT NULL'}.	{}
foreign_keys	dict[str, str]	a dict where each key is the name of a column in this table, and each value represents a column in another table. Values can be specified as Column objects or as strings like 'users(id)' or 'users.id'.	{}
primary_key	tuple[str]	optional tuple with the names of multiple columns that should be combined to form the primary key.	()

Source code in hissdb/db.py

def create_table(self,
    name: str,
    columns: dict[str, str] = {},
    foreign_keys: dict[str, str] = {},
    primary_key: tuple[str] = (),
    if_not_exist: bool = False,
    **kwargs,
) -> Table:
    """
    Add a table to the database.

    Arguments:
        name: the name of the new table
        columns: a dict where each key is the name of a column, and
            each value contains the column constraints, e.g.
            {'id': 'INTEGER PRIMARY KEY', 'title': 'TEXT NOT NULL'}.
        foreign_keys: a dict where each key is the name of a column
            in this table, and each value represents a column in
            another table. Values can be specified as Column objects
            or as strings like 'users(id)' or 'users.id'.
        primary_key: optional tuple with the names of multiple
            columns that should be combined to form the primary key.
    """
    if if_not_exist and name in self._tables:
        return self._tables[name]
    table = Table(columns, foreign_keys, primary_key, **kwargs)
    self[name] = table
    return table

disconnect(commit='AUTO')

Close the connection to the database. If commit is False, roll back any changes. If commit is True, commit them.

If commit is 'AUTO', only commit if self._autocommit is True, but don't rollback either way.

Source code in hissdb/db.py

def disconnect(self, commit: bool = 'AUTO'):
    """
    Close the connection to the database. If commit is False, roll
    back any changes. If commit is True, commit them.

    If commit is 'AUTO', only commit if self._autocommit is True, but
    don't rollback either way.
    """
    if commit == True or (commit == 'AUTO' and self._autocommit):
        self.commit()
    elif commit == False:
        self.rollback()
    self.connection.close()
    self._connection = None

drop_table(name)

Delete the given table and its contents. Same as 'del self[name]'

Source code in hissdb/db.py

def drop_table(self, name: str):
    """
    Delete the given table and its contents.
    Same as 'del self[name]'
    """
    del self[name]

execute(statement, placeholders={}, many=False)

Feed the given statement and placeholders to the execute() or executemany() method of this database's SQLite3 connection. If self._verbose is True, also print the executed statement.

Source code in hissdb/db.py

def execute(self,
    statement: str,
    placeholders: dict = {},
    many: bool = False,
):
    """
    Feed the given statement and placeholders to the execute() or
    executemany() method of this database's SQLite3 connection. If
    self._verbose is True, also print the executed statement.
    """
    if hasattr(statement, 'placeholders'):
        placeholders = copy(placeholders)
        placeholders.update(statement.placeholders)
    if many:
        func = self.connection.executemany
    else:
        func = self.connection.execute
    if self._verbose:
        render = statement
        if type(placeholders) is dict:
            for k, v in placeholders.items():
                render = sub(rf':{k}(\b|$)', repr(v), render)
        print(f'{render};\n')

    return func(statement, placeholders)

rollback()

Undo all changes since the last commit

Source code in hissdb/db.py

def rollback(self):
    "Undo all changes since the last commit"
    self.connection.rollback()    

Table

Source code in hissdb/table.py

class Table:
    def __init__(
        self,
        columns: dict[str, Column] = {},
        foreign_keys: dict[str, Column] = {},
        primary_key: tuple[Column] = (),
        **kwargs,
    ):
        self._columns = {}
        kwargs.update(columns)
        for k, v in kwargs.items():
            col = Column(name = k, constraints = v)
            col._table = self
            self._columns[k] = col

        self._foreign_keys = foreign_keys
        self._primary_key = primary_key

        # these values are set when the table is assigned to a db
        self._name = None
        self._db = None


    @classmethod
    def _from_schema(cls, schema: str):
        # normalize schema to something parseable with small regex
        schema = sub(r'\s+', ' ', schema)
        schema = schema.replace('"', '').replace("'", '')
        schema = schema.replace('( ', '(').replace(' )', ')')

        # read column definitions and constraints from the schema
        name = schema.split(' ', 3)[2]
        columns, foreign_keys, = {}, {}
        primary_key = None
        for clause in schema.split('(', 1)[1][:-1].split(', '):
            if clause.startswith('FOREIGN KEY'):
                m = match(
                    r'FOREIGN KEY ?\(?(.+?)\)? REFERENCES (.+?) ?\((.+?)\)',
                    clause
                )
                key_name = m.group(1)
                foreign_table = m.group(2)
                foreign_key = m.group(3)
                foreign_keys[key_name] = f'{foreign_table}({foreign_key})'

            elif clause.startswith('PRIMARY KEY'):
                keys_str = schema.split('(', 1)[1][:-1].strip("'\"")
                primary_key = tuple(keys_str.split(', '))

            else:
                parts = clause.split(' ', 1)
                col_name = parts[0]
                constraints = parts[1] if len(parts) > 1 else None
                columns[col_name] = constraints

        table = cls(
            foreign_keys = foreign_keys,
            primary_key = primary_key,
            **columns
        )
        table._name = name
        return table


    def __getitem__(self, item):
        result = self._columns.get(item)
        if result:
            return result
        raise AttributeError(
            f"{self} has no column named '{item}'"
        )


    def __getattr__(self, attr):
        return self[attr]


    def __setitem__(self, item, value):
        value._table = self
        if not value._name:
            value._name = item
        self._db.execute(
            f'ALTER TABLE {self} ADD COLUMN {value._name}'
            + (f' {value._constraints}' if value._constraints else '')
        )
        self._columns[item] = value
        self._clear_cache()


    def __setattr__(self, attr, value):
        if type(value) is Column:
            self[attr] = value
        else:
            super().__setattr__(attr, value)


    def __delattr__(self, attr):
        if attr in self.__dict__:
            super().__delattr__(attr)
        else:
            del self[attr]


    def __delitem__(self, item):
        self._columns.pop(item)
        self._db.execute(f'ALTER TABLE {self} DROP COLUMN {item}')
        self._clear_cache()


    def __contains__(self, item):
        return item in self._columns


    def __str__(self):
        return self._name


    def __repr__(self):
        return f"{repr(self._db)}['{self._name}']"


    def count(self, where: Expression = None, **kwargs) -> int:
        """
        Get the number of rows in the table, optionally restricted to
        those that meet the given criteria.
        """
        return self.select(
            count(),
            where,
            **kwargs,
        ).execute().fetchone()[0]


    def fetchone(self, cols = '*', where = None, **kwargs):
        """
        Make and execute a Select statement from
        this table, and return the first result.
        """
        return self.select(cols, where, **kwargs).execute().fetchone()


    def fetchall(self, cols = '*', where = None, **kwargs):
        """
        Make and execute a Select statement from
        this table, and return a list of all results.
        """
        return self.select(cols, where, **kwargs).execute().fetchall()


    def insert(self, row: dict = {}, **kwargs) -> int:
        """
        Make and execute an Insert statement into this table, and return
        the index of the of the new row.
        """
        return Insert(table = self, row = row, **kwargs).execute().lastrowid


    def insertmany(
        self,
        cols: tuple[Column],
        rows: list[tuple],
        or_: str = None,
        **kwargs
    ) -> int:
        """
        Make and execute an InsertMany statement, and return the number
        of rows added.
        """
        return InsertMany(
            table = self,
            cols = cols,
            rows = rows,
            or_ = or_,
            **kwargs
        ).execute().rowcount


    def update(self,
        updates: dict[Column, Expression] = {},
        where: Expression = None,
        **kwargs,
    ) -> int:
        """
        Make and execute an Update statement from this table,
        and return the number of rows modified.
        """ 
        return Update(
            table = self,
            updates = updates,
            where = where,
            **kwargs,
        ).execute().rowcount


    def delete(self, where: Expression = None, **kwargs) -> int:
        """
        Make and execute a Delete statement from this table, and return
        the number of rows deleted.
        """
        return Delete(
            table = self,
            where = where,
            **kwargs
        ).execute().rowcount


    def select(self,
        cols: list[Column] = '*',
        where: Expression = None,
        **kwargs,
    ) -> Cursor:
        """
        Make and return a Select statement starting from this table.
        """
        statement = Select(
            table = self,
            cols = cols,
            where = where,
            **kwargs,
        )
        return statement


    @cached_property
    def _info(self):
        return self._db.connection.execute(
            f"PRAGMA TABLE_INFO({self._name})"
        ).fetchall()


    @cached_property
    def _schema(self):
        """
        If this table is in a database, get the schema from there.
        Otherwise, generate a schema from the list of columns, etc.
        """
        if self._db: # load schema from DB
            cur = self._db.connection.execute(
                "SELECT sql FROM sqlite_schema WHERE tbl_name = ?",
                (self._name,)
            )
            return cur.fetchone()[0]

        # if no DB, derive schema from values provided in __init__
        cols = []
        foreign_keys = []
        for col in self._columns.values():
            if col._constraints:
                cols.append(f'{col._name} {col._constraints}')
            else:
                cols.append(col._name)

        for key, value in self._foreign_keys.items():
            if type(value) is str:
                if '.' in value: # e.g. users.id
                    parts = value.split('.')
                elif '(' in value: # e.g. users(id)
                    parts = value.split('(')
                    parts[1] = parts[1][:-1] # strip closing paren
                else:
                    raise SyntaxError(
                        f'Unrecognized column reference "{value}"'
                    )
                col_ref = f'{parts[0]}({parts[1]})'
            else:
                col_ref = (
                    f'{value._table._name}'
                    f'({value._name})'
                )
            foreign_keys.append(
                f'FOREIGN KEY ({key}) REFERENCES {col_ref}'
            )

        clauses = cols + foreign_keys

        if self._primary_key:
            primary_key_cols = [str(col) for col in self._primary_key]
            clauses.append(f'PRIMARY KEY ({", ".join(primary_key_cols)})')

        return f'CREATE TABLE {self._name} ({", ".join(clauses)})'


    @cached_property
    def _foreign_keys(self):
        results = {}
        matches = finditer(
            (
                fr'FOREIGN KEY \((.+?)\) '
                r'REFERENCES ([^\s]+?) ?\((.+?)\)'
            ),
            self._schema
        )
        for match in matches:
            table = match.group(2)
            column = match.group(3)
            results[self[match.group(1)]] = self._db[table][column]
        return results


    def _clear_cache(self):
        for prop in ['_foreign_keys', '_schema', '_info']:
            if prop in self.__dict__:
                self.__dict__.pop(prop)

_schema() property cached

If this table is in a database, get the schema from there. Otherwise, generate a schema from the list of columns, etc.

Source code in hissdb/table.py

@cached_property
def _schema(self):
    """
    If this table is in a database, get the schema from there.
    Otherwise, generate a schema from the list of columns, etc.
    """
    if self._db: # load schema from DB
        cur = self._db.connection.execute(
            "SELECT sql FROM sqlite_schema WHERE tbl_name = ?",
            (self._name,)
        )
        return cur.fetchone()[0]

    # if no DB, derive schema from values provided in __init__
    cols = []
    foreign_keys = []
    for col in self._columns.values():
        if col._constraints:
            cols.append(f'{col._name} {col._constraints}')
        else:
            cols.append(col._name)

    for key, value in self._foreign_keys.items():
        if type(value) is str:
            if '.' in value: # e.g. users.id
                parts = value.split('.')
            elif '(' in value: # e.g. users(id)
                parts = value.split('(')
                parts[1] = parts[1][:-1] # strip closing paren
            else:
                raise SyntaxError(
                    f'Unrecognized column reference "{value}"'
                )
            col_ref = f'{parts[0]}({parts[1]})'
        else:
            col_ref = (
                f'{value._table._name}'
                f'({value._name})'
            )
        foreign_keys.append(
            f'FOREIGN KEY ({key}) REFERENCES {col_ref}'
        )

    clauses = cols + foreign_keys

    if self._primary_key:
        primary_key_cols = [str(col) for col in self._primary_key]
        clauses.append(f'PRIMARY KEY ({", ".join(primary_key_cols)})')

    return f'CREATE TABLE {self._name} ({", ".join(clauses)})'

count(where=None, **kwargs)

Get the number of rows in the table, optionally restricted to those that meet the given criteria.

Source code in hissdb/table.py

def count(self, where: Expression = None, **kwargs) -> int:
    """
    Get the number of rows in the table, optionally restricted to
    those that meet the given criteria.
    """
    return self.select(
        count(),
        where,
        **kwargs,
    ).execute().fetchone()[0]

delete(where=None, **kwargs)

Make and execute a Delete statement from this table, and return the number of rows deleted.

Source code in hissdb/table.py

def delete(self, where: Expression = None, **kwargs) -> int:
    """
    Make and execute a Delete statement from this table, and return
    the number of rows deleted.
    """
    return Delete(
        table = self,
        where = where,
        **kwargs
    ).execute().rowcount

fetchall(cols='*', where=None, **kwargs)

Make and execute a Select statement from this table, and return a list of all results.

Source code in hissdb/table.py

def fetchall(self, cols = '*', where = None, **kwargs):
    """
    Make and execute a Select statement from
    this table, and return a list of all results.
    """
    return self.select(cols, where, **kwargs).execute().fetchall()

fetchone(cols='*', where=None, **kwargs)

Make and execute a Select statement from this table, and return the first result.

Source code in hissdb/table.py

def fetchone(self, cols = '*', where = None, **kwargs):
    """
    Make and execute a Select statement from
    this table, and return the first result.
    """
    return self.select(cols, where, **kwargs).execute().fetchone()

insert(row={}, **kwargs)

Make and execute an Insert statement into this table, and return the index of the of the new row.

Source code in hissdb/table.py

def insert(self, row: dict = {}, **kwargs) -> int:
    """
    Make and execute an Insert statement into this table, and return
    the index of the of the new row.
    """
    return Insert(table = self, row = row, **kwargs).execute().lastrowid

insertmany(cols, rows, or_=None, **kwargs)

Make and execute an InsertMany statement, and return the number of rows added.

Source code in hissdb/table.py

def insertmany(
    self,
    cols: tuple[Column],
    rows: list[tuple],
    or_: str = None,
    **kwargs
) -> int:
    """
    Make and execute an InsertMany statement, and return the number
    of rows added.
    """
    return InsertMany(
        table = self,
        cols = cols,
        rows = rows,
        or_ = or_,
        **kwargs
    ).execute().rowcount

select(cols='*', where=None, **kwargs)

Make and return a Select statement starting from this table.

Source code in hissdb/table.py

def select(self,
    cols: list[Column] = '*',
    where: Expression = None,
    **kwargs,
) -> Cursor:
    """
    Make and return a Select statement starting from this table.
    """
    statement = Select(
        table = self,
        cols = cols,
        where = where,
        **kwargs,
    )
    return statement

update(updates={}, where=None, **kwargs)

Make and execute an Update statement from this table, and return the number of rows modified.

Source code in hissdb/table.py

def update(self,
    updates: dict[Column, Expression] = {},
    where: Expression = None,
    **kwargs,
) -> int:
    """
    Make and execute an Update statement from this table,
    and return the number of rows modified.
    """ 
    return Update(
        table = self,
        updates = updates,
        where = where,
        **kwargs,
    ).execute().rowcount

Column

Bases: Expression

A Column is a reference to a column in a SQLite database. Because it is also an Expression, many of its logical operators are overridden

so that you can build SQL Statements via Python logic like this

john_does = db.people.select(where= db.people.first_name == 'John' & db.people.last_name == 'Doe' )

For more information on this, see the Expression documentation.

Attributes:

Name	Type	Description
cid		column index number
name		name of the column
type		string containing the SQL datatype of this column
notnull		int representing whether the column disallows null vals
dflt_value		the column's default value
pk		int representing whether the column is a primary key

Source code in hissdb/column.py

class Column(Expression):
    """
    A Column is a reference to a column in a SQLite database. Because it
    is also an Expression, many of its logical operators are overridden
    so that you can build SQL Statements via Python logic like this:

        john_does = db.people.select(where=
            db.people.first_name == 'John'
            & db.people.last_name == 'Doe'
        )

    For more information on this, see the Expression documentation.

    Attributes:
        cid: column index number
        name: name of the column
        type: string containing the SQL datatype of this column
        notnull: int representing whether the column disallows null vals
        dflt_value: the column's default value
        pk: int representing whether the column is a primary key
    """

    _pragma_cols = ['cid', 'name', 'type', 'notnull', 'dflt_value', 'pk',]
    placeholders = {}

    def __init__(
        self,
        constraints: str = None,
        table = None,
        name: str = None,
    ):
        """
        Column object constructor.

        Arguments:
            constraints: a SQL expression defining this column, like
                'TEXT NOT NULL' or 'INTEGER PRIMARY KEY'
            table: the table containing this column. If not provided, it
                will be set when the column is assigned to a table with
                Table.__setattr__ or Table.__setitem__.
            name: the name of this column. If not provided, it will be
                set when this column is assigned to a table.

        """
        self._name = name
        self._constraints = constraints

        if table:
            table[name] = self

    def select(self, where: Expression = None, **kwargs) -> Cursor:
        """
        Convenience method to make a Select statement targeting only
        this column.
        """
        return self._table.select(
            cols = [self],
            where = where,
            **kwargs,
        )

    def fetchone(self, where: Expression = None, **kwargs):
        """
        Convenience method to execute a Select statement targeting only
        this column, and return the single resulting value (rather than
        a tuple with one item in it).
        """
        val = self.select(where, **kwargs).execute().fetchone()
        return val[0] if val else None

    def fetchall(self, where: Expression = None, **kwargs):
        """
        Convenience method to execute a Select statement targeting only
        this column, and return a list of the resulting values (rather
        than a list of one-item tuples).
        """
        vals = self.select(where, **kwargs).execute().fetchall()
        return [val[0] for val in vals]

    def update(self,
        new_value: Expression,
        where: Expression = None,
        **kwargs
    ):
        """
        Convenience method to execute an Update statement setting the
        value of this column, and return the number of rows modified.
        """
        return self._table.update(
            updates={self: new_value},
            where = where,
            **kwargs
        )

    def __str__(self):
        return f'{self._table}.{self._name}'

    def __repr__(self):
        return f"{repr(self._table)}['{self._name}']"

    def __hash__(self):
        return hash(str(self))

    def __getattr__(self, attr: str):
        if attr in self._pragma_cols:
            return self._info[self._pragma_cols.index(attr)]

    @property
    def _necessary_tables(self):
        return [self._table]

    @cached_property
    def _info(self):
        col_names = [r[1] for r in self._table._info]
        col_index = col_names.index(self._name)
        return self._table._info[col_index]

    @cached_property
    def _foreign_key(self):
        if self._name in self._table._foreign_keys:
            return self._table._foreign_keys[self]
        else:
            return None

__init__(constraints=None, table=None, name=None)

Column object constructor.

Parameters:

Name	Type	Description	Default
constraints	str	a SQL expression defining this column, like 'TEXT NOT NULL' or 'INTEGER PRIMARY KEY'	None
table		the table containing this column. If not provided, it will be set when the column is assigned to a table with Table.setattr or Table.setitem.	None
name	str	the name of this column. If not provided, it will be set when this column is assigned to a table.	None

Source code in hissdb/column.py

def __init__(
    self,
    constraints: str = None,
    table = None,
    name: str = None,
):
    """
    Column object constructor.

    Arguments:
        constraints: a SQL expression defining this column, like
            'TEXT NOT NULL' or 'INTEGER PRIMARY KEY'
        table: the table containing this column. If not provided, it
            will be set when the column is assigned to a table with
            Table.__setattr__ or Table.__setitem__.
        name: the name of this column. If not provided, it will be
            set when this column is assigned to a table.

    """
    self._name = name
    self._constraints = constraints

    if table:
        table[name] = self

fetchall(where=None, **kwargs)

Convenience method to execute a Select statement targeting only this column, and return a list of the resulting values (rather than a list of one-item tuples).

Source code in hissdb/column.py

def fetchall(self, where: Expression = None, **kwargs):
    """
    Convenience method to execute a Select statement targeting only
    this column, and return a list of the resulting values (rather
    than a list of one-item tuples).
    """
    vals = self.select(where, **kwargs).execute().fetchall()
    return [val[0] for val in vals]

fetchone(where=None, **kwargs)

Convenience method to execute a Select statement targeting only this column, and return the single resulting value (rather than a tuple with one item in it).

Source code in hissdb/column.py

def fetchone(self, where: Expression = None, **kwargs):
    """
    Convenience method to execute a Select statement targeting only
    this column, and return the single resulting value (rather than
    a tuple with one item in it).
    """
    val = self.select(where, **kwargs).execute().fetchone()
    return val[0] if val else None

select(where=None, **kwargs)

Convenience method to make a Select statement targeting only this column.

Source code in hissdb/column.py

def select(self, where: Expression = None, **kwargs) -> Cursor:
    """
    Convenience method to make a Select statement targeting only
    this column.
    """
    return self._table.select(
        cols = [self],
        where = where,
        **kwargs,
    )

update(new_value, where=None, **kwargs)

Convenience method to execute an Update statement setting the value of this column, and return the number of rows modified.

Source code in hissdb/column.py

def update(self,
    new_value: Expression,
    where: Expression = None,
    **kwargs
):
    """
    Convenience method to execute an Update statement setting the
    value of this column, and return the number of rows modified.
    """
    return self._table.update(
        updates={self: new_value},
        where = where,
        **kwargs
    )

Expressions

Expressions are the building blocks of Statements. They provide two key features: First, they parameterize input values to avoid SQL injection; and second, they provide a set of methods and overloaded logical operators (e.g. '==', '&', or '+') that allow users to build compound Expressions with Python syntax.

Attributes:

Name	Type	Description
tokens		the results of parameterizing each of the provided args. Any item that is not an Expression, Table, or Column object will be converted to a placeholder unless it is in the _literals whitelist.
placeholders		a dictionary of parameters that would need to be provided to sqlite3.execute() if this expression were a Statement of its own
necessary_tables		a list of Tables this expression references

Source code in hissdb/expression.py

class Expression:
    """
    Expressions are the building blocks of Statements. They provide two
    key features: First, they parameterize input values to avoid SQL
    injection; and second, they provide a set of methods and overloaded
    logical operators (e.g. '==', '&', or '+') that allow users to
    build compound Expressions with Python syntax.

    Attributes:
        tokens: the results of parameterizing each of the provided args.
            Any item that is not an Expression, Table, or Column object
            will be converted to a placeholder unless it is in the
            _literals whitelist.
        placeholders: a dictionary of parameters that would need to be
            provided to sqlite3.execute() if this expression were a
            Statement of its own
        necessary_tables: a list of Tables this expression references
    """    
    # strings that will not be converted to placeholders
    _literals = [
        '=', '==', '%', '>', '>=', '<', '<=', '!=', '!<', '!>', '~', '<>',
        '&', '||', '+', '-', '/', '*', '>>', '<<', 'LIKE', 'NOT', 'NOT LIKE',
        'NOT IN', 'SELECT', 'FROM', 'WHERE', 'IN', 'BETWEEN', 'NOT BETWEEN',
        'GLOB', 'EXISTS', 'NOT EXISTS', 'UNIQUE', 'NULL', 'NOT NULL', 'AND',
        'OR', 'AS', '(', ')', 'DISTINCT', 'ALL', 'ASC', 'DESC',
    ]

    def __init__(self, *args, func: str = None, prefix: str = None):
        """
        Arguments:
            args: the list of words or other values, in order, in the
                expression. Each one will be converted to a placeholder
                unless it is an Expression, Column, or Table object or
                is otherwise in the _literals whitelist. When the
                expression is converted to a string, the args (or their
                placeholders) will be joined with spaces.
            func: the SQL function, if any, that the expression
                implements. If a func is provided, then the rendered
                expression will be enclosed in parentheses, preceded by
                the func string, and each arg will be separated by
                commas instead of spaces, to make them into function
                arguments rather than just words.
            prefix: only relevant if func is provided. The prefix will
                be rendered just after the opening parentheses, before
                the first arg, with no separating comma. This is meant
                to allow for syntax like the "DISTINCT" argument for
                aggregate functions.
        """
        self.placeholders = {}
        self._necessary_tables = []
        self.args = args
        self.tokens = []
        self.func = func
        self.prefix = prefix

        for arg in args:
            if issubclass(arg.__class__, __class__):
                self.placeholders.update(arg.placeholders)
                self._necessary_tables += arg._necessary_tables
            elif arg.__class__.__module__ == 'hissdb.column':
                self._necessary_tables.append(arg._table)
            elif arg.__class__.__module__ == 'hissdb.table':
                self._necessary_tables.append(arg)

            elif arg is None:
                arg = 'NULL'
            elif type(arg) not in (int, float, str):
                raise SyntaxError(
                    f'Couldn\'t include "{arg}" in expression; '
                    f'no support for objects of type: {type(arg)}'
                )
            elif arg not in self._literals:
                placeholder = next_placeholder()
                self.placeholders[placeholder[1:]] = arg
                arg = placeholder

            # add parentheses to statements
            if arg.__class__.__module__ == 'hissdb.statements':
                arg = f'({str(arg)})'

            self.tokens.append(arg)

        for i, token in enumerate (self.tokens):
            if (
                token is '='
                and i+1 < len(self.tokens)
                and self.tokens[i+1] in ['NULL', 'NOT']
            ):
                self.tokens[i] = 'IS'

        self._necessary_tables = list(set(self._necessary_tables))


    def __str__(self):
        """
        Text of the expression that will be inserted into a SQL
        statement, with strings and integers replaced with placeholders
        unless they are listed in Expression._literals
        """
        joiner = ', ' if self.func else ' '
        output = joiner.join([str(t) for t in self.tokens])
        output = output.replace('( ', '(').replace(' )', ')')
        if self.prefix:
            output = f'{self.prefix} {output}'
        if self.func is not None:
            return f'{self.func}({output})'
        else:
            return output


    def __repr__(self):
        return (
            'Expression('
            + ', '.join([repr(t) for t in self.args])
            + (f", func='{self.func}'" if self.func else '')
            + (f", prefix='{self.prefix}'" if self.prefix else '')
            + ')'
        )

    @cached_property
    def _db(self):
        "Find the Database that this expression relates to"
        for token in [t for t in self.tokens if hasattr(t, '_db')]:
            if token._db:
                return token._db

    def render(self):
        "Text of the expression with placeholders filled in"
        text = str(self)
        for k, v in self.placeholders.items():
            if type(v) is str:
                v = f"'{v}'"
            elif type(v) is int:
                v = str(v)
            text = text.replace(f':{k}', v)
        return text


    # BITWISE OPERATORS

    def __and__(self, other):
        return __class__(self, 'AND', other)

    def __or__(self, other):
        return __class__('(', self, 'OR', other, ')')

    def __rshift__(self, other):
        return __class__(self, '>>', other)

    def __lshift__(self, other):
        return __class__(self, '<<', other)

    def __invert__(self):
        """
        Return an expression that is True if and only if this one is not
        True. This works by replacing operators with their inverses
        (e.g. replacing '>' with '<=').

        When the expression to be inverted contains two sub-expressions
        joined with AND or OR, the sub-expressions are both inverted,
        and the AND is replaced with OR, or vice versa.
        """
        args = list(copy(self.args))
        func = copy(self.func)
        if func:
            operator = func
        elif len(args) == 3 and type(args[1]) is str:
            operator = args[1]
        elif len(args) == 7 and args[3] == 'OR':
            operator = args[3]
            args = [args[1], args[3], args[5]]

        else:
            operator = None

        opposites = (
            ('LIKE', 'NOT LIKE'),
            ('IN', 'NOT IN'),
            ('BETWEEN', 'NOT BETWEEN'),
            ('IS', 'IS NOT'),
            ('EXISTS', 'NOT EXISTS'),
            ('<>', '='),
            ('==', '<>'),
            ('<', '>='),
            ('>', '<='),
            ('AND', 'OR'),  # also inverts sub-expressions, see below
        )
        for a, b in opposites:
            if operator not in (a, b):
                continue
            new_op = a if operator == b else b
            if func:
                func = new_op
            else:
                args[1] = new_op
            break
        else:
            raise NotImplementedError(
                f"Unsure how to invert expression '{str(self)}'"
            )

        if operator == 'OR':
            return ~args[0] & ~args[2]
        elif operator == 'AND':
            return ~args[0] | ~args[2]
        else:
            return __class__(*args, func=func)


    # COMPARISONS

    def __eq__(self, other):
        return __class__(self, '=', other)

    def __ne__(self, other):
        return __class__(self, '<>', other)

    def __gt__(self, other):
        return __class__(self, '>', other)

    def __lt__(self, other):
        return __class__(self, '<', other)

    def __ge__(self, other):
        return __class__(self, '>=', other)

    def __le__(self, other):
        return __class__(self, '<=', other)


    # ARITHMETIC OPERATORS

    def __add__(self, other):
        """
        Add two numbers or concatenate two strings
        """
        othertype = type_(other)
        selftype = type_(self)

        if selftype is str and othertype is str:
            return __class__(self, '||', other)
        else:
            return __class__(self, '+', other)

    def __sub__(self, other):
        return __class__(self, '-', other)

    def __mul__(self, other):
        return __class__(self, '*', other)

    def __div__(self, other):
        return __class__(self, '/', other)

    def __mod__(self, other):
        "LIKE operator for strings, modulo operator otherwise"
        if type_(self) is str:
            return __class__(self, 'LIKE', other)
        else:
            return __class__(self, '%', other)

    def __abs__(self):
        "SQLite ABS() function"
        return __class__(self, func='ABS')

    def max(self, distinct: bool = False):
        "SQLite MAX() function"
        return __class__(
            self, func='MAX', prefix='DISTINCT' if distinct else None
        )

    def min(self, distinct: bool = False):
        "SQLite MIN() function"
        return __class__(self, func='MIN')

    def avg(self, distinct: bool = False):
        "SQLite AVG() function"
        return __class__(
            self, func='AVG', prefix='DISTINCT' if distinct else None
        )

    def round(self):
        "SQLite ROUND() function"
        return __class__(self, func='ROUND')

    def ceil(self):
        "SQLite CEIL() function"
        return __class__(self, func='CEIL')

    def floor(self):
        "SQLite FLOOR() function"
        return __class__(self, func='FLOOR')

    def ln(self):
        "SQLite LN() function"
        return __class__(self, func='LN')

    def sqrt(self):
        "SQLite SQRT() function"
        return __class__(self, func='SQRT')

    def exp(self):
        "SQLite EXP() function"
        return __class__(self, func='EXP')

    def pow(self, exponent: int):
        "SQLite POWER() function"
        return __class__(self, exponent, func='POWER')

    # ROW OPERATORS

    def count(self, distinct: bool = False):
        "SQLite COUNT() function"
        prefix = 'DISTINCT' if distinct else None
        return __class__(self, func='COUNT', prefix=prefix)

    def exists(self):
        "SQLite EXISTS() function"
        return __class__(self, func='EXISTS')

    def in_(self, vals: list):
        if type(vals) in [list, tuple, set]:
            return __class__(self, 'IN', '(', *vals, ')')
        else:
            return __class__(self, 'IN', vals)

    # STRING OPERATORS

    def startswith(self, other):
        return __class__(self, 'LIKE', f'{other}%')

    def endswith(self, other):
        return __class__(self, 'LIKE', f'%{other}')

    def replace(self, find, repl):
        return __class__(self, find, repl, func='REPLACE')

    def length(self):
        return __class__(self, func='LENGTH')

    def lower(self):
        return __class__(self, func='LOWER')

    def upper(self):
        return __class__(self, func='UPPER')

    def substr(self, start: int, length: int):
        return __class__(self, start, length, func='SUBSTR')

    def strip(self, character: str = ' '):
        if character == ' ':
            return __class__(self, func='TRIM')
        else:
            return __class__(self, character, func='TRIM')

    def lstrip(self, character: str = ' '):
        if character == ' ':
            return __class__(self, func='LTRIM')
        else:
            return __class__(self, character, func='LTRIM')

    def rstrip(self, character: str = ' '):
        if character == ' ':
            return __class__(self, func='RTRIM')
        else:
            return __class__(self, character, func='RTRIM')

    def index(self, substr: str):
        if type_(self) is str:
            return __class__(self, substr, func='INSTR')
        else:
            raise NotImplementedError

    # MISC. CONVENIENCES

    @property
    def desc(self):
        "Shortcut for use in ORDER BY clauses"
        return __class__(self, 'DESC')

__abs__()

SQLite ABS() function

Source code in hissdb/expression.py

def __abs__(self):
    "SQLite ABS() function"
    return __class__(self, func='ABS')

__add__(other)

Add two numbers or concatenate two strings

Source code in hissdb/expression.py

def __add__(self, other):
    """
    Add two numbers or concatenate two strings
    """
    othertype = type_(other)
    selftype = type_(self)

    if selftype is str and othertype is str:
        return __class__(self, '||', other)
    else:
        return __class__(self, '+', other)

__init__(*args, func=None, prefix=None)

Parameters:

Name	Type	Description	Default
args		the list of words or other values, in order, in the expression. Each one will be converted to a placeholder unless it is an Expression, Column, or Table object or is otherwise in the _literals whitelist. When the expression is converted to a string, the args (or their placeholders) will be joined with spaces.	required
func	str	the SQL function, if any, that the expression implements. If a func is provided, then the rendered expression will be enclosed in parentheses, preceded by the func string, and each arg will be separated by commas instead of spaces, to make them into function arguments rather than just words.	None
prefix	str	only relevant if func is provided. The prefix will be rendered just after the opening parentheses, before the first arg, with no separating comma. This is meant to allow for syntax like the "DISTINCT" argument for aggregate functions.	None

Source code in hissdb/expression.py

def __init__(self, *args, func: str = None, prefix: str = None):
    """
    Arguments:
        args: the list of words or other values, in order, in the
            expression. Each one will be converted to a placeholder
            unless it is an Expression, Column, or Table object or
            is otherwise in the _literals whitelist. When the
            expression is converted to a string, the args (or their
            placeholders) will be joined with spaces.
        func: the SQL function, if any, that the expression
            implements. If a func is provided, then the rendered
            expression will be enclosed in parentheses, preceded by
            the func string, and each arg will be separated by
            commas instead of spaces, to make them into function
            arguments rather than just words.
        prefix: only relevant if func is provided. The prefix will
            be rendered just after the opening parentheses, before
            the first arg, with no separating comma. This is meant
            to allow for syntax like the "DISTINCT" argument for
            aggregate functions.
    """
    self.placeholders = {}
    self._necessary_tables = []
    self.args = args
    self.tokens = []
    self.func = func
    self.prefix = prefix

    for arg in args:
        if issubclass(arg.__class__, __class__):
            self.placeholders.update(arg.placeholders)
            self._necessary_tables += arg._necessary_tables
        elif arg.__class__.__module__ == 'hissdb.column':
            self._necessary_tables.append(arg._table)
        elif arg.__class__.__module__ == 'hissdb.table':
            self._necessary_tables.append(arg)

        elif arg is None:
            arg = 'NULL'
        elif type(arg) not in (int, float, str):
            raise SyntaxError(
                f'Couldn\'t include "{arg}" in expression; '
                f'no support for objects of type: {type(arg)}'
            )
        elif arg not in self._literals:
            placeholder = next_placeholder()
            self.placeholders[placeholder[1:]] = arg
            arg = placeholder

        # add parentheses to statements
        if arg.__class__.__module__ == 'hissdb.statements':
            arg = f'({str(arg)})'

        self.tokens.append(arg)

    for i, token in enumerate (self.tokens):
        if (
            token is '='
            and i+1 < len(self.tokens)
            and self.tokens[i+1] in ['NULL', 'NOT']
        ):
            self.tokens[i] = 'IS'

    self._necessary_tables = list(set(self._necessary_tables))

__invert__()

Return an expression that is True if and only if this one is not True. This works by replacing operators with their inverses (e.g. replacing '>' with '<=').

When the expression to be inverted contains two sub-expressions joined with AND or OR, the sub-expressions are both inverted, and the AND is replaced with OR, or vice versa.

Source code in hissdb/expression.py

def __invert__(self):
    """
    Return an expression that is True if and only if this one is not
    True. This works by replacing operators with their inverses
    (e.g. replacing '>' with '<=').

    When the expression to be inverted contains two sub-expressions
    joined with AND or OR, the sub-expressions are both inverted,
    and the AND is replaced with OR, or vice versa.
    """
    args = list(copy(self.args))
    func = copy(self.func)
    if func:
        operator = func
    elif len(args) == 3 and type(args[1]) is str:
        operator = args[1]
    elif len(args) == 7 and args[3] == 'OR':
        operator = args[3]
        args = [args[1], args[3], args[5]]

    else:
        operator = None

    opposites = (
        ('LIKE', 'NOT LIKE'),
        ('IN', 'NOT IN'),
        ('BETWEEN', 'NOT BETWEEN'),
        ('IS', 'IS NOT'),
        ('EXISTS', 'NOT EXISTS'),
        ('<>', '='),
        ('==', '<>'),
        ('<', '>='),
        ('>', '<='),
        ('AND', 'OR'),  # also inverts sub-expressions, see below
    )
    for a, b in opposites:
        if operator not in (a, b):
            continue
        new_op = a if operator == b else b
        if func:
            func = new_op
        else:
            args[1] = new_op
        break
    else:
        raise NotImplementedError(
            f"Unsure how to invert expression '{str(self)}'"
        )

    if operator == 'OR':
        return ~args[0] & ~args[2]
    elif operator == 'AND':
        return ~args[0] | ~args[2]
    else:
        return __class__(*args, func=func)

__mod__(other)

LIKE operator for strings, modulo operator otherwise

Source code in hissdb/expression.py

def __mod__(self, other):
    "LIKE operator for strings, modulo operator otherwise"
    if type_(self) is str:
        return __class__(self, 'LIKE', other)
    else:
        return __class__(self, '%', other)

__str__()

Text of the expression that will be inserted into a SQL statement, with strings and integers replaced with placeholders unless they are listed in Expression._literals

Source code in hissdb/expression.py

def __str__(self):
    """
    Text of the expression that will be inserted into a SQL
    statement, with strings and integers replaced with placeholders
    unless they are listed in Expression._literals
    """
    joiner = ', ' if self.func else ' '
    output = joiner.join([str(t) for t in self.tokens])
    output = output.replace('( ', '(').replace(' )', ')')
    if self.prefix:
        output = f'{self.prefix} {output}'
    if self.func is not None:
        return f'{self.func}({output})'
    else:
        return output

_db() property cached

Find the Database that this expression relates to

Source code in hissdb/expression.py

@cached_property
def _db(self):
    "Find the Database that this expression relates to"
    for token in [t for t in self.tokens if hasattr(t, '_db')]:
        if token._db:
            return token._db

avg(distinct=False)

SQLite AVG() function

Source code in hissdb/expression.py

def avg(self, distinct: bool = False):
    "SQLite AVG() function"
    return __class__(
        self, func='AVG', prefix='DISTINCT' if distinct else None
    )

ceil()

SQLite CEIL() function

Source code in hissdb/expression.py

def ceil(self):
    "SQLite CEIL() function"
    return __class__(self, func='CEIL')

count(distinct=False)

SQLite COUNT() function

Source code in hissdb/expression.py

def count(self, distinct: bool = False):
    "SQLite COUNT() function"
    prefix = 'DISTINCT' if distinct else None
    return __class__(self, func='COUNT', prefix=prefix)

desc() property

Shortcut for use in ORDER BY clauses

Source code in hissdb/expression.py

@property
def desc(self):
    "Shortcut for use in ORDER BY clauses"
    return __class__(self, 'DESC')

exists()

SQLite EXISTS() function

Source code in hissdb/expression.py

def exists(self):
    "SQLite EXISTS() function"
    return __class__(self, func='EXISTS')

exp()

SQLite EXP() function

Source code in hissdb/expression.py

def exp(self):
    "SQLite EXP() function"
    return __class__(self, func='EXP')

floor()

SQLite FLOOR() function

Source code in hissdb/expression.py

def floor(self):
    "SQLite FLOOR() function"
    return __class__(self, func='FLOOR')

ln()

SQLite LN() function

Source code in hissdb/expression.py

def ln(self):
    "SQLite LN() function"
    return __class__(self, func='LN')

max(distinct=False)

SQLite MAX() function

Source code in hissdb/expression.py

def max(self, distinct: bool = False):
    "SQLite MAX() function"
    return __class__(
        self, func='MAX', prefix='DISTINCT' if distinct else None
    )

min(distinct=False)

SQLite MIN() function

Source code in hissdb/expression.py

def min(self, distinct: bool = False):
    "SQLite MIN() function"
    return __class__(self, func='MIN')

pow(exponent)

SQLite POWER() function

Source code in hissdb/expression.py

def pow(self, exponent: int):
    "SQLite POWER() function"
    return __class__(self, exponent, func='POWER')

render()

Text of the expression with placeholders filled in

Source code in hissdb/expression.py

def render(self):
    "Text of the expression with placeholders filled in"
    text = str(self)
    for k, v in self.placeholders.items():
        if type(v) is str:
            v = f"'{v}'"
        elif type(v) is int:
            v = str(v)
        text = text.replace(f':{k}', v)
    return text

round()

SQLite ROUND() function

Source code in hissdb/expression.py

def round(self):
    "SQLite ROUND() function"
    return __class__(self, func='ROUND')

sqrt()

SQLite SQRT() function

Source code in hissdb/expression.py

def sqrt(self):
    "SQLite SQRT() function"
    return __class__(self, func='SQRT')

BaseStatement

Bases: Expression

base class that all Statements inherit from

Source code in hissdb/statements.py

class BaseStatement(Expression):
    "base class that all Statements inherit from"
    def __init__(self,
        table,
        where: Expression = None,
        join: dict = {},
        order_by: tuple[Expression] = None,
        limit: int = None,
        offset: int = None,
        # union: __class__ = None,
        # union_all: __class__ = None,
        # intersect: __class__ = None,
        autojoin: bool = True,
        **kwargs,
    ):
        """
        BaseStatement class constructor. There should be no need for a
        user to use this directly rather than a subclass, but all
        subclasses accept the following arguments in their constructors.

        Arguments:
            table: the Table object this statement relates to
            where: an Expression object representing a WHERE clause
                in SQL.
            join: a dictionary where each key is a Column object to join
                on the condition that the corresponding Expression is
                met.
            order_by: a Column, an Expression object representing a
                virtual column, or a tuple containing multiple of either
                of those things. To sort descending, you can use the
                'desc' property of any Column or Expression, e.g.
                "order_by=db.users.first_name.desc"
            limit: an int or an Expression representing an int, setting
                the maximum number of rows to select or modify
            offset: an int or an Expression representing an int, which
                sets which row to on.
            autojoin: whether HissDB should use the database's foreign
                key relationships to automatically join any tables that
                the statement requires. Defaults to True.
        """
        self.table = table
        self.where: Expression = where
        self.join = join
        self.order_by: tuple[Expression] = None
        self.limit: int = limit
        self.offset: int = offset
        # self.union = union
        # self.union_all = union_all
        # self.intersect = intersect

        self.autojoin: bool = autojoin
        self.unknown_kwargs = kwargs

        if order_by and type(order_by) not in [list, tuple, set]:
            self.order_by = [order_by]
        else:
            self.order_by = order_by

    def execute(self) -> Cursor:
        return self._db.execute(
            statement = str(self),
            placeholders = self.placeholders
        )

    def __str__(self):
        return '\n'.join(self.clauses)

    def __repr__(self):
        return (
            f'{self._db}.execute('
            f"'{self}', {self.placeholders})"
        )

    def __call__(self) -> Cursor:
        return self.execute()

    def contains(self, other):
        return Expression(other, Expression(self, func='IN '))

    @property
    def _db(self):
        return self.table._db

    @property
    def clauses(self) -> list[str]:
        if self.autojoin:
            joins = implicit_join(
                start_table = self.table,
                target_tables = self._necessary_tables,
                prior_joins = self.join,
            )
        else:
            joins = self.join

        if self.order_by:
            order_str = ', '.join([str(o) for o in self.order_by])

        return list(filter(lambda x: bool(x), [
            f'FROM {self.table}',
            *[f'JOIN {k} ON {v}' for k,v in joins.items()],
            (f'WHERE {self.where}' if self.where else ''),
            (f'ORDER BY {order_str}' if self.order_by else ''),
            (f'LIMIT {self.limit}' if self.limit else ''),
            (f'OFFSET {self.offset}' if self.offset else ''),
            # (f'UNION\n{self.union}' if self.union else ''),
            # (f'UNION ALL\n{self.union_all}' if self.union_all else ''),
            # (f'INTERSECT\n{self.intersect}' if self.intersect else ''),
        ]))

    @property
    def _necessary_tables(self) -> list:
        necessary_tables = []
        for val in self.__dict__.values():
            if not val:
                pass
            elif val.__class__.__module__ == 'hissdb.table':
                necessary_tables.append(val)
            elif hasattr(val, '_necessary_tables'):
                necessary_tables += val._necessary_tables
            elif type(val) is list and hasattr(val[0], '_necessary_tables'):
                for item in val:
                    necessary_tables += item._necessary_tables
            elif type(val) is dict:
                for item in [*val.keys(), *val.values()]:
                    if not hasattr(item, '_necessary_tables'):
                        continue
                    necessary_tables += item._necessary_tables
        return list(set(necessary_tables))

    @property
    def placeholders(self) -> dict:
        placeholders = {}
        for key, val in self.__dict__.items():
            if not val:
                pass
            elif hasattr(val, 'placeholders'):
                placeholders.update(val.placeholders)
            elif type(val) is list and hasattr(val[0], 'placeholders'):
                for item in val:
                    placeholders.update(item.placeholders)
            elif (
                type(val) is list and 
                len(val) > 0 and
                hasattr(val[0], 'placeholders')
            ):
                for item in val:
                    placeholders.update(item.placeholders)
            elif (
                type(val) is dict and
                len(val) > 0 and
                hasattr(list(val.values())[0], 'placeholders')
            ):
                for item in val.values():
                    placeholders.update(item.placeholders)
        return placeholders

    def _resolve_column(self, col_name: str) -> Column:
        if type(col_name) is str:
            if '.' in col_name:
                table, col = col_name.split('.')
                return self._db[table][col]
            else:
                return self.table[col_name]
        return col_name

__init__(table, where=None, join={}, order_by=None, limit=None, offset=None, autojoin=True, **kwargs)

BaseStatement class constructor. There should be no need for a user to use this directly rather than a subclass, but all subclasses accept the following arguments in their constructors.

Parameters:

Name	Type	Description	Default
table		the Table object this statement relates to	required
where	Expression	an Expression object representing a WHERE clause in SQL.	None
join	dict	a dictionary where each key is a Column object to join on the condition that the corresponding Expression is met.	{}
order_by	tuple[Expression]	a Column, an Expression object representing a virtual column, or a tuple containing multiple of either of those things. To sort descending, you can use the 'desc' property of any Column or Expression, e.g. "order_by=db.users.first_name.desc"	None
limit	int	an int or an Expression representing an int, setting the maximum number of rows to select or modify	None
offset	int	an int or an Expression representing an int, which sets which row to on.	None
autojoin	bool	whether HissDB should use the database's foreign key relationships to automatically join any tables that the statement requires. Defaults to True.	True

Source code in hissdb/statements.py

def __init__(self,
    table,
    where: Expression = None,
    join: dict = {},
    order_by: tuple[Expression] = None,
    limit: int = None,
    offset: int = None,
    # union: __class__ = None,
    # union_all: __class__ = None,
    # intersect: __class__ = None,
    autojoin: bool = True,
    **kwargs,
):
    """
    BaseStatement class constructor. There should be no need for a
    user to use this directly rather than a subclass, but all
    subclasses accept the following arguments in their constructors.

    Arguments:
        table: the Table object this statement relates to
        where: an Expression object representing a WHERE clause
            in SQL.
        join: a dictionary where each key is a Column object to join
            on the condition that the corresponding Expression is
            met.
        order_by: a Column, an Expression object representing a
            virtual column, or a tuple containing multiple of either
            of those things. To sort descending, you can use the
            'desc' property of any Column or Expression, e.g.
            "order_by=db.users.first_name.desc"
        limit: an int or an Expression representing an int, setting
            the maximum number of rows to select or modify
        offset: an int or an Expression representing an int, which
            sets which row to on.
        autojoin: whether HissDB should use the database's foreign
            key relationships to automatically join any tables that
            the statement requires. Defaults to True.
    """
    self.table = table
    self.where: Expression = where
    self.join = join
    self.order_by: tuple[Expression] = None
    self.limit: int = limit
    self.offset: int = offset
    # self.union = union
    # self.union_all = union_all
    # self.intersect = intersect

    self.autojoin: bool = autojoin
    self.unknown_kwargs = kwargs

    if order_by and type(order_by) not in [list, tuple, set]:
        self.order_by = [order_by]
    else:
        self.order_by = order_by

Insert

Bases: BaseStatement

SQL statement to insert a single row into a table

Source code in hissdb/statements.py

class Insert(BaseStatement):
    "SQL statement to insert a single row into a table"
    def __init__(self, table, row: dict = {}, or_: str = None, **kwargs):
        """
        Insert statement constructor. Any unknown keyword arguments will
        be added to the row dict.

        Arguments:
            table: Table object to insert the row into
            row: dict of values representing the row to insert
            or_: what to do when the insert statement fails due to a
                table constraint. Options are 'ABORT', 'FAIL', 'IGNORE',
                'REPLACE', and 'ROLLBACK'.
        """
        super().__init__(table=table, **kwargs)
        self.or_ = or_
        if self.unknown_kwargs:
            row = copy(row)
            row.update(self.unknown_kwargs)
        self.row = {k: Expression(v) for k,v in row.items()}


    @property
    def placeholders(self):
        placeholders = {}
        for val in self.row.values():
            if hasattr(val, 'placeholders'):
                placeholders.update(val.placeholders)
        return placeholders


    @property
    def clauses(self):
        return [
            'INSERT'
            + (f' OR {self.or_}' if self.or_ else '')
            + f' INTO {self.table} ({", ".join(self.row.keys())})'
            + f' VALUES ({", ".join([str(v) for v in self.row.values()])})'
        ] + super().clauses[1:] # skip the FROM clause

__init__(table, row={}, or_=None, **kwargs)

Insert statement constructor. Any unknown keyword arguments will be added to the row dict.

Parameters:

Name	Type	Description	Default
table		Table object to insert the row into	required
row	dict	dict of values representing the row to insert	{}
or_	str	what to do when the insert statement fails due to a table constraint. Options are 'ABORT', 'FAIL', 'IGNORE', 'REPLACE', and 'ROLLBACK'.	None

Source code in hissdb/statements.py

def __init__(self, table, row: dict = {}, or_: str = None, **kwargs):
    """
    Insert statement constructor. Any unknown keyword arguments will
    be added to the row dict.

    Arguments:
        table: Table object to insert the row into
        row: dict of values representing the row to insert
        or_: what to do when the insert statement fails due to a
            table constraint. Options are 'ABORT', 'FAIL', 'IGNORE',
            'REPLACE', and 'ROLLBACK'.
    """
    super().__init__(table=table, **kwargs)
    self.or_ = or_
    if self.unknown_kwargs:
        row = copy(row)
        row.update(self.unknown_kwargs)
    self.row = {k: Expression(v) for k,v in row.items()}

Select

Bases: BaseStatement

SQL statement to return some or all rows meeting given criteria

Source code in hissdb/statements.py

class Select(BaseStatement):
    "SQL statement to return some or all rows meeting given criteria"
    def __init__(self,
        table,
        cols: list[Column] = '*',
        where: Expression = None,
        group_by: list[Column] = None,
        having: Expression = None,
        **kwargs
    ):
        """
        Select statement constructor.

        Any unknown keyword arguments are interpreted as WHERE
        conditions constraining the value of a column in the given
        table. For instance, 'first_name="Jerry"' is equivalent to
        'where=[table].first_name == "Jerry"'

        Arguments:
            table: the Table object from which to select values
            cols: list of Column objects, or Expressions representing
                virtual columns, that the statement should select. For
                ease of use, columns can also be referenced via names
                rather than Column objects.
            where: Expression constraining which rows to select
            group_by: equivalent to SQL 'GROUP BY' clause
            having: equivalent to SQL 'HAVING' clause
        """
        super().__init__(table = table, where = where, **kwargs)
        self.raw_columns = cols
        for key, val in self.unknown_kwargs.items():
            new_criteria = Expression(self.table._columns[key], '=', val)
            if self.where:
                self.where = self.where & new_criteria
            else:
                self.where = new_criteria

        if self.raw_columns == '*' or type(self.raw_columns) is Expression:
            self.columns = self.raw_columns
        else:
            self.columns = [self._resolve_column(c) for c in self.raw_columns]

        if group_by:
            if type(group_by) in [list, set, tuple]:
                self.group_by = [self._resolve_column(c) for c in group_by]
            else:
                self.group_by = self._resolve_column(group_by)
        else:
            self.group_by = None
            if having:
                raise SyntaxError(
                    "statements can't include `having` without `group_by`"
                )
        self.having = having

    @property
    def clauses(self):
        if type(self.columns) in [list, tuple, set]:
            select = f'SELECT {", ".join([str(c) for c in self.columns])}'
        else:
            select = f'SELECT {self.columns}'
        clauses = [select] + super().clauses

        if self.group_by:
            if self.group_by:
                if type(self.group_by) in [list, tuple, set]:
                    group_by_clause = f'GROUP BY ({", ".join(self.group_by)})'
                else:
                    group_by_clause = f'GROUP BY {self.group_by}'

            # insert *before* LIMIT or ORDER BY clauses, if present
            group_by_clause = f'GROUP BY {self.group_by}'            
            for i, cl in enumerate(clauses):
                if cl.startswith('ORDER BY') or cl.startswith('LIMIT'):
                    clauses.insert(i, group_by_clause)
                    if self.having is not None:
                        clauses.insert(i+1, f'HAVING {self.having}')
                    break
            else:
                clauses.append(group_by_clause)
                if self.having is not None:
                    clauses.append(f'HAVING {self.having}')

        return clauses

    def __and__(self, other):
        return Intersect(self, other)

    def __or__(self, other):
        return Union(self, other)

__init__(table, cols='*', where=None, group_by=None, having=None, **kwargs)

Select statement constructor.

Any unknown keyword arguments are interpreted as WHERE conditions constraining the value of a column in the given table. For instance, 'first_name="Jerry"' is equivalent to 'where=[table].first_name == "Jerry"'

Parameters:

Name	Type	Description	Default
table		the Table object from which to select values	required
cols	list[Column]	list of Column objects, or Expressions representing virtual columns, that the statement should select. For ease of use, columns can also be referenced via names rather than Column objects.	'*'
where	Expression	Expression constraining which rows to select	None
group_by	list[Column]	equivalent to SQL 'GROUP BY' clause	None
having	Expression	equivalent to SQL 'HAVING' clause	None

Source code in hissdb/statements.py

def __init__(self,
    table,
    cols: list[Column] = '*',
    where: Expression = None,
    group_by: list[Column] = None,
    having: Expression = None,
    **kwargs
):
    """
    Select statement constructor.

    Any unknown keyword arguments are interpreted as WHERE
    conditions constraining the value of a column in the given
    table. For instance, 'first_name="Jerry"' is equivalent to
    'where=[table].first_name == "Jerry"'

    Arguments:
        table: the Table object from which to select values
        cols: list of Column objects, or Expressions representing
            virtual columns, that the statement should select. For
            ease of use, columns can also be referenced via names
            rather than Column objects.
        where: Expression constraining which rows to select
        group_by: equivalent to SQL 'GROUP BY' clause
        having: equivalent to SQL 'HAVING' clause
    """
    super().__init__(table = table, where = where, **kwargs)
    self.raw_columns = cols
    for key, val in self.unknown_kwargs.items():
        new_criteria = Expression(self.table._columns[key], '=', val)
        if self.where:
            self.where = self.where & new_criteria
        else:
            self.where = new_criteria

    if self.raw_columns == '*' or type(self.raw_columns) is Expression:
        self.columns = self.raw_columns
    else:
        self.columns = [self._resolve_column(c) for c in self.raw_columns]

    if group_by:
        if type(group_by) in [list, set, tuple]:
            self.group_by = [self._resolve_column(c) for c in group_by]
        else:
            self.group_by = self._resolve_column(group_by)
    else:
        self.group_by = None
        if having:
            raise SyntaxError(
                "statements can't include `having` without `group_by`"
            )
    self.having = having

Update

Delete

Bases: BaseStatement

SQL statement to delete some or all rows meeting given criteria

Source code in hissdb/statements.py

class Delete(BaseStatement):
    "SQL statement to delete some or all rows meeting given criteria"
    @property
    def clauses(self):
        clauses = super().clauses
        clauses[0] = 'DELETE ' + clauses[0] # makes 'DELETE FROM ...'
        return clauses

InsertMany

Bases: BaseStatement

SQL statement to efficiently insert a list or generator of rows

Source code in hissdb/statements.py

class InsertMany(BaseStatement):
    "SQL statement to efficiently insert a list or generator of rows"
    def __init__(self,
        table,
        cols: tuple[Column],
        rows: list[tuple],
        or_: str = None,
        **kwargs
    ):
        """
        InsertMany statement constructor.

        Arguments:
            cols: tuple of Column objects (or strings representing them)
                corresponding to the columns for which values will
                be provided.
            rows: list or generator containing each row to insert. A row
                is a tuple whose values each represent the corresponding
                value in cols.
            or_: what to do when the insert statement fails due to a
                table constraint. Options are 'ABORT', 'FAIL', 'IGNORE',
                'REPLACE', and 'ROLLBACK'.
        """
        super().__init__(table = table, **kwargs)
        self.or_ = or_
        self.cols = [self._resolve_column(col) for col in cols]
        self.rows = rows


    def execute(self) -> Cursor:
        return self._db.execute(
            statement = str(self),
            placeholders = self.rows,
            many=True
        )

    @property
    def clauses(self):
        return [
            'INSERT'
            + (f' OR {self.or_}' if self.or_ else '')
            +f' INTO {self.table} ({", ".join([c._name for c in self.cols])})'
            + f' VALUES ({", ".join(["?" for col in self.cols])})'
        ] + super().clauses[1:] # skip the FROM clause

__init__(table, cols, rows, or_=None, **kwargs)

InsertMany statement constructor.

Parameters:

Name	Type	Description	Default
cols	tuple[Column]	tuple of Column objects (or strings representing them) corresponding to the columns for which values will be provided.	required
rows	list[tuple]	list or generator containing each row to insert. A row is a tuple whose values each represent the corresponding value in cols.	required
or_	str	what to do when the insert statement fails due to a table constraint. Options are 'ABORT', 'FAIL', 'IGNORE', 'REPLACE', and 'ROLLBACK'.	None

Source code in hissdb/statements.py

def __init__(self,
    table,
    cols: tuple[Column],
    rows: list[tuple],
    or_: str = None,
    **kwargs
):
    """
    InsertMany statement constructor.

    Arguments:
        cols: tuple of Column objects (or strings representing them)
            corresponding to the columns for which values will
            be provided.
        rows: list or generator containing each row to insert. A row
            is a tuple whose values each represent the corresponding
            value in cols.
        or_: what to do when the insert statement fails due to a
            table constraint. Options are 'ABORT', 'FAIL', 'IGNORE',
            'REPLACE', and 'ROLLBACK'.
    """
    super().__init__(table = table, **kwargs)
    self.or_ = or_
    self.cols = [self._resolve_column(col) for col in cols]
    self.rows = rows