Conexión con MS Access desde Python

Posted by in Microsoft Access, Python

Desde que me inicié con #Python, una de los primeros objetivos que tengo es deshacerme de MS Access como base de datos. Pero mientras, tengo que trabajar con ella. A pesar de que me ha sorprendido mucho la cantidad de módulos y paquetes para casi todo, parece que hay pocas opciones para trabajar con las bases de datos MS Access de Microsoft …. por aquello de que se les tienen un poco de manía. Así que te presento la alternativa que estoy usando.

He encontrado en ActiveState una clase (http://code.activestate.com/recipes/528868/) que facilita la conexión con MS Access, mediante el uso de Python for Windows Extension. Sin embargo, y pese a todo su potencial, hay que andar compilando una serie de librerías hasta conseguir que funcione. Yo aún no lo he probado, pero … si buscaba algo «simple», parece que éste no es mi camino!

Adodb

Entre las alternativas que he encontrado, destacar adodb.py. una librería abstracta para bases de datos, originalmente pensada para PHP que fué convertida a Python. Este proyecto lleva bastante tiempo estancado, pues desde 2008 no ha recibido ninguna mejora. Lo podeis descargar aquí.

La parte positiva es que esta clase facilita una rápida conexión con distintas bases de datos, ya sea MS Access, MySQL PostgreSQL, ORACLE, SQLite, … Pero entre las contras, yo destaco la necesidad de crear la cadena de conexión necesaria para cada caso, (ya os conté que odio las cadenas de conexión), y por otro lado, es un paquete que no está incluido ni en Python Portable ni en Anaconda, y que por tanto, hay que instalar previo a su utilización.

El tutorial para Python lo teneis aquí, y es …. bastante breve. Tengo que probarla, pero me da pereza que instalarla

Probando adodbpy

Quizás, la mejor alternativa que he encontrado es adodbpy, que ésta sí está incluida en la distribución Anaconda de Python, como podeis ver en la imagen siguiente.

adodb y adodbapi en Anaconda

adodb y adodbapi en Anaconda

Si sólo teneis instalado Python, podeis descargar el paquete aquí: adodbapi.sourceforge.net.

Adodbpy es un módulo similar a Microsoft ADO, pero para Python, y también requiere de las cadenas de conexión para su correcto funcionamiento.

Y para probar, que mejor que un pequeño ejemplo que encontré en Github, modificando la ruta de la base de datos y la consulta SQL.

usando adodbapi

usando adodbapi

y en acción:

probando adodbapi

probando adodbapi

Y … mi propia clase, basandome en adodbpy

Dado que el uso que voy a hacerle a adodbpy es muy similar en los proyectos con lo que estoy empezando a trabajar con bases de datos y Python, qué mejor que tener mi propia clase para reutilizarla. La podeis descargar de GitHub.


# coding: utf-8
__author__ = 'dtrillo'
import adodbapi

version = "0.8.3 - 20140605"
nl = "\n"

class MSAccessConnector():
    """ Clase para trabajar con bases de datos Microsoft Access en Python usando 'adodbapi'    """
    def __init__(self, fileMdb, debug=False):
        """ Inicializa OBJETO MSAccess """
        self.debug = debug
        self.ErrMsg, self.file, self.connString = '', '', ''
        self.consultas = []     # Listado de consultas realizadas
        self.isConnected = False
        self.openConnection(fileMdb)

    def openConnection(self, fileMdb):
        if fileMdb != self.file and len(fileMdb) > 0:
            if self.isConnected: self.close()
            self.file = fileMdb
            self.__createConnString()   # Crea conexion
            self.isConnected = self.__openConn()

    def __openConn(self):
        """ Abre Conexion con BdD para empezar a trabajar """
        if len(self.connString) == 0:
            self.ErrMsg = 'Sin Cadena de Conexion'
            self.conn = None
            return False
        if len(self.ErrMsg)>0: return False

        try:
            self.ErrMsg = ''
            self.conn = adodbapi.connect(self.connString, timeout=15) # connect to the database
            if self.debug: print "Conectado!"
            self.SQLList_empty()       # Reinicia Listado de consutlas SQL
            return True
        except Exception, e:
            self.ErrMsg = 'Error al abrir Conexion' + nl + str(e)
            self.conn = None
            return False

    def __createConnString(self):
        """ Crea Cadena de Conexion con MDB """
        if len(self.file) == 0:
            self.connString = ''
            return False
        else:
            #self.connString = 'Provider=Microsoft.Jet.OLEDB.4.0; Data Source=%s' % self.file
            self.connString = 'Provider=Microsoft.ACE.OLEDB.12.0; Data Source=%s;Persist Security Info=False;' % self.file
            return True

    def __del__(self):
        """ Al eliminar el OBJETO, cierra la conexion y elimina la conexion        """
        self.close()
        self.conn = None
        self.consultas = []

    def close(self):
        """ Cierra la conexion """
        try:
            self.conn.close()
        except:
            pass
        finally:
            self.SQLList_empty()
            self.isConnected = False
            if self.debug: print "Fin de Conexion!"

    @staticmethod
    def verificaSQLisSELECT(sql):
        word = "SELECT"
        tmp = sql[0:len(word)].upper()
        return True if word == tmp else False

    def query(self, sql):
        """ consultas SELECT """
        if self.isConnected == False or len(self.ErrMsg)>0:
            return [], True
        try:
            if self.verificaSQLisSELECT(sql) == False: # Versión 0.8.1
                self.ErrMsg = "Se esperaba una sentencia SELECT"
                return [], True

            self.ErrMsg = ''
            cur = self.conn.cursor()
            try:
                cur.execute(sql)
                result = cur.fetchall() # show the result
                cur.close() # close the cursor and connection
                self.__SQLList_add(sql)
                return result, False
            except Exception, f:
                self.ErrMsg = 'Error al ejecutar SQL ' + nl + str(f)
                return [], True
        except Exception, e:
            self.ErrMsg = 'Error al iniciar cursor!' + nl + str(e)
            return [], True
    def query2(self, sql):
        res, bln = self.query(sql)
        return res
    def query1value(self, sql, default = None):
        """ Devuelve el UNICO valor esperado de una consulta SQL """
        res = self.query2(sql)
        try:
            valor = res[0][0]
        except:
            valor = default
        return valor

    def execute(self, sql):
        """ consultas de accion """
        if self.isConnected == False or len(self.ErrMsg)>0: return False
        if len(sql) == 0: return False
        try:
            cur = self.conn.cursor()
            try:
                self.ErrMsg = ''
                cur.execute(sql)
                self.conn.commit()
                cur.close() # close the cursor and connection
                #self.__SQLList_add(sql)
                return True
            except Exception, f:
                self.ErrMsg = 'Error al ejecutar SQL ' + nl + str(f)
                return False
        except Exception, e:
            self.ErrMsg = 'Error al iniciar cursor!' + nl + str(e)
            return False

    # Trabajando con Lista de SQL realizadas
    def __SQLList_add(self, qsql):
        """ La consulta SQL se ha realizado con exito, y la guardo en el listado """
        if len(qsql) > 0:
            if not (qsql in self.consultas):
                self.consultas.append(qsql)
            else:
                self.consultas.remove(qsql)  # La quito de donde esta
                self.consultas.append(qsql)  # Pero la agrego al final

    def SQLList_last(self):
        """ Devuelve la ultima SQL realizada """
        return self.consultas[-1]

    def SQLList(self):
        """ Devuelve la lista de consultas SQL realizadas con la conexion ABIERTA """
        return self.consultas

    def SQLList_empty(self):
        """ Vacia el contenido de consultas SQL """
        self.consultas = []

Y un pequeño ejemplo: acceder a una base de datos MS Access y mostrar todos los registros de dicha tabla, según una consulta SQL. El código lo teneis disponible en GitHub además.


# coding: utf-8
""" Ejemplo de Conexion con MS Access """
miVersion = "Test de Conexion con MS Access"
nl = "\n"

from MSAccessConnector import *

database = "l:/sef_py2/sef2.8.mdb"
access = MSAccessConnector(database)
sql = "select * from equipos3 where equipo like 'd%'"

lista, error = access.query(sql)

if error:
print access.ErrMsg
else:
print "Encontrados %s equipos!!" % (len(lista))
for equipo in lista:
print equipo.equipo

Y en funcionamiento:

Conectando Python con MS Access

Conectando Python con MS Access

Si os salta algún error en windows 7 al trabajar con MS Access, aquí podeis ver cómo lo he resuelto en mi caso.

Espero que os sirva, y seguimos aprendiendo!