0001 from win32com.client import Dispatch, DispatchWithEvents, constants 0002 import pythoncom 0003 import os 0004 import time 0005 0006 finished = 0 # Flag for the wait loop from (3) to test 0007 0008 class ADOEvents: # event handler class 0009 def OnWillConnect(self, str, user, pw, opt, sts, cn): 0010 # Must have this event, as if it is not handled, ADO assumes the 0011 # operation is cancelled, and raises an error (Operation cancelled 0012 # by the user) 0013 pass 0014 def OnConnectComplete(self, error, status, connection): 0015 # Assume no errors, until we have the basic stuff 0016 # working. Now, "connection" should be an open 0017 # connection to my data source 0018 # Do the "something" from (2). For now, just 0019 # print the connection data source 0020 print "connection is", connection 0021 print "Connected to", connection.Properties("Data Source") 0022 # OK, our work is done. Let the main loop know 0023 global finished 0024 finished = 1 0025 def OnCommitTransComplete(self, pError, adStatus, pConnection): 0026 pass 0027 def OnInfoMessage(self, pError, adStatus, pConnection): 0028 pass 0029 def OnDisconnect(self, adStatus, pConnection): 0030 pass 0031 def OnBeginTransComplete(self, TransactionLevel, pError, adStatus, pConnection): 0032 pass 0033 def OnRollbackTransComplete(self, pError, adStatus, pConnection): 0034 pass 0035 def OnExecuteComplete(self, RecordsAffected, pError, adStatus, pCommand, pRecordset, pConnection): 0036 pass 0037 def OnWillExecute(self, Source, CursorType, LockType, Options, adStatus, pCommand, pRecordset, pConnection): 0038 pass 0039 0040 def TestConnection(dbname): 0041 # Create the ADO connection object, and link the event 0042 # handlers into it 0043 c = DispatchWithEvents("ADODB.Connection", ADOEvents) 0044 0045 # Initiate the asynchronous open 0046 dsn = "Driver={Microsoft Access Driver (*.mdb)};Dbq=%s" % dbname 0047 user = "system" 0048 pw = "manager" 0049 c.Open(dsn, user, pw, constants.adAsyncConnect) 0050 0051 # Sit in a loop, until our event handler (above) sets the 0052 # "finished" flag or we time out. 0053 end_time = time.clock() + 10 0054 while time.clock() < end_time: 0055 # Pump messages so that COM gets a look in 0056 pythoncom.PumpWaitingMessages() 0057 if not finished: 0058 print "XXX - Failed to connect!" 0059 0060 def Test(): 0061 import testAccess 0062 try: 0063 testAccess.GenerateSupport() 0064 except pythoncom.com_error: 0065 print "*** Can not import the MSAccess type libraries - tests skipped" 0066 return 0067 dbname = testAccess.CreateTestAccessDatabase() 0068 try: 0069 TestConnection(dbname) 0070 finally: 0071 os.unlink(dbname) 0072 0073 if __name__=='__main__': 0074 Test() 0075
Generated by PyXR 0.9.4