Jugando con SQLite3 y Python

written by uve 2 March 2011

Aunque hay muchos ejemplos por ahí, ahora me toca a mi poner uno :) Básicamente he hecho un pequeño wrapper sobre SQLite3 para almacenar información de forma persistente. La idea es muy simple: vamos a gestionar una lista de nodos (dirección MAC y dirección lógica).

La clase con la que vamos a trabajar es la siguiente:

class Node:
    def __init__(self, mac, address):
        self.mac = mac
        self.address = address

Lo primero, vamos a crear la base de datos desde el intérprete de SQLite3. Es posible que necesitemos instalarlo primero:

$ sudo apt-get install sqlite3

Las bibliotecas para trabajar con SQLite3 vienen instaladas siempre, pero el intérprete no tiene por qué. Una vez instalado, ahora sí vamos a crear la base de datos, node.db:

$ sqlite3 nodes.db
SQLite version 3.7.3
Enter ".help" for instructions
Enter SQL statements terminated with a ";"
sqlite> CREATE TABLE nodes (mac VARCHAR(8) PRIMARY KEY, address VARCHAR(2));
sqlite> .quit

También se puede crear desde Python, pero por comodidad lo he hecho de esta forma. Cada cuál que se sienta libre de hacerlo como quiera.

Python y SQLilte3

Y ahora vamos a ver como trabajar realmente con SQLite3.

Conexión con la base de datos

conn = sqlite3.connect('nodes.db') # Conexión con la base de datos
...
conn.commit() # Ver las notas finales
conn.close() # Se cierra la conexión

Consultas

Se hace uso del método execute del objeto Cursor. Este método nos permite ejecutar una única sentencia SQL.

cursor = conn.cursor()
cursor.execute('SELECT * FROM nodes')

Además, se pueden ejecutar sentencias parametrizadas:

params = ('1111',) # Lista de parámetros
cursor = conn.cursor()
cursor.execute('SELECT * FROM nodes WHERE mac=?', params)

La recuperación de los resultados se puede realizar de varias maneras:

cursor.exectue('SELECT ...')
print cursor.rowcount # Número de tuplas recuperadas
for row in cursor:
    ...

cursor.execute('SELECT ...')
row = cursor.fetchone() # Obtiene la siguiente tupla o None si no hay más

cursor.execute('SELECT ...')
rows = cursor.fetchmany(5) # Obtiene las 5 tuplas siguientes

cursor.execute('SELECT ...')
all_rows = cursor.fetchall() # Obtiene todas las tuplas restantes

El objeto sqlite3.Row

Por defecto, las tuplas recuperadas están representadas con el tipo de dato tuple de Python, es decir, disponemos de acceso a cada columna a través de su índice. Veamos un ejemplo:

cursor.execute('SELECT mac, address FROM node')

row = cursor.fetchone()
row[0] # Campo mac de la tupla
row[1] # Campo address de la tupla

Esto es poco versatil y reutilizable. Si cambiamos la consulta por SELECT address, mac FROM node') necesitaríamos cambiar el código, para acceder a las columnas de forma distinta. Para evitar esto, disponemos del objeto Row, que nos permite acceso a través del nombre del campo:

conn = sqlite3.connect('nodes.db')
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('SELECT mac, address FROM node')
row = cursor.fetchone()
row['mac']
row['address']
row.keys() # Lista con los campos disponibles: ['mac', 'address']

Otras sentencias SQL

A través del método execute, explicado anteriormente, se puede ejecutar cualquier tipo de sentencia SQL. Por ejemplo, DELETEs e INSERTs:

values = ('1122', '12',)
cursor = self.conn.cursor()
cursor.execute('INSERT INTO nodes VALUES(?,?)', values)

values = ('1122',)
cursor = self.conn.cursor()
cursor.execute('DELETE FROM nodes WHERE mac=?', values)

Además hay disponibles otros métodos disponibles para ejecutar sentencias, como executescript ó executemany, los cuales no voy a entrar en detalle.

Preparando un wrapper

Para tratar de abstraer todo el tema de acceso a la base de datos, propongo una clase que internamente se encargue de toda la gestión de SQLite3. De esta forma, si el día de mañana decidimos cambiar el mecanismo de almacenamiento persistente, será tan sencillo como cambiar la clase Database por otra.

class Database:
    def __init__(self, filename='nodes.db'):
        self.conn = sqlite3.connect(filename)
        self.conn.row_factory = sqlite3.Row
    
    def close(self):
        if self.conn:
            self.conn.commit()
            self.conn.close()
            self.conn = None
    
    def add_node(self, node):
        values = (node.mac, node.address,)
        cursor = self.conn.cursor()
        cursor.execute('INSERT INTO nodes VALUES(?,?)', values)
    
    def remove_node(self, mac):
        values = (mac,)
        cursor = self.conn.cursor()
        cursor.execute('DELETE FROM nodes WHERE mac=?', values)
    
    def node_list(self):
        cursor = self.conn.cursor()
        cursor.execute('SELECT * FROM nodes')
        return [ Node(row['mac'], row['address']) for row in cursor.fetchall() ]
    
    def locate_node(self, mac):
        values = (mac,)
        cursor = self.conn.cursor()
        cursor.execute('SELECT * FROM nodes WHERE mac=?', values)
        node = cursor.fetchone()
        if node:
            node = Node(node['mac'], node['address'])
        return node

Algunas consideraciones a tener en cuenta:

Si no se realiza un commit antes de cerrar la base de datos, no tendrá efecto ninguna operación realizada
Hasta que no se realice un commit, ninguna otra conexión a la base de datos podrá ver los cambios que se han realizado