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

Diff of /gnue-common/src/datasources/drivers/oracle/Base/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:23 2003 UTC revision 1.1.2.2 by jcater, Wed Nov 19 03:00:24 2003 UTC
# Line 32  Line 32 
32  #     service=    This is the Oracle TNS name for your connection  (required)  #     service=    This is the Oracle TNS name for your connection  (required)
33  #  #
34    
35  import sys, string  __all__ = ['DataObject_SQL','DataObject_Object']
 from gnue.common.datasources import GDataObjects, GConditions  
 from gnue.common.apps import GDebug  
   
 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])  
36    
       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  
37    
38    from gnue.common.datasources.drivers import DBSIG2
39    
40    class _Base:
41      _escapeSingleQuote = ''''
42      _dateTimeFormat = "'%Y-%m-%d %H:%M:%S'"
43    
44    class DataObject_SQL(_Base, DBSIG2.DataObject_SQL):
45      pass
46    
47    class DataObject_Object(_Base, DBSIG2.DataObject_Object):
48      pass

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