create.sql

INSERT INTO users (email, password) VALUES (%s, %s);

create_database.sql

DROP DATABASE IF EXISTS stupid_simple_test;

CREATE DATABASE stupid_simple_test;

USE stupid_simple_test;

CREATE TABLE users (
    id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
    email VARCHAR(255) NOT NULL,
    password VARCHAR(255) -- Should be properly encrypted in a real application!
);

get_by_email.sql

SELECT * FROM users WHERE email = %s

sss.py

"""
Stupid-simple SQL
=================

This library is an experiment in taking the exact opposite approach of popular
ORMs and query-building libraries. Rather than build APIs and abstractions that
represent SQL queries and database records with objects in the host-language,
we keep SQL seperate from the host language and offer a simple API for
executing previously written queries with parameters.

Synopsis
--------

This gist has the following directory structure::

    queries/
      migrations/
        create_database.sql
      users/
        create.sql
        get_by_email.sql

We can create a :class:`Library` representing these scripts and execute them
like so::

    from os.path import dirname
    from MySQLdb import connect

    email, password = 'user@someplace.com', 'mypassword'
    queries = Library(join(dirname(__file__), 'queries'))
    conn = connect(user='root')

    queries.migrations.create_database(conn)
    queries.users.create(conn, email, password)
    user = queries.users.get_by_email(conn, email).one()

    print user
    assert user['email'] == email
    assert user['password'] == password

Implicit Connections
--------------------

Say you're using a web framework that provides some sort of thread-local
request context (e.g. - Flask). You could subclass :class:`Query` to avoid
having to pass an explicit connection when executing a query::

    class MyQuery(Query):
        def __call__(self, *args, **kwargs):
            conn = self._get_connection()
            super(MyQuery, self).__call__(conn, *args, **kwargs)

        @static
        def _get_connection():
            # get or create a thread lcoal connection
            pass

    queries = Library('queries', query_class=MyQuery)

Fun ideas for the future
------------------------

  * Integrate a SQL parser like sqlparse_. Could extend Query objects with
    some query building methods for programmatically refining queries.

  * Alternatively, use a more powerful templating library in SQL files.
    We could escape input using MySQLdb.Connection.escape or
    psycopg2.extensions.adapt

_sqlparse: http://pypi.python.org/pypi/sqlparse>

"""
import re
import os
from os.path import join


class Query(object):
    """
    A Query represents a SQL script. Calling it will execute the SQL using a
    connection and any additional parameters passed in.

    >>> from minimock import Mock
    >>> conn = Mock('connection')
    >>> cursor = Mock('cursor')
    >>> cursor.description = []
    >>> conn.cursor.mock_returns = cursor
    >>>
    >>> q = Query('SELECT * FROM some_table WHERE that_thing = ?')
    >>> result = q(conn, 12)
    Called connection.cursor()
    Called cursor.execute('SELECT * FROM some_table WHERE that_thing = ?', (12,))

    The ``result`` object is an instance of :class:`ResultSet`. In addition to
    :meth:`ResultSet.one` and :meth:`ResultSet.all` it can also be iterated::

        for i, user in enumerate(queries.users.top_ten_this_week(conn)):
            print "%d. %s" % (i + 1, user['name'])

    """
    def __init__(self, sql):
        self._sql = sql

    def __str__(self):
        return self._sql

    def __call__(self, conn, *args, **kwargs):
        """
        Execute the query using a (connected) database cursor.

        Note: it probably makes sense to replace the cursor parameter with some
        sort of thread-local object or connection pool, but that's really down
        to what libraries you're already using.
        """
        cursor = conn.cursor()
        cursor.execute(self._sql, args or kwargs)
        return ResultSet(cursor)


class Library(object):
    """
    Collection of :class:`Query` objects.

    A QueryLibrary recursively loads all of the .sql files in a directory,
    attaching :class:`Query` objects to itself for each one.

    :param basedir: The directory to load queries from.
    :param query_class: The constructor for query instances. You may want to
      override this to allow queries to get their connection implicitly.
    """
    def __init__(self, basedir, query_class=Query):
        self._query_class = query_class
        self.load(basedir)

    def load(self, basedir):
        for path in os.listdir(basedir):
            if path.startswith('.'):
                continue

            fullpath = join(basedir, path)

            if os.path.isdir(fullpath):
                name = self._normalize(path)
                self._safe_setattr(name, Library(fullpath, self._query_class))
            elif path.endswith('.sql') and os.path.isfile(fullpath):
                name = self._normalize(path[:-4])
                with open(fullpath) as f:
                    self._safe_setattr(name, self._query_class(f.read()))

    def _safe_setattr(self, name, val):
        """
        wrap setattr in a check for existing attribute
        """
        if hasattr(self, name):
            raise TypeError("Name %s already exists on object!" % name)
        setattr(self, name, val)

    def _normalize(self, path):
        """ turn paths into attribute names, in no way is this robust """
        return re.sub("\\W", '_', path)


class ResultSet(object):
    """
    Wraps a cursor to provide a sane iterator interface to query results.

    ``cursor`` **must** have just executed a query
    """
    def __init__(self, cursor):
        self._cursor = cursor
        if cursor.description:
            self._column_names = tuple(c[0] for c in cursor.description)
        else:
            self._column_names = ()

    def _dictify(self, row):
        return dict(zip(self._column_names, row))

    def __len__(self):
        return self._cursor.rowcount

    def __iter__(self):
        return self

    def next(self):
        return self._dictify(self._cursor.fetchone())

    first = next

    def all(self):
        return map(self._dictify, self._cursor.fetchall())

    def one(self):
        if not len(self):
            raise Exception("No result found:\n" + self._cursor._executed)
        elif len(self) > 1:
            raise Exception("More than one record returned:\n" + self._cursor._executed)
        return self.next()

    def __getattr__(self, key):
        return getattr(self._cursor, key)


# smoke test, run the example from the synopsis
if __name__ == '__main__':
    source = []
    store = False
    for line in open(__file__):
        line = line.rstrip()
        if line.endswith('like so::'):
            store = True
        elif store:
            if line == '' or line.startswith('    '):
                source.append(line.strip())
            else:
                break

    print "\n".join(source)
    exec "\n".join(source) in globals()