Usage¶
It is assumed that you have at least a little familiarity with sqlite. If you’re unfamiliar, consider checking out sqlitetutorial.net.
If you’re familiar, note that we turn on foreign key enforcement!
coolqlite should feel relatively familiar those used to the
stdlib’s sqlite3 module.
First, we connect(),
passing a filesystem path, or :memory:
for an in-memory database.
import coolqlite
db = coolqlite.connect(":memory:")
Let’s create a simple table to work on for some TTRPG nerdery.
What does it look like?¶
Let’s play with a database for TTRPG nerdery.
db.run(t"""create table CharacterClass
( id INTEGER NOT NULL PRIMARY KEY
, name TEXT NOT NULL UNIQUE
) strict""")
Note the following:
Even though we didn’t bind any parameters, we still used a t-string. Coolqlite insists on this to avoid the mistake of using an f-string when you do mean to bind parameters.
We made a strict table, because type checking is our friend.
We said
NOT NULLfor the primary key.
Sqlite allows null primary keys by default.. Technically this isn’t necessary when using a strict table, but let’s call it out.We used
run(). Userun()when you don’t care about the output.
Let’s insert some data:
for class_ in "manifold timepiece growth drain dream".split():
db.run(t"insert into CharacterClass(name) values ({class_})")
We did some interpolation there, but coolqlite analyzes the t-string to properly parameterize the query.
How does reading data work?¶
Putting data into a database and never reading it isn’t so useful. So let’s see all the ways to get the data out!
from coolqlite import Results
def get_all_classes() -> Results:
return db.query(t"select id, name from CharacterClass")
def get_classes_like(pattern: str) -> Results:
return db.query(t"select * from CharacterClass where name LIKE {pattern}")
def get_class_named(name: str) -> Results:
return db.query(t"select id, name from CharacterClass where name = {pattern}")
Right now, we just care about looking at every row in the
CharacterClass table. But many times, you care about
the number of results you get, and anything different is an error.
Resultshas many ways to look at the rows:
Iterate over however many rows there are, because it’s an iterable.
Get whatever the next one is (or
Noneif done), withResults.next().Get exactly one row with
Results.one(), no more and no less, throwing an exception otherwise.Get exactly one or exactly none with
Results.one_or_none(). The signature looks the same asResults.next(), but it will throw an exception if there are two or more rows.Get at least one row, allowing more, with
Results.at_least_one().
The above mentioned exception would be a
QueryResultCountError.
coolqlite.Row is a subclass of
sqlite3.Row, but it has a nicer repr and
will properly throw a KeyError if a column is missing.
Let’s try some of them!
from contextlib import suppressing
from coolqlite import QueryResultCountError
id_ = get_class_named("manifold").one()["id"]
# uhh there might have been one named this?
result = get_class_named("timekeeper").one_or_none()
assert result is None # guess not.
with suppressing(QueryResultCountError):
get_class_named("gunk").one()
assert False
# no gunk :(
# there might be a lot of these, let's do it one at a time.
results = get_all_classes()
row = results.next()
if row is not None:
row = results.next()
# this is technically the same as next(iter(results), None),
# but a little more convenient.
# Maybe a user types `d` in a search box
# and you want to suggest completions:
for id_, name in get_classes_like("d%"):
print(f"{name} has id {id_}")
(It’s technically a subclass of Results called QueryResults, which
adds access to rowcount and lastrowid.)
What about inserting things other than strs and ints?¶
Sqlite understands some basic python types natively,
but others need converting.
A Connection keeps track of converter functions
that work either by type or by name.
Type Converters¶
This works the same way as a functools.singledispatch().
For a simple case, let’s store UUIDs as strings in the database:
from coolqlite import ToSqlFnRegistry, connect
from uuid import UUID
funcs = ToSqlFnRegistry()
# Also works as a decorator, see singledispatch docs.
funcs.register(UUID, str)
db = connect(":memory":, to_sql_fns=funcs)
Named Converters¶
Named converters called by name in the t-string, where the “format specifier” would be normally:
db.run(t"insert into Table(column) values({some_value:converter_name_here})")
Let’s start with a common example–
very few applications can avoid datetimes.
Here’s where we encounter an opinionated default of coolqlite:
“In the face of ambiguity, refuse the temptation to guess.”
– The Zen Of Python
We offer easy-to-use helpers that are named converters
by default. They turn temporal values into iso-formatted
strings. (With a space instead of T for datetimes).
You could easily wire them up as type converters if you want!
This is a format sqlite recognizes as a time value, and is easy to read when querying the database.
There’s some reasons behind this, but the short version is:
Sqlite temporal functions assume everything is UTC unless there’s an explicit offset. They also assume an integer is a julian day counter instead of a unix epoch, unless explicitly told otherwise.
For storing certain events, especially in the future, the wall-clock time might actually be the only correct one! (Read that link for details.)
Maybe this feels like a bit much for “Basic Usage”, but it’s unavoidable.
Date & Time Examples¶
Let’s get set up:
>>> from zoneinfo import ZoneInfo
>>> import coolqlite
>>> import datetype
>>>
>>> def set_up_table(db: coolqlite.Connection):
... db.run(t"""create table JournalEntry
... ( created_at TEXT NOT NULL default (concat(current_timestamp, 'Z'))
... , about_dt TEXT NOT NULL
... , note TEXT NOT NULL default ''
... ) strict""")
>>>
>>> db = coolqlite.connect(":memory:")
>>> set_up_table(db)
>>>
>>> NY = ZoneInfo("America/New_York")
>>> def add_entries(db: coolqlite.Connection):
... # db would handle this for us, but why not.
... created_at = datetype.AwareDateTime.now(NY)
... about_dt = datetime(2024, 6, 25, 9, 23, tzinfo=NY)
... note = "A really cool game came out on this day!"
...
... db.run(
... t"insert into JournalEntry values ({created_at:z}, {about_dt:keeptz}, {note})"
... )
...
... created_at = datetype.AwareDateTime.now(NY)
... # events far in the future will likely have timezone changes by then,
... # so if the wall-clock time is important, DON'T store it as offset-aware!
... about_dt = datetime(2112, 6, 7, 12, 00)
... note = "At noon, meet friend coming back from mars, where they drive in fancy cars."
...
... db.run(
... t"insert into JournalEntry values ({created_at:z}, {about_dt:naive}, {note})"
... )
>>>
>>> add_entries(db)
And let’s get the data back out:
>>> for journal_entry in db.query(t"select * from JournalEntry"):
... print(journal_entry)
<row with columns created_at='2026-02-23 17:29:03.168339Z', about_dt='2024-06-25 09:23:00-04:00', note='A really cool game came out on this day!'>
<row with columns created_at='2026-02-23 17:29:03.168514Z', about_dt='2112-06-07 12:00:00', note='At noon, meet friend coming back from mars, where they drive in fancy cars.'>
Look at what each converter did:
zconverted the aware datetime to UTCkeeptzkept the offsetnaiveallowed a datetime without an offset
There’s also ANY_temporal for when you just want
it in iso-format, aware or not.
We put in temporal types, and got back out strings. Can we get those back to how they were?
Converting when reading¶
You could convert manually on each field when you access them on the Row:
>>> from coolqlite.from_sql.datetime_ import aware_dt_from_iso, dt_from_iso
>>>
>>> def describe_entry(created_at: datetime, about_dt: datetime, note: str):
... print(f"On {created_at:%c}, we wrote the following about {about_dt:%c}:\n{note}")
>>>
>>> for row in db.query(t"select created_at, about_dt, note from JournalEntry"):
... created_at = aware_dt_from_iso(row["created_at"])
... about_dt = dt_from_iso(row["about_dt"])
... note = row["note"]
... describe_entry(created_at, about_dt, note)
On Mon Feb 23 17:29:03 2026, we wrote the following about Tue Jun 25 09:23:00 2024:
A really cool game came out on this day!
On Mon Feb 23 17:29:03 2026, we wrote the following about Tue Jun 7 12:00:00 2112:
At noon, meet friend coming back from mars, where they drive in fancy cars.
But that’s a little tedious.
Luckily, we have the power of cattrs on our side.
Anything it can convert from a mapping or tuple, we can convert a row into.
>>> from dataclasses import dataclass
>>> @dataclass
>>> class JournalEntry:
... created_at: datetype.AwareDateTime
... # could also just say datetype.DateTime.
... about_dt: datetype.AwareDateTime | datetype.NaiveDateTime
... note: str
... def __str__(self):
... return f"On {self.created_at:%c}, we wrote the following about {self.about_dt:%c}:\n{self.note}"
>>> for entry in db.query(t"select * from JournalEntry", JournalEntry):
... print(entry)
On Mon Feb 23 17:29:03 2026, we wrote the following about Tue Jun 25 09:23:00 2024:
A really cool game came out on this day!
On Mon Feb 23 17:29:03 2026, we wrote the following about Tue Jun 7 12:00:00 2112:
At noon, meet friend coming back from mars, where they drive in fancy cars.
Just one column¶
Many times, you’ll want to select just a single column. There’s a shortcut for that:
>>> from datetype import DateTime
>>> future_entry = db.colquery(
... t"select about_dt from JournalEntry where note like '%mars%'", DateTime
>>> ).one()
>>> print(future_entry)
2112-06-07 12:00:00
If you don’t specify a type at the end, it will fall back to a SqliteType,
i.e. any of the primitives sqlite supports.
Type Checking¶
We try to make sure type checkers stay happy. Unfortunately, there’s 2 weaknesses we’ve found so far.
First, typing.NewType.
Cattrs can handle them,
but since they’re not a “type” per se, type checkers will complain when you
use them for colquery.
You’ve got two ways around this:
map:
UserId = typing.NewType("UserId", int) results = db.colquery(t"select id from users", int).map(UserId) typing.assert_type(results, QueryResults[UserId])
map, but lambda:
results = db.colquery(t"select id from users", int).map(lambda x: UserId(x))
MyPy will accept either, but pyright doesn’t like the first, sadly.
Second, unions. They’re also not a “real type” apparently.
Luckily, we can “trick” the type checkers by
query()ing a tuple
of that type and unpacking the first column (not colquerying).
We even offer a helper for that.
from coolqlite import first_col
results = db.query(t"select 1", tuple[int | None]).map(first_col)
typing.assert_type(results, QueryResults[int | None])
Transactions¶
You are a regular guy! You can do that transaction! – Jacob Wysocki
Because the connection is in autocommit mode, you don’t need transactions for one-off changes.
However, if you’re making multiple changes that need to either all succeed or all fail, you’ll want to do that in a transaction.
coolqlite does transactions through sqlite’s
savepoints.
They’re more flexible and powerful than plain
BEGIN..COMMIT/ROLLBACK.
At its simplest, it works the same as how the stdlib module
uses connections for transactions, but instead with a separate object.
For the following examples, let’s use a simple setup:
from coolqlite import connect
db = connect(":memory:")
db.run(t"create table Names(name TEXT NOT NULL UNIQUE) strict")
def has_name(name: str) -> bool:
return 1 == db.colquery(t"select count(*) from Names where name = {name}", int).one()
Using a Savepoint context manager
from with_savepoint()
will “commit” the data if it exits without error.
with db.with_savepoint():
db.run(t"insert into Names values ('foo')")
assert has_name("foo")
If there’s an exception, the transaction will be rolled back.
try:
with db.with_savepoint():
db.run(t"insert into Names values ('bar')")
raise Exception("something bad!")
except Exception:
pass
assert not has_name("bar")
Inspecting Queries And A Neat Shortcut¶
Queries first get “analyzed”, turning a template into something structured. They are then “built” by converting each value.
You can inspect at either stage of the journey, and supply either result as an argument for query methods. Names may be mangled.
>>> from coolqlite.query import analyze_query
>>> from pprint import pprint
>>> dt = datetime(2112, 6, 7, 12, 00)
>>> analyzed = analyze_query(t"select {dt:naive}")
>>> print(f"query: {analyzed.query_str!r}")
>>> for param in analyzed.params.values():
... pprint(param)
>>> # or could put the template here directly.
>>> built = db.build(analyzed)
>>> assert analyzed.query_str == built.query_str
>>> for param in built.params.values():
... pprint(param)
query: 'select :_CQL_dt_naive'
Param(param_name='_CQL_dt_naive',
original_param_expr='dt',
converter_name='naive',
original_value=datetime.datetime(2112, 6, 7, 12, 0))
ConvertedParam(param_name='_CQL_dt_naive',
original_param_expr='dt',
converter_name='naive',
original_value=datetime.datetime(2112, 6, 7, 12, 0),
converted_value='2112-06-07 12:00:00')
This is mainly useful for debugging, but there’s a neat trick you can do: make functions that mark the expected input types and names, and give you an analyzed query and the type it should return.
def get_ids_and_names(status: str):
query = analyze_query(t"select id, name from Table where status = {status}")
row_type = tuple[int, str]
return query, row_type
for id_, name in db.query(*get_ids_and_names("active")):
...
Programmactically Modifying Query Text¶
For when you programmactically need to build a query, you can get around the parameter-binding behavior.
The built-in named converter DANGER will let you do this.
Danger
If you directly use user input for this, you are likely vulnerable to SQL Injection.
If you’re using user input for something like selecting columns to filter by, always always always parse that input into something controlled and limited, like an enum.
You have been warned.
>>> from enum import StrEnum
>>> class SelectableColumn(StrEnum):
... CREATED_AT = "created_at"
... ABOUT_DT = "about_dt"
... NOTE = "note"
>>> class ComparisonOperator(StrEnum):
... EQ = "="
... LIKE = "like"
>>> # say this comes from a web form somewhere.
>>> def select_from_user_condition(column_UNTRUSTED: str, op_UNTRUSTED: str, condition: str):
... column = SelectableColumn(column_UNTRUSTED)
... op = ComparisonOperator(op_UNTRUSTED)
... return analyze_query(t"""select * from JournalEntry
... where {column:DANGER} {op:DANGER} {condition}""")
>>> select_from_user_condition("note", "like", "%mars%").query_str
'select * from JournalEntry where note like :condition'
>>> select_from_user_condition("' or TRUE; --", "", "").query_str
ValueError: "' or TRUE; --" is not a valid SelectableColumn
You could even put all of the inputs into a dataclass, and have cattrs convert the input into it, so you can get the errors in parallel.
Neat!
Reading Redux¶
Taking Advantage of Results¶
If you need to do some more advanced logic than SQL will allow,
you might iterate over rows and yield some derived data.
What’s annoying is you then lose the Results niceities from that.
But guess what?
You can use any iterator (and thus iterable or generator) with Results.
We offer a
as_results()decorator to make that even easier.
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()
Note that information from QueryResults can’t be passed along,
but the information might not be accurate anymore based on your transformation anyway.