Welcome, guest | Sign In | My Account | Store | Cart

Python DAO interface class (Python recipe) by Larry Bates
ActiveState Code (http://code.activestate.com/recipes/303349/)

This is a simple base class that can be used to quickly get connected to any database that has DAO interface. I've used it quite a lot with MS Access databases and it has worked quite well for me. It really is just a Python wrapper around the DAO COM object. Hasn't been tested extensively, but worked in the projects that I've used it in so far.

Larry Bates

Python, 126 lines
Download
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
import os
import sys
import win32com.client

class daoBaseClass:
    '''
    This base class is used to read/write data from/to daoDatabase.  It
    is necessarry that makepy be run against Microsoft DAO 3.6 Object
    Library (5.0) prior to using this class.
    '''
    #
    # Completely free, no warranties.
    #
    def __init__(self, databasepath, SQL_query=None, logf=None, _trace=0, _debug=0):
        #----------------------------------------------------------------------
        # Create an attribute to hold EOF status
        #----------------------------------------------------------------------
        self.EOF=0
        self._AddNew=0
        self.path=databasepath
        self.logf=logf
        self._trace=_trace
        self._debug=_debug
        #
        # Support both MoveNext and fetchone with same method
        #
        self.fetchone=self.MoveNext
        #
        # Dispatch the DAO Engine
        #
        try:    self.daoEngine=win32com.client.Dispatch("DAO.DBEngine.36")
        except:
            emsg="Unable to dispatch DAO.DBEngine.36"
            self.abort(emsg)
            
        #----------------------------------------------------------------------
        # Try to open the database
        #----------------------------------------------------------------------
        try:     self.daoDB = self.daoEngine.OpenDatabase(self.path)
        except: 
            emsg="Unable open database file=%s" % self.path
            self.abort(emsg)

        #----------------------------------------------------------------------
        # If query is specified, open a recordset with it
        #----------------------------------------------------------------------
        if SQL_query: self.execute(SQL_query)
        return

    def __getitem__(self, key):
        #----------------------------------------------------------------------
        # Try to get the Value for the field requested from the recordset
        #----------------------------------------------------------------------
        try:    return self.daoRS.Fields(key).Value
        except:
            emsg="Field name '%s' not found in current record" % key
            if self.logf: self.logf.writelines("E", emsg)
            else:         print emsg
            return None

    def __setitem__(self, key, value):
        #----------------------------------------------------------------------
        # If I'm adding new information to a record, skip call to Edit()
        #----------------------------------------------------------------------
        if not self._AddNew: self.daoRS.Edit()
        #----------------------------------------------------------------------
        # Try to set the value for the field requested in the recordset
        #----------------------------------------------------------------------
        try:    self.daoRS.Fields(key).Value=value
        except:
            emsg="Unable to set Field name '%s' to value=%s" % (key, str(value))
            if self.logf: self.logf.writelines("E", emsg)
            else:         print emsg
        
        else:
            #----------------------------------------------------------------------
            # If I'm adding new information to a record, skip call to Update()
            #----------------------------------------------------------------------
            if not self._AddNew:  self.daoRS.Update()
            
        return

    def execute(self, SQL_query):
        '''
        This method is used to execute a SQL_query against the TimePilot
        database.
        '''
        try: self.daoRS = self.daoDB.OpenRecordset(SQL_query)
        except:
            emsg="Unable execute SQL_query=%s" % SQL_query
            if self.logf: self.logf.writelines("E", emsg)
            else:         print emsg

        self.EOF=self.daoRS.EOF            
        return
    
    def MoveNext(self):
        self.daoRS.MoveNext()
        self.EOF=self.daoRS.EOF
        return

    def AddNew(self):
        #----------------------------------------------------------------------
        # Sets a flag so that __setitem__ will know that I'm adding new
        # information not editing existing information.
        #----------------------------------------------------------------------
        self._AddNew=1
        self.daoRS.AddNew()
        return

    def Update(self):
        #----------------------------------------------------------------------
        # Resets flag so I'm no longer in AddNew mode
        #----------------------------------------------------------------------
        self._AddNew=0
        self.daoRS.Update()
        return

    def close(self):
        self.daoRS.Close()
        self.daoDB.Close()

    def abort(self, emsg):
        if self.logf: self.logf.writelines("E", emsg)
        else:         print emsg
        sys.exit(emsg)
Tags: database

3 comments

Ramon Felciano 19 years, 7 months ago  # | flag

Examples? Any chance of getting a short sample snippet of how to use it (e.g. to connect to an Access DB)?

Larry Bates (author) 19 years, 6 months ago  # | flag

Example of using DAO class. Here is a class/example main program that reads/writes data from/to a TimePilot (electronic timeclock application) MS Access database files.

Hope it helps.

Larry Bates

import os

import daoBaseClass

class TimePilot(daoBaseClass):

'''

This class is used to read/write data from/to TimePilot data files (MS Access).

'''

def __init__(self, timepilotdatapath,         SQL_query=None):

    #

    # Call base class' __init__ method

    #

    daoBaseClass.__init__(self, timepilotdatapath,

SQL_query=SQL_query)

    return

if __name__ == "__main__":

SQL_query='SELECT [USER ID] as userid, [ACTIVE], ' \

        '[OT1 Total] as ot1, [OT2 Total] as ot2, ' \

        '[Week0] as hours, [In Late Count] as latecount ' \

        'FROM Employee '



timepilotdatapath=r"C:\Timepilot\TA\Test\TimePilot.mdb"

#

# Create a TimePilot class instance by passing it the path

# to where data is stored.

#

TP=TimePilot(timepilotdatapath, SQL_query=SQL_query)



while not TP.EOF:

    print "userid=",TP['userid'], "hours=", TP['hours'], \

        "OT1=",TP['ot1'], "OT2=",TP['ot2'], \

        "latecount=",TP['latecount']

    TP.MoveNext()

    if TP.EOF:

        break



TP.close()

#----------------------------------------------------------------------

# Create a TimePilot class instance by passing it the path

# to where master data is stored.

#----------------------------------------------------------------------

timepilotdatapath=r"C:\Timepilot\TA\Test\TimePilot.mdb"

TP=TimePilot(timepilotdatapath)

#----------------------------------------------------------------------

# Create a new SQL_query that positions us at test record

# (userid=2) and updates some information there.

#----------------------------------------------------------------------

SQL_query="SELECT * FROM employee WHERE [USER ID]=2"

TP.execute(SQL_query)

TP['Address1']="111 Fair Oaks Lane"

TP['City']="Birmingham"

TP['State']="AL"

TP['Zip']="35406"

TP['Social Security']="421XXXXXX"

TP['Title']="Vice President"

#----------------------------------------------------------------------

# Create a new record in the database.

#----------------------------------------------------------------------

TP.AddNew()

TP['User ID']=5

TP['Display Name']="Unit test"

TP['Company']="Test"

TP['Department']="test1"

TP['Pay Type']="Hourly"

TP['Shift']="test"

TP['Holiday']="Holidays"

TP['Active']=-1

TP.Update()

TP.close()
Eric Walker 18 years, 7 months ago  # | flag

example error. when I put the example in a file called test.py and but the daoBaseClass in a file called daoBaseClass.py I get the following error: Traceback (most recent call last): File "C:\mypythontest\test.py", line 4, in ? class TimePilot(daoBaseClass): TypeError: Error when calling the metaclass bases module.__init__() takes at most 2 arguments (3 given)\

I made a sample access db called db1 and it doesn't work. any ideas? below is the code that I have.. import os import daoBaseClass

class TimePilot(daoBaseClass): ''' This class is used to read/write data from/to TimePilot data files (MS Access). ''' def __init__(self, timepilotdatapath, SQL_query=None): # # Call base class' __init__ method # daoBaseClass.__init__(timepilotdatapath, SQL_query=SQL_query) return

if __name__ == "__main__": SQL_query='SELECT [Field1] as sample1 FROM Employee '

timepilotdatapath=r"C:\db1.mdb"
#
# Create a TimePilot class instance by passing it the path
# to where data is stored.
#
TP=TimePilot(timepilotdatapath, SQL_query)

while not TP.EOF:
    print "sample1=",TP['sample1']
    TP.MoveNext()
    if TP.EOF:
        break

TP.close()
#----------------------------------------------------------------------
# Create a TimePilot class instance by passing it the path
# to where master data is stored.
#----------------------------------------------------------------------
timepilotdatapath=r"C:\Timepilot\TA\Test\TimePilot.mdb"
#TP=TimePilot(timepilotdatapath)
#----------------------------------------------------------------------
# Create a new SQL_query that positions us at test record
# (userid=2) and updates some information there.
#----------------------------------------------------------------------
#SQL_query="SELECT * FROM employee WHERE [USER ID]=2"
#TP.execute(SQL_query)
#TP['Address1']="111 Fair Oaks Lane"
#TP['City']="Birmingham"
#TP['State']="AL"
#TP['Zip']="35406"
#TP['Social Security']="421XXXXXX"
#TP['Title']="Vice President"
#----------------------------------------------------------------------
# Create a new record in the database.
#----------------------------------------------------------------------
#TP.AddNew()
#TP['User ID']=5
#TP['Display Name']="Unit test"
#TP['Company']="Test"
#TP['Department']="test1"
#TP['Pay Type']="Hourly"
#TP['Shift']="test"
#TP['Holiday']="Holidays"
#TP['Active']=-1
#TP.Update()
#TP.close()