bn8/python-mylib
1
0
Fork 0
Code Issues Pull requests Projects Releases 12 Wiki Activity Actions

oracle: fix closing cursors

Browse source
This commit is contained in:
Benjamin Renard 2022-12-09 12:20:01 +01:00
parent 651e1a1a6c
commit c643fd30ac
2 changed files with 24 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
mylib/oracle.py
View file

@ -125,7 +125,6 @@ class OracleDB:
log.debug("Just-try mode : do not really execute SQL query '%s'", sql)
return True
cursor = self._conn.cursor()
try:
log.debug(
'Run SQL query "%s" %s',
@ -135,6 +134,7 @@ class OracleDB:
for key, value in params.items()
]) if params else "without params"
)
with self._conn.cursor() as cursor:
if isinstance(params, dict):
cursor.execute(sql, **params)
else:
@ -164,7 +164,6 @@ class OracleDB:
:return: List of selected rows as dict on success, False otherwise
:rtype: list, bool
"""
cursor = self._conn.cursor()
try:
log.debug(
'Run SQL SELECT query "%s" %s',
@ -174,11 +173,14 @@ class OracleDB:
for key, value in params.items()
]) if params else "without params"
)
with self._conn.cursor() as cursor:
if isinstance(params, dict):
cursor.execute(sql, **params)
else:
cursor.execute(sql)
cursor.rowfactory = lambda *args: dict(zip([d[0] for d in cursor.description], args))
cursor.rowfactory = lambda *args: dict(
zip([d[0] for d in cursor.description], args)
)
results = cursor.fetchall()
return results
except Exception:

10
tests/test_oracle.py
View file

@ -15,8 +15,10 @@ class FakeCXOracleCursor:
self.expected_return = expected_return
self.expected_just_try = expected_just_try
self.expected_exception = expected_exception
self.opened = True
def execute(self, sql, **params):
assert self.opened
if self.expected_exception:
raise Exception("%s.execute(%s, %s): expected exception" % (self, sql, params))
if self.expected_just_try and not sql.lower().startswith('select '):
@ -26,8 +28,16 @@ class FakeCXOracleCursor:
return self.expected_return
def fetchall(self):
assert self.opened
return self.expected_return
def __enter__(self):
self.opened = True
return self
def __exit__(self, *args):
self.opened = False
def __repr__(self):
return "FakeCXOracleCursor(%s, %s, %s, %s)" % (
self.expected_sql, self.expected_params,