/[gnue]/gnue-common/src/datasources/drivers/interbase/interbase/ResultSet.py
ViewVC logotype

Diff of /gnue-common/src/datasources/drivers/interbase/interbase/ResultSet.py

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.1.2.1 by jcater, Fri Oct 10 01:21:19 2003 UTC revision 1.1.2.2 by siesel, Wed Nov 19 22:05:47 2003 UTC
# Line 27  Line 27 
27  #  #
28  # NOTES:  # NOTES:
29  #  #
 #   Supported attributes (via connections.conf or <database> tag)  
 #  
 #     host=      This is the Interbase host for your connection  (required)  
 #     dbame=      This is the Interbase database to use (required)  
 #  
   
   
   
 from string import upper, lower, rstrip  
 import sys  
 from gnue.common.datasources import GDataObjects, GConditions  
 from gnue.common.apps import GDebug  
 from gnue.common.datasources.drivers.DBSIG2.Driver \  
    import DBSIG2.RecordSet, DBSIG2.ResultSet, DBSIG2.DataObject, \  
           DBSIG2.DataObject_SQL, DBSIG2.DataObject_Object  
   
 try:  
   import kinterbasdb as SIG2api  
 except ImportError, message:  
   tmsg = _("Driver not installed: Kinterbasdb for Interbase [%s]") % message  
   raise GConnections.AdapterNotInstalled, tmsg  
   
   
 class Interbase_RecordSet(DBSIG2.RecordSet):  
   pass  
   
   
 class Interbase_ResultSet(DBSIG2.ResultSet):  
   def __init__(self, dataObject, cursor=None, defaultValues={}, masterRecordSet=None):  
     DBSIG2.ResultSet.__init__(self, dataObject, \  
             cursor, defaultValues, masterRecordSet)  
     self._recordSetClass = Interbase_RecordSet  
   
   
 class Interbase_DataObject(DBSIG2.DataObject):  
   def __init__(self):  
     DBSIG2.DataObject.__init__(self)  
     self._DatabaseError = SIG2api.DatabaseError  
     self._resultSetClass = Interbase_ResultSet  
     self._primaryKeyFields = []  
   
     # The date/time format used in insert/select statements  
     # (based on format used for time.strftime())  
     self._dateTimeFormat = "cast('%Y-%m-%d %H:%M:%S' as timestamp)"  
   
   def connect(self, connectData={}):  
     GDebug.printMesg(1,"Interbase database driver initializing")  
   
     try:  
       charset = connectData['charset']  
     except KeyError:  
       charset = ""  
   
     try:  
       self._dataConnection = SIG2api.connect( \  
                  user=str(connectData['_username']), \  
                  password=str(connectData['_password']), \  
                  charset=charset, \  
                  database=connectData['dbname'], \  
                  host=connectData['host'])  
     except self._DatabaseError, value:  
       raise GDataObjects.LoginError, value  
   
     self._postConnect()  
   
   
   def _createResultSet(self, conditions={}, readOnly=0, masterRecordSet=None,sql=""):  
   
     # Used by drivers with a unique id (like rowid)  
     if not self._primaryIdChecked: self._checkForPrimaryId()  
   
     try:  
       cursor = self._dataConnection.cursor()  
   
       cursor.arraysize = self.cache  
       cursor.execute(self._buildQuery(conditions, additionalSQL=sql))  
   
       # pull a record count  
       if self._strictQueryCount:  
 #        recordCount = cursor.rowcount  
 #        #disable the count query and see if anyone screams  
 #        #recordCount = self._getQueryCount(conditions,sql)  
   
         #kinterbasdb screams :(  
         recordCount = self._getQueryCount(conditions,sql)  
           
     except self._DatabaseError, err:  
       raise GDataObjects.ConnectionError, err  
   
     rs = self._resultSetClass(self, cursor=cursor, masterRecordSet=masterRecordSet)  
     if self._strictQueryCount:  
       rs._recordCount = recordCount  
     if readOnly:  
       rs._readonly = readOnly  
   
     return rs  
   
   
   
   #  
   # Schema (metadata) functions  
   #  
   
   # Return a list of the types of Schema objects this driver provides  
   def getSchemaTypes(self):  
     return [('view',_('Views'),1),  
             ('table',_('Tables'),1)]  
   
   # Return a list of Schema objects  
   def getSchemaList(self, type=None):  
   
   # This excludes any system tables and views.  
     statement = "select rdb$relation_name, rdb$view_source "+\  
                         "from rdb$relations " + \  
                         "where rdb$system_flag=0 " + \  
                         "order by rdb$relation_name"  
   
     cursor = self._dataConnection.cursor()  
     cursor.execute(statement)  
   
   # TODO: rdb$view_source is null for table and rdb$view_source is not null for view  
     list = []  
     for rs in cursor.fetchall():  
       list.append(GDataObjects.Schema(attrs={'id':rs[0], 'name':rstrip(rs[0]),  
                          'type':'table',  
                          'primarykey': self.__getPrimaryKey(rstrip(rs[0]))},  
                          getChildSchema=self.__getFieldSchema))  
   
     cursor.close()  
     return list  
   
   # Find a schema object with specified name  
   def getSchemaByName(self, name, type=None):  
   
     statement = "select rdb$relation_name, rdb$view_source "+\  
                         "from rdb$relations " + \  
                         "where rdb$relation_name = '%s'" % (name)  
30    
31      cursor = self._dataConnection.cursor()  __all__ = ['RecordSet']
     cursor.execute(statement)  
32    
33      rs = cursor.fetchone()  import string
34      if rs:  from gnue.common.datasources.drivers import DBSIG2
       schema = GDataObjects.Schema(attrs={'id':rs[0], 'name':rstrip(rs[0]),  
                            'type':'table',  
                            'primarykey': self.__getPrimaryKey(rstrip(rs[0]))},  
                            getChildSchema=self.__getFieldSchema)  
     else:  
       schema = None  
35    
     cursor.close()  
     return schema  
36    
   # Return a list of fields (for _buildDeleteStatement and for _buildUpdateStatement)  
   def __getPrimaryKey(self, relname):  
     statement = "select rdb$relation_name, rdb$field_name, "+\  
                                    "rdb$constraint_name, rdb$field_position "+\  
                                     "from rdb$relation_constraints rc, rdb$index_segments ri "+\  
                                     "where ri.rdb$index_name = rc.rdb$index_name "+\  
                                             "and rc.rdb$constraint_type = 'PRIMARY KEY' "+\  
                                             "and rc.rdb$relation_name = '%s' " % (relname)+\  
                                     "order by ri.rdb$field_position"  
   
     cursor = self._dataConnection.cursor()  
     cursor.execute(statement)  
   
     list = []  
     for rs in cursor.fetchall():  
       list.append(lower(rstrip(rs[1])))  
   
     cursor.close()  
     return list  
   
   # Get fields for a table  
   def __getFieldSchema(self, parent):  
   
     statement = "select * from %s"%(parent.name) + " where (0=1)"  
   
     cursor = self._dataConnection.cursor()  
     cursor.execute(statement)  
   
     list = []  
   
     for d in cursor.description:  
       try:  
         nativetype = lower(d[SIG2api.DESCRIPTION_TYPE_CODE].__name__)  
       except AttributeError:  
         nativetype='unknown'  
         
       attrs={'id':d[SIG2api.DESCRIPTION_NAME],  
                  'name':lower(d[SIG2api.DESCRIPTION_NAME]),  
                  'type':'field',  
                  'nativetype': nativetype,  
                  'required': d[SIG2api.DESCRIPTION_NULL_OK]==0,  
                  'length': d[SIG2api.DESCRIPTION_DISPLAY_SIZE]}  
   
       if nativetype in ('int','float','long'):  
         attrs['datatype']='number'  
         attrs['precision']=d[SIG2api.DESCRIPTION_SCALE]  
       elif nativetype == 'tuple':  
         attrs['datatype']='date'  
       else:  
         attrs['datatype']='text'  
   
       cursor.execute("select rdb$default_source from rdb$relation_fields"+ \  
                   " where rdb$relation_name = '%s' " % (parent.name)+ \  
                   " and rdb$field_name = '%s'" % (upper(attrs['name'])))  
       defrs = cursor.fetchone()  
       if defrs[0]:  
         dflt = defrs[0]  
         if dflt[9:12] == "NOW":  
           attrs['defaulttype'] = 'timestamp'  
         else:  
           attrs['defaulttype'] = 'constant'  
           attrs['defaultval'] = dflt[8:]  
   
       list.append(GDataObjects.Schema(attrs=attrs))  
   
     cursor.close()  
     return list  
   
   def _postConnect(self):  
     self.triggerExtensions = TriggerExtensions(self._dataConnection)  
   
   
 class Interbase_DataObject_Object(Interbase_DataObject, \  
       DBSIG2.DataObject_Object):  
   
   def __init__(self):  
     Interbase_DataObject.__init__(self)  
   
   def _buildQuery(self, conditions={},forDetail=None,additionalSQL=""):  
     return DBSIG2.DataObject_Object._buildQuery(self, conditions,forDetail,\  
                                                 additionalSQL)  
   
   
 class Interbase_DataObject_SQL(Interbase_DataObject, \  
       DBSIG2.DataObject_SQL):  
   def __init__(self):  
     # Call DBSIG init first because Interbase_DataObject needs to overwrite  
     # some of its values  
     DBSIG2.DataObject_SQL.__init__(self)  
     Interbase_DataObject.__init__(self)  
   
   def _buildQuery(self, conditions={},forDetail=None,additionalSQL=""):  
     return DBSIG2.DataObject_SQL._buildQuery(self, conditions, forDetail,\  
                                             additionalSQL)  
   
   
 #  
 #  Extensions to Trigger Namespaces  
 #  
 class TriggerExtensions:  
   
   def __init__(self, connection):  
     self.__connection = connection  
   
   # Return the current date, according to database  
   def getTimeStamp(self):  
     return self.__singleQuery("select cast('now' as date) from rdb$database")  
   
   # Return a sequence number from sequence 'name'  
   def getSequence(self, name):  
     return self.__singleQuery("select gen_id(%s,1) from rdb$database" % name)  
   
   # Run the SQL statement 'statement'  
   def sql(self, statement):  
     cursor = self.__connection.cursor()  
     try:  
       cursor.execute(statement)  
       cursor.close()  
     except:  
       cursor.close()  
       raise  
   
   # Used internally  
   def __singleQuery(self, statement):  
     cursor = self.__connection.cursor()  
     try:  
       cursor.execute(statement)  
       rv = cursor.fetchone()  
       cursor.close()  
     except mesg:  
       GDebug.printMesg(1,"**** Unable to execute extension query")  
       GDebug.printMesg(1,"**** %s" % mesg)  
       cursor.close()  
       return None  
   
     try:  
       return rv[0]  
     except:  
       return None  
   
 ######################################  
37  #  #
 #  The following hashes describe  
 #  this driver's characteristings.  
 #  
 ######################################  
   
38  #  #
 #  All datasouce "types" and corresponding DataObject class  
39  #  #
40  supportedDataObjects = {  class RecordSet(DBSIG2.RecordSet):
41    'object': Interbase_DataObject_Object,    pass
   'sql':    Interbase_DataObject_SQL  
 }  
   
   
   

Legend:
Removed from v.1.1.2.1  
changed lines
  Added in v.1.1.2.2

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26