Re: pysqlite design decisions

#!/usr/bin/env python

# A ctypes binding to SQLite

import ctypes

def utf8me(s):
    return unicode(s).encode("utf_8")

def fromutf8(s):
    return s.decode("utf_8")

SQLITE_OK=0
SQLITE_ERROR=1
SQLITE_ROW=100
SQLITE_DONE=101
# todo: fill out rest and make distinct exceptions for each
def SQLiteError(Exception): pass

def getexception(code, connection):
    if code==SQLITE_OK: return None
    # todo: this would need to switch on the error code to get the right exception
    return SQLiteError(fromutf8(connection.library.dll.sqlite3_errmsg(connection.db)))

def raiseexception(code, db):
    e=getexception(code, db)
    if e is not None:
        raise e

class CSqlite:

    # SQlite constants
    SQLITE_STATIC=ctypes.c_void_p(0)
    SQLITE_TRANSIENT=ctypes.c_void_p(-1)

    class Statement:
        # states we can be in
        STATE_END=0
        STATE_FIRST_ROW=1
        STATE_ROW=2
        STATE_BEGIN=3
        
        def __init__(self, connection, stmt):
            self.connection=connection
            self.dll=self.connection.library.dll
            self.stmt=stmt
            self.db=connection.db
            self.state=self.STATE_BEGIN

        def __del__(self):
            if self.stmt:
                self.dll.sqlite3_finalize(self.stmt)
                self.stmt=None

        def __iter__(self):
            return self

        def reset(self):
            self.state=self.STATE_BEGIN
            res=self.dll.sqlite3_reset(self.stmt)
            raiseexception(res, self.connection)
            try:
                res=self.dll.sqlite3_clear_bindings(self.stmt)
                raiseexception(res, self.connection)
            except AttributeError:
                pass # needs more recent version of SQLite

        def _bind(self, n, value):
            # haven't done blob/buffer
            if isinstance(value, (unicode, str)):
                utfvalue=utf8me(value)
                res=self.dll.sqlite3_bind_text(self.stmt, n, utfvalue, len(utfvalue), CSqlite.SQLITE_TRANSIENT)
            elif isinstance(value, int):
                res=self.dll.sqlite3_bind_int(self.stmt, n, value)
            elif isinstance(value, long):
                res=self.dll.sqlite3_bind_int64(self.stmt, n, ctypes.c_longlong(value))
            elif isinstance(value, float):
                res=self.dll.sqlite3_bind_double(self.stmt, n, ctypes.c_double(value))
            elif value is None:
                res=self.dll.sqlite3_bind_null(self.stmt, n)
            else:
                raise Exception("unsupported data type "+`value`)
            raiseexception(res, self.connection)

        def execute(self, *bindings):
            self.reset()
            for i in range(self.dll.sqlite3_bind_parameter_count(self.stmt)):
                name=self.dll.sqlite3_bind_parameter_name(self.stmt, i+1)
                if name:
                    name=fromutf8(name)
                    self._bind(i+1, bindings[0]['name']) # must be dict like
                else:
                    self._bind(i+1, bindings[i])
            self.state=self.STATE_BEGIN
            try:
                self.next()  # greedy execution since user won't call next() with statements that don't return data
            except StopIteration:
                pass
            return self

        def _getfield(self, field):
            t=self.dll.sqlite3_column_type(self.stmt, field)
            if t==1: # integer
                return self.dll.sqlite3_column_int64(self.stmt, field)
            elif t==2: # float
                return self.dll.sqlite3_column_double(self.stmt, field)
            elif t==3: # text
                # this should use sqlite3_column_bytes so that embedded nulls are retained
                return fromutf8(self.dll.sqlite3_column_text(self.stmt, field))
            elif t==4: # blob
                pass
            else: # 5 = null
                return None

        def next(self):
            if self.state==self.STATE_END:
                raise StopIteration()

            if self.state!=self.STATE_FIRST_ROW:
                res=self.dll.sqlite3_step(self.stmt)
                if res==SQLITE_DONE:
                    self.state=self.STATE_END
                    raise StopIteration()
                if res==SQLITE_ERROR:
                    # get actual error code from sqlite3_reset
                    res=self.dll.sqlite3_reset(self.stmt)
                if res!=SQLITE_ROW:
                    raiseexception(res, self.connection)

            if self.state==self.STATE_BEGIN:
                self.state=self.STATE_FIRST_ROW
                return
            self.state=self.STATE_ROW
            return tuple([self._getfield(field) for field in range(self.dll.sqlite3_column_count(self.stmt))])

    class Connection:
        def __init__(self, library, filename):
            self.library=library
            self.db=ctypes.c_void_p()
            res=self.library.dll.sqlite3_open(utf8me(filename), ctypes.byref(self.db))
            e=getexception(res, self.db) # db is always allocated even on failure
            if e:
                self.library.dll.sqlite3_close(self.db)
                self.db=None
                raise e

        def __del__(self):
            if self.db:
                self.library.dll.sqlite3_close(self.db)
                self.db=None
                self.library=None

        def prepare(self, sql):
            zsql=utf8me(sql)
            stmt=ctypes.c_void_p()
            tail=ctypes.c_char_p()
            res=self.library.dll.sqlite3_prepare(self.db, zsql, -1, ctypes.byref(stmt), ctypes.byref(tail))
            if len(tail.value):
                if stmt: self.library.dll.sqlite3_finalize(stmt)
                raise Exception("Only one SQL statement can be provided: left "+`tail`)
            raiseexception(res, self)
            return CSqlite.Statement(self, stmt)
            

    def __init__(self, filename="libsqlite3.so"):
        dll=ctypes.CDLL(filename)
        # some annotations
        dll.sqlite3_errmsg.restype = ctypes.c_char_p
        dll.sqlite3_bind_parameter_name.restype = ctypes.c_char_p
        dll.sqlite3_column_double.restype = ctypes.c_double
        dll.sqlite3_column_int64.restype = ctypes.c_longlong
        dll.sqlite3_column_text.restype = ctypes.c_char_p
        self.dll=dll

    def connection(self, filename):
        return CSqlite.Connection(self, filename)

if __name__=='__main__':
    import os
    try: os.remove("testdb")
    except: pass
    
    c=CSqlite()
    connection=c.connection("testdb")
    stmt=connection.prepare("create table foo(x,y)")
    stmt.execute()
    stmt=connection.prepare("insert into foo values(?,?)")
    stmt.execute("some text", u"\N{BLACK STAR} \N{WHITE STAR} \N{LIGHTNING} \N{COMET} ")
    stmt.execute(1, 2)
    stmt.execute(1.1, 2.1)
    stmt.execute(None, None)
    stmt=connection.prepare("select * from foo")
    for row in stmt.execute():
        print row

    try:
        # check exceptions
        stmt=connection.prepare("blah blah blah")
    except SQLiteError:
        pass