Skip to content

Instantly share code, notes, and snippets.

@michalc
Last active April 3, 2024 17:26
Show Gist options
  • Star 18 You must be signed in to star a gist
  • Fork 4 You must be signed in to fork a gist
  • Save michalc/a3147997e21665896836e0f4157975cb to your computer and use it in GitHub Desktop.
Save michalc/a3147997e21665896836e0f4157975cb to your computer and use it in GitHub Desktop.
Use libsqlite3 directly from Python with ctypes: without using the built-in sqlite3 Python package, and without compiling anything
# From https://stackoverflow.com/a/68876046/1319998, which is itself inspired by https://stackoverflow.com/a/68814418/1319998
from contextlib import contextmanager
from collections import namedtuple
from ctypes import cdll, byref, string_at, c_char_p, c_int, c_double, c_int64, c_void_p
from ctypes.util import find_library
from sys import platform
def query(db_file, sql, params=()):
libsqlite3 = cdll.LoadLibrary(find_library('sqlite3'))
libsqlite3.sqlite3_errstr.restype = c_char_p
libsqlite3.sqlite3_errmsg.restype = c_char_p
libsqlite3.sqlite3_column_name.restype = c_char_p
libsqlite3.sqlite3_column_double.restype = c_double
libsqlite3.sqlite3_column_int64.restype = c_int64
libsqlite3.sqlite3_column_blob.restype = c_void_p
libsqlite3.sqlite3_column_bytes.restype = c_int64
SQLITE_ROW = 100
SQLITE_DONE = 101
SQLITE_TRANSIENT = -1
SQLITE_OPEN_READWRITE = 0x00000002
bind = {
type(0): libsqlite3.sqlite3_bind_int64,
type(0.0): libsqlite3.sqlite3_bind_double,
type(''): lambda pp_stmt, i, value: libsqlite3.sqlite3_bind_text(pp_stmt, i, value.encode('utf-8'), len(value.encode('utf-8')), SQLITE_TRANSIENT),
type(b''): lambda pp_stmt, i, value: libsqlite3.sqlite3_bind_blob(pp_stmt, i, value, len(value), SQLITE_TRANSIENT),
type(None): lambda pp_stmt, i, _: libsqlite3.sqlite3_bind_null(pp_stmt, i),
}
extract = {
1: libsqlite3.sqlite3_column_int64,
2: libsqlite3.sqlite3_column_double,
3: lambda pp_stmt, i: string_at(
libsqlite3.sqlite3_column_blob(pp_stmt, i),
libsqlite3.sqlite3_column_bytes(pp_stmt, i),
).decode(),
4: lambda pp_stmt, i: string_at(
libsqlite3.sqlite3_column_blob(pp_stmt, i),
libsqlite3.sqlite3_column_bytes(pp_stmt, i),
),
5: lambda pp_stmt, i: None,
}
def run(func, *args):
res = func(*args)
if res != 0:
raise Exception(libsqlite3.sqlite3_errstr(res).decode())
def run_with_db(db, func, *args):
if func(*args) != 0:
raise Exception(libsqlite3.sqlite3_errmsg(db).decode())
@contextmanager
def get_db(db_file):
db = c_void_p()
run(libsqlite3.sqlite3_open_v2, db_file.encode(), byref(db), SQLITE_OPEN_READWRITE, None)
try:
yield db
finally:
run_with_db(db, libsqlite3.sqlite3_close, db)
@contextmanager
def get_pp_stmt(db, sql):
pp_stmt = c_void_p()
run_with_db(db, libsqlite3.sqlite3_prepare_v3, db, sql.encode(), -1, 0, byref(pp_stmt), None)
try:
yield pp_stmt
finally:
run_with_db(db, libsqlite3.sqlite3_finalize, pp_stmt)
with \
get_db(db_file) as db, \
get_pp_stmt(db, sql) as pp_stmt:
for i, param in enumerate(params):
run_with_db(db, bind[type(param)], pp_stmt, i + 1, param)
row_constructor = namedtuple('Row', (
libsqlite3.sqlite3_column_name(pp_stmt, i).decode()
for i in range(0, libsqlite3.sqlite3_column_count(pp_stmt))
))
while True:
res = libsqlite3.sqlite3_step(pp_stmt)
if res == SQLITE_DONE:
break
if res != SQLITE_ROW:
raise Exception(libsqlite3.sqlite3_errstr(res).decode())
yield row_constructor(*(
extract[libsqlite3.sqlite3_column_type(pp_stmt, i)](pp_stmt, i)
for i in range(0, len(row_constructor._fields))
))
@michalc
Copy link
Author

michalc commented May 12, 2023

How about this? Strongly based on https://gist.github.com/jsok/2936764 (the main non-language difference is looping while the results is SQLITE_ROW to not go really wrong in error cases)

import sys
from ctypes import cdll, byref, c_double, c_void_p, string_at
from ctypes.util import find_library

class Codes():
    SQLITE_ROW = 100
    SQLITE_INTEGER = 1
    SQLITE_FLOAT = 2
    SQLITE_TEXT = 3

libsqlite3 = cdll.LoadLibrary(find_library('sqlite3'))
libsqlite3.sqlite3_column_text.restype = c_void_p
libsqlite3.sqlite3_column_double.restype = c_double

db = c_void_p()
sqlite3_stmt = c_void_p()

libsqlite3.sqlite3_open("expenses.db".encode(), byref(db));
if not db:
    print("Failed to open DB");
    sys.exit(1)

print("Performing query...");

libsqlite3.sqlite3_prepare_v2(db, "select * from expenses".encode(), -1, byref(sqlite3_stmt), None)

print("Got results:");
while libsqlite3.sqlite3_step(sqlite3_stmt) == Codes.SQLITE_ROW:
    num_cols = libsqlite3.sqlite3_column_count(sqlite3_stmt)

    for i in range(0, num_cols):
        match libsqlite3.sqlite3_column_type(sqlite3_stmt, i):
            case Codes.SQLITE_TEXT:
                print("{}, ".format(string_at(libsqlite3.sqlite3_column_text(sqlite3_stmt, i))), end="");
            case Codes.SQLITE_INTEGER:
                print("{:d}, ".format(libsqlite3.sqlite3_column_int(sqlite3_stmt, i)), end="");
            case Codes.SQLITE_FLOAT:
                print("{:e}, ".format(libsqlite3.sqlite3_column_double(sqlite3_stmt, i)), end="");
            case _:
                pass
    print("")

libsqlite3.sqlite3_finalize(sqlite3_stmt);
libsqlite3.sqlite3_close(db);

@Axle-Ozz-i-sofT
Copy link

Thank you Michal
I did look at that which I found beneficial the other day :) That code is sitting next a few decent methods including yours.
That being said, I think I just nutted it out (almost) in a procedural methodology. I Still have to check that the the pointers are being carried by reference correct to each function and some error handling etc.
Not using too many of the function calls from the C API this isn't too over blown and still illustrates the ctypes conversions.

sql3_tests.py (Version hello world)

#-------------------------------------------------------------------------------
# Name:         sql3_tests.py (based upon basics_2.c, ozz_sql3.h)
# Purpose:      SQLite3 Basic examples tests.
#               Tests for convenience wrapper functions for SQLite version 3.
#
# Platform:     Win64, Ubuntu64
# Depends:      SQLite v3.34.1 plus (dll/so), ctypes, ozz_sql3.py
#
# Author:       Axle
#
# Created:      12/05/2023
# Updated:
# Copyright:    (c) Axle 2023
# Licence:      MIT-0 No Attribution
#-------------------------------------------------------------------------------
# Notes:
# Using the SQLite shared object (.dll, .so) directly as a Run-time library. The
# sqlite3.dll/.so must be in the system or application path.
#
# Python 3 built in SQLite3 library is a better/safer approach but uses a
# distinctly different API to the default C API which goes against the
# primary goal of exemplifying the same code routines in all 3 languages.
# As such I am using the Ctypes module for direct access to the shared libraries
# (.dll, .so) exposed C API. In essence python types are translated to C types
# for use by the C based shared object, and then C types are converted back to
# Python types when data is returned. This happens by default with most Python
# library modules but occurs in a more opeque manner in the background.
#
#-------------------------------------------------------------------------------

import ctypes, sys, os
import ozz_sql3

# https://stephenscotttucker.medium.com/interfacing-python-with-c-using-ctypes-classes-and-arrays-42534d562ce7
# https://realpython.com/pointers-in-python/
# https://dbader.org/blog/python-ctypes-tutorial
# https://github.com/trolldbois/ctypeslib
# https://www.scaler.com/topics/python-ctypes/
# https://solarianprogrammer.com/2019/07/18/python-using-c-cpp-libraries-ctypes/


def main():
    pass
    return_code = 0  # error codes returned from functions.
    ver_buffer = []  # empty list (strings are imutable when passed to a function in python).
    db_filename = ":memory:"  # Using a temporary "In RAM" database.

    # I could move these to the module and just return a single ID for the
    # loaded dll/so session.
    f_library = ozz_sql3.get_libsql3_path()  # Get the path for sqlite.dll/so
    idlib_sql3  = ozz_sql3.load_libsql3(f_library)  # Load the sqlite3 shared library.
    #print(type(idlib_sql3))  # DEBUG

    hlib_sql3 = ozz_sql3.handle_libsql3(idlib_sql3)  # get the OS handle of the CDLL object (Not used here).
    #print(type(hlib_sql3))  # DEBUG

    # Get our SQLite version. Confirmation that sqlite 3 is installed as a
    # shared library and compiling/working correctly.
    # Ensure that sqlite3.dll is in the system path or in the working directory
    # of the project python script at run-time.
    # NOTE: I am using the C API interface directly and not as a query. SQLite
    # provides a limited number of helper MACROS that can be accessed directly
    # without opening a databse.

    # The return is already converted to Python UTF-8 string byt the function.
    buffer1 = ozz_sql3.sqlite3_libversion(idlib_sql3)
    print("1 SQLite Version:" + buffer1)
    print("===========================================")

    # Get version long
    p_db = ozz_sql3.p_sqlite3()
    p_stmt = ozz_sql3.p_sqlite3_stmt()

    ozz_sql3.sqlite3_open(idlib_sql3, db_filename, p_db)

    sql_query = "SELECT SQLITE_VERSION()"

    ozz_sql3.sqlite3_prepare_v2(idlib_sql3, p_db, sql_query, -1, p_stmt, None)

    # Call once only. No loop of rows.
    ozz_sql3.sqlite3_step(idlib_sql3, p_db, p_stmt)

    buffer2 = ozz_sql3.sqlite3_column_text(idlib_sql3, p_db, p_stmt, 0)
    print("1 SQLite Version:" + buffer2)

    ozz_sql3.sqlite3_finalize(idlib_sql3, p_db, p_stmt)

    ozz_sql3.sqlite3_close(idlib_sql3, p_db)

    return None
## END main

# Console Pause wrapper.
def Con_Pause():
    dummy = ""
    print("")
    dummy = input("Press [Enter] key to continue...")
    return None

if __name__ == '__main__':
    main()

ozz_sql3.py (Basic wrapper)

#-------------------------------------------------------------------------------
# Name:         ozz_sql3.py (based upon basics_2.c, ozz_sql3.h)
# Purpose:      SQLite3 Basic examples module.
#               Convenience wrapper functions for SQLite version 3.
#
# Platform:     Win64, Ubuntu64
# Depends:      SQLite v3.34.1 plus (dll/so), ctypes, sys, os
#
# Author:       Axle
#
# Created:      12/05/2023
# Updated:
# Copyright:    (c) Axle 2023
# Licence:      MIT-0 No Attribution
#-------------------------------------------------------------------------------
# Notes:
# Using the SQLite shared object (.dll, .so) directly as a Run-time library. The
# sqlite3.dll/.so must be in the system or application path.
#
# Python 3 built in SQLite3 library is a better/safer approach but uses a
# distinctly different API to the default C API which goes against the
# primary goal of exemplifying the same code routines in all 3 languages.
# As such I am using the Ctypes module for direct access to the shared libraries
# (.dll, .so) exposed C API. In essence python types are translated to C types
# for use by the C based shared object, and then C types are converted back to
# Python types when data is returned. This happens by default with most Python
# library modules but occurs in a more opeque manner in the background.
#
#-------------------------------------------------------------------------------

import ctypes, sys, os

# https://www.digitalocean.com/community/tutorials/how-to-write-modules-in-python-3
# https://gist.github.com/michalc/a3147997e21665896836e0f4157975cb

## ====>> Error Constants
# Create full list of error constants.
SQLITE_OK = 0  # Define the sqlite error codes
SQLITE_ERROR = 1   # Generic error
SQLITE_MISUSE = 21   # Library used incorrectly
SQLITE_ROW = 100  # Define the sqlite error codes
SQLITE_DONE = 101
SQLITE_TRANSIENT = -1

SQLITE_OPEN_READONLY = 0x00000001  # Ok for sqlite3_open_v2()
SQLITE_OPEN_READWRITE = 0x00000002  # Ok for sqlite3_open_v2()
SQLITE_OPEN_CREATE = 0x00000004  # Ok for sqlite3_open_v2()

## ====>> Ctype structures
# Create an sqlite3 class (struct)
# sqlite3 *p_db;  # database handle (structure).
class sqlite3(ctypes.Structure):
    _fields_ = ()  # opaque structure
# !!! I am uncertain if I should make this part of the class !!!
def p_sqlite3():  # p_db
    return ctypes.POINTER(sqlite3)()  # Create a C pointer to the class (struct)


# Create an sqlite3_stmt class (struct)
#sqlite3_stmt *statement;  # structure represents a single SQL statement
class sqlite3_stmt(ctypes.Structure):
    _fields_ = ()  # opaque structure
# !!! I am uncertain if I should make this part of the class !!!
def p_sqlite3_stmt():  # p_stmt
    return ctypes.POINTER(sqlite3_stmt)()  # Create a C pointer to the class (struct)

## ====>> Load SQLite 3 library

# The correct library path must be provided here.
def get_libsql3_path():
    #print(sys.path)
    #print(sys.path[0])
    # for windows
    if os.name == 'nt':
        f_library = os.path.join(sys.path[0], "sqlite3.dll")
        #f_library = "D:\\SQLite3Tests\\Py\\sqlite3.dll"
    # for mac and linux
    elif os.name == 'posix':
        #f_library = os.path.join(sys.path[0], "libsqlite3.so.0.8.6")  # Not recomended
        f_library = "libsqlite3.so"
    else:  # Other OS
        pass
    return f_library

def load_libsql3(f_library):
    return ctypes.cdll.LoadLibrary(f_library)

def handle_libsql3(idlib_sql3):  # hlib_sql3
    return idlib_sql3._handle

## ====>> Start SQLite3 C API ctype conversions for each function.
# define the types for the C functions. Returns are converted to/from bbyte/string

# functions called from a shared library (dll, so) must be defined using Ctypes.
# This is somewhat advanced but wanted to show both the background methods
# of constructing a basic binder interface between Python and a C
# shared object (dll, so) as well as keep the same sqlite C API for the example.
# In practice it will be far easier to use the built in, well tested and
# safer Python DB-API 2.0 interface or APSW, although the API differs
# somewhat from the native C API.

def sqlite3_libversion(idlib_sql3):
    # const char *sqlite3_libversion(void);
    idlib_sql3.sqlite3_libversion.argtypes = None  # No argements are sent to the C function (aka function(void);)
    idlib_sql3.sqlite3_libversion.restype = ctypes.c_char_p  # returns char* = ctypes.c_char_p
    return idlib_sql3.sqlite3_libversion().decode('utf-8')  # Convert b'' to utf-8 str

# note that the first 2 functions require access to the class (structures)
# "By Reference" so that the class (strcut) can be assigned data values.
# the following fuctions only have to see/read the data thus the
# ctypes.POINTER(ctypes.POINTER(sqlite3)) vs ctypes.c_void_p
def sqlite3_open(idlib_sql3, db_filename, p_db):
    # global p_db
    # int sqlite3_open(
    #                   const char *filename,   /* Database filename (UTF-8) */
    #                   sqlite3 **ppDb          /* OUT: SQLite db handle */
    #                   );
    idlib_sql3.sqlite3_open.argtypes = [ctypes.c_char_p, ctypes.POINTER(ctypes.POINTER(sqlite3))]
    idlib_sql3.sqlite3_open.restype = ctypes.c_int

    # const char *sqlite3_errmsg(sqlite3*);
    idlib_sql3.sqlite3_errmsg.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_errmsg.restype = ctypes.c_char_p

    b_db_filename = db_filename.encode('utf-8')  # encode our string to C byte array.

    return_code = idlib_sql3.sqlite3_open(b_db_filename, ctypes.byref(p_db))  # Open Memory (RAM) data base.
    if return_code != SQLITE_OK:  # int 0
        print("Cannot open database: " + str(idlib_sql3.sqlite3_errmsg(p_db).decode('utf-8')) + " | " + str(return_code), file=sys.stderr);  # DEBUG
        return -1
    return 1

def sqlite3_prepare_v2(idlib_sql3, p_db, sql_query, nByte, p_stmt, pzTail):
    b_sql1 = sql_query.encode('utf-8')

    # int sqlite3_prepare_v2(
    #                           sqlite3 *db,            /* Database handle */
    #                           const char *zSql,       /* SQL statement, UTF-8 encoded */
    #                           int nByte,              /* Maximum length of zSql in bytes. */
    #                           sqlite3_stmt **ppStmt,  /* OUT: Statement handle */
    #                           const char **pzTail     /* OUT: Pointer to unused portion of zSql */
    #                           );
    idlib_sql3.sqlite3_prepare_v2.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int, ctypes.POINTER(ctypes.POINTER(sqlite3_stmt)), ctypes.POINTER(ctypes.c_char_p)]
    idlib_sql3.sqlite3_prepare_v2.restype = ctypes.c_int

    # const char *sqlite3_errmsg(sqlite3*);
    idlib_sql3.sqlite3_errmsg.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_errmsg.restype = ctypes.c_char_p

    return_code = idlib_sql3.sqlite3_prepare_v2(p_db, b_sql1, nByte, ctypes.byref(p_stmt), pzTail)
    # On success, sqlite3_prepare_v2 returns SQLITE_OK; otherwise an error code
    # is returned.
    if return_code != SQLITE_OK:
        # This is error handling code for the sqlite3_prepare_v2 function call.
        print("Failed to prepare data: " + str(idlib_sql3.sqlite3_errmsg(p_db).decode('utf-8')), file=sys.stderr);  # DEBUG
        idlib_sql3.sqlite3_close(p_db)
        return -1

    return 1

def sqlite3_step(idlib_sql3, p_db, p_stmt):


    # int sqlite3_step(sqlite3_stmt*);
    idlib_sql3.sqlite3_step.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_step.restype = ctypes.c_int

    # const char *sqlite3_errmsg(sqlite3*);
    idlib_sql3.sqlite3_errmsg.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_errmsg.restype = ctypes.c_char_p

    return_code = idlib_sql3.sqlite3_step(p_stmt)
    if return_code == SQLITE_ROW:
        return 0
    else:
        print("Step error: " + str(idlib_sql3.sqlite3_errmsg(p_db).decode('utf-8')) + " | " + str(return_code), file=sys.stderr);  # DEBUG
        idlib_sql3.sqlite3_close(p_db)
        return 0
    return 1

def sqlite3_column_text(idlib_sql3, p_db, p_stmt, iCol):

    # const unsigned char *sqlite3_column_text(sqlite3_stmt*, int iCol);
    idlib_sql3.sqlite3_column_text.argtypes = [ctypes.c_void_p, ctypes.c_int]
    idlib_sql3.sqlite3_column_text.restype = ctypes.c_char_p  # c_ubyte

    # const unsigned char *sqlite3_column_text(sqlite3_stmt*, int iCol);
    # iCol refers to the current column in the return data. In this case
    # there is only one column of return value, so we know the zero column
    # contains the version number.
    return idlib_sql3.sqlite3_column_text(p_stmt, iCol).decode('utf-8')
    #ret_version.append( str(str_buffer.decode('utf-8')))
    #print("DEBUG" + ret_version[0])


def sqlite3_finalize(idlib_sql3, p_db, p_stmt):

    # int sqlite3_finalize(sqlite3_stmt *pStmt);
    idlib_sql3.sqlite3_finalize.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_finalize.restype = ctypes.c_int

    # const char *sqlite3_errmsg(sqlite3*);
    idlib_sql3.sqlite3_errmsg.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_errmsg.restype = ctypes.c_char_p

    # The sqlite3_finalize function destroys the prepared statement object and
    # commits the changes to the databse file.
    return_code = idlib_sql3.sqlite3_finalize(p_stmt)
    if return_code != SQLITE_OK:
        # This is error handling code.
        print("Failed to finalize data: " + str(idlib_sql3.sqlite3_errmsg(p_db).decode('utf-8')) + " | " + str(return_code), file=sys.stderr);  # DEBUG
        idlib_sql3.sqlite3_close(p_db)
        return -1
    return 1

def sqlite3_close(idlib_sql3, p_db):
    # int sqlite3_close(sqlite3*);
    idlib_sql3.sqlite3_close.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_close.restype = ctypes.c_int

    # const char *sqlite3_errmsg(sqlite3*);
    idlib_sql3.sqlite3_errmsg.argtypes = [ctypes.c_void_p]
    idlib_sql3.sqlite3_errmsg.restype = ctypes.c_char_p

    # The sqlite3_close function closes the database connection.
    return_code = idlib_sql3.sqlite3_close(p_db)
    if return_code != SQLITE_OK:
        # This is error handling code. NOTE! As p_db is closed the error code may not be available!
        print("Failed to close database: " + str(idlib_sql3.sqlite3_errmsg(p_db).decode('utf-8')) + " | " + str(return_code), file=sys.stderr);  # DEBUG
        return -1
    return 1

Output:

>>> 
*** Remote Interpreter Reinitialized ***
1 SQLite Version:3.34.1
===========================================
1 SQLite Version:3.34.1
>>>

I think this will work and be extendable to other shared libraries as a naive example and it follows with my C examples (almost) lol
Thank you very much for the suggestions Michal as they were useful. I will look more closely at getting this wrapped up a bit tighter before posting it up on github.
Axle

@Axle-Ozz-i-sofT
Copy link

P.S. I have gained a lot from your example /jsok/ above. That has offered a lot of hints :)
Thank you.

@michalc
Copy link
Author

michalc commented May 12, 2023

No problem!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment