Python cx_Oracle.connect() 예제

[참조] https://www.programcreek.com/python/example/55416/cx_Oracle.connect

 

Python cx_Oracle.connect() 예제

 

[ Example 1 ]
----------------------------------------------------------------------
def _open_connection(self):
        return cx_Oracle.connect(self.user,
                     self.password,
                     "%s:%d/%s" % (self.host, self.port, self._oracle_database)) 

[ Example 2]
----------------------------------------------------------------------
def execute(self, host, port='1521', user='', password='', sid='', service_name=''):

    if sid:
      dsn = cx_Oracle.makedsn(host=host, port=port, sid=sid)
    elif service_name:
      dsn = cx_Oracle.makedsn(host=host, port=port, service_name=service_name)
    else:
      raise ValueError('Options sid and service_name cannot be both empty')

    try:
      with Timing() as timing:
        fp = cx_Oracle.connect(user, password, dsn, threaded=True)

      code, mesg = '0', fp.version

    except cx_Oracle.DatabaseError as e:
      code, mesg = e.args[0].message[:-1].split(': ', 1)

    return self.Response(code, mesg, timing)

[ Example 3 ]
----------------------------------------------------------------------
def oracle(self, ip):
for i in range(1, len(oracle_user)):
  try:
    user = oracle_user[i]
    pwd  = oracle_pass_default[i]
    conn = cx_Oracle.connect(user, pwd, ip+':1521/orcl')
    print u'{}[+] {}:1521  Oracle: {} {}{}'.format(G, ip, user, pwd, W)
    conn.close()
  except Exception as e:
    pass
  for pwd in passwd:
    try:
      pwd = pwd.replace('{user}', 'sys')
      conn = cx_Oracle.connect('sys', pwd, ip+':1521/orcl')
      print u'{}[+] {}:1521  Oracle: sys {}{}'.format(G, ip, pwd, W)
      conn.close()
    except Exception as e:
      pass 

[ Example 4 ]
----------------------------------------------------------------------
def execute(self, host, port='1521', user='', password='', sid='', service_name=''):
    if sid:
      dsn = cx_Oracle.makedsn(host=host, port=port, sid=sid)
    elif service_name:
      dsn = cx_Oracle.makedsn(host=host, port=port, service_name=service_name)
    else:
      raise ValueError('Options sid and service_name cannot be both empty')

    try:
      with Timing() as timing:
        fp = cx_Oracle.connect(user, password, dsn, threaded=True)

      code, mesg = '0', fp.version

    except cx_Oracle.DatabaseError as e:
      code, mesg = e.args[0].message[:-1].split(': ', 1)

    return self.Response(code, mesg, timing)

[ Example 5 ]
----------------------------------------------------------------------
def connect(self):
  self.initConnection()
  self.__dsn = cx_Oracle.makedsn(self.hostname, self.port, self.db)
  self.__dsn = utf8encode(self.__dsn)
  self.user = utf8encode(self.user)
  self.password = utf8encode(self.password)

  try:
    self.connector = cx_Oracle.connect(dsn=self.__dsn, user=self.user, password=self.password,

                            mode=cx_Oracle.SYSDBA)
    logger.info("successfully connected as SYSDBA")
  except (cx_Oracle.OperationalError, cx_Oracle.DatabaseError, cx_Oracle.InterfaceError), ex:
    if "Oracle Client library" in str(ex):
      msg = re.sub(r"DPI-\d+:\s+", "", str(ex))
      msg = re.sub(r': ("[^"]+")', r" (\g<1>)", msg)
      msg = re.sub(r". See (http[^ ]+)", r'. See "\g<1>"', msg)
      raise SqlmapConnectionException(msg)

  try:
    self.connector = cx_Oracle.connect(dsn=self.__dsn, user=self.user, password=self.password)
  except (cx_Oracle.OperationalError, cx_Oracle.DatabaseError, cx_Oracle.InterfaceError), msg:
    raise SqlmapConnectionException(msg) 

[ Example 6 ]
----------------------------------------------------------------------
def db_connect(args):
    if args.type == "mysql" or args.type == "mariadb":
        import mysql.connector
        try:
            connection = mysql.connector.connect(
                user=args.user,
                password=args.password,
                database=args.db)
        except mysql.connector.Error as err:
            print(colorize("red", "[ERROR] {}".format(err)))
            return None
    elif args.type == "mssql":
        import pymssql
        try:
            connection = pymssql.connect(server="localhost", database=args.db)
        except pymssql.Error as err:
            print(colorize("red", "[ERROR] {}".format(err)))
            return None
    elif args.type == "pgsql":
        import psycopg2
        try:
            connection = psycopg2.connect(
                "dbname='{}' user='{}' password='{}'".format(
                    args.db, args.user, args.password))
        except psycopg2.Error as err:
            print(colorize("red", "[ERROR] {}".format(err)))
            return None
    elif args.type == "oracle":
        import cx_Oracle
        try:
            connection = cx_Oracle.connect(
                args.user, args.password, cx_Oracle.makedsn(
                    '127.0.0.1', 1521, args.db), mode=cx_Oracle.SYSDBA)
        except cx_Oracle.Error as err:
            print(colorize("red", "[ERROR] {}".format(err)))
            return None

    return connection