Coverage for mindsdb / integrations / handlers / oracle_handler / oracle_handler.py: 78%
267 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1from typing import Any, Dict, List, Optional, Text
3import oracledb
4import pandas as pd
5from oracledb import connect, Connection, DatabaseError, Cursor
6from mindsdb_sql_parser.ast.base import ASTNode
8from mindsdb.integrations.libs.base import MetaDatabaseHandler
9from mindsdb.integrations.libs.response import (
10 HandlerStatusResponse as StatusResponse,
11 HandlerResponse as Response,
12 RESPONSE_TYPE,
13)
14from mindsdb.utilities import log
15from mindsdb.utilities.render.sqlalchemy_render import SqlalchemyRender
16import mindsdb.utilities.profiler as profiler
17from mindsdb.api.mysql.mysql_proxy.libs.constants.mysql import MYSQL_DATA_TYPE
20oracledb.defaults.fetch_lobs = False # Return LOBs directly as strings or bytes.
21logger = log.getLogger(__name__)
24def _map_type(internal_type_name: str) -> MYSQL_DATA_TYPE:
25 """Map Oracle types to MySQL types.
26 List of types: https://docs.oracle.com/en/database/oracle/oracle-database/23/sqlrf/Data-Types.html
28 Args:
29 internal_type_name (str): The name of the Oracle type to map.
31 Returns:
32 MYSQL_DATA_TYPE: The MySQL type that corresponds to the Oracle type.
33 """
34 internal_type_name = internal_type_name.upper()
35 types_map = {
36 (
37 "VARCHAR2",
38 "NVARCHAR2",
39 "CHARACTER VARYING",
40 "CHAR VARYING",
41 "NATIONAL CHARACTER",
42 "NATIONAL CHAR",
43 "VARCHAR",
44 "NATIONAL CHARACTER VARYING",
45 "NATIONAL CHAR VARYING",
46 "NCHAR VARYING",
47 "LONG VARCHAR",
48 ): MYSQL_DATA_TYPE.VARCHAR,
49 ("INTEGER", "INT"): MYSQL_DATA_TYPE.INT,
50 ("SMALLINT",): MYSQL_DATA_TYPE.SMALLINT,
51 ("NUMBER", "DECIMAL"): MYSQL_DATA_TYPE.DECIMAL,
52 ("FLOAT", "BINARY_FLOAT", "REAL"): MYSQL_DATA_TYPE.FLOAT,
53 ("BINARY_DOUBLE",): MYSQL_DATA_TYPE.DOUBLE,
54 ("LONG",): MYSQL_DATA_TYPE.BIGINT,
55 ("DATE",): MYSQL_DATA_TYPE.DATE,
56 (
57 "HOUR",
58 "MINUTE",
59 "SECOND",
60 "TIMEZONE_HOUR",
61 "TIMEZONE_MINUTE",
62 ): MYSQL_DATA_TYPE.SMALLINT,
63 (
64 "TIMESTAMP",
65 "TIMESTAMP WITH TIME ZONE",
66 "TIMESTAMP WITH LOCAL TIME ZONE",
67 ): MYSQL_DATA_TYPE.TIMESTAMP,
68 ("RAW", "LONG RAW", "BLOB", "BFILE"): MYSQL_DATA_TYPE.BINARY,
69 ("ROWID", "UROWID"): MYSQL_DATA_TYPE.TEXT,
70 ("CHAR", "NCHAR", "CLOB", "NCLOB", "CHARACTER"): MYSQL_DATA_TYPE.CHAR,
71 ("VECTOR",): MYSQL_DATA_TYPE.VECTOR,
72 ("JSON",): MYSQL_DATA_TYPE.JSON,
73 }
75 for db_types_list, mysql_data_type in types_map.items(): 75 ↛ 79line 75 didn't jump to line 79 because the loop on line 75 didn't complete
76 if internal_type_name in db_types_list:
77 return mysql_data_type
79 logger.debug(f"Oracle handler type mapping: unknown type: {internal_type_name}, use VARCHAR as fallback.")
80 return MYSQL_DATA_TYPE.VARCHAR
83def _make_table_response(result: list[tuple[Any]], cursor: Cursor) -> Response:
84 """Build response from result and cursor.
86 Args:
87 result (list[tuple[Any]]): result of the query.
88 cursor (oracledb.Cursor): cursor object.
90 Returns:
91 Response: response object.
92 """
93 description: list[tuple[Any]] = cursor.description
94 mysql_types: list[MYSQL_DATA_TYPE] = []
95 for column in description:
96 db_type = column[1]
97 precision = column[4]
98 scale = column[5]
99 if db_type is oracledb.DB_TYPE_JSON:
100 mysql_types.append(MYSQL_DATA_TYPE.JSON)
101 elif db_type is oracledb.DB_TYPE_VECTOR:
102 mysql_types.append(MYSQL_DATA_TYPE.VECTOR)
103 elif db_type is oracledb.DB_TYPE_NUMBER:
104 if scale != 0:
105 mysql_types.append(MYSQL_DATA_TYPE.FLOAT)
106 else:
107 # python max int is 19 digits, oracle can return more
108 if precision > 18:
109 mysql_types.append(MYSQL_DATA_TYPE.DECIMAL)
110 else:
111 mysql_types.append(MYSQL_DATA_TYPE.INT)
112 elif db_type is oracledb.DB_TYPE_BINARY_FLOAT: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 mysql_types.append(MYSQL_DATA_TYPE.FLOAT)
114 elif db_type is oracledb.DB_TYPE_BINARY_DOUBLE: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 mysql_types.append(MYSQL_DATA_TYPE.FLOAT)
116 elif db_type is oracledb.DB_TYPE_BINARY_INTEGER: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 mysql_types.append(MYSQL_DATA_TYPE.INT)
118 elif db_type is oracledb.DB_TYPE_BOOLEAN:
119 mysql_types.append(MYSQL_DATA_TYPE.BOOLEAN)
120 elif db_type in (
121 oracledb.DB_TYPE_CHAR,
122 oracledb.DB_TYPE_NCHAR,
123 oracledb.DB_TYPE_LONG,
124 oracledb.DB_TYPE_NVARCHAR,
125 oracledb.DB_TYPE_VARCHAR,
126 oracledb.DB_TYPE_LONG_NVARCHAR,
127 ):
128 mysql_types.append(MYSQL_DATA_TYPE.TEXT)
129 elif db_type in (oracledb.DB_TYPE_RAW, oracledb.DB_TYPE_LONG_RAW):
130 mysql_types.append(MYSQL_DATA_TYPE.BINARY)
131 elif db_type is oracledb.DB_TYPE_DATE:
132 mysql_types.append(MYSQL_DATA_TYPE.DATE)
133 elif db_type is oracledb.DB_TYPE_TIMESTAMP:
134 mysql_types.append(MYSQL_DATA_TYPE.TIMESTAMP)
135 else:
136 # fallback
137 mysql_types.append(MYSQL_DATA_TYPE.TEXT)
139 # region cast int and bool to nullable types
140 serieses = []
141 for i, mysql_type in enumerate(mysql_types):
142 expected_dtype = None
143 if mysql_type in (
144 MYSQL_DATA_TYPE.SMALLINT,
145 MYSQL_DATA_TYPE.INT,
146 MYSQL_DATA_TYPE.MEDIUMINT,
147 MYSQL_DATA_TYPE.BIGINT,
148 MYSQL_DATA_TYPE.TINYINT,
149 ):
150 expected_dtype = "Int64"
151 elif mysql_type in (MYSQL_DATA_TYPE.BOOL, MYSQL_DATA_TYPE.BOOLEAN):
152 expected_dtype = "boolean"
153 serieses.append(pd.Series([row[i] for row in result], dtype=expected_dtype, name=description[i][0]))
154 df = pd.concat(serieses, axis=1, copy=False)
155 # endregion
157 return Response(RESPONSE_TYPE.TABLE, data_frame=df, mysql_types=mysql_types)
160class OracleHandler(MetaDatabaseHandler):
161 """
162 This handler handles connection and execution of SQL queries on Oracle.
163 """
165 name = "oracle"
167 def __init__(self, name: Text, connection_data: Optional[Dict], **kwargs) -> None:
168 """
169 Initializes the handler.
171 Args:
172 name (Text): The name of the handler instance.
173 connection_data (Dict): The connection data required to connect to OracleDB.
174 kwargs: Arbitrary keyword arguments.
175 """
176 super().__init__(name)
177 self.connection_data = connection_data
178 self.kwargs = kwargs
180 self.connection = None
181 self.is_connected = False
183 def connect(self) -> Connection:
184 """
185 Establishes a connection to the Oracle database.
187 Raises:
188 ValueError: If the expected connection parameters are not provided.
190 Returns:
191 oracledb.Connection: A connection object to the Oracle database.
192 """
193 if self.is_connected is True: 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true
194 return self.connection
196 # Mandatory connection parameters.
197 if not all(key in self.connection_data for key in ["user", "password"]):
198 raise ValueError("Required parameters (user, password) must be provided.")
200 if self.connection_data.get("thick_mode", False):
201 oracle_client_lib_dir = self.connection_data.get("oracle_client_lib_dir")
202 if isinstance(oracle_client_lib_dir, str) and oracle_client_lib_dir.strip():
203 try:
204 oracledb.init_oracle_client(lib_dir=oracle_client_lib_dir)
205 except Exception as e:
206 raise ValueError(f"Failed to initialize Oracle client: {e}")
207 else:
208 raise ValueError(
209 "Parameter 'oracle_client_lib_dir' must be provided as a non-empty string when using thick_mode."
210 )
212 config = {
213 "user": self.connection_data["user"],
214 "password": self.connection_data["password"],
215 }
217 # If 'dsn' is given, use it. Otherwise, use the individual connection parameters.
218 if "dsn" in self.connection_data:
219 config["dsn"] = self.connection_data["dsn"]
221 else:
222 if "host" not in self.connection_data and not any( 222 ↛ 229line 222 didn't jump to line 229 because the condition on line 222 was always true
223 key in self.connection_data for key in ["sid", "service_name"]
224 ):
225 raise ValueError(
226 "Required parameter host and either sid or service_name must be provided. Alternatively, dsn can be provided."
227 )
229 config["host"] = self.connection_data.get("host")
231 # Optional connection parameters when 'dsn' is not given.
232 optional_parameters = ["port", "sid", "service_name"]
233 for parameter in optional_parameters:
234 if parameter in self.connection_data:
235 config[parameter] = self.connection_data[parameter]
237 # Other optional connection parameters.
238 if "disable_oob" in self.connection_data: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 config["disable_oob"] = self.connection_data["disable_oob"]
241 if "auth_mode" in self.connection_data: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 mode_name = "AUTH_MODE_" + self.connection_data["auth_mode"].upper()
243 if not hasattr(oracledb, mode_name):
244 raise ValueError(f"Unknown auth mode: {mode_name}")
245 config["mode"] = getattr(oracledb, mode_name)
247 try:
248 connection = connect(
249 **config,
250 )
252 if "session_variables" in self.connection_data: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 with connection.cursor() as cur:
254 for key, value in self.connection_data["session_variables"].items():
255 cur.execute(f"ALTER SESSION SET {key} = {repr(value)}")
257 except DatabaseError as database_error:
258 logger.error(f"Error connecting to Oracle, {database_error}!")
259 raise
261 except Exception as unknown_error:
262 logger.error(f"Unknown error when connecting to Oracle: {unknown_error}")
263 raise
265 self.is_connected = True
266 self.connection = connection
267 return self.connection
269 def disconnect(self):
270 """
271 Closes the connection to the Oracle database if it's currently open.
272 """
273 if self.is_connected is False:
274 return
275 self.connection.close()
276 self.is_connected = False
278 def check_connection(self) -> StatusResponse:
279 """
280 Checks the status of the connection to the Oracle database.
282 Returns:
283 StatusResponse: An object containing the success status and an error message if an error occurs.
284 """
285 response = StatusResponse(False)
286 need_to_close = self.is_connected is False
288 try:
289 con = self.connect()
290 con.ping()
291 response.success = True
292 except (ValueError, DatabaseError) as known_error:
293 logger.error(f"Connection check to Oracle failed, {known_error}!")
294 response.error_message = str(known_error)
295 except Exception as unknown_error:
296 logger.error(f"Connection check to Oracle failed due to an unknown error, {unknown_error}!")
297 response.error_message = str(unknown_error)
299 if response.success and need_to_close:
300 self.disconnect()
302 elif not response.success and self.is_connected: 302 ↛ 303line 302 didn't jump to line 303 because the condition on line 302 was never true
303 self.is_connected = False
305 return response
307 @profiler.profile()
308 def native_query(self, query: Text) -> Response:
309 """
310 Executes a SQL query on the Oracle database and returns the result.
312 Args:
313 query (Text): The SQL query to be executed.
315 Returns:
316 Response: A response object containing the result of the query or an error message.
317 """
318 need_to_close = self.is_connected is False
320 connection = self.connect()
321 with connection.cursor() as cur:
322 try:
323 cur.execute(query)
324 if cur.description is None:
325 response = Response(RESPONSE_TYPE.OK, affected_rows=cur.rowcount)
326 else:
327 result = cur.fetchall()
328 response = _make_table_response(result, cur)
329 connection.commit()
330 except DatabaseError as database_error:
331 logger.error(f"Error running query: {query} on Oracle, {database_error}!")
332 response = Response(
333 RESPONSE_TYPE.ERROR,
334 error_message=str(database_error),
335 )
336 connection.rollback()
338 except Exception as unknown_error:
339 logger.error(f"Unknwon error running query: {query} on Oracle, {unknown_error}!")
340 response = Response(
341 RESPONSE_TYPE.ERROR,
342 error_message=str(unknown_error),
343 )
344 connection.rollback()
346 if need_to_close is True: 346 ↛ 348line 346 didn't jump to line 348 because the condition on line 346 was always true
347 self.disconnect()
348 return response
350 def query_stream(self, query: ASTNode, fetch_size: int = 1000):
351 """
352 Executes a SQL query represented by an ASTNode and retrieves the data in a streaming fashion.
354 Args:
355 query (ASTNode): An ASTNode representing the SQL query to be executed.
356 fetch_size (int): The number of rows to fetch in each batch.
357 Yields:
358 pd.DataFrame: A DataFrame containing a batch of rows from the query result.
359 Response: In case of an error, yields a Response object with the error details.
360 """
361 query_str = SqlalchemyRender("oracle").get_string(query, with_failback=True)
362 need_to_close = self.is_connected is False
364 connection = self.connect()
365 with connection.cursor() as cur:
366 try:
367 cur.execute(query_str)
368 while True:
369 result = cur.fetchmany(fetch_size)
370 if not result:
371 break
372 df = pd.DataFrame(result, columns=[col[0] for col in cur.description])
373 yield df
374 connection.commit()
375 finally:
376 connect
377 if need_to_close is True:
378 self.disconnect()
380 def insert(self, table_name: str, df: pd.DataFrame) -> Response:
381 """
382 Inserts data from a DataFrame into a specified table in the Oracle database.
384 Args:
385 table_name (str): The name of the table where the data will be inserted.
386 df (pd.DataFrame): The DataFrame containing the data to be inserted.
387 Returns:
388 Response: A response object indicating the success or failure of the insert operation.
389 """
390 need_to_close = self.is_connected is False
391 connection = self.connect()
392 columns = list(df.columns)
393 placeholders = ", ".join([f":{i + 1}" for i in range(len(columns))])
394 insert_query = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})"
396 with connection.cursor() as cur:
397 try:
398 cur.executemany(insert_query, df.values.tolist())
399 connection.commit()
400 rowcount = cur.rowcount
401 except DatabaseError as database_error:
402 logger.error(f"Error inserting data into table {table_name} on Oracle, {database_error}!")
403 connection.rollback()
404 raise
405 if need_to_close is True: 405 ↛ 408line 405 didn't jump to line 408 because the condition on line 405 was always true
406 self.disconnect()
408 return Response(RESPONSE_TYPE.OK, affected_rows=rowcount)
410 @profiler.profile()
411 def query(self, query: ASTNode) -> Response:
412 """
413 Executes a SQL query represented by an ASTNode and retrieves the data.
415 Args:
416 query (ASTNode): An ASTNode representing the SQL query to be executed.
418 Returns:
419 Response: The response from the `native_query` method, containing the result of the SQL query execution.
420 """
421 renderer = SqlalchemyRender("oracle")
422 query_str = renderer.get_string(query, with_failback=True)
423 return self.native_query(query_str)
425 def get_tables(self) -> Response:
426 """
427 Retrieves a list of all non-system tables and views in the current schema of the Oracle database.
429 Returns:
430 Response: A response object containing the list of tables and views, formatted as per the `Response` class.
431 """
432 query = """
433 SELECT
434 owner AS table_schema,
435 table_name AS table_name,
436 'BASE TABLE' AS table_type
437 FROM all_tables t
438 JOIN all_users u ON t.owner = u.username
439 WHERE t.tablespace_name = 'USERS'
441 UNION ALL
443 SELECT
444 v.owner AS table_schema,
445 v.view_name AS table_name,
446 'VIEW' AS table_type
447 FROM all_views v
448 JOIN all_users u ON v.owner = u.username
449 WHERE v.owner IN (
450 SELECT DISTINCT owner
451 FROM all_tables
452 WHERE tablespace_name = 'USERS'
453 )
454 """
455 return self.native_query(query)
457 def get_columns(self, table_name: Text) -> Response:
458 """
459 Retrieves column details for a specified table in the Oracle database.
461 Args:
462 table_name (Text): The name of the table for which to retrieve column information.
464 Returns:
465 Response: A response object containing the column details, formatted as per the `Response` class.
466 Raises:
467 ValueError: If the 'table_name' is not a valid string.
468 """
469 query = f"""
470 SELECT
471 COLUMN_NAME,
472 DATA_TYPE,
473 COLUMN_ID AS ORDINAL_POSITION,
474 DATA_DEFAULT AS COLUMN_DEFAULT,
475 CASE NULLABLE WHEN 'Y' THEN 'YES' ELSE 'NO' END AS IS_NULLABLE,
476 CHAR_LENGTH AS CHARACTER_MAXIMUM_LENGTH,
477 NULL AS CHARACTER_OCTET_LENGTH,
478 DATA_PRECISION AS NUMERIC_PRECISION,
479 DATA_SCALE AS NUMERIC_SCALE,
480 NULL AS DATETIME_PRECISION,
481 CHARACTER_SET_NAME,
482 NULL AS COLLATION_NAME
483 FROM USER_TAB_COLUMNS
484 WHERE table_name = '{table_name}'
485 ORDER BY TABLE_NAME, COLUMN_ID
486 """
487 result = self.native_query(query)
488 if result.resp_type is RESPONSE_TYPE.TABLE: 488 ↛ 490line 488 didn't jump to line 490 because the condition on line 488 was always true
489 result.to_columns_table_response(map_type_fn=_map_type)
490 return result
492 def meta_get_tables(self, table_names: Optional[List[str]]) -> Response:
493 """
494 Retrieves metadata about all non-system tables and views in the current schema of the Oracle database.
496 Returns:
497 list[dict[str, Any]]: A list of dictionaries, each containing metadata about a table or view.
498 """
499 query = """
500 SELECT
501 o.object_name AS table_name,
502 USER AS table_schema,
503 o.object_type AS table_type,
504 c.comments AS table_description,
505 t.num_rows AS row_count
506 FROM
507 user_objects o
508 LEFT JOIN
509 user_tab_comments c ON o.object_name = c.table_name
510 LEFT JOIN
511 user_tables t ON o.object_name = t.table_name AND o.object_type = 'TABLE'
512 WHERE
513 o.object_type IN ('TABLE', 'VIEW')
514 """
515 if table_names is not None and len(table_names) > 0: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true
516 table_names = [f"'{t.upper()}'" for t in table_names]
517 query += f" AND o.object_name IN ({','.join(table_names)})"
519 query += " ORDER BY o.object_name"
521 result = self.native_query(query)
522 return result
524 def meta_get_columns(self, table_names: Optional[List[str]]) -> Response:
525 """Retrieves metadata about the columns of specified tables in the Oracle database.
527 Args:
528 table_names (list[str]): A list of table names for which to retrieve column metadata.
530 Returns:
531 list[dict[str, Any]]: A list of dictionaries, each containing metadata about a column.
532 """
533 query = """
534 SELECT
535 utc.table_name,
536 utc.column_name,
537 utc.data_type,
538 ucc.comments AS column_description,
539 utc.data_default AS column_default,
540 CASE
541 WHEN utc.nullable = 'Y' THEN 1
542 ELSE 0
543 END AS is_nullable
544 FROM
545 user_tab_columns utc
546 JOIN
547 user_tables ut ON utc.table_name = ut.table_name
548 LEFT JOIN
549 user_col_comments ucc ON utc.table_name = ucc.table_name AND utc.column_name = ucc.column_name
550 """
551 if table_names is not None and len(table_names) > 0: 551 ↛ 554line 551 didn't jump to line 554 because the condition on line 551 was always true
552 table_names = [f"'{t.upper()}'" for t in table_names]
553 query += f" WHERE utc.table_name IN ({','.join(table_names)})"
554 query += " ORDER BY utc.table_name, utc.column_id"
555 result = self.native_query(query)
556 return result
558 def meta_get_column_statistics(self, table_names: Optional[List[str]]) -> Response:
559 """Retrieves statistics about the columns of specified tables in the Oracle database.
561 Args:
562 table_names (list[str]): A list of table names for which to retrieve column statistics.
564 Returns:
565 list[dict[str, Any]]: A list of dictionaries, each containing statistics about a column.
566 """
567 table_filter = ""
568 if table_names is not None and len(table_names) > 0: 568 ↛ 572line 568 didn't jump to line 572 because the condition on line 568 was always true
569 quoted_names = [f"'{t.upper()}'" for t in table_names]
570 table_filter = f" WHERE cs.table_name IN ({','.join(quoted_names)})"
572 query = (
573 """
574 SELECT
575 cs.table_name AS TABLE_NAME,
576 cs.column_name AS COLUMN_NAME,
577 CASE
578 WHEN cs.sample_size > 0 THEN ROUND((cs.num_nulls / cs.sample_size) * 100, 2)
579 ELSE NULL
580 END AS NULL_PERCENTAGE,
581 cs.num_distinct AS DISTINCT_VALUES_COUNT,
582 NULL AS MOST_COMMON_VALUES,
583 NULL AS MOST_COMMON_FREQUENCIES,
584 cs.histogram AS HISTOGRAM_TYPE,
585 h.bounds AS HISTOGRAM_BOUNDS
586 FROM
587 user_tab_col_statistics cs
588 LEFT JOIN (
589 SELECT
590 table_name,
591 column_name,
592 LISTAGG(endpoint_value, ', ') WITHIN GROUP (ORDER BY endpoint_number) AS bounds
593 FROM
594 user_tab_histograms
595 GROUP BY
596 table_name,
597 column_name
598 ) h ON cs.table_name = h.table_name AND cs.column_name = h.column_name
599 """
600 + table_filter
601 + """
602 ORDER BY
603 cs.table_name,
604 cs.column_name
605 """
606 )
608 result = self.native_query(query)
610 if result.resp_type is RESPONSE_TYPE.TABLE and result.data_frame is not None: 610 ↛ 628line 610 didn't jump to line 628 because the condition on line 610 was always true
611 df = result.data_frame
613 def extract_min_max(
614 histogram_str: str,
615 ) -> tuple[Optional[float], Optional[float]]:
616 if histogram_str and str(histogram_str).lower() not in ["nan", "none"]: 616 ↛ 622line 616 didn't jump to line 622 because the condition on line 616 was always true
617 values = str(histogram_str).split(",")
618 if values: 618 ↛ 622line 618 didn't jump to line 622 because the condition on line 618 was always true
619 min_val = values[0].strip(" '\"")
620 max_val = values[-1].strip(" '\"")
621 return min_val, max_val
622 return None, None
624 min_max_values = df["HISTOGRAM_BOUNDS"].apply(extract_min_max)
625 df["MINIMUM_VALUE"] = min_max_values.apply(lambda x: x[0])
626 df["MAXIMUM_VALUE"] = min_max_values.apply(lambda x: x[1])
627 df.drop(columns=["HISTOGRAM_BOUNDS"], inplace=True)
628 return result
630 def meta_get_primary_keys(self, table_names: Optional[List[str]]) -> Response:
631 """
632 Retrieves the primary keys for the specified tables in the Oracle database.
634 Args:
635 table_names (list[str]): A list of table names for which to retrieve primary keys.
637 Returns:
638 list[dict[str, Any]]: A list of dictionaries, each containing information about a primary key.
639 """
641 query = """
642 SELECT
643 cols.table_name,
644 cols.column_name,
645 cols.position AS ordinal_position,
646 cons.constraint_name
647 FROM
648 all_constraints cons
649 JOIN
650 all_cons_columns cols ON cons.constraint_name = cols.constraint_name AND cons.owner = cols.owner
651 WHERE
652 cons.constraint_type = 'P'
653 AND cons.owner = SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA')
654 """
655 if table_names is not None and len(table_names) > 0: 655 ↛ 659line 655 didn't jump to line 659 because the condition on line 655 was always true
656 quoted_names = [f"'{t.upper()}'" for t in table_names]
657 query += f" AND cols.table_name IN ({','.join(quoted_names)})"
659 query += " ORDER BY cols.table_name, cols.position"
661 result = self.native_query(query)
662 return result
664 def meta_get_foreign_keys(self, table_names: Optional[List[str]]) -> Response:
665 """
666 Retrieves the foreign keys for the specified tables in the Oracle database.
668 Args:
669 table_names (list[str]): A list of table names for which to retrieve foreign keys.
671 Returns:
672 list[dict[str, Any]]: A list of dictionaries, each containing information about a foreign key.
673 """
675 query = """
676 SELECT
677 pk_cols.table_name AS parent_table_name,
678 pk_cols.column_name AS parent_column_name,
679 fk_cols.table_name AS child_table_name,
680 fk_cols.column_name AS child_column_name,
681 fk_cons.constraint_name
682 FROM
683 all_constraints fk_cons
684 JOIN
685 all_cons_columns fk_cols ON fk_cons.owner = fk_cols.owner AND fk_cons.constraint_name = fk_cols.constraint_name
686 JOIN
687 all_cons_columns pk_cols ON fk_cons.owner = pk_cols.owner AND fk_cons.r_constraint_name = pk_cols.constraint_name
688 WHERE
689 fk_cons.constraint_type = 'R'
690 AND fk_cons.owner = SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA')
691 """
692 if table_names is not None and len(table_names) > 0: 692 ↛ 696line 692 didn't jump to line 696 because the condition on line 692 was always true
693 quoted_names = [f"'{t.upper()}'" for t in table_names]
694 query += f" AND fk_cols.table_name IN ({','.join(quoted_names)})"
696 query += " ORDER BY fk_cols.table_name, fk_cols.position"
697 result = self.native_query(query)
698 return result