/[gnue]/gnue-common/src/datasources/drivers/oracle/Base/RecordSet.py
ViewVC logotype

Diff of /gnue-common/src/datasources/drivers/oracle/Base/RecordSet.py

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

revision 1.1.2.1 by jcater, Fri Oct 10 01:21:23 2003 UTC revision 1.1.2.2 by jcater, Wed Nov 19 03:00:24 2003 UTC
# Line 36  import sys, string Line 36  import sys, string
36  from gnue.common.datasources import GDataObjects, GConditions  from gnue.common.datasources import GDataObjects, GConditions
37  from gnue.common.apps import GDebug  from gnue.common.apps import GDebug
38    
 class Oracle_Schema:  
   
   def __init__(self):  
     pass  
   
   def _postConnect(self):  
     self.triggerExtensions = TriggerExtensions(self._dataConnection)  
   
   #  
   # Schema (metadata) functions  
   #  
   
   # Return a list of the types of Schema objects this driver provides  
   def getSchemaTypes(self):  
     return [ ('user_table',   _('User Tables'),1),  
              ('user_view',    _('User Views'),1),  
              ('user_synonym', _('User Synonyms'),1),  
              ('all_table',    _('System Tables'),1),  
              ('all_view',     _('System Views'),1),  
              ('all_synonym',  _('System Synonyms'),1) ]  
   
   # Return a list of Schema objects  
   def getSchemaList(self, type=None):  
   
     where_user = ""  
     if type == None:  
       where_type = ['TABLE', 'VIEW', 'SYNONYM']  
     else:  
       scope, type = string.split(type,'_')  
       where_type = [string.upper(type)]  
       if scope == 'user':  
         where_user = " AND OWNER = USER"  
   
   
     statement = \  
       "select owner||'.'||table_name||'.'||table_type full_name, \n" + \  
       "  decode(owner,user,null,owner||'.')||table_name table_name, \n" + \  
       "  decode(owner,user,'user_','all_')||lower(table_type) table_type \n" + \  
       "  from all_catalog where table_type in ('%s') %s \n" \  
               % (string.join(where_type,"','"), where_user) + \  
       "  order by table_name "  
   
     GDebug.printMesg(5,statement)  
   
     cursor = self._dataConnection.cursor()  
     cursor.execute(statement)  
   
     list = []  
     for rs in cursor.fetchall():  
       list.append(GDataObjects.Schema(attrs={'id':rs[0], 'name':string.lower(rs[1]),  
                          'type':rs[2]},  
                          getChildSchema=self.__getFieldSchema))  
   
     cursor.close()  
     return list  
   
   
   # Find a schema object with specified name  
   def getSchemaByName(self, name, type=None):  
   
     spl = string.split(string.upper(name),'.')  
     where = "TABLE_NAME='%s'" % spl[-1]  
     if len(spl) > 1:  
       where += " AND OWNER='%s'" % spl[-2]  
   
     statement = \  
       "select owner||'.'||table_name||'.'||table_type full_name, \n" + \  
       "  decode(owner,user,null,owner||'.')||table_name table_name, \n" + \  
       "  decode(owner,user,'user_','all_')||lower(table_type) table_type \n" + \  
       "  from all_catalog where %s " \  
               % (where)  
   
     GDebug.printMesg(5,statement)  
   
     cursor = self._dataConnection.cursor()  
     cursor.execute(statement)  
   
     list = []  
     rs = cursor.fetchone()  
     if rs:  
       rv = GDataObjects.Schema(attrs={'id':rs[0], 'name':string.lower(rs[1]),  
                          'type':rs[2]},  
                          getChildSchema=self.__getFieldSchema)  
     else:  
       rv = None  
   
     cursor.close()  
     return rv  
   
   
   # Get fields for a table  
   def __getFieldSchema(self, parent):  
   
     owner, name, type = string.split(parent.id,'.')  
   
     cursor = self._dataConnection.cursor()  
   
     if type == 'SYNONYM':  
       statement = "select table_owner, table_name, " + \  
                   "decode(db_link,null,'','@'||db_link) name " + \  
                   "from all_synonyms " + \  
                   "where owner = '%s' and synonym_name='%s'" % (owner, name)  
   
       GDebug.printMesg(5,statement)  
   
       cursor.execute(statement)  
       rs = cursor.fetchone()  
       owner, name, link = rs  
       if link is None:  
         link = ""  
     else:  
       link = ""  
   
     statement = \  
        "select owner||'.'||table_name||'.'||column_name||'.%s', " % (link) + \  
        "column_name, data_type, nullable, data_length, data_scale, data_precision " + \  
        "from all_tab_columns%s " % (link) + \  
        "where owner = '%s' and table_name = '%s' " % (owner, name) + \  
        "order by column_id"  
   
     GDebug.printMesg(5,statement)  
   
     cursor.execute(statement)  
   
     list = []  
     for rs in cursor.fetchall():  
   
       attrs={'id': rs[0], 'name': string.lower(rs[1]),  
              'type':'field', 'nativetype': rs[2],  
              'required': rs[3] == 'N'}  
   
       if rs[2] in ('NUMBER',):  
         attrs['precision'] = int(rs[5])  
         attrs['datatype'] = 'number'  
         attrs['length'] = int(rs[6])  
       elif rs[2] in ('DATE',):  
         attrs['datatype'] = 'date'  
       else:  
         attrs['datatype'] = 'text'  
         if int(rs[4]):  
           attrs['length'] = int(rs[4])  
   
       list.append(GDataObjects.Schema(attrs=attrs))  
   
     cursor.close()  
     return tuple(list)  
   
   
 #  
 #  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 sysdate from dual")  
   
   # Return a sequence number from sequence 'name'  
   def getSequence(self, name):  
     return self.__singleQuery("select %s.nextval from dual" % name)  
   
   # Run the SQL statement 'statement'  
   def sql(self, statement):  
     cursor = self.__connection.cursor()  
     try:  
       cursor.execute(statement)  
       cursor.close()  
     except:  
       cursor.close()  
       raise  
     try:  
       return cursor.fetchall()  
     except:  
       return None  
   
   # 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  
   
   
   
   

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