How-to write Python for Oracle

First of all, you need to install the Python driver for Oracle module, cx_Oracle. The easiest way is to use pip or easy_install to install it.

$ pip install cx_Oracle
or
$ easy_install cx_Oracle

Then, you can start to write scripts. Here is an example:

#!/usr/bin/python
'''
Create a user, drop a user
'''

    def __do_create_oracleuser(self):
        isSuccess = False
        
        #password equal to username
        oracle_pass = self.newuser  
        privileges = self.privileges

        check_user_sql = "select count(*) from dba_users \
                          where username = upper('%s') " \
                          % (self.newuser, )                
        get_temptbs_sql = '''select decode(PROPERTY_VALUE,'SYSTEM', tablespace_name, PROPERTY_VALUE)
                            from 
                            (select tablespace_name 
                            from (select tablespace_name,sum(bytes) 
                                  from dba_temp_files 
                                  group by tablespace_name 
                                  order by 2 desc)
                             where rownum=1
                            ) a,
                            (select PROPERTY_VALUE 
                             FROM DATABASE_PROPERTIES 
                             where PROPERTY_NAME='DEFAULT_TEMP_TABLESPACE'
                            ) b'''
        create_user_sql = "create user %s identified by %s        \
                           default tablespace %s quota %dG on %s" \
                           % (self.newuser, oracle_pass, self.tablespace,
                              self.space, self.tablespace )
        #permission = "connect, resource"
        #grant_sql = "grant %s to %s" % (permission, self.newuser)
         
        try:
            source = cx_Oracle.makedsn(self.vip, self.port, self.sid)
            db =  cx_Oracle.connect(self.sysuser,self.syspass, source)
            cursor = db.cursor()
            #1 check user exist
            cursor.execute(check_user_sql)
            count = cursor.fetchone()     
            if count[0] > 0:
                raise DbUserCreateError('DbUserCreateError: In Oracle DB User [' + 
                                    self.newuser + '] Already Exists.')        
            cursor.execute(get_temptbs_sql)
            temptbs = cursor.fetchone()
            create_user_sql += " temporary tablespace %s" %temptbs;
            cursor.execute(create_user_sql)
            if privileges:
                for privilege in privileges.strip().split(';'):
                    if privilege:
                        privilege_sql = "grant %s to %s" % (privilege, self.newuser)
                        cursor.execute(privilege_sql)
            
        except cx_Oracle.DatabaseError as exc:
            error, = exc.args
            self.logger.error(exc)
            raise DbUserCreateError(error) 
        else:
            isSuccess = True
            self.logger.info('Create Oracle Db User [' + self.newuser + '] Success!')
        finally:
            try:
                cursor.close()           
                db.close()
            except:
                self.logger.error('cx_Oracle.connect or db.cursor')
            
        return { 
                'username': self.newuser,
                'password' : oracle_pass,
                'db' : self.sid,
                'success' : isSuccess ,
                } 


    def __do_drop_oracleuser(self, db_item):
        sysuser = db_item['admin_user']
        syspass = db_item['admin_passwd']
        vip = db_item['vip']
        port = db_item['db_port']
        sid = db_item['sid']
        olduser = db_item['olduser']

        check_user_sql = "select count(*) from dba_users \
                          where username = upper('%s') " \
                          % (olduser)  
        drop_sql = "drop user %s cascade" % olduser
        #TO-DO: lock user when decomm
        #lock_sql = "alter user %s account lock" % olduser
        try:
            
            source = cx_Oracle.makedsn(vip, port, sid)
            db =  cx_Oracle.connect(sysuser, syspass, source)
            cursor = db.cursor()
            #drop oracle user
            try:
                #1 check user exist
                cursor.execute(check_user_sql)
                count = cursor.fetchone()     
                if count[0] > 0:
                    cursor.execute(drop_sql)
               
            except cx_Oracle.DatabaseError as exc:
                #self.logger.error(exc)
                error, = exc.args
                raise DbUserDropError("OracleDbUserDropError:" + error)
            else:
                self.logger.info('Drop Oracle User[' + olduser + '] Success!')     
            
        except cx_Oracle.DatabaseError as exc:
            self.logger.error(exc)
            error, = exc.args
            raise DbUserDropError("OracleDbUserDropError:" + error)
        finally:
            try:
                cursor.close()           
                db.close()
            except:
                self.logger.error('cx_Oracle.connect or db.cursor')

Of course, it’s a quite simple example. But it’s a good and great start

Advertisement