-
Notifications
You must be signed in to change notification settings - Fork 22
Closed
Description
Hi,
THe DB Health provider is not working correctly with Oracle Database.
This is because the query used in the code to check if DB is up is not a valid SQL Statement for Oracle :
res = self.engine.execute(f"SELECT {expected}")should be :
res = self.engine.execute(f"SELECT {expected} from dual")Instead of submitting a statement which can not guaranteed to be correct for all SQLAlchemy backend, may I suggest to use the do_ping function provided in the SQL Alchemy plumbing for each dialect, which will take care of using the correct query :
Instead of :
def get_health(self) -> DbHealthStatus:
expected = int(time.time() * 1000)
try:
res = self.engine.execute(f"SELECT {expected}")
actual = next(res)[0]
if expected == actual:
return DbHealthStatus(status=Status.UP, details=DbHealthDetails(self.engine.name))
return DbHealthStatus(
status=Status.UNKNOWN,
details=DbHealthDetails(self.engine.name, f"Selected {expected}, got {actual}"))
except OperationalError as e:
return DbHealthStatus(status=Status.DOWN, details=DbHealthDetails(self.engine.name, str(e)))do :
def get_health(self) -> DbHealthStatus:
try:
conn = self.engine.raw_connection()
if self.engine.dialect.do_ping(conn):
return DbHealthStatus(status=Status.UP, details=DbHealthDetails(self.engine.name))
return DbHealthStatus(
status=Status.UNKNOWN,
details=DbHealthDetails(self.engine.name, f"Selected {expected}, got {actual}"))
except (OperationalError, DatabaseError) as e:
return DbHealthStatus(status=Status.DOWN, details=DbHealthDetails(self.engine.name, str(e)))Rgds
Jerome
Metadata
Metadata
Assignees
Labels
No labels