Connect to MS Access with Python

Since I started to program with #Python, one of the first objectives I have is to stop using MS Access. But, until them, I must work with it. Although I am surprised for the huge amount of modules and packages of Python, no one seems to fit with MS Access, and the few that I found doesn’t seem easy …. at first. So, let’s go!

I found a class on ActiveState (http://code.activestate.com/recipes/528868/) to connect to MS Access, by using Python for Windows Extension, but although it’s very powerful, you need to compile a few libraries before start. Too complicated for beginers, so I haven’t tried, because I am looking for something “simple”, and that’s not what I expected.

Adodb

Another option I found is adodb.py, an abstract library for databases, originally was developed for PHP, but it was migrated yo Python. The project seems not to be active, because since 2008 there is no update, but you can download it here.

The positive aspect of this classs is the hability to create a connection very easy with several types of databases, like MS Access, MySQL PostgreSQL, ORACLE, SQLite, … but one of the aspect I don’t like is that you need a connection string (and I already told you that I hate connection strings). Another issue is that the library is not included neither in Python Portable ni en Anaconda, so, you need to install it before.

Here you have a tutorial for Python … not very large. I should try, but I was not in the mood …. maybe other day ….

Trying adodbpy

Maybe, the best alternative I found is adodbpy, included on the Anaconda distribution, as you can see in the next image.

If you have only Python installed (no Python distribution), you can download the package here: adodbapi.sourceforge.net.

Adodbpy is a module close to Microsoft ADO, but specifically for Python, and to use it, you need connection strings.

Let’s try it with a small example I found on Github, just modifying the database path and the SQL query.

and … in action:

And … I create my own class, based on adodbpy

Because I will use adodbpy a lot, and almost in the same way in several projects (and of course, because I’m starting with databases and Python), it’s a good idea to create my own class for working with MS Access and Python, and reuse it in several projects. You can download the class on my 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 = []

And of course, a small examples: connect to a MS Access database and show the records based on a SQL query. The code is also available on GitHub.

# 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

And .. running:

If you got an error on Windows 7 when working with MS Access, here you can read how I fixed it!

I hope it helps you, and have a nice day!