uri = user/password@host[:port]/sid[ as {sysdba|sysoper}] Return cx_Oracle.Connection instance
(uri)
| 86 | |
| 87 | |
| 88 | def _connect(uri): |
| 89 | """ |
| 90 | uri = user/password@host[:port]/sid[ as {sysdba|sysoper}] |
| 91 | |
| 92 | Return cx_Oracle.Connection instance |
| 93 | """ |
| 94 | # cx_Oracle.Connection() not support 'as sysdba' syntax |
| 95 | uri_l = uri.rsplit(" as ", 1) |
| 96 | if len(uri_l) == 2: |
| 97 | credentials, mode = uri_l |
| 98 | mode = MODE[mode] |
| 99 | else: |
| 100 | credentials = uri_l[0] |
| 101 | mode = 0 |
| 102 | |
| 103 | serv_name = False |
| 104 | userpass, hostportsid = credentials.split("@") |
| 105 | user, password = userpass.split("/") |
| 106 | hostport, sid = hostportsid.split("/") |
| 107 | if "servicename" in sid: |
| 108 | serv_name = True |
| 109 | sid = sid.split("servicename")[0].strip() |
| 110 | hostport_l = hostport.split(":") |
| 111 | if len(hostport_l) == 2: |
| 112 | host, port = hostport_l |
| 113 | else: |
| 114 | host = hostport_l[0] |
| 115 | port = 1521 |
| 116 | log.debug("connect: %s", (user, password, host, port, sid, mode)) |
| 117 | # force UTF-8 client encoding |
| 118 | os.environ["NLS_LANG"] = ".AL32UTF8" |
| 119 | if serv_name: |
| 120 | conn = cx_Oracle.connect( |
| 121 | user, password, cx_Oracle.makedsn(host, port, service_name=sid), mode |
| 122 | ) |
| 123 | else: |
| 124 | conn = cx_Oracle.connect( |
| 125 | user, password, cx_Oracle.makedsn(host, port, sid), mode |
| 126 | ) |
| 127 | conn.outputtypehandler = _unicode_output |
| 128 | return conn |
| 129 | |
| 130 | |
| 131 | @depends("cx_Oracle", fallback_function=_cx_oracle_req) |