anosql

NOTICE: This project is now deprecated in favor of aiosql.

Unfortunately, I no longer have the time to devote to this project, and aiosql is now a lot more popular. I don't think it makes sense to maintain both. Open source ftw! Thanks for your hard work, Will!

A Python library for using SQL

Inspired by the excellent Yesql library by Kris Jenkins. In my mother tongue, ano means yes.

If you are on python3.6+ or need anosql to work with asyncio-based database drivers, see the related project, aiosql.

Complete documentation is available at Read The Docs.

Installation

$ pip install anosql

Usage

Basics

Given a queries.sql file:

-- name: get-all-greetings
-- Get all the greetings in the database
SELECT * FROM greetings;

-- name: select-users
-- Get all the users from the database,
-- and return it as a dict
SELECT * FROM USERS;

We can issue SQL queries, like so:

import anosql
import psycopg2
import sqlite3

# PostgreSQL
conn = psycopg2.connect('...')
queries = anosql.from_path('queries.sql', 'psycopg2')

# Or, Sqlite3...
conn = sqlite3.connect('cool.db')
queries = anosql.from_path('queries.sql', 'sqlite3')

queries.get_all_greetings(conn)
# => [(1, 'en', 'Hi')]

queries.get_all_greetings.__doc__
# => Get all the greetings in the database

queries.get_all_greetings.sql
# => SELECT * FROM greetings;

queries.available_queries
# => ['get_all_greetings']

Parameters

Often, you want to change parts of the query dynamically, particularly values in the WHERE clause. You can use parameters to do this:

-- name: get-greetings-for-language
-- Get all the greetings in the database for given language
SELECT *
FROM greetings
WHERE lang = %s;

And they become positional parameters:

visitor_language = "en"
queries.get_greetings_for_language(conn, visitor_language)
# => [(1, 'en', 'Hi')]

One Row Query

Often, you would expect at most one row from a query, so that getting a list is not convenient. Appending ? to the query name makes it return either one tuple if it returned one row, or None in other cases.

-- name: get-a-greeting?
-- Get a greeting based on its id
SELECT *
FROM greetings
WHERE id = %s;

Then a tuple is returned:

queries.get_a_greeting(conn, 1)
# => (1, 'en', 'Hi')

Named Parameters

To make queries with many parameters more understandable and maintainable, you can give the parameters names:

-- name: get-greetings-for-language-and-length
-- Get all the greetings in the database for given language and length
SELECT *
FROM greetings
WHERE lang = :lang
AND len(greeting) <= :length_limit;

If you were writing a Postgresql query, you could also format the parameters as %s(lang) and %s(length_limit).

Then, call your queries like you would any Python function with named parameters:

visitor_language = "en"

greetings_for_texting = queries.get_greetings_for_language_and_length(
              conn, lang=visitor_language, length_limit=140)

Update/Insert/Delete

In order to run UPDATE, INSERT, or DELETE statements, you need to add ! to the end of your query name. Anosql will then execute it properly. It will also return the number of affected rows.

Insert queries returning autogenerated values

If you want the auto-generated primary key to be returned after you run an insert query, you can add <! to the end of your query name.

-- name: create-user<!
INSERT INTO person (name) VALUES (:name)

Adding custom query loaders.

Out of the box, anosql supports SQLite and PostgreSQL via the stdlib sqlite3 database driver and psycopg2. If you would like to extend anosql to communicate with other types of databases, you may create a driver adapter class and register it with anosql.core.register_driver_adapter().

Driver adapters are duck-typed classes which adhere to the below interface. Looking at anosql/adapters package is a good place to get started by looking at how the psycopg2 and sqlite3 adapters work.

To register a new loader:

import anosql
import anosql.core

class MyDbAdapter():
    def process_sql(self, name, op_type, sql):
        pass

    def select(self, conn, sql, parameters):
        pass

    @contextmanager
    def select_cursor(self, conn, sql, parameters):
        pass

    def insert_update_delete(self, conn, sql, parameters):
        pass

    def insert_update_delete_many(self, conn, sql, parameters):
        pass

    def insert_returning(self, conn, sql, parameters):
        pass

    def execute_script(self, conn, sql):
        pass


anosql.core.register_driver_adapter("mydb", MyDbAdapter)

# To use make a connection to your db, and pass "mydb" as the db_type:
import mydbdriver
conn = mydbriver.connect("...")

anosql.load_queries("path/to/sql/", "mydb")
greetings = anosql.get_greetings(conn)

conn.close()

If your adapter constructor takes arguments, you can register a function which can build your adapter instance:

def adapter_factory():
    return MyDbAdapter("foo", 42)

anosql.register_driver_adapter("mydb", adapter_factory)

Tests

$ pip install tox
$ tox

License

BSD, short and sweet