Convenient Wrapper around aiosql and a Database Connection.

Status Tests Coverage Python Databases Drivers Version Badges License


This class creates a persistent database connection and imports SQL queries from a file as simple Python functions.

If the connection is broken, a new connection is attempted with increasing throttling delays.

Compared to aiosql, the point is not to need to pass a connection as an argument on each call: The DB class embeds both connection and query methods.

For concurrent programming (threads, greenlets…), a relevant setup should also consider thread-locals and pooling issues at some higher level.


Install the module with pip install anodb or whatever method you like. Once available:

import anodb
# parameters: driver, connection string, SQL file
db = anodb.DB("sqlite3", "test.db", "test.sql")

db.do_some_insert(key=1, val="hello")
db.do_some_update(key=1, val="world")
print("data", db.do_some_select(key=1))


With file test.sql containing something like:

-- name: do_some_select
SELECT * FROM Stuff WHERE key = :key;

-- name: do_some_insert!
INSERT INTO Stuff(key, val) VALUES (:key, :val);

-- name: do_some_update!
UPDATE Stuff SET val = :val WHERE key = :key;


The anodb module provides the DB class which embeds both a PEP 249 database connection (providing methods commit, rollback, cursor, close and its connect counterpart to re-connect) and SQL queries wrapped into dynamically generated functions by aiosql. Such functions may be loaded from a string (add_queries_from_str) or a path (add_queries_from_path).

The DB constructor parameters are:

import anodb

db = anodb.DB("sqlite3", "acme.db", "acme-queries.sql")
db = anodb.DB("duckdb", "acme.db", "acme-queries.sql")
db = anodb.DB("psycopg", "host=localhost dbname=acme", "acme-queries.sql")
db = anodb.DB("psycopg", None, "acme-queries.sql", host="localhost", user="calvin", password="...", dbname="acme")
db = anodb.DB("psycopg2", "host=localhost dbname=acme", "acme-queries.sql")
db = anodb.DB("pygresql", None, "acme-queries.sql", host="localhost:5432", user="calvin", password="...", database="acme")
db = anodb.DB("pg8000", None, "acme-queries.sql", host="localhost", port=5432, user="calvin", password="...", database="acme")
db = anodb.DB("MySQLdb", None, "acme-queries.sql", host="localhost", port=3306, user="calvin", password="...", database="acme")
db = anodb.DB("pymysql", None, "acme-queries.sql", host="localhost", port=3306, user="calvin", password="...", database="acme")
db = anodb.DB("mysql-connector", None, "acme-queries.sql", host="localhost", port=3306, user="calvin", password="...", database="acme")
db = anodb.DB("mariadb", None, "acme-queries.sql", host="localhost", port=3306, user="calvin", password="...", database="acme")


This code is Public Domain.

All software has bug, this is software, hence… Beware that you may lose your hairs or your friends because of it. If you like it, feel free to send a postcard to the author.


Sources, documentation and issues are available on GitHub.

See all versions and get packages from PyPI.