Welcome to quma’s docs¶

Warning

This is alpha software and subject to change!

quma is a small SQL database library for Python and PyPy version 3.5 and higher. It maps object methods to SQL script files and supports SQLite, PostgreSQL, MySQL and MariaDB.

Unlike ORMs, it allows to write SQL as it was intended and to use all features the DBMS provides. As it uses plain SQL files you can fully utilize your database editor or IDE tool to author your queries.

It also provides a simple connection pool.

Learn more¶

If you want to know how to install quma and its dependencies, see Installation.
To get started read How to use quma from start to finish.
To see what you can do with query objects read The Query class.
In Connecting you learn how to connect to SQLite, PostgreSQL and MySQL/MariaDB databases.
Connection pool.
Learn what a changling cursor is and how it enables you to access result data in three different ways.
Database management systems have different ways of parameter binding. Passing parameters shows how it works in quma.
SQL doesn’t support every kind of dynamic queries. If you reach its limits you can circumvent this by using Templates.
If you pass more than one directory to the constructor, quma shadows duplicate files. See how this works in Shadowing.
You can add custom methods to namespaces. Learn how to do it in Custom namespaces. You will also learn about aliasing.
If you like to work on quma itself, Testing has the information on how to run its tests.

License¶

quma is released under the MIT license.

Installation¶

If you like to use quma with SQLite Python has everything covered and you only need to install quma itself:

pip install quma

To access a PostgreSQL or MySQL/MariaDB database you need to install the matching driver:

# PostgreSQL
pip install quma psycopg2
# or
pip install quma psycopg2cffi


# MySQL/MariaDB
pip install quma mysqlclient

Templates¶

You need to install the Mako template library if you want to use dynamic sql scripts using templates.

pip install mako

Development¶

git clone https://github.com/ebenefuenf/quma
cd quma
pip install -e '.[test,templates,postgres,mysql]'

How to use quma¶

quma reads sql files from the file system and makes them accessible as script objects. It passes the content of the files to a connected database management system (DBMS) when these objects are called.

Throughout this document we assume a directory with the following structure:

Scripts                │   Content
───────────────────────│─────────────────────────────────────────
path/to/sql/scripts    │
├─ users               │
│   ├─ all.sql         │   SELECT * FROM users
│   ├─ by_city.sql     │   SELECT * FROM users WHERE city = :city
│   ├─ by_id.sql       │   SELECT * FROM users WHERE id = :id
│   ├─ remove.sql      │   DELETE FROM users WHERE id = :id
│   └─ rename.sql      │   UPDATE TABLE users
│                      │       SET name = :name WHERE id = :id
└─ get_admin.sql       │   SELECT * FROM users WHERE admin = 1

After initialization you can run these scripts by calling members of a Database or a Cursor instance. Using the example above the following members are available:

# 'cur' is a cursor instance
cur.users.all(...
cur.users.by_city(...
cur.users.by_id(...
cur.users.rename(...
cur.users.remove(...
cur.get_admin(...

Read on to see how this works.

Opening a connection¶

To connect to a DBMS you need to instantiate an object of the Database class and provide a connection string and either a single path or a list of paths to your SQL scripts.

from quma import Database
db = Database('sqlite:///:memory:', '/path/to/sql-scripts')

Note: Database instances are threadsafe.

For more connection examples (e. g. PostgreSQL or MySQL/MariaDB) and parameters see Connecting. quma also supports connection pooling.

From now on we assume an established connection in the examples.

Creating a cursor¶

DBAPI libs like psycopg2 or sqlite3 have the notion of a cursor, which is used to manage the context of a fetch operation. quma is similar in that way. To execute queries you need to create a cursor instance.

quma provides two ways to create a cursor object. Either by using a context manager:

with db.cursor as cur:
    ...

Or by calling the cursor method of the Database instance:

try:
    cur = db.cursor()
finally:
    cur.close()

Running queries¶

To run the query in a sql script from the path(s) you passed to the Database constructor you call members of the Database instance or the cursor (db and cur from now on).

Scripts and directories at the root of the path are translated to direct members of db or cur. After initialisation of our example dir above, the script /get_admin.sql is available as Script instance db.get_admin or cur.get_admin and the directory /users as instance of Namespace, i. e. db.users or cur.users. Scripts in subfolders will create script objects as members of the corresponding namespace: /users/all will be db.users.all or cur.users.all.

When you call a Script object, as in cur.user.all() where all is the mentioned object, you get back a Query instance. The simplest use is to iterate over it (see below for more information about the Query class):

with db.cursor as cur:
    all_users = cur.users.all()
    for user in all_users:
        print(user['name'])

The same using the db API:

with db.cursor as cur:
    all_users = db.users.all(cur)

To learn what you can do with Query objects see The Query class.

Note

As you can see cur provides a nicer API where you don’t have to pass the cursor when you call a script or a method. Then again the db API has the advantage of being around 30% faster. But this should only be noticeable if you run hundreds or thousands of queries in a row for example in a loop.

If you have cloned the quma repository from github you can see the difference when you run the script bin/cursor_vs_db.py.

Committing changes and rollback¶

quma does not automatically commit by default. You have to manually commit all changes as well as rolling back if an error occurs using the commit() and rollback() methods of the cursor.

try:
    cur.users.remove(id=13).run()
    cur.users.rename(id=14, name='New Name').run()
    cur.commit()
except Exception:
    cur.rollback()

If db is initialized with the flag contextcommit set to True and a context manager is used, quma will automatically commit when the context manager ends. So you don’t need to call cur.commit().

db = Database('sqlite:///:memory:', contextcommit=True)

with db.cursor as cur:
    cur.users.remove(id=13).run()
    cur.users.rename(id=14, name='New Name').run()
    # no need to call cur.commit()

Note: If you are using MySQL or SQLite some statements will automatically cause a commit. See the MySQL docs and SQLite docs

Autocommit¶

If you pass autocommit=True when you initialize a cursor, each query will be executed in its own transaction that is implicitly committed.

with db(autocommit=True).cursor as cur:
    cur.users.remove(id=13).run()

try:
    cur = db.cursor(autocommit=True)
    cur.users.remove(id=13).run()
finally:
    cur.close()

Executing literal statements¶

Database instances provide the method execute(). You can pass arbitrary sql strings. Each call will be automatically committed. If there is a result it will be returned otherwise it returns None.

db.execute('CREATE TABLE users ...')
users = db.execute('SELECT * FROM users')
for user in users:
    print(user.name)

If you want to execute statements in the context of a transaction use the execute() method of the cursor:

with db.cursor as cur:
    cur.execute('DELETE FROM users WHERE id = 13');
    cur.commit()

Accessing the DBAPI cursor and connection¶

The underlying DBAPI connection and cursor objects are available as members of the cursor instance. The connection object is raw_conn and the cursor raw_cursor.cursor.

# The connection
dbapi_cursor = cur.raw_conn.autocommit = True
dbapi_cursor = cur.raw_conn.cursor()

# The cursor
cur.raw_cursor.cursor.execute('SELECT * FROM users;')
users = cur.raw_cursor.cursor.fetchall()
# raw_cursor wraps the real cursor. This would work also
cur.raw_cursor.execute('SELECT * FROM users;')
users = cur.raw_cursor.fetchall()

All members of the raw_cursor.cursor object are also available as members of cur. Hence there should be no need to use it directly:

cur.execute('SELECT * FROM users;')
users = cur.fetchall()

The Query class¶

When you call a script object it returns an instance of the Query class which holds the code from the script file and the parameters passed to the script call.

Queries are executed lazily. This means you have to either call a method of the query object or to iterate over it to cause the execution of the query against the DBMS.

Getting multiple rows from a query¶

You can either iterate directly over the query object or call its all() method to get a list of the all the rows.

with db.cursor as cur:
    result = cur.users.by_city(city='City 1')
    for row in result:
        print(row.name)

    # calling the .all() method to get a materialized list/tuple
    user_list = cur.users.by_city(city='City 1').all()
    # is a bit faster than
    user_list = list(cur.users.by_city(city='City 1'))

When calling all() MySQL and MariaDB will return a tuple, PostgreSQL and SQLite will return a list.

Note

If you are using PyPy with the sqlite3 driver the cast using the list() function does not currently work and will always result in an empty list.

Getting a single row¶

If you know there will be only one row in the result of a query you can use the one() method to get it. quma will raise a DoesNotExistError error if there is no row in the result and a MultipleRowsError if there are returned more than one row.

from quma import (
    DoesNotExistError,
    MultipleRowsError,
)
...

with db.cursor as cur:
    try:
        user = cur.users.by_id(id=13).one()
    except DoesNotExistError:
        print('The user does not exist')
    except MultipleRowsError:
        print('There are multiple users with the same id')

DoesNotExistError and MultipleRowsError are also attached to the Database class so you can access it from the db instance. For example:

with db.cursor as cur:
    try:
        user = cur.users.by_id(id=13).one()
    except db.DoesNotExistError:
        print('The user does not exist')
    except db.MultipleRowsError:
        print('There are multiple users with the same id')

It is also possible to get a single row by accessing its index on the result set:

user = cur.users.by_id(id=13)[0]
# or
users = cur.users.by_id(id=13)
user = users[0]

If you want the first row of a result set which may have more than one row or none at all you can use the first() method:

# "user" will be None if there are no rows in the result.
user = cur.users.all().first()

The method value() invokes the one() method, and upon success returns the value of the first column of the row (i. e. fetchall()[0][0]). This comes in handy if you are using a RETURNING clause, for example, or return the last inserted id after an insert.

last_inserted_id = cur.users.insert().value()

Execute only¶

To simply execute a query without needing its result you call the run() method:

with db.cursor as cur:
    cur.user.add(name='User',
                 email='user@example.com',
                 city='City').run()

    # or
    query = cur.user.add(name='User',
                         email='user@example.com',
                         city='City')
    query.run()

This is handy if you only want to execute the query, e. g. DML statements like INSERT, UPDATE or DELETE where you don’t need a fetch call.

Getting data in chunks¶

quma supports the fetchmany method of Python’s DBAPI by providing the many() method of Query. many() returns an instance of ManyResult which implements the get() method which internally calls the fetchmany method of the underlying cursor.

many_users = cur.users.by_city(city='City').many()
first_two = manyusers.get(2) # the first call of get executes the query
next_three = manyusers.get(3)
next_two = manyusers.get(2)

Another example:

def users_generator()
    with db.cursor as cur:
        many_users = cur.users.all().many()
        batch = many_users.get(3) # the first call of get executes the query
        while batch:
            for result in batch:
                yield result
            batch = many_users.get(3)

for user in users_generator():
    print(user.name)

Note

In contrast to all other fetching methods of the query object, like all(), first(), or one(), a call of many() will not execute the query. Instead, the first call of the get() method of an many result object will cause the execution. Also, results of many calls are not cached and if a query was already executed the many mechanism will execute it again anyway. So keep in mind that already executed queries will be re-executed when many() is called after the first execution, as in:

all_users = cur.users.all()
first_user = allusers.first() # query executed the first time
many_users = allusers.many()
first_two = manyusers.get(2) # query executed a second time

Additionally, the cache of all_users from the last example will be be invalidated after the first call of get(). So you should avoid to mix many queries with “normal” queries.

A simpler version of `many()`¶

If your expected result set is too large for simply iterating over the query object or calling all() (as they call fetchall internally) but you like to work with the result in a single simple loop instead of using many(), you can use the method unbunch(). It is a convenience method which internally calls fetchmany with the given size. Using unbunch() we can simplify the many() example with the users_generator from the last section:

with db.cursor as cur:
    for user in cur.users.all().unbunch(3)
        print(user.name)

unbunch() re-excutes the query and invalidates the cache on each call, just like many().

Getting the number of rows¶

If you are only interested in the number of row in a result you can pass a Query object to the len() function. quma also includes a convenience method called count(). Some drivers (like pycopg2) support the rowcount property of PEP249 which specifies the number of rows that the last execute produced. If it is available it will be used to determine the number of rows, otherwise a fetchall will be executed and passed to len() to get the number.

number_of_users = len(cur.users.all())
number_of_users = cur.users.all().count()
number_of_users = db.users.all(cur).count()

Note

len() or count() calls must occure before fetch calls like one() or all(). This has to do with the internals of the DBAPI drivers. A fetch would overwrite the value of rowcount which would return -1 afterwards.

Checking if a result exists¶

To check if a query has a result or not call the exists() method.

has_users = cur.users.all().exists()

You can also use the query object itself for truth value testing:

all_users = cur.users.all()
if all_users:
    user1 = allusers.first()

Results are cached¶

As described above, quma executes queries lazily. Only after the first call of a method or when an iteration over the query object is started, the data will be fetched. The fetched result will be cached in the query object. This means you can perform more than one operation on the object while the query will not be re-executed. If you want to re-execute it, you need to call run() manually.

with db.cursor as cur:
    all_users = cur.users.all()

    for user in all_users:
        # the result is fetched and cached on the first iteration
        print(user.name)

    # get a list of all users from the cache
    all_users.all()
    # get the first user from the cache
    all_users.first()

    # re-execute the query
    all_users.run()

    # fetch and cache the new result of the re-executed query
    all_users.all()

Access the underlying cursor¶

You can access the attributes of the cursor which is used to execute the query directly on the query object.

with db.cursor as cur:
    added = cur.users.add(name='User', email='user.1@example.com').run()
    if added.lastrowid:
        user = cur.user.by_id(id=added.lastrowid).run()
        user.fetchone()

Overview¶

Class Query¶

Class ManyResult¶

Connecting¶

You connect to a server/database by creating an instance of the class quma.Database. You have to at least provide a valid database URL and the path to your sql scripts. See below for the details.

Connection Examples¶

sqldir = '/path/to/sql/scripts'

# can also be a list of paths:
sqldir = [
  '/path/to/sql/scripts',
  '/another/path/to/sql/scripts',
]

# SQLite
db = Database('sqlite:////path/to/db.sqlite', sqldir)
# SQLite in memory db
db = Database('sqlite:///:memory:', sqldir)

# PostgreSQL localhost
db = Database('postgresql://username:password@/db_name', sqldir)
# PostgreSQL network server
db = Database('postgresql://username:password@10.0.0.1:5432/db_name', sqldir)

# MySQL/MariaDB localhost
db = Database('mysql://username:password@/db_name', sqldir)
# MySQL/MariaDB network server
db = Database('mysql://username:password@192.168.1.1:5432/db_name', sqldir)

The Database class¶

Connection pool¶

quma provides a connection pool implementation (PostgreSQL and MySQL only) similar to sqlalchemy’s and even borrows code and ideas from it.

Setup a pool:

# PostgreSQL pool (keeps 5 connections open and allows 10 more)
db = Database('postgresql+pool://username:password@/db_name', sqldir,
              size=5, overflow=10)

# MySQL/MariaDB pool
db = Database('mysql+pool://username:password@/db_name', sqldir,
              size=5, overflow=10)

For a description of the parameters see Connecting.

Reusing connections¶

To reuse connections you can pass a carrier object to db when you create a cursor. quma then creates the attribute __quma_conn__ on the carrier holding the connection object. You should only use this feature if that fact doesn’t lead to problems in your application. Only objects which allow adding attributes at runtime are supported. A good example is the request object in web applications:

from pyramid.view import view_config
from quma import Database

db = Database('sqlite:////path/to/db.sqlite', sqldir)


def do_more(request, user_id):
    # reuses the same connection which was opened
    # in user_view.
    with db(request).cursor as cur:
        cur.user.remove(id=user_id)


@view_config(route_name='user')
def user_view(request):
    with db(request).cursor as cur:
        user = cur.user.by_name(name='Username').one()
        do_more(request, user['id'])

    with db(request).cursor as cur:
        # reuses the connection
        user = cur.user.rename(id=13, name='New Username')
        # commit every statement previously executed
        cur.commit()
        # exlicitly close the cursor
        cur.close()

Note: It is always a good idea to close a connection if you’re done. If you are using a carrier and a connection pool it is absolutely necessary and you have to explicitly close the cursor or release the carrier. You can do it using cur.close() or by passing the carrier to db.release(carrier), otherwise the connection would not be returned to the pool.

Changling Cursor¶

If you are using SQLite or PostgreSQL you can access result object attributes using three different methods if you pass changling=True on db initialization. (MySQL does not support it. See below)

db = Database('sqlite:///:memory:', sqldir, changeling=True)

with db.cursor as c:
    user = db.users.by_id(c, 13).one()
    name = user[0]       # by index
    name = user['name']  # by key
    name = user.name     # by attribute

Shadowed superclass members¶

If a query result has a field with the same name as a member of the superclass of the changeling (sqlite: sqlite3.Row, psycopg2: psycopg2.extras.DictRow) it shadows the original member. This means the original member isn’t accessible. You can access it anyway if you prefix it with an underscore ‘_’.

The sqlite3.Row, for example, has a method keys() which lists all field names. If a query returns a field with the name ‘keys’ the method is shadowed:

-- /path/to/sql/scripts/users/by_id.sql
SELECT name, email, 'the keys' AS keys FROM users WHERE id = :id;

row = cur.users.by_id(13).one()
assert row.keys == 'the keys'

# If you want to call the keys method of row prefix it with _
print(row._keys()) # ['name', 'email', 'keys']

Performance¶

By default, changling is False which is slightly faster. Then SQLite supports access by index only. PostgreSQL by key and index (we use psycopg.extras.DictCursor internally).

MySQL/MariaDB¶

MySQL/MariaDB supports access by index only, except you pass dict_cursor=True on initialization. Then it supports access by key only.

Passing Parameters to SQL Queries¶

SQLite supports two kinds of placeholders: question marks (qmark style) and named placeholders (named style). PostgreSQL/MySQL/MariaDB support simple (%s) and named (%(name)s) pyformat placeholders:

-- SQLite qmark
SELECT name, email FROM users WHERE id = ?
-- named
SELECT name, email FROM users WHERE id = :id

-- PostgreSQL/MySQL/MariaDB pyformat
SELECT name, email FROM users WHERE id = %s
-- named
SELECT name, email FROM users WHERE id = %(id)s

# simple style (? or %s)
cur.users.by_id(1)
db.users.by_id(cur, 1)

# named style (:name or %(name)s)
cur.users.by_id(id=1)
db.users.by_id(cur, id=1)

Templates¶

quma supports rendering templates using the Mako template library. By default, template files must have the file extension *.msql, which can be overwritten.

Using this feature you are able to write dynamic queries which would not be possible with SQL alone. Beware of SQL injections.

A very simple example:

-- sql/users/by_group.msql
SELECT
    name,
% if admin:
    birthday,
% endif
    city
FROM users
% if not admin:
WHERE
    group = 'public'
% endif

In Python you call it the same way like simple SQL queries:

cur.users.by_group(admin=True)

Shadowing¶

If you pass a list of two or more directories to the Database constructor, the order is important. Files from subsequent directories in the list with the same relative path will shadow (or overwrite) files from preceding directories.

Let’s say you have two different directories with SQL scripts you like to use with quma. For example directory one:

/path/to/sql/scripts/one
├── addresses
│    ├── all.sql
│    └── remove.sql
├── users
│    ├── all.sql
│    └── remove.sql
├── get_admin.sql
└── remove_admin.sql

and directory two:

/path/to/sql/scripts/two
├── users
│    ├── all.sql
│    └── rename.sql
├── create_admin.sql
└── get_admin.sql

When you initialize quma with both directories like this:

from quma import Database

db = Database('sqlite:///:memory:', [
    '/path/to/sql/scripts/one',
    '/path/to/sql/scripts/two',
])

quma creates the following members:

                         # From directory:
cur.addresses.all        #   one
cur.addresses.remove     #   one
cur.users.all            #   two (shadows all.sql from dir one)
cur.users.remove         #   one
cur.users.rename         #   two
cur.create_admin         #   two
cur.get_admin            #   two (shadows get_admin.sql from dir one)
cur.remove_admin         #   one

Custom namespaces¶

quma automatically creates namespace objects when it reads in your sql scripts. Each subfolder in the script directory will result in a namespace object as a direct member of db or cur.

You can add custom methods to these objects by putting a __init__.py file into the subfolder which is your namespace and by adding a subclass of quma.Namespace to it. The class must have the same name as the folder with the first letter uppercase.

path/to/sql/scripts
├─ users
│   ├─ __init__.py
│   ├─ all.sql
│   ├─ by_city.sql
..

from quma import Namespace

# If the subfolder's name is 'users' the
# class must be named 'Users'.
class Users(Namespace):
    # the method must accept the cursor as its first parameter
    def get_test(self, cur):
        return 'Test'

    def get_address(self, cur, username):
        user = cur.user.by_username(username=username)
        return cur.address.by_user(user.id)

Public methods of the namespace must be definied with the cursor as second parameter. It will automatically be passed when you use the cur api.

Now you can call the method the same way as you would call scripts:

db.users.get_test(cur)
cur.users.get_test() # no need to pass cur
address = cur.users.get_address('testuser')

Root members¶

If you want to add root level methods you need to add __init__.py to the root of your script directory and name the subclass Root.

path/to/sql/scripts
├─ __init__.py
├─ users
│   ├─ all.sql
..

class Root(Namespace):
    def root_method(self, cursor):
        return 'Test'

db.root_method()
cur.root_method()

Aliasing¶

If you add the class level attribute alias to your custom namespace, you can call it by that name too.

from quma import Namespace

class Users(Namespace):
    alias = 'user'

cur.user.all()
# This is the same as.
cur.users.all()

Importable database¶

Sometimes it isn’t enough to create a global Database instance and import it into other modules. For example, if you read the database credentials from a configuration file at runtime and then initialize the instance while the uninitialized global is already imported elsewhere. The following code shows a way to keep the quma API in place and allows to import the db wrapper class even if the connection is not established yet.

#####  my_db_module.py

import quma

_db = None

class MetaDB(type):
    def __getattr__(cls, attr):
        return getattr(_db, attr)

class db(object, metaclass=MetaDB):
    def __init__(self, carrier=None, autocommit=None):
        self.carrier = carrier
        self.autocommit = autocommit

    def __getattr__(self, attr):
        return getattr(_db(carrier=self.carrier,
                        autocommit=self.autocommit), attr)

def connect():
    global _db
    sqldir = '/path/to/sql/scripts'

    _db = quma.Database(uri, sqldir)

Create the instance in your main module:

#####  main.py

import my_db_module

my_db_module.connect()

Now you can import the class my_db_module.db from everywhere and use it the same way as a usual instance of quma.Database.

#####  e. g. model.py

from my_db_module import db

with db.cursor as cur:
    cur.users.all()

Testing¶

Prerequisites: In order to run the tests for MySQL or PostgreSQL you need to create a test database:

PostgreSQL:

CREATE USER quma_test_user WITH PASSWORD 'quma_test_password';
CREATE DATABASE quma_test_db;
GRANT ALL PRIVILEGES ON DATABASE quma_test_db to quma_test_user;

MySQL/MariaDB:

CREATE DATABASE quma_test_db;
CREATE USER quma_test_user@localhost IDENTIFIED BY 'quma_test_password';
GRANT ALL ON quma_test_db.* TO quma_test_user@localhost;

How to run the tests¶

Run pytest or py.test to run all tests. pytest -m "not postgres and not mysql" for all general tests. And pytest -m "postgres" or pytest -m "mysql" for DBMS specific tests.

Overwrite credentials¶

If you like to use your own test database and user you can overwrite the default credentials by setting environment variables

PostgreSQL:

QUMA_PGSQL_USER
QUMA_PGSQL_PASS
QUMA_PGSQL_DB

MySQL/MariaDB:

QUMA_MYSQL_USER
QUMA_MYSQL_PASS
QUMA_MYSQL_DB

Welcome to quma’s docs¶

Learn more¶

License¶

Installation¶

Templates¶

Development¶

How to use quma¶

Opening a connection¶

Creating a cursor¶

Running queries¶

Committing changes and rollback¶

Autocommit¶

Executing literal statements¶

Accessing the DBAPI cursor and connection¶

The Query class¶

Getting multiple rows from a query¶

Getting a single row¶

Execute only¶

Getting data in chunks¶

A simpler version of many()¶

Getting the number of rows¶

Checking if a result exists¶

Results are cached¶

Access the underlying cursor¶

Overview¶

Class Query¶

Class ManyResult¶

Connecting¶

Connection Examples¶

The Database class¶

Connection pool¶

Reusing connections¶

Changling Cursor¶

Shadowed superclass members¶

Performance¶

MySQL/MariaDB¶

Passing Parameters to SQL Queries¶

Templates¶

Shadowing¶

Custom namespaces¶

Root members¶

Aliasing¶

Importable database¶

Testing¶

How to run the tests¶

Overwrite credentials¶

A simpler version of `many()`¶