Finally a PostgreSQL database library for the 2020s
Querying, paging, schema syncing, pubsub, creating, deleting, subsetting. Maximum enjoyment, minimum effort, minimum boilerplate.
Free for up to 5 users under the Fair Use 5 license. Or get a commercial subscription for more features and >5 users. More at Licensing & pricing.
A zero-boilerplate database object unlocks a heap of useful functionality.
import databaseci
# Use a local database name...
db = databaseci.db('example')
# Or any database URL
db = databaseci.db('postgresql:///example')
rows = db.q('select * from table')
print(rows)
Querying
Query the database directly from the object, or create transactions explicitly as required.
db.q('select * from table')
with db.t() as t:
rows = t.q('select * from table')
Schema syncing
Create/setup/populate/migrate databases idempotently using schema diffs.
# connect to the local database instance and create a new db
db = databaseci.db().create_db('newdb')
SCHEMA_SQL = """
create table names(
id serial primary key,
name text unique
);
"""
db.sync_schema_to(SCHEMA_SQL)
(Use with care - schema syncing can make potentially destructive schema changes to your database, such as dropping tables).
Paging
Offset-free, multi-column, multi-direction.
Using the table above, let's insert some sample rows:
db.insert(
'names',
[
dict(name='Alex'),
dict(name='Bob'),
dict(name='Charlie'),
dict(name='Dirk'),
dict(name='Emily')
],
upsert_on='name'
)
Then query a page at a time with a page size of two:
bookmark = None
while True:
page = db.q(
'select * from names',
paging=dict(
order_by='name desc, id',
per_page=2,
bookmark=bookmark
)
)
print(page)
if not page.paging.has_next:
break
bookmark = page.paging.next
"Keyset" paging queries pages via bookmarks, rather than a numeric offset, meaning the millionth page is just as fast to query as the first. For more, see the paging docs.
Pub/sub aka listen/notify
Listening requires a dedicated listening connection.
Simply create it and start looping through notifications.
db = databaseci.db('example')
with db.listen(
["channel_name"]
) as listening:
for n in db.notifications(
listening,
timeout=5
):
print(n)
Then notify:
db = databaseci.db('example')
db.notify('channel_name', 'payload')
Temporary databases
Create a temporary database using a local database:
from databaseci import temporary_local_db as temp_db
with temp_db() as db:
...
Or pull the databaseci/tempdb-pg
image to use the docker equivalent.
from databaseci import temporary_docker_db as temp_db
with temp_db() as db:
...