coolqlite

A cool sqlite library.

coolqlite.connect(database: str, *, uri: bool = False, readonly: Literal[True] | None = None, row_factory: RowFactory[RowType] = Row, to_sql_fns: ToSqlFnRegistry | None = None, converter: BaseConverter | None = None) → Connection[RowType]

coolqlite.connect(database: Path, *, readonly: Literal[True] | None = None, row_factory: RowFactory[RowType] = Row, to_sql_fns: ToSqlFnRegistry | None = None, converter: BaseConverter | None = None) → Connection[RowType]

Connect to a sqlite database.

Works similarly to sqlite3.connect(), but has fewer flags exposed for the sake of good-practices defaults.

See package docs for details.

Parameters:

• database (str | Path) – The path to the sqlite db file, or you can use ":memory:" for an in-memory db.

• row_factory (RowFactory) – A custom RowFactory, defaults to Row.

• to_sql_fns (ToSqlFnRegistry | None) – The registry of sql serialization functions. If unspecified / None, uses the defaults for ToSqlConfig .

Returns:

The connection to the database, with the correct Row type as per row_factory .

Return type:

Connection

class coolqlite.Connection(private, connec: Connection, to_sql_fns: ToSqlFnRegistry, converter: BaseConverter)

Bases: Generic

A connection to a sqlite database.

Connect with connect(), not the constructor.

It is generic over the RowType configured with connect(), which defaults to Row.

Todo

More comprehensive Connection docs? Or should that be at the module level / in prose?

query(query: Template | AnalyzedQuery | BuiltQuery, type_: type[T], /) → QueryResults[T]

query(query: Template | AnalyzedQuery | BuiltQuery, /) → QueryResults[RowType]

Run the given query.

colquery(query: Template | AnalyzedQuery | BuiltQuery, /) → QueryResults[None | int | float | str | bytes]

colquery(query: Template | AnalyzedQuery | BuiltQuery, type_: type[T], /) → QueryResults[T]

run(query: Template | AnalyzedQuery | BuiltQuery)

Run a sql statement without looking at the results.

property user_version: int

Access to the user_version sqlite pragma.

You can use this to manage migrations in a simple fashion:

MIGRATIONS = [file.read_text() for file in MIGRATIONS_DIR.iterdir()]
with db:
    for migration in MIGRATIONS[db.user_version:]:
        db.run_script(migration)

    db.user_version = len(MIGRATIONS)

close()

Close the connection to the database. This will roll back any currently active transaction.

Note that using the Connection as a context manager will not close it at the end, since context management is used for transactions.

Like the stdlib suggests, you can instead use contextlib.closing() :

from contextlib import closing
from coolqlite import connect

with closing(connect(":memory:")) as db:
    with db: # transaction!
        ... # do stuff here

RUN_SCRIPT_DONT_USE_TEMPLATES(script: str)

Execute multiple statements. Does not give access to the cursor.

Warning

This may be removed or reworked in the future; I’m unsure about its design.

build(query: Template | AnalyzedQuery) → BuiltQuery

Build the query, converting its values.

See coolqlite.query for terminology, and coolqlite.query.build_query() for details.

with_savepoint() → Savepoint

Create a new savepoint and a context manager for it.

See the coolqlite.transactions module docs and Savepoint for details.

with_EXPERIMENTAL_SAVEPOINT() → AdvancedSavepoint

class coolqlite.Results(inner: Iterator[RowType] | Iterable[RowType])

class coolqlite.Results(inner: Iterator[Inner] | Iterable[Inner], mapping_fn: Callable[[Inner], RowType])

Bases: Iterator, Generic

An iterator of rows that offers easy ways to validate the number of returned rows, and is parameterized over the type of the row (from the row factory).

next() → RowType | None

Fetch the next row, or None if no rows remain.

one() → RowType

Return exactly one row, raising a QueryResultCountError if there’s zero or more than one.

If you want at least one, then use next(cursor.any()).

one_or_none() → RowType | None

Return exactly one or zero rows, raising a QueryResultCountError if there’s two or more.

Useful for checking for the presence or absence of something.

at_least_one() → Iterator

Get an iterator that is guaranteed to yield at least one row.

Raises a QueryResultCountError if there are none.

map(map: Callable[[RowType], R], /) → Results

Return a new Results that applies the given function to every row.

close()

If the underlying iterator is closeable (has a close method), and hasn’t been closed yet, close it.

See sqlite3.Cursor.close() if this was based off of a sqlite cursor..

coolqlite.as_results(fn: Callable[[P], Iterable]) → Callable[[P], Results]

Wrap a generator function so it returns a Results instead.

Useful for when you want to do some in-python processing over query results, but still want the caller to have control over how many rows they care about.

from coolqlite import as_results, Connection

@as_results
def advanced_query(db: Connection):
    # imagine this is something more complex
    # than could easily be done in SQL.
    for name in db.colquery(t"select name from Table", str):
        if name.startswith("Alex"):
            yield name[::-1]
        else:
            yield name

only_one = advanced_query(db).one()

class coolqlite.QueryResults(private, cursor: ~sqlite3.Cursor, mapping_fn: ~collections.abc.Callable[[...], RowType] = <function _identity_fn>)

Bases: Results, Generic

An extension of Results that allows access to cursor-specific information, like row count and lastrowid.

Note that the same semantics as in the sqlite3 module apply!

property rowcount

The underlying cursor’s rowcount.

property lastrowid

The underlying cursor’s lastrowid.

map(map: Callable[[RowType], R], /) → QueryResults

Return a new QueryResults that applies the given function to every row.

coolqlite.first_col(x: tuple[T, ...]) → T

Get the first item in a tuple.

This is only here to help with type checkers complaining about being passed a union type for colquery() :

You use query() instead, wrap the union in a tuple, and then give this to Results.map() .

result = db.query(t"select 1", tuple[int | None]).map(first_col)
typing.assert_type(result, QueryResults[int | None])

class coolqlite.Row

Bases: Row

Extends a sqlite3.Row with the following niceities:

• Easily allows accessing through a namespace

• Requesting a missing column by name raises a KeyError instead of an IndexError

• The repr is more readable

exception coolqlite.CoolqliteError

Bases: Exception

A base exception for all standard coolqlite errors.

exception coolqlite.CoolqliteExceptionGroup

Bases: ExceptionGroup[E], CoolqliteError, Generic

A base exceptiongroup for all standard coolqlite errors.

class coolqlite.ToSqlFnRegistry(*, config: ToSqlConfig = ToSqlConfig(configure_datetime_fns=True))

Bases: object

Dispatch to serializers based on name or type.

Register type-based dispatch as you would a functools.singledispatch():

registry = ToSqlFnRegistry()
@registry.register
def _handle_int(x: int) -> str:
    return str(x)

Or register by name:

registry = ToSqlFnRegistry()

@registry.add_named("reversed-int")
def _reversed_int(x: int) -> str:
    return str(x).reverse()

registry.add_named("upper", str.upper)

add_named(name: str) → Callable[[F], F]

add_named(name: str, fn: ToSqliteConversionFn) → None

Add a named converter. Usable as a decorator.

type coolqlite.RowFactory = Callable[[Cursor, tuple[None | int | float | str | bytes, ...]], Result]

exception coolqlite.ResultCountError(*, expected: _RowCount, actual: _RowCount)

Bases: CoolqliteError

A different number of rows than expected were returned.