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

Diff of /gnue-common/src/datasources/drivers/interbase/interbase/DataObject.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 26  Line 26 
26  # Requires Kinterbasdb > 3.0 (http://kinterbasdb.sourceforge.net/)  # Requires Kinterbasdb > 3.0 (http://kinterbasdb.sourceforge.net/)
27  #  #
28  # NOTES:  # NOTES:
 #  
 #   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)  
 #  
29    
30    
   
 from string import upper, lower, rstrip  
31  import sys  import sys
32  from gnue.common.datasources import GDataObjects, GConditions  from gnue.common.datasources import GDataObjects
33  from gnue.common.apps import GDebug  from gnue.common.apps import GDebug
34  from gnue.common.datasources.drivers.DBSIG2.Driver \  from gnue.common.datasources.drivers import DBSIG2
    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()  
35    
36    class _Base(DBSIG2.DataObject):  
37      # The date/time format used in insert/select statements
38      # (based on format used for time.strftime())
39      _dateTimeFormat = "cast('%Y-%m-%d %H:%M:%S' as timestamp)"
40    
41    def _createResultSet(self, conditions={}, readOnly=0, masterRecordSet=None,sql=""):    def _createResultSet(self, conditions={}, readOnly=0, masterRecordSet=None,sql=""):
42    
# Line 125  class Interbase_DataObject(DBSIG2.DataOb Line 70  class Interbase_DataObject(DBSIG2.DataOb
70      return rs      return rs
71    
72    
73    class DataObject_Object(_Base, DBSIG2.DataObject_Object):
74      pass
75    
76    #  class DataObject_SQL(_Base, DBSIG2.DataObject_SQL):
77    # Schema (metadata) functions    pass
   #  
   
   # 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)  
   
     cursor = self._dataConnection.cursor()  
     cursor.execute(statement)  
   
     rs = cursor.fetchone()  
     if rs:  
       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  
   
     cursor.close()  
     return schema  
   
   # 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  
   
 ######################################  
 #  
 #  The following hashes describe  
 #  this driver's characteristings.  
 #  
 ######################################  
   
 #  
 #  All datasouce "types" and corresponding DataObject class  
 #  
 supportedDataObjects = {  
   'object': Interbase_DataObject_Object,  
   '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