Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/oracle/base.py: 25%

956 `CREATE VECTOR INDEX <https://www.oracle.com/pls/topic/lookup?ctx=dblatest&id=GUID-B396C369-54BB-4098-A0DD-7C54B3A0D66F>`_ - in the Oracle documentation

957

958

959

960Similarity Searching

961~~~~~~~~~~~~~~~~~~~~

962

963When using the :class:`_oracle.VECTOR` datatype with a :class:`.Column` or similar

964ORM mapped construct, additional comparison functions are available, including:

965

966* ``l2_distance``

967* ``cosine_distance``

968* ``inner_product``

969

970Example Usage::

971

972 result_vector = connection.scalars(

973 select(t1).order_by(t1.embedding.l2_distance([2, 3, 4])).limit(3)

974 )

975

976 for user in vector:

977 print(user.id, user.embedding)

978

979FETCH APPROXIMATE support

980~~~~~~~~~~~~~~~~~~~~~~~~~

981

982Approximate vector search can only be performed when all syntax and semantic

983rules are satisfied, the corresponding vector index is available, and the

984query optimizer determines to perform it. If any of these conditions are

985unmet, then an approximate search is not performed. In this case the query

986returns exact results.

987

988To enable approximate searching during similarity searches on VECTORS, the

989``oracle_fetch_approximate`` parameter may be used with the :meth:`.Select.fetch`

990clause to add ``FETCH APPROX`` to the SELECT statement::

991

992 select(users_table).fetch(5, oracle_fetch_approximate=True)

993

994""" # noqa

995

996from __future__ import annotations

997

998from collections import defaultdict

999from dataclasses import fields

1000from functools import lru_cache

1001from functools import wraps

1002import re

1003from typing import Any

1004from typing import Callable

1005from typing import TYPE_CHECKING

1006

1007from . import dictionary

1008from .json import JSON

1009from .json import JSONIndexType

1010from .json import JSONPathType

1011from .types import _OracleBoolean

1012from .types import _OracleDate

1013from .types import BFILE

1014from .types import BINARY_DOUBLE

1015from .types import BINARY_FLOAT

1016from .types import BOOLEAN

1017from .types import DATE

1018from .types import FLOAT

1019from .types import INTERVAL

1020from .types import LONG

1021from .types import NCLOB

1022from .types import NUMBER

1023from .types import NVARCHAR2 # noqa

1024from .types import OracleRaw # noqa

1025from .types import RAW

1026from .types import ROWID # noqa

1027from .types import TIMESTAMP

1028from .types import VARCHAR2 # noqa

1029from .vector import VECTOR

1030from .vector import VectorIndexConfig

1031from .vector import VectorIndexType

1032from ... import Computed

1033from ... import exc

1034from ... import schema as sa_schema

1035from ... import sql

1036from ... import util

1037from ...engine import default

1038from ...engine import ObjectKind

1039from ...engine import ObjectScope

1040from ...engine import reflection

1041from ...engine.reflection import ReflectionDefaults

1042from ...sql import and_

1043from ...sql import bindparam

1044from ...sql import compiler

1045from ...sql import elements

1046from ...sql import expression

1047from ...sql import func

1048from ...sql import null

1049from ...sql import or_

1050from ...sql import select

1051from ...sql import selectable as sa_selectable

1052from ...sql import sqltypes

1053from ...sql import util as sql_util

1054from ...sql import visitors

1055from ...sql.base import NO_ARG

1056from ...sql.compiler import AggregateOrderByStyle

1057from ...sql.visitors import InternalTraversal

1058from ...types import BLOB

1059from ...types import CHAR

1060from ...types import CLOB

1061from ...types import DOUBLE_PRECISION

1062from ...types import INTEGER

1063from ...types import NCHAR

1064from ...types import NVARCHAR

1065from ...types import REAL

1066from ...types import VARCHAR

1067

1068if TYPE_CHECKING:

1069 from ...sql.sqltypes import _JSON_VALUE

1070

1071RESERVED_WORDS = set(

1072 "SHARE RAW DROP BETWEEN FROM DESC OPTION PRIOR LONG THEN "

1073 "DEFAULT ALTER IS INTO MINUS INTEGER NUMBER GRANT IDENTIFIED "

1074 "ALL TO ORDER ON FLOAT DATE HAVING CLUSTER NOWAIT RESOURCE "

1075 "ANY TABLE INDEX FOR UPDATE WHERE CHECK SMALLINT WITH DELETE "

1076 "BY ASC REVOKE LIKE SIZE RENAME NOCOMPRESS NULL GROUP VALUES "

1077 "AS IN VIEW EXCLUSIVE COMPRESS SYNONYM SELECT INSERT EXISTS "

1078 "NOT TRIGGER ELSE CREATE INTERSECT PCTFREE DISTINCT USER "

1079 "CONNECT SET MODE OF UNIQUE VARCHAR2 VARCHAR LOCK OR CHAR "

1080 "DECIMAL UNION PUBLIC AND START UID COMMENT CURRENT LEVEL".split()

1081)

1082

1083NO_ARG_FNS = set(

1084 "UID CURRENT_DATE SYSDATE USER CURRENT_TIME CURRENT_TIMESTAMP".split()

1085)

1086

1087

1088colspecs = {

1089 sqltypes.Boolean: _OracleBoolean,

1090 sqltypes.Interval: INTERVAL,

1091 sqltypes.DateTime: DATE,

1092 sqltypes.Date: _OracleDate,

1093 sqltypes.JSON: JSON,

1094 sqltypes.JSON.JSONIndexType: JSONIndexType,

1095 sqltypes.JSON.JSONPathType: JSONPathType,

1096}

1097

1098ischema_names = {

1099 "VARCHAR2": VARCHAR,

1100 "NVARCHAR2": NVARCHAR,

1101 "CHAR": CHAR,

1102 "NCHAR": NCHAR,

1103 "DATE": DATE,

1104 "NUMBER": NUMBER,

1105 "BLOB": BLOB,

1106 "BFILE": BFILE,

1107 "CLOB": CLOB,

1108 "NCLOB": NCLOB,

1109 "TIMESTAMP": TIMESTAMP,

1110 "TIMESTAMP WITH TIME ZONE": TIMESTAMP,

1111 "TIMESTAMP WITH LOCAL TIME ZONE": TIMESTAMP,

1112 "INTERVAL DAY TO SECOND": INTERVAL,

1113 "RAW": RAW,

1114 "FLOAT": FLOAT,

1115 "DOUBLE PRECISION": DOUBLE_PRECISION,

1116 "REAL": REAL,

1117 "LONG": LONG,

1118 "BINARY_DOUBLE": BINARY_DOUBLE,

1119 "BINARY_FLOAT": BINARY_FLOAT,

1120 "ROWID": ROWID,

1121 "BOOLEAN": BOOLEAN,

1122 "VECTOR": VECTOR,

1123 "JSON": JSON,

1124}

1125

1126

1127class OracleTypeCompiler(compiler.GenericTypeCompiler):

1128 # Note:

1129 # Oracle DATE == DATETIME

1130 # Oracle does not allow milliseconds in DATE

1131 # Oracle does not support TIME columns

1132

1133 def visit_datetime(self, type_, **kw):

1134 return self.visit_DATE(type_, **kw)

1135

1136 def visit_float(self, type_, **kw):

1137 return self.visit_FLOAT(type_, **kw)

1138

1139 def visit_double(self, type_, **kw):

1140 return self.visit_DOUBLE_PRECISION(type_, **kw)

1141

1142 def visit_unicode(self, type_, **kw):

1143 if self.dialect._use_nchar_for_unicode:

1144 return self.visit_NVARCHAR2(type_, **kw)

1145 else:

1146 return self.visit_VARCHAR2(type_, **kw)

1147

1148 def visit_INTERVAL(self, type_, **kw):

1149 return "INTERVAL DAY%s TO SECOND%s" % (

1150 type_.day_precision is not None

1151 and "(%d)" % type_.day_precision

1152 or "",

1153 type_.second_precision is not None

1154 and "(%d)" % type_.second_precision

1155 or "",

1156 )

1157

1158 def visit_LONG(self, type_, **kw):

1159 return "LONG"

1160

1161 def visit_TIMESTAMP(self, type_, **kw):

1162 if getattr(type_, "local_timezone", False):

1163 return "TIMESTAMP WITH LOCAL TIME ZONE"

1164 elif type_.timezone:

1165 return "TIMESTAMP WITH TIME ZONE"

1166 else:

1167 return "TIMESTAMP"

1168

1169 def visit_DOUBLE_PRECISION(self, type_, **kw):

1170 return self._generate_numeric(type_, "DOUBLE PRECISION", **kw)

1171

1172 def visit_BINARY_DOUBLE(self, type_, **kw):

1173 return self._generate_numeric(type_, "BINARY_DOUBLE", **kw)

1174

1175 def visit_BINARY_FLOAT(self, type_, **kw):

1176 return self._generate_numeric(type_, "BINARY_FLOAT", **kw)

1177

1178 def visit_FLOAT(self, type_, **kw):

1179 kw["_requires_binary_precision"] = True

1180 return self._generate_numeric(type_, "FLOAT", **kw)

1181

1182 def visit_NUMBER(self, type_, **kw):

1183 return self._generate_numeric(type_, "NUMBER", **kw)

1184

1185 def _generate_numeric(

1186 self,

1187 type_,

1188 name,

1189 precision=None,

1190 scale=None,

1191 _requires_binary_precision=False,

1192 **kw,

1193 ):

1194 if precision is None:

1195 precision = getattr(type_, "precision", None)

1196

1197 if _requires_binary_precision:

1198 binary_precision = getattr(type_, "binary_precision", None)

1199

1200 if precision and binary_precision is None:

1201 # https://www.oracletutorial.com/oracle-basics/oracle-float/

1202 estimated_binary_precision = int(precision / 0.30103)

1203 raise exc.ArgumentError(

1204 "Oracle Database FLOAT types use 'binary precision', "

1205 "which does not convert cleanly from decimal "

1206 "'precision'. Please specify "

1207 "this type with a separate Oracle Database variant, such "

1208 f"as {type_.__class__.__name__}(precision={precision})."

1209 f"with_variant(oracle.FLOAT"

1210 f"(binary_precision="

1211 f"{estimated_binary_precision}), 'oracle'), so that the "

1212 "Oracle Database specific 'binary_precision' may be "

1213 "specified accurately."

1214 )

1215 else:

1216 precision = binary_precision

1217

1218 if scale is None:

1219 scale = getattr(type_, "scale", None)

1220

1221 if precision is None:

1222 return name

1223 elif scale is None:

1224 n = "%(name)s(%(precision)s)"

1225 return n % {"name": name, "precision": precision}

1226 else:

1227 n = "%(name)s(%(precision)s, %(scale)s)"

1228 return n % {"name": name, "precision": precision, "scale": scale}

1229

1230 def visit_string(self, type_, **kw):

1231 return self.visit_VARCHAR2(type_, **kw)

1232

1233 def visit_VARCHAR2(self, type_, **kw):

1234 return self._visit_varchar(type_, "", "2")

1235

1236 def visit_NVARCHAR2(self, type_, **kw):

1237 return self._visit_varchar(type_, "N", "2")

1238

1239 visit_NVARCHAR = visit_NVARCHAR2

1240

1241 def visit_VARCHAR(self, type_, **kw):

1242 return self._visit_varchar(type_, "", "")

1243

1244 def _visit_varchar(self, type_, n, num):

1245 if not type_.length:

1246 return "%(n)sVARCHAR%(two)s" % {"two": num, "n": n}

1247 elif not n and self.dialect._supports_char_length:

1248 varchar = "VARCHAR%(two)s(%(length)s CHAR)"

1249 return varchar % {"length": type_.length, "two": num}

1250 else:

1251 varchar = "%(n)sVARCHAR%(two)s(%(length)s)"

1252 return varchar % {"length": type_.length, "two": num, "n": n}

1253

1254 def visit_text(self, type_, **kw):

1255 return self.visit_CLOB(type_, **kw)

1256

1257 def visit_unicode_text(self, type_, **kw):

1258 if self.dialect._use_nchar_for_unicode:

1259 return self.visit_NCLOB(type_, **kw)

1260 else:

1261 return self.visit_CLOB(type_, **kw)

1262

1263 def visit_large_binary(self, type_, **kw):

1264 return self.visit_BLOB(type_, **kw)

1265

1266 def visit_big_integer(self, type_, **kw):

1267 return self.visit_NUMBER(type_, precision=19, **kw)

1268

1269 def visit_boolean(self, type_, **kw):

1270 if self.dialect.supports_native_boolean:

1271 return self.visit_BOOLEAN(type_, **kw)

1272 else:

1273 return self.visit_SMALLINT(type_, **kw)

1274

1275 def visit_RAW(self, type_, **kw):

1276 if type_.length:

1277 return "RAW(%(length)s)" % {"length": type_.length}

1278 else:

1279 return "RAW"

1280

1281 def visit_ROWID(self, type_, **kw):

1282 return "ROWID"

1283

1284 def visit_VECTOR(self, type_, **kw):

1285 dim = type_.dim if type_.dim is not None else "*"

1286 storage_format = (

1287 type_.storage_format.value

1288 if type_.storage_format is not None

1289 else "*"

1290 )

1291 storage_type = (

1292 type_.storage_type.value if type_.storage_type is not None else "*"

1293 )

1294 return f"VECTOR({dim},{storage_format},{storage_type})"

1295

1296 def visit_JSON(self, type_: JSON, **kw: Any) -> str:

1297 use_blob = (

1298 not self.dialect._supports_oracle_json

1299 if getattr(type_, "use_blob", NO_ARG) is NO_ARG

1300 else type_.use_blob

1301 )

1302

1303 if use_blob:

1304 return "BLOB"

1305 else:

1306 return "JSON"

1307

1308

1309class OracleCompiler(compiler.SQLCompiler):

1310 """Oracle compiler modifies the lexical structure of Select

1311 statements to work under non-ANSI configured Oracle databases, if

1312 the use_ansi flag is False.

1313 """

1314

1315 compound_keywords = util.update_copy(

1316 compiler.SQLCompiler.compound_keywords,

1317 {expression.CompoundSelect.EXCEPT: "MINUS"},

1318 )

1319

1320 def __init__(self, *args, **kwargs):

1321 self.__wheres = {}

1322 super().__init__(*args, **kwargs)

1323

1324 def visit_mod_binary(self, binary, operator, **kw):

1325 return "mod(%s, %s)" % (

1326 self.process(binary.left, **kw),

1327 self.process(binary.right, **kw),

1328 )

1329

1330 def visit_now_func(self, fn, **kw):

1331 return "CURRENT_TIMESTAMP"

1332

1333 def visit_char_length_func(self, fn, **kw):

1334 return "LENGTH" + self.function_argspec(fn, **kw)

1335

1336 def visit_pow_func(self, fn, **kw):

1337 return f"POWER{self.function_argspec(fn)}"

1338

1339 def visit_match_op_binary(self, binary, operator, **kw):

1340 return "CONTAINS (%s, %s)" % (

1341 self.process(binary.left),

1342 self.process(binary.right),

1343 )

1344

1345 def visit_true(self, expr, **kw):

1346 return "1"

1347

1348 def visit_false(self, expr, **kw):

1349 return "0"

1350

1351 def visit_cast(self, cast, **kwargs):

1352 # Oracle requires VARCHAR2 to have a length in CAST expressions

1353 # Adapt String types to VARCHAR2 with appropriate length

1354 type_ = cast.typeclause.type

1355 if isinstance(type_, sqltypes.String) and not isinstance(

1356 type_, (sqltypes.Text, sqltypes.CLOB)

1357 ):

1358 adapted = VARCHAR2._adapt_string_for_cast(type_)

1359 type_clause = self.dialect.type_compiler_instance.process(adapted)

1360 else:

1361 type_clause = cast.typeclause._compiler_dispatch(self, **kwargs)

1362

1363 return "CAST(%s AS %s)" % (

1364 cast.clause._compiler_dispatch(self, **kwargs),

1365 type_clause,

1366 )

1367

1368 def get_cte_preamble(self, recursive):

1369 return "WITH"

1370

1371 def get_select_hint_text(self, byfroms):

1372 return " ".join("/*+ %s */" % text for table, text in byfroms.items())

1373

1374 def function_argspec(self, fn, **kw):

1375 if len(fn.clauses) > 0 or fn.name.upper() not in NO_ARG_FNS:

1376 return compiler.SQLCompiler.function_argspec(self, fn, **kw)

1377 else:

1378 return ""

1379

1380 def visit_function(self, func, **kw):

1381 text = super().visit_function(func, **kw)

1382 if kw.get("asfrom", False) and func.name.lower() != "table":

1383 text = "TABLE (%s)" % text

1384 return text

1385

1386 def visit_table_valued_column(self, element, **kw):

1387 text = super().visit_table_valued_column(element, **kw)

1388 text = text + ".COLUMN_VALUE"

1389 return text

1390

1391 def default_from(self):

1392 """Called when a ``SELECT`` statement has no froms,

1393 and no ``FROM`` clause is to be appended.

1394

1395 The Oracle compiler tacks a "FROM DUAL" to the statement.

1396 """

1397

1398 return " FROM DUAL"

1399

1400 def visit_join(self, join, from_linter=None, **kwargs):

1401 if self.dialect.use_ansi:

1402 return compiler.SQLCompiler.visit_join(

1403 self, join, from_linter=from_linter, **kwargs

1404 )

1405 else:

1406 if from_linter:

1407 from_linter.edges.add((join.left, join.right))

1408

1409 kwargs["asfrom"] = True

1410 if isinstance(join.right, expression.FromGrouping):

1411 right = join.right.element

1412 else:

1413 right = join.right

1414 return (

1415 self.process(join.left, from_linter=from_linter, **kwargs)

1416 + ", "

1417 + self.process(right, from_linter=from_linter, **kwargs)

1418 )

1419

1420 def _get_nonansi_join_whereclause(self, froms):

1421 clauses = []

1422

1423 def visit_join(join):

1424 if join.isouter:

1425 # https://docs.oracle.com/database/121/SQLRF/queries006.htm#SQLRF52354

1426 # "apply the outer join operator (+) to all columns of B in

1427 # the join condition in the WHERE clause" - that is,

1428 # unconditionally regardless of operator or the other side

1429 def visit_binary(binary):

1430 if isinstance(

1431 binary.left, expression.ColumnClause

1432 ) and join.right.is_derived_from(binary.left.table):

1433 binary.left = _OuterJoinColumn(binary.left)

1434 elif isinstance(

1435 binary.right, expression.ColumnClause

1436 ) and join.right.is_derived_from(binary.right.table):

1437 binary.right = _OuterJoinColumn(binary.right)

1438

1439 clauses.append(

1440 visitors.cloned_traverse(

1441 join.onclause, {}, {"binary": visit_binary}

1442 )

1443 )

1444 else:

1445 clauses.append(join.onclause)

1446

1447 for j in join.left, join.right:

1448 if isinstance(j, expression.Join):

1449 visit_join(j)

1450 elif isinstance(j, expression.FromGrouping):

1451 visit_join(j.element)

1452

1453 for f in froms:

1454 if isinstance(f, expression.Join):

1455 visit_join(f)

1456

1457 if not clauses:

1458 return None

1459 else:

1460 return sql.and_(*clauses)

1461

1462 def visit_outer_join_column(self, vc, **kw):

1463 return self.process(vc.column, **kw) + "(+)"

1464

1465 def visit_sequence(self, seq, **kw):

1466 return self.preparer.format_sequence(seq) + ".nextval"

1467

1468 def get_render_as_alias_suffix(self, alias_name_text):

1469 """Oracle doesn't like ``FROM table AS alias``"""

1470

1471 return " " + alias_name_text

1472

1473 def returning_clause(

1474 self, stmt, returning_cols, *, populate_result_map, **kw

1475 ):

1476 columns = []

1477 binds = []

1478

1479 for i, column in enumerate(

1480 expression._select_iterables(returning_cols)

1481 ):

1482 if (

1483 self.isupdate

1484 and isinstance(column, sa_schema.Column)

1485 and isinstance(column.server_default, Computed)

1486 and not self.dialect._supports_update_returning_computed_cols

1487 ):

1488 util.warn(

1489 "Computed columns don't work with Oracle Database UPDATE "

1490 "statements that use RETURNING; the value of the column "

1491 "*before* the UPDATE takes place is returned. It is "

1492 "advised to not use RETURNING with an Oracle Database "

1493 "computed column. Consider setting implicit_returning "

1494 "to False on the Table object in order to avoid implicit "

1495 "RETURNING clauses from being generated for this Table."

1496 )

1497 if column.type._has_column_expression:

1498 col_expr = column.type.column_expression(column)

1499 else:

1500 col_expr = column

1501

1502 outparam = sql.outparam("ret_%d" % i, type_=column.type)

1503 self.binds[outparam.key] = outparam

1504 binds.append(

1505 self.bindparam_string(self._truncate_bindparam(outparam))

1506 )

1507

1508 # has_out_parameters would in a normal case be set to True

1509 # as a result of the compiler visiting an outparam() object.

1510 # in this case, the above outparam() objects are not being

1511 # visited. Ensure the statement itself didn't have other

1512 # outparam() objects independently.

1513 # technically, this could be supported, but as it would be

1514 # a very strange use case without a clear rationale, disallow it

1515 if self.has_out_parameters:

1516 raise exc.InvalidRequestError(

1517 "Using explicit outparam() objects with "

1518 "UpdateBase.returning() in the same Core DML statement "

1519 "is not supported in the Oracle Database dialects."

1520 )

1521

1522 self._oracle_returning = True

1523

1524 columns.append(self.process(col_expr, within_columns_clause=False))

1525 if populate_result_map:

1526 self._add_to_result_map(

1527 getattr(col_expr, "name", col_expr._anon_name_label),

1528 getattr(col_expr, "name", col_expr._anon_name_label),

1529 (

1530 column,

1531 getattr(column, "name", None),

1532 getattr(column, "key", None),

1533 ),

1534 column.type,

1535 )

1536

1537 return "RETURNING " + ", ".join(columns) + " INTO " + ", ".join(binds)

1538

1539 def _row_limit_clause(self, select, **kw):

1540 """Oracle Database 12c supports OFFSET/FETCH operators

1541 Use it instead subquery with row_number

1542

1543 """

1544

1545 if (

1546 select._fetch_clause is not None

1547 or not self.dialect._supports_offset_fetch

1548 ):

1549 return super()._row_limit_clause(

1550 select, use_literal_execute_for_simple_int=True, **kw

1551 )

1552 else:

1553 return self.fetch_clause(

1554 select,

1555 fetch_clause=self._get_limit_or_fetch(select),

1556 use_literal_execute_for_simple_int=True,

1557 **kw,

1558 )

1559

1560 def _get_limit_or_fetch(self, select):

1561 if select._fetch_clause is None:

1562 return select._limit_clause

1563 else:

1564 return select._fetch_clause

1565

1566 def fetch_clause(

1567 self,

1568 select,

1569 fetch_clause=None,

1570 require_offset=False,

1571 use_literal_execute_for_simple_int=False,

1572 **kw,

1573 ):

1574 text = super().fetch_clause(

1575 select,

1576 fetch_clause=fetch_clause,

1577 require_offset=require_offset,

1578 use_literal_execute_for_simple_int=(

1579 use_literal_execute_for_simple_int

1580 ),

1581 **kw,

1582 )

1583

1584 if select.dialect_options["oracle"]["fetch_approximate"]:

1585 text = re.sub("FETCH FIRST", "FETCH APPROX FIRST", text)

1586

1587 return text

1588

1589 def translate_select_structure(self, select_stmt, **kwargs):

1590 select = select_stmt

1591

1592 if not getattr(select, "_oracle_visit", None):

1593 if not self.dialect.use_ansi:

1594 froms = self._display_froms_for_select(

1595 select, kwargs.get("asfrom", False)

1596 )

1597 whereclause = self._get_nonansi_join_whereclause(froms)

1598 if whereclause is not None:

1599 select = select.where(whereclause)

1600 select._oracle_visit = True

1601

1602 # if fetch is used this is not needed

1603 if (

1604 select._has_row_limiting_clause

1605 and not self.dialect._supports_offset_fetch

1606 and select._fetch_clause is None

1607 ):

1608 limit_clause = select._limit_clause

1609 offset_clause = select._offset_clause

1610

1611 if select._simple_int_clause(limit_clause):

1612 limit_clause = limit_clause.render_literal_execute()

1613

1614 if select._simple_int_clause(offset_clause):

1615 offset_clause = offset_clause.render_literal_execute()

1616

1617 # currently using form at:

1618 # https://blogs.oracle.com/oraclemagazine/\

1619 # on-rownum-and-limiting-results

1620

1621 orig_select = select

1622 select = select._generate()

1623 select._oracle_visit = True

1624

1625 # add expressions to accommodate FOR UPDATE OF

1626 for_update = select._for_update_arg

1627 if for_update is not None and for_update.of:

1628 for_update = for_update._clone()

1629 for_update._copy_internals()

1630

1631 for elem in for_update.of:

1632 if not select.selected_columns.contains_column(elem):

1633 select = select.add_columns(elem)

1634

1635 # Wrap the middle select and add the hint

1636 inner_subquery = select.alias()

1637 limitselect = sql.select(

1638 *[

1639 c

1640 for c in inner_subquery.c

1641 if orig_select.selected_columns.corresponding_column(c)

1642 is not None

1643 ]

1644 )

1645

1646 if (

1647 limit_clause is not None

1648 and self.dialect.optimize_limits

1649 and select._simple_int_clause(limit_clause)

1650 ):

1651 limitselect = limitselect.prefix_with(

1652 expression.text(

1653 "/*+ FIRST_ROWS(%s) */"

1654 % self.process(limit_clause, **kwargs)

1655 )

1656 )

1657

1658 limitselect._oracle_visit = True

1659 limitselect._is_wrapper = True

1660

1661 # add expressions to accommodate FOR UPDATE OF

1662 if for_update is not None and for_update.of:

1663 adapter = sql_util.ClauseAdapter(inner_subquery)

1664 for_update.of = [

1665 adapter.traverse(elem) for elem in for_update.of

1666 ]

1667

1668 # If needed, add the limiting clause

1669 if limit_clause is not None:

1670 if select._simple_int_clause(limit_clause) and (

1671 offset_clause is None

1672 or select._simple_int_clause(offset_clause)

1673 ):

1674 max_row = limit_clause

1675

1676 if offset_clause is not None:

1677 max_row = max_row + offset_clause

1678

1679 else:

1680 max_row = limit_clause

1681

1682 if offset_clause is not None:

1683 max_row = max_row + offset_clause

1684 limitselect = limitselect.where(

1685 sql.literal_column("ROWNUM") <= max_row

1686 )

1687

1688 # If needed, add the ora_rn, and wrap again with offset.

1689 if offset_clause is None:

1690 limitselect._for_update_arg = for_update

1691 select = limitselect

1692 else:

1693 limitselect = limitselect.add_columns(

1694 sql.literal_column("ROWNUM").label("ora_rn")

1695 )

1696 limitselect._oracle_visit = True

1697 limitselect._is_wrapper = True

1698

1699 if for_update is not None and for_update.of:

1700 limitselect_cols = limitselect.selected_columns

1701 for elem in for_update.of:

1702 if (

1703 limitselect_cols.corresponding_column(elem)

1704 is None

1705 ):

1706 limitselect = limitselect.add_columns(elem)

1707

1708 limit_subquery = limitselect.alias()

1709 origselect_cols = orig_select.selected_columns

1710 offsetselect = sql.select(

1711 *[

1712 c

1713 for c in limit_subquery.c

1714 if origselect_cols.corresponding_column(c)

1715 is not None

1716 ]

1717 )

1718

1719 offsetselect._oracle_visit = True

1720 offsetselect._is_wrapper = True

1721

1722 if for_update is not None and for_update.of:

1723 adapter = sql_util.ClauseAdapter(limit_subquery)

1724 for_update.of = [

1725 adapter.traverse(elem) for elem in for_update.of

1726 ]

1727

1728 offsetselect = offsetselect.where(

1729 sql.literal_column("ora_rn") > offset_clause

1730 )

1731

1732 offsetselect._for_update_arg = for_update

1733 select = offsetselect

1734

1735 return select

1736

1737 def limit_clause(self, select, **kw):

1738 return ""

1739

1740 def visit_empty_set_expr(self, type_, **kw):

1741 return "SELECT 1 FROM DUAL WHERE 1!=1"

1742

1743 def for_update_clause(self, select, **kw):

1744 if self.is_subquery():

1745 return ""

1746

1747 tmp = " FOR UPDATE"

1748

1749 if select._for_update_arg.of:

1750 tmp += " OF " + ", ".join(

1751 self.process(elem, **kw) for elem in select._for_update_arg.of

1752 )

1753

1754 if select._for_update_arg.nowait:

1755 tmp += " NOWAIT"

1756 if select._for_update_arg.skip_locked:

1757 tmp += " SKIP LOCKED"

1758

1759 return tmp

1760

1761 def visit_is_distinct_from_binary(self, binary, operator, **kw):

1762 return "DECODE(%s, %s, 0, 1) = 1" % (

1763 self.process(binary.left),

1764 self.process(binary.right),

1765 )

1766

1767 def visit_is_not_distinct_from_binary(self, binary, operator, **kw):

1768 return "DECODE(%s, %s, 0, 1) = 0" % (

1769 self.process(binary.left),

1770 self.process(binary.right),

1771 )

1772

1773 def visit_regexp_match_op_binary(self, binary, operator, **kw):

1774 string = self.process(binary.left, **kw)

1775 pattern = self.process(binary.right, **kw)

1776 flags = binary.modifiers["flags"]

1777 if flags is None:

1778 return "REGEXP_LIKE(%s, %s)" % (string, pattern)

1779 else:

1780 return "REGEXP_LIKE(%s, %s, %s)" % (

1781 string,

1782 pattern,

1783 self.render_literal_value(flags, sqltypes.STRINGTYPE),

1784 )

1785

1786 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):

1787 return "NOT %s" % self.visit_regexp_match_op_binary(

1788 binary, operator, **kw

1789 )

1790

1791 def visit_regexp_replace_op_binary(self, binary, operator, **kw):

1792 string = self.process(binary.left, **kw)

1793 pattern_replace = self.process(binary.right, **kw)

1794 flags = binary.modifiers["flags"]

1795 if flags is None:

1796 return "REGEXP_REPLACE(%s, %s)" % (

1797 string,

1798 pattern_replace,

1799 )

1800 else:

1801 return "REGEXP_REPLACE(%s, %s, %s)" % (

1802 string,

1803 pattern_replace,

1804 self.render_literal_value(flags, sqltypes.STRINGTYPE),

1805 )

1806

1807 def visit_aggregate_strings_func(self, fn, **kw):

1808 return super().visit_aggregate_strings_func(

1809 fn, use_function_name="LISTAGG", **kw

1810 )

1811

1812 def _visit_bitwise(self, binary, fn_name, custom_right=None, **kw):

1813 left = self.process(binary.left, **kw)

1814 right = self.process(

1815 custom_right if custom_right is not None else binary.right, **kw

1816 )

1817 return f"{fn_name}({left}, {right})"

1818

1819 def visit_bitwise_xor_op_binary(self, binary, operator, **kw):

1820 return self._visit_bitwise(binary, "BITXOR", **kw)

1821

1822 def visit_bitwise_or_op_binary(self, binary, operator, **kw):

1823 return self._visit_bitwise(binary, "BITOR", **kw)

1824

1825 def visit_bitwise_and_op_binary(self, binary, operator, **kw):

1826 return self._visit_bitwise(binary, "BITAND", **kw)

1827

1828 def visit_bitwise_rshift_op_binary(self, binary, operator, **kw):

1829 raise exc.CompileError("Cannot compile bitwise_rshift in oracle")

1830

1831 def visit_bitwise_lshift_op_binary(self, binary, operator, **kw):

1832 raise exc.CompileError("Cannot compile bitwise_lshift in oracle")

1833

1834 def visit_bitwise_not_op_unary_operator(self, element, operator, **kw):

1835 raise exc.CompileError("Cannot compile bitwise_not in oracle")

1836

1837 def _render_json_extract_from_binary(self, binary, operator, **kw):

1838 literal_kw = kw.copy()

1839 literal_kw["literal_binds"] = True

1840

1841 left = self.process(binary.left, **kw)

1842 right = self.process(binary.right, **literal_kw)

1843

1844 if binary.type._type_affinity is sqltypes.Boolean:

1845 # RETURNING clause doesn't handle true/false to 1/0

1846 # mapping, so use CASE expression for boolean

1847 return (

1848 f"CASE JSON_VALUE({left}, {right})"

1849 f" WHEN 'true' THEN 1"

1850 f" WHEN 'false' THEN 0"

1851 f" ELSE CAST(JSON_VALUE({left}, {right})"

1852 f" AS NUMBER(1)) END"

1853 )

1854 elif binary.type._type_affinity is sqltypes.Integer:

1855 json_value_returning = "INTEGER"

1856 elif binary.type._type_affinity in (

1857 sqltypes.Numeric,

1858 sqltypes.Float,

1859 ):

1860 if isinstance(binary.type, sqltypes.Float):

1861 json_value_returning = "FLOAT"

1862 else:

1863 json_value_returning = (

1864 f"NUMBER({binary.type.precision}, {binary.type.scale})"

1865 )

1866 elif binary.type._type_affinity is sqltypes.String:

1867 json_value_returning = "VARCHAR2(4000)"

1868 else:

1869 # binary.type._type_affinity is sqltypes.JSON

1870 # or other

1871 return f"JSON_QUERY({left}, {right})"

1872

1873 return (

1874 f"JSON_VALUE({left}, {right}"

1875 f" RETURNING {json_value_returning} ERROR ON ERROR)"

1876 )

1877

1878 def visit_json_getitem_op_binary(

1879 self, binary: elements.BinaryExpression[Any], operator: Any, **kw: Any

1880 ) -> str:

1881 return self._render_json_extract_from_binary(binary, operator, **kw)

1882

1883 def visit_json_path_getitem_op_binary(

1884 self, binary: elements.BinaryExpression[Any], operator: Any, **kw: Any

1885 ) -> str:

1886 return self._render_json_extract_from_binary(binary, operator, **kw)

1887

1888

1889class OracleDDLCompiler(compiler.DDLCompiler):

1890

1891 def _build_vector_index_config(

1892 self, vector_index_config: VectorIndexConfig

1893 ) -> str:

1894 parts = []

1895 sql_param_name = {

1896 "hnsw_neighbors": "neighbors",

1897 "hnsw_efconstruction": "efconstruction",

1898 "ivf_neighbor_partitions": "neighbor partitions",

1899 "ivf_sample_per_partition": "sample_per_partition",

1900 "ivf_min_vectors_per_partition": "min_vectors_per_partition",

1901 }

1902 if vector_index_config.index_type == VectorIndexType.HNSW:

1903 parts.append("ORGANIZATION INMEMORY NEIGHBOR GRAPH")

1904 elif vector_index_config.index_type == VectorIndexType.IVF:

1905 parts.append("ORGANIZATION NEIGHBOR PARTITIONS")

1906 if vector_index_config.distance is not None:

1907 parts.append(f"DISTANCE {vector_index_config.distance.value}")

1908

1909 if vector_index_config.accuracy is not None:

1910 parts.append(

1911 f"WITH TARGET ACCURACY {vector_index_config.accuracy}"

1912 )

1913

1914 parameters_str = [f"type {vector_index_config.index_type.name}"]

1915 prefix = vector_index_config.index_type.name.lower() + "_"

1916

1917 for field in fields(vector_index_config):

1918 if field.name.startswith(prefix):

1919 key = sql_param_name.get(field.name)

1920 value = getattr(vector_index_config, field.name)

1921 if value is not None:

1922 parameters_str.append(f"{key} {value}")

1923

1924 parameters_str = ", ".join(parameters_str)

1925 parts.append(f"PARAMETERS ({parameters_str})")

1926

1927 if vector_index_config.parallel is not None:

1928 parts.append(f"PARALLEL {vector_index_config.parallel}")

1929

1930 return " ".join(parts)

1931

1932 def define_constraint_cascades(self, constraint):

1933 text = ""

1934 if constraint.ondelete is not None:

1935 text += " ON DELETE %s" % constraint.ondelete

1936

1937 # oracle has no ON UPDATE CASCADE -

1938 # its only available via triggers

1939 # https://web.archive.org/web/20090317041251/https://asktom.oracle.com/tkyte/update_cascade/index.html

1940 if constraint.onupdate is not None:

1941 util.warn(

1942 "Oracle Database does not contain native UPDATE CASCADE "

1943 "functionality - onupdates will not be rendered for foreign "

1944 "keys. Consider using deferrable=True, initially='deferred' "

1945 "or triggers."

1946 )

1947

1948 return text

1949

1950 def visit_drop_table_comment(self, drop, **kw):

1951 return "COMMENT ON TABLE %s IS ''" % self.preparer.format_table(

1952 drop.element

1953 )

1954

1955 def visit_create_index(self, create, **kw):

1956 index = create.element

1957 self._verify_index_table(index)

1958 preparer = self.preparer

1959 text = "CREATE "

1960 if index.unique:

1961 text += "UNIQUE "

1962 if index.dialect_options["oracle"]["bitmap"]:

1963 text += "BITMAP "

1964 vector_options = index.dialect_options["oracle"]["vector"]

1965 if vector_options:

1966 text += "VECTOR "

1967 text += "INDEX %s ON %s (%s)" % (

1968 self._prepared_index_name(index, include_schema=True),

1969 preparer.format_table(index.table, use_schema=True),

1970 ", ".join(

1971 self.sql_compiler.process(

1972 expr, include_table=False, literal_binds=True

1973 )

1974 for expr in index.expressions

1975 ),

1976 )

1977 if index.dialect_options["oracle"]["compress"] is not False:

1978 if index.dialect_options["oracle"]["compress"] is True:

1979 text += " COMPRESS"

1980 else:

1981 text += " COMPRESS %d" % (

1982 index.dialect_options["oracle"]["compress"]

1983 )

1984 if vector_options:

1985 if vector_options is True:

1986 vector_options = VectorIndexConfig()

1987

1988 text += " " + self._build_vector_index_config(vector_options)

1989 return text

1990

1991 def post_create_table(self, table):

1992 table_opts = []

1993 opts = table.dialect_options["oracle"]

1994

1995 if opts["on_commit"]:

1996 on_commit_options = opts["on_commit"].replace("_", " ").upper()

1997 table_opts.append("\n ON COMMIT %s" % on_commit_options)

1998

1999 if opts["compress"]:

2000 if opts["compress"] is True:

2001 table_opts.append("\n COMPRESS")

2002 else:

2003 table_opts.append("\n COMPRESS FOR %s" % (opts["compress"]))

2004 if opts["tablespace"]:

2005 table_opts.append(

2006 "\n TABLESPACE %s" % self.preparer.quote(opts["tablespace"])

2007 )

2008 return "".join(table_opts)

2009

2010 def get_identity_options(self, identity_options):

2011 text = super().get_identity_options(identity_options)

2012 text = text.replace("NO MINVALUE", "NOMINVALUE")

2013 text = text.replace("NO MAXVALUE", "NOMAXVALUE")

2014 text = text.replace("NO CYCLE", "NOCYCLE")

2015 options = identity_options.dialect_options["oracle"]

2016 if options.get("order") is not None:

2017 text += " ORDER" if options["order"] else " NOORDER"

2018 return text.strip()

2019

2020 def visit_computed_column(self, generated, **kw):

2021 text = "GENERATED ALWAYS AS (%s)" % self.sql_compiler.process(

2022 generated.sqltext, include_table=False, literal_binds=True

2023 )

2024 if generated.persisted is True:

2025 raise exc.CompileError(

2026 "Oracle Database computed columns do not support 'stored' "

2027 "persistence; set the 'persisted' flag to None or False for "

2028 "Oracle Database support."

2029 )

2030 elif generated.persisted is False:

2031 text += " VIRTUAL"

2032 return text

2033

2034 def visit_identity_column(self, identity, **kw):

2035 if identity.always is None:

2036 kind = ""

2037 else:

2038 kind = "ALWAYS" if identity.always else "BY DEFAULT"

2039 text = "GENERATED %s" % kind

2040 if identity.dialect_options["oracle"].get("on_null"):

2041 text += " ON NULL"

2042 text += " AS IDENTITY"

2043 options = self.get_identity_options(identity)

2044 if options:

2045 text += " (%s)" % options

2046 return text

2047

2048

2049class OracleIdentifierPreparer(compiler.IdentifierPreparer):

2050 reserved_words = {x.lower() for x in RESERVED_WORDS}

2051 illegal_initial_characters = {str(dig) for dig in range(0, 10)}.union(

2052 ["_", "$"]

2053 )

2054

2055 def _bindparam_requires_quotes(self, value):

2056 """Return True if the given identifier requires quoting."""

2057 lc_value = value.lower()

2058 return (

2059 lc_value in self.reserved_words

2060 or value[0] in self.illegal_initial_characters

2061 or not self.legal_characters.match(str(value))

2062 )

2063

2064 def format_savepoint(self, savepoint):

2065 name = savepoint.ident.lstrip("_")

2066 return super().format_savepoint(savepoint, name)

2067

2068

2069class OracleExecutionContext(default.DefaultExecutionContext):

2070 def fire_sequence(self, seq, type_):

2071 return self._execute_scalar(

2072 "SELECT "

2073 + self.identifier_preparer.format_sequence(seq)

2074 + ".nextval FROM DUAL",

2075 type_,

2076 )

2077

2078 def pre_exec(self):

2079 if self.statement and "_oracle_dblink" in self.execution_options:

2080 self.statement = self.statement.replace(

2081 dictionary.DB_LINK_PLACEHOLDER,

2082 self.execution_options["_oracle_dblink"],

2083 )

2084

2085

2086class OracleDialect(default._BackendsMultiReflection, default.DefaultDialect):

2087 name = "oracle"

2088 supports_statement_cache = True

2089 supports_alter = True

2090 max_identifier_length = 128

2091

2092 _supports_offset_fetch = True

2093

2094 insert_returning = True

2095 update_returning = True

2096 delete_returning = True

2097

2098 div_is_floordiv = False

2099

2100 supports_simple_order_by_label = False

2101 cte_follows_insert = True

2102 returns_native_bytes = True

2103

2104 supports_native_boolean = True

2105 supports_sequences = True

2106 sequences_optional = False

2107 postfetch_lastrowid = False

2108

2109 default_paramstyle = "named"

2110 colspecs = colspecs

2111 ischema_names = ischema_names

2112 requires_name_normalize = True

2113

2114 supports_comments = True

2115

2116 supports_default_values = False

2117 supports_default_metavalue = True

2118 supports_empty_insert = False

2119 supports_identity_columns = True

2120

2121 _supports_oracle_json = True

2122

2123 aggregate_order_by_style = AggregateOrderByStyle.WITHIN_GROUP

2124

2125 statement_compiler = OracleCompiler

2126 ddl_compiler = OracleDDLCompiler

2127 type_compiler_cls = OracleTypeCompiler

2128 preparer = OracleIdentifierPreparer

2129 execution_ctx_cls = OracleExecutionContext

2130

2131 reflection_options = ("oracle_resolve_synonyms",)

2132

2133 _use_nchar_for_unicode = False

2134

2135 construct_arguments = [

2136 (

2137 sa_schema.Table,

2138 {

2139 "resolve_synonyms": False,

2140 "on_commit": None,

2141 "compress": False,

2142 "tablespace": None,

2143 },

2144 ),

2145 (

2146 sa_schema.Index,

2147 {

2148 "bitmap": False,

2149 "compress": False,

2150 "vector": False,

2151 },

2152 ),

2153 (sa_schema.Sequence, {"order": None}),

2154 (sa_schema.Identity, {"order": None, "on_null": None}),

2155 (sa_selectable.Select, {"fetch_approximate": False}),

2156 (sa_selectable.CompoundSelect, {"fetch_approximate": False}),

2157 ]

2158

2159 @util.deprecated_params(

2160 use_binds_for_limits=(

2161 "1.4",

2162 "The ``use_binds_for_limits`` Oracle Database dialect parameter "

2163 "is deprecated. The dialect now renders LIMIT / OFFSET integers "

2164 "inline in all cases using a post-compilation hook, so that the "

2165 "value is still represented by a 'bound parameter' on the Core "

2166 "Expression side.",

2167 )

2168 )

2169 def __init__(

2170 self,

2171 use_ansi=True,

2172 optimize_limits=False,

2173 use_binds_for_limits=None,

2174 use_nchar_for_unicode=False,

2175 exclude_tablespaces=("SYSTEM", "SYSAUX"),

2176 enable_offset_fetch=True,

2177 json_serializer: Callable[[_JSON_VALUE], str] | None = None,

2178 json_deserializer: Callable[[str], _JSON_VALUE] | None = None,

2179 **kwargs,

2180 ):

2181 default.DefaultDialect.__init__(self, **kwargs)

2182 self._use_nchar_for_unicode = use_nchar_for_unicode

2183 self.use_ansi = use_ansi

2184 self.optimize_limits = optimize_limits

2185 self.exclude_tablespaces = exclude_tablespaces

2186 self.enable_offset_fetch = self._supports_offset_fetch = (

2187 enable_offset_fetch

2188 )

2189 self._json_serializer = json_serializer

2190 self._json_deserializer = json_deserializer

2191

2192 def initialize(self, connection):

2193 super().initialize(connection)

2194

2195 # Oracle 8i has RETURNING:

2196 # https://docs.oracle.com/cd/A87860_01/doc/index.htm

2197

2198 # so does Oracle8:

2199 # https://docs.oracle.com/cd/A64702_01/doc/index.htm

2200

2201 if self._is_oracle_8:

2202 self.colspecs = self.colspecs.copy()

2203 self.colspecs.pop(sqltypes.Interval)

2204 self.use_ansi = False

2205

2206 self._supports_oracle_json = self.server_version_info >= (21,)

2207 self.supports_native_boolean = self.server_version_info >= (23,)

2208 self.supports_identity_columns = self.server_version_info >= (12,)

2209 self._supports_offset_fetch = (

2210 self.enable_offset_fetch and self.server_version_info >= (12,)

2211 )

2212

2213 def _get_effective_compat_server_version_info(self, connection):

2214 # dialect does not need compat levels below 12.2, so don't query

2215 # in those cases

2216

2217 if self.server_version_info < (12, 2):

2218 return self.server_version_info

2219 try:

2220 compat = connection.exec_driver_sql(

2221 "SELECT value FROM v$parameter WHERE name = 'compatible'"

2222 ).scalar()

2223 except exc.DBAPIError:

2224 compat = None

2225

2226 if compat:

2227 try:

2228 return tuple(int(x) for x in compat.split("."))

2229 except:

2230 return self.server_version_info

2231 else:

2232 return self.server_version_info

2233

2234 @property

2235 def _is_oracle_8(self):

2236 return self.server_version_info and self.server_version_info < (9,)

2237

2238 @property

2239 def _supports_table_compression(self):

2240 return self.server_version_info and self.server_version_info >= (10, 1)

2241

2242 @property

2243 def _supports_table_compress_for(self):

2244 return self.server_version_info and self.server_version_info >= (11,)

2245

2246 @property

2247 def _supports_char_length(self):

2248 return not self._is_oracle_8

2249

2250 @property

2251 def _supports_update_returning_computed_cols(self):

2252 # on version 18 this error is no longet present while it happens on 11

2253 # it may work also on versions before the 18

2254 return self.server_version_info and self.server_version_info >= (18,)

2255

2256 @property

2257 def _supports_except_all(self):

2258 return self.server_version_info and self.server_version_info >= (21,)

2259

2260 def do_release_savepoint(self, connection, name):

2261 # Oracle does not support RELEASE SAVEPOINT

2262 pass

2263

2264 def _check_max_identifier_length(self, connection):

2265 if self._get_effective_compat_server_version_info(connection) < (

2266 12,

2267 2,

2268 ):

2269 return 30

2270 else:

2271 # use the default

2272 return None

2273

2274 def get_isolation_level_values(self, dbapi_connection):

2275 return ["READ COMMITTED", "SERIALIZABLE"]

2276

2277 def get_default_isolation_level(self, dbapi_conn):

2278 try:

2279 return self.get_isolation_level(dbapi_conn)

2280 except NotImplementedError:

2281 raise

2282 except:

2283 return "READ COMMITTED"

2284

2285 def _execute_reflection(

2286 self, connection, query, dblink, returns_long, params=None

2287 ):

2288 if dblink and not dblink.startswith("@"):

2289 dblink = f"@{dblink}"

2290 execution_options = {

2291 # handle db links

2292 "_oracle_dblink": dblink or "",

2293 # override any schema translate map

2294 "schema_translate_map": None,

2295 }

2296

2297 if dblink and returns_long:

2298 # Oracle seems to error with

2299 # "ORA-00997: illegal use of LONG datatype" when returning

2300 # LONG columns via a dblink in a query with bind params

2301 # This type seems to be very hard to cast into something else

2302 # so it seems easier to just use bind param in this case

2303 def visit_bindparam(bindparam):

2304 bindparam.literal_execute = True

2305

2306 query = visitors.cloned_traverse(

2307 query, {}, {"bindparam": visit_bindparam}

2308 )

2309 return connection.execute(

2310 query, params, execution_options=execution_options

2311 )

2312

2313 @util.memoized_property

2314 def _has_multi_table_query(self):

2315 # materialized views are returned by all_tables

2316 tables = (

2317 select(

2318 dictionary.all_tables.c.table_name,

2319 dictionary.all_tables.c.owner,

2320 )

2321 .union_all(

2322 select(

2323 dictionary.all_views.c.view_name.label("table_name"),

2324 dictionary.all_views.c.owner,

2325 )

2326 )

2327 .subquery("tables_and_views")

2328 )

2329

2330 query = select(tables.c.table_name).where(

2331 tables.c.table_name.in_(bindparam("table_names")),

2332 tables.c.owner == bindparam("owner"),

2333 )

2334 return query

2335

2336 def has_multi_table(

2337 self, connection, table_names, schema=None, dblink=None, **kw

2338 ):

2339 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2340

2341 owner = schema or self.default_schema_name

2342 db_tns = [self.denormalize_name(tn) for tn in table_names]

2343 params = {

2344 "table_names": db_tns,

2345 "owner": self.denormalize_schema_name(owner),

2346 }

2347 cursor = self._execute_reflection(

2348 connection,

2349 self._has_multi_table_query,

2350 dblink,

2351 returns_long=False,

2352 params=params,

2353 )

2354 existing = set(cursor.scalars().all())

2355 retval = {

2356 (schema, table): db_tn in existing

2357 for table, db_tn in zip(table_names, db_tns)

2358 }

2359 return retval.items()

2360

2361 @reflection.cache

2362 def has_sequence(

2363 self, connection, sequence_name, schema=None, dblink=None, **kw

2364 ):

2365 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2366 if not schema:

2367 schema = self.default_schema_name

2368

2369 query = select(dictionary.all_sequences.c.sequence_name).where(

2370 dictionary.all_sequences.c.sequence_name

2371 == self.denormalize_schema_name(sequence_name),

2372 dictionary.all_sequences.c.sequence_owner

2373 == self.denormalize_schema_name(schema),

2374 )

2375

2376 cursor = self._execute_reflection(

2377 connection, query, dblink, returns_long=False

2378 )

2379 return bool(cursor.scalar())

2380

2381 def _get_default_schema_name(self, connection):

2382 return self.normalize_name(

2383 connection.exec_driver_sql(

2384 "select sys_context( 'userenv', 'current_schema' ) from dual"

2385 ).scalar()

2386 )

2387

2388 def denormalize_schema_name(self, name):

2389 # look for quoted_name

2390 force = getattr(name, "quote", None)

2391 if force is None and name == "public":

2392 # look for case insensitive, no quoting specified, "public"

2393 return "PUBLIC"

2394 return super().denormalize_name(name)

2395

2396 @reflection.flexi_cache(

2397 ("schema", InternalTraversal.dp_string),

2398 ("filter_names", InternalTraversal.dp_string_list),

2399 ("dblink", InternalTraversal.dp_string),

2400 )

2401 def _get_synonyms(self, connection, schema, filter_names, dblink, **kw):

2402 owner = self.denormalize_schema_name(

2403 schema or self.default_schema_name

2404 )

2405

2406 has_filter_names, params = self._prepare_filter_names(filter_names)

2407 query = select(

2408 dictionary.all_synonyms.c.synonym_name,

2409 dictionary.all_synonyms.c.table_name,

2410 dictionary.all_synonyms.c.table_owner,

2411 dictionary.all_synonyms.c.db_link,

2412 ).where(dictionary.all_synonyms.c.owner == owner)

2413 if has_filter_names:

2414 query = query.where(

2415 dictionary.all_synonyms.c.synonym_name.in_(

2416 params["filter_names"]

2417 )

2418 )

2419 result = self._execute_reflection(

2420 connection, query, dblink, returns_long=False

2421 ).mappings()

2422 return result.all()

2423

2424 @lru_cache()

2425 def _all_objects_query(

2426 self, owner, scope, kind, has_filter_names, has_mat_views

2427 ):

2428 query = (

2429 select(dictionary.all_objects.c.object_name)

2430 .select_from(dictionary.all_objects)

2431 .where(dictionary.all_objects.c.owner == owner)

2432 )

2433

2434 # NOTE: materialized views are listed in all_objects twice;

2435 # once as MATERIALIZE VIEW and once as TABLE

2436 if kind is ObjectKind.ANY:

2437 # materilaized view are listed also as tables so there is no

2438 # need to add them to the in_.

2439 query = query.where(

2440 dictionary.all_objects.c.object_type.in_(("TABLE", "VIEW"))

2441 )

2442 else:

2443 object_type = []

2444 if ObjectKind.VIEW in kind:

2445 object_type.append("VIEW")

2446 if (

2447 ObjectKind.MATERIALIZED_VIEW in kind

2448 and ObjectKind.TABLE not in kind

2449 ):

2450 # materilaized view are listed also as tables so there is no

2451 # need to add them to the in_ if also selecting tables.

2452 object_type.append("MATERIALIZED VIEW")

2453 if ObjectKind.TABLE in kind:

2454 object_type.append("TABLE")

2455 if has_mat_views and ObjectKind.MATERIALIZED_VIEW not in kind:

2456 # materialized view are listed also as tables,

2457 # so they need to be filtered out

2458 # EXCEPT ALL / MINUS profiles as faster than using

2459 # NOT EXISTS or NOT IN with a subquery, but it's in

2460 # general faster to get the mat view names and exclude

2461 # them only when needed

2462 query = query.where(

2463 dictionary.all_objects.c.object_name.not_in(

2464 bindparam("mat_views")

2465 )

2466 )

2467 query = query.where(

2468 dictionary.all_objects.c.object_type.in_(object_type)

2469 )

2470

2471 # handles scope

2472 if scope is ObjectScope.DEFAULT:

2473 query = query.where(dictionary.all_objects.c.temporary == "N")

2474 elif scope is ObjectScope.TEMPORARY:

2475 query = query.where(dictionary.all_objects.c.temporary == "Y")

2476

2477 if has_filter_names:

2478 query = query.where(

2479 dictionary.all_objects.c.object_name.in_(

2480 bindparam("filter_names")

2481 )

2482 )

2483 return query

2484

2485 @reflection.flexi_cache(

2486 ("schema", InternalTraversal.dp_string),

2487 ("scope", InternalTraversal.dp_plain_obj),

2488 ("kind", InternalTraversal.dp_plain_obj),

2489 ("filter_names", InternalTraversal.dp_string_list),

2490 ("dblink", InternalTraversal.dp_string),

2491 )

2492 def _get_all_objects(

2493 self, connection, schema, scope, kind, filter_names, dblink, **kw

2494 ):

2495 owner = self.denormalize_schema_name(

2496 schema or self.default_schema_name

2497 )

2498

2499 has_filter_names, params = self._prepare_filter_names(filter_names)

2500 has_mat_views = False

2501 if (

2502 ObjectKind.TABLE in kind

2503 and ObjectKind.MATERIALIZED_VIEW not in kind

2504 ):

2505 # see note in _all_objects_query

2506 mat_views = self.get_materialized_view_names(

2507 connection, schema, dblink, _normalize=False, **kw

2508 )

2509 if mat_views:

2510 params["mat_views"] = mat_views

2511 has_mat_views = True

2512

2513 query = self._all_objects_query(

2514 owner, scope, kind, has_filter_names, has_mat_views

2515 )

2516

2517 result = self._execute_reflection(

2518 connection, query, dblink, returns_long=False, params=params

2519 ).scalars()

2520

2521 return result.all()

2522

2523 def _handle_synonyms_decorator(fn):

2524 @wraps(fn)

2525 def wrapper(self, *args, **kwargs):

2526 return self._handle_synonyms(fn, *args, **kwargs)

2527

2528 return wrapper

2529

2530 def _handle_synonyms(self, fn, connection, *args, **kwargs):

2531 if not kwargs.get("oracle_resolve_synonyms", False):

2532 return fn(self, connection, *args, **kwargs)

2533

2534 original_kw = kwargs.copy()

2535 schema = kwargs.pop("schema", None)

2536 result = self._get_synonyms(

2537 connection,

2538 schema=schema,

2539 filter_names=kwargs.pop("filter_names", None),

2540 dblink=kwargs.pop("dblink", None),

2541 info_cache=kwargs.get("info_cache", None),

2542 )

2543

2544 dblinks_owners = defaultdict(dict)

2545 for row in result:

2546 key = row["db_link"], row["table_owner"]

2547 tn = self.normalize_name(row["table_name"])

2548 dblinks_owners[key][tn] = row["synonym_name"]

2549

2550 if not dblinks_owners:

2551 # No synonym, do the plain thing

2552 return fn(self, connection, *args, **original_kw)

2553

2554 data = {}

2555 for (dblink, table_owner), mapping in dblinks_owners.items():

2556 call_kw = {

2557 **original_kw,

2558 "schema": table_owner,

2559 "dblink": self.normalize_name(dblink),

2560 "filter_names": mapping.keys(),

2561 }

2562 call_result = fn(self, connection, *args, **call_kw)

2563 for (_, tn), value in call_result:

2564 synonym_name = self.normalize_name(mapping[tn])

2565 data[(schema, synonym_name)] = value

2566 return data.items()

2567

2568 @reflection.cache

2569 def get_schema_names(self, connection, dblink=None, **kw):

2570 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2571 query = select(dictionary.all_users.c.username).order_by(

2572 dictionary.all_users.c.username

2573 )

2574 result = self._execute_reflection(

2575 connection, query, dblink, returns_long=False

2576 ).scalars()

2577 return [self.normalize_name(row) for row in result]

2578

2579 @reflection.cache

2580 def get_table_names(self, connection, schema=None, dblink=None, **kw):

2581 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2582 # note that table_names() isn't loading DBLINKed or synonym'ed tables

2583 if schema is None:

2584 schema = self.default_schema_name

2585

2586 den_schema = self.denormalize_schema_name(schema)

2587 if kw.get("oracle_resolve_synonyms", False):

2588 tables = (

2589 select(

2590 dictionary.all_tables.c.table_name,

2591 dictionary.all_tables.c.owner,

2592 dictionary.all_tables.c.iot_name,

2593 dictionary.all_tables.c.duration,

2594 dictionary.all_tables.c.tablespace_name,

2595 )

2596 .union_all(

2597 select(

2598 dictionary.all_synonyms.c.synonym_name.label(

2599 "table_name"

2600 ),

2601 dictionary.all_synonyms.c.owner,

2602 dictionary.all_tables.c.iot_name,

2603 dictionary.all_tables.c.duration,

2604 dictionary.all_tables.c.tablespace_name,

2605 )

2606 .select_from(dictionary.all_tables)

2607 .join(

2608 dictionary.all_synonyms,

2609 and_(

2610 dictionary.all_tables.c.table_name

2611 == dictionary.all_synonyms.c.table_name,

2612 dictionary.all_tables.c.owner

2613 == func.coalesce(

2614 dictionary.all_synonyms.c.table_owner,

2615 dictionary.all_synonyms.c.owner,

2616 ),

2617 ),

2618 )

2619 )

2620 .subquery("available_tables")

2621 )

2622 else:

2623 tables = dictionary.all_tables

2624

2625 query = select(tables.c.table_name)

2626 if self.exclude_tablespaces:

2627 query = query.where(

2628 func.coalesce(

2629 tables.c.tablespace_name, "no tablespace"

2630 ).not_in(self.exclude_tablespaces)

2631 )

2632 query = query.where(

2633 tables.c.owner == den_schema,

2634 tables.c.iot_name.is_(null()),

2635 tables.c.duration.is_(null()),

2636 )

2637

2638 # remove materialized views

2639 mat_query = select(

2640 dictionary.all_mviews.c.mview_name.label("table_name")

2641 ).where(dictionary.all_mviews.c.owner == den_schema)

2642

2643 query = (

2644 query.except_all(mat_query)

2645 if self._supports_except_all

2646 else query.except_(mat_query)

2647 )

2648

2649 result = self._execute_reflection(

2650 connection, query, dblink, returns_long=False

2651 ).scalars()

2652 return [self.normalize_name(row) for row in result]

2653

2654 @reflection.cache

2655 def get_temp_table_names(self, connection, dblink=None, **kw):

2656 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2657 schema = self.denormalize_schema_name(self.default_schema_name)

2658

2659 query = select(dictionary.all_tables.c.table_name)

2660 if self.exclude_tablespaces:

2661 query = query.where(

2662 func.coalesce(

2663 dictionary.all_tables.c.tablespace_name, "no tablespace"

2664 ).not_in(self.exclude_tablespaces)

2665 )

2666 query = query.where(

2667 dictionary.all_tables.c.owner == schema,

2668 dictionary.all_tables.c.iot_name.is_(null()),

2669 dictionary.all_tables.c.duration.is_not(null()),

2670 )

2671

2672 result = self._execute_reflection(

2673 connection, query, dblink, returns_long=False

2674 ).scalars()

2675 return [self.normalize_name(row) for row in result]

2676

2677 @reflection.cache

2678 def get_materialized_view_names(

2679 self, connection, schema=None, dblink=None, _normalize=True, **kw

2680 ):

2681 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2682 if not schema:

2683 schema = self.default_schema_name

2684

2685 query = select(dictionary.all_mviews.c.mview_name).where(

2686 dictionary.all_mviews.c.owner

2687 == self.denormalize_schema_name(schema)

2688 )

2689 result = self._execute_reflection(

2690 connection, query, dblink, returns_long=False

2691 ).scalars()

2692 if _normalize:

2693 return [self.normalize_name(row) for row in result]

2694 else:

2695 return result.all()

2696

2697 @reflection.cache

2698 def get_view_names(self, connection, schema=None, dblink=None, **kw):

2699 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2700 if not schema:

2701 schema = self.default_schema_name

2702

2703 query = select(dictionary.all_views.c.view_name).where(

2704 dictionary.all_views.c.owner

2705 == self.denormalize_schema_name(schema)

2706 )

2707 result = self._execute_reflection(

2708 connection, query, dblink, returns_long=False

2709 ).scalars()

2710 return [self.normalize_name(row) for row in result]

2711

2712 @reflection.cache

2713 def get_sequence_names(self, connection, schema=None, dblink=None, **kw):

2714 """Supported kw arguments are: ``dblink`` to reflect via a db link."""

2715 if not schema:

2716 schema = self.default_schema_name

2717 query = select(dictionary.all_sequences.c.sequence_name).where(

2718 dictionary.all_sequences.c.sequence_owner

2719 == self.denormalize_schema_name(schema)

2720 )

2721

2722 result = self._execute_reflection(

2723 connection, query, dblink, returns_long=False

2724 ).scalars()

2725 return [self.normalize_name(row) for row in result]

2726

2727 def _value_or_raise(self, data, table, schema):

2728 table = self.normalize_name(str(table))

2729 try:

2730 return dict(data)[(schema, table)]

2731 except KeyError:

2732 raise exc.NoSuchTableError(

2733 f"{schema}.{table}" if schema else table

2734 ) from None

2735

2736 def _prepare_filter_names(self, filter_names):

2737 if filter_names:

2738 fn = [self.denormalize_name(name) for name in filter_names]

2739 return True, {"filter_names": fn}

2740 else:

2741 return False, {}

2742

2743 @lru_cache()

2744 def _table_options_query(

2745 self, owner, scope, kind, has_filter_names, has_mat_views

2746 ):

2747 query = select(

2748 dictionary.all_tables.c.table_name,

2749 (

2750 dictionary.all_tables.c.compression

2751 if self._supports_table_compression

2752 else sql.null().label("compression")

2753 ),

2754 (

2755 dictionary.all_tables.c.compress_for

2756 if self._supports_table_compress_for

2757 else sql.null().label("compress_for")

2758 ),

2759 dictionary.all_tables.c.tablespace_name,

2760 ).where(dictionary.all_tables.c.owner == owner)

2761 if has_filter_names:

2762 query = query.where(

2763 dictionary.all_tables.c.table_name.in_(

2764 bindparam("filter_names")

2765 )

2766 )

2767 if scope is ObjectScope.DEFAULT:

2768 query = query.where(dictionary.all_tables.c.duration.is_(null()))

2769 elif scope is ObjectScope.TEMPORARY:

2770 query = query.where(

2771 dictionary.all_tables.c.duration.is_not(null())

2772 )

2773

2774 if (

2775 has_mat_views

2776 and ObjectKind.TABLE in kind

2777 and ObjectKind.MATERIALIZED_VIEW not in kind

2778 ):

2779 # can't use EXCEPT ALL / MINUS here because we don't have an

2780 # excludable row vs. the query above

2781 # outerjoin + where null works better on oracle 21 but 11 does

2782 # not like it at all. this is the next best thing

2783

2784 query = query.where(

2785 dictionary.all_tables.c.table_name.not_in(

2786 bindparam("mat_views")

2787 )

2788 )

2789 elif (

2790 ObjectKind.TABLE not in kind

2791 and ObjectKind.MATERIALIZED_VIEW in kind

2792 ):

2793 query = query.where(

2794 dictionary.all_tables.c.table_name.in_(bindparam("mat_views"))

2795 )

2796 return query

2797

2798 @_handle_synonyms_decorator

2799 def get_multi_table_options(

2800 self,

2801 connection,

2802 *,

2803 schema,

2804 filter_names,

2805 scope,

2806 kind,

2807 dblink=None,

2808 **kw,

2809 ):

2810 """Supported kw arguments are: ``dblink`` to reflect via a db link;

2811 ``oracle_resolve_synonyms`` to resolve names to synonyms

2812 """

2813 owner = self.denormalize_schema_name(

2814 schema or self.default_schema_name

2815 )

2816

2817 has_filter_names, params = self._prepare_filter_names(filter_names)

2818 has_mat_views = False

2819

2820 if (

2821 ObjectKind.TABLE in kind

2822 and ObjectKind.MATERIALIZED_VIEW not in kind

2823 ):

2824 # see note in _table_options_query

2825 mat_views = self.get_materialized_view_names(

2826 connection, schema, dblink, _normalize=False, **kw

2827 )

2828 if mat_views:

2829 params["mat_views"] = mat_views

2830 has_mat_views = True

2831 elif (

2832 ObjectKind.TABLE not in kind

2833 and ObjectKind.MATERIALIZED_VIEW in kind

2834 ):

2835 mat_views = self.get_materialized_view_names(

2836 connection, schema, dblink, _normalize=False, **kw

2837 )

2838 params["mat_views"] = mat_views

2839

2840 options = {}

2841 default = ReflectionDefaults.table_options

2842

2843 if ObjectKind.TABLE in kind or ObjectKind.MATERIALIZED_VIEW in kind:

2844 query = self._table_options_query(

2845 owner, scope, kind, has_filter_names, has_mat_views

2846 )

2847 result = self._execute_reflection(

2848 connection, query, dblink, returns_long=False, params=params

2849 )

2850

2851 for table, compression, compress_for, tablespace in result:

2852 data = default()

2853 if compression == "ENABLED":

2854 data["oracle_compress"] = compress_for

2855 if tablespace:

2856 data["oracle_tablespace"] = tablespace

2857 options[(schema, self.normalize_name(table))] = data

2858 if ObjectKind.VIEW in kind and ObjectScope.DEFAULT in scope:

2859 # add the views (no temporary views)

2860 for view in self.get_view_names(connection, schema, dblink, **kw):

2861 if not filter_names or view in filter_names:

2862 options[(schema, view)] = default()

2863

2864 return options.items()

2865

2866 def _run_batches(

2867 self, connection, query, dblink, returns_long, mappings, all_objects

2868 ):

2869 each_batch = 500

2870 batches = list(all_objects)

2871 while batches:

2872 batch = batches[0:each_batch]

2873 batches[0:each_batch] = []

2874

2875 result = self._execute_reflection(

2876 connection,

2877 query,

2878 dblink,

2879 returns_long=returns_long,

2880 params={"all_objects": batch},

2881 )

2882 if mappings:

2883 yield from result.mappings()

2884 else:

2885 yield from result

2886

2887 @lru_cache()

2888 def _column_query(self, owner):

2889 all_cols = dictionary.all_tab_cols

2890 all_comments = dictionary.all_col_comments

2891 all_ids = dictionary.all_tab_identity_cols

2892

2893 if self.server_version_info >= (12,):

2894 add_cols = (

2895 all_cols.c.default_on_null,

2896 sql.case(

2897 (all_ids.c.table_name.is_(None), sql.null()),

2898 else_=all_ids.c.generation_type

2899 + ","

2900 + all_ids.c.identity_options,

2901 ).label("identity_options"),

2902 )

2903 join_identity_cols = True

2904 else:

2905 add_cols = (

2906 sql.null().label("default_on_null"),

2907 sql.null().label("identity_options"),

2908 )

2909 join_identity_cols = False

2910

2911 # NOTE: on oracle cannot create tables/views without columns and

2912 # a table cannot have all column hidden:

2913 # ORA-54039: table must have at least one column that is not invisible

2914 # all_tab_cols returns data for tables/views/mat-views.

2915 # all_tab_cols does not return recycled tables

2916

2917 query = (

2918 select(

2919 all_cols.c.table_name,

2920 all_cols.c.column_name,

2921 all_cols.c.data_type,

2922 all_cols.c.char_length,

2923 all_cols.c.data_length,

2924 all_cols.c.data_precision,

2925 all_cols.c.data_scale,

2926 all_cols.c.nullable,

2927 all_cols.c.data_default,

2928 all_comments.c.comments,

2929 all_cols.c.virtual_column,

2930 *add_cols,

2931 ).select_from(all_cols)

2932 # NOTE: all_col_comments has a row for each column even if no

2933 # comment is present, so a join could be performed, but there

2934 # seems to be no difference compared to an outer join

2935 .outerjoin(

2936 all_comments,

2937 and_(

2938 all_cols.c.table_name == all_comments.c.table_name,

2939 all_cols.c.column_name == all_comments.c.column_name,

2940 all_cols.c.owner == all_comments.c.owner,

2941 ),

2942 )

2943 )

2944 if join_identity_cols:

2945 query = query.outerjoin(

2946 all_ids,

2947 and_(

2948 all_cols.c.table_name == all_ids.c.table_name,

2949 all_cols.c.column_name == all_ids.c.column_name,

2950 all_cols.c.owner == all_ids.c.owner,

2951 ),

2952 )

2953

2954 query = query.where(

2955 all_cols.c.table_name.in_(bindparam("all_objects")),

2956 all_cols.c.hidden_column == "NO",

2957 all_cols.c.owner == owner,

2958 ).order_by(all_cols.c.table_name, all_cols.c.column_id)

2959 return query

2960

2961 @_handle_synonyms_decorator

2962 def get_multi_columns(

2963 self,

2964 connection,

2965 *,

2966 schema,

2967 filter_names,

2968 scope,

2969 kind,

2970 dblink=None,

2971 **kw,

2972 ):

2973 """Supported kw arguments are: ``dblink`` to reflect via a db link;

2974 ``oracle_resolve_synonyms`` to resolve names to synonyms

2975 """

2976 owner = self.denormalize_schema_name(

2977 schema or self.default_schema_name

2978 )

2979 query = self._column_query(owner)

2980

2981 if (

2982 filter_names

2983 and kind is ObjectKind.ANY

2984 and scope is ObjectScope.ANY

2985 ):

2986 all_objects = [self.denormalize_name(n) for n in filter_names]

2987 else:

2988 all_objects = self._get_all_objects(

2989 connection, schema, scope, kind, filter_names, dblink, **kw

2990 )

2991

2992 columns = defaultdict(list)

2993

2994 # all_tab_cols.data_default is LONG

2995 result = self._run_batches(

2996 connection,

2997 query,

2998 dblink,

2999 returns_long=True,

3000 mappings=True,

3001 all_objects=all_objects,

3002 )

3003

3004 def maybe_int(value):

3005 if isinstance(value, float) and value.is_integer():

3006 return int(value)

3007 else:

3008 return value

3009

3010 remove_size = re.compile(r"$\d+$")

3011

3012 for row_dict in result:

3013 table_name = self.normalize_name(row_dict["table_name"])

3014 orig_colname = row_dict["column_name"]

3015 colname = self.normalize_name(orig_colname)

3016 coltype = row_dict["data_type"]

3017 precision = maybe_int(row_dict["data_precision"])

3018

3019 if coltype == "NUMBER":

3020 scale = maybe_int(row_dict["data_scale"])

3021 if precision is None and scale == 0:

3022 coltype = INTEGER()

3023 else:

3024 coltype = NUMBER(precision, scale)

3025 elif coltype == "FLOAT":

3026 # https://docs.oracle.com/cd/B14117_01/server.101/b10758/sqlqr06.htm

3027 if precision == 126:

3028 # The DOUBLE PRECISION datatype is a floating-point

3029 # number with binary precision 126.

3030 coltype = DOUBLE_PRECISION()

3031 elif precision == 63:

3032 # The REAL datatype is a floating-point number with a

3033 # binary precision of 63, or 18 decimal.

3034 coltype = REAL()

3035 else:

3036 # non standard precision

3037 coltype = FLOAT(binary_precision=precision)

3038

3039 elif coltype in ("VARCHAR2", "NVARCHAR2", "CHAR", "NCHAR"):

3040 char_length = maybe_int(row_dict["char_length"])

3041 coltype = self.ischema_names.get(coltype)(char_length)

3042 elif coltype == "RAW":

3043 data_length = maybe_int(row_dict["data_length"])

3044 coltype = RAW(data_length)

3045 elif "WITH TIME ZONE" in coltype:

3046 coltype = TIMESTAMP(timezone=True)

3047 elif "WITH LOCAL TIME ZONE" in coltype:

3048 coltype = TIMESTAMP(local_timezone=True)

3049 else:

3050 coltype = re.sub(remove_size, "", coltype)

3051 try:

3052 coltype = self.ischema_names[coltype]

3053 except KeyError:

3054 util.warn(

3055 "Did not recognize type '%s' of column '%s'"

3056 % (coltype, colname)

3057 )

3058 coltype = sqltypes.NULLTYPE

3059

3060 default = row_dict["data_default"]

3061 if row_dict["virtual_column"] == "YES":

3062 computed = dict(sqltext=default)

3063 default = None

3064 else:

3065 computed = None

3066

3067 identity_options = row_dict["identity_options"]

3068 if identity_options is not None:

3069 identity = self._parse_identity_options(

3070 identity_options, row_dict["default_on_null"]

3071 )

3072 default = None

3073 else:

3074 identity = None

3075

3076 cdict = {

3077 "name": colname,

3078 "type": coltype,

3079 "nullable": row_dict["nullable"] == "Y",

3080 "default": default,

3081 "comment": row_dict["comments"],

3082 }

3083 if orig_colname.lower() == orig_colname:

3084 cdict["quote"] = True

3085 if computed is not None:

3086 cdict["computed"] = computed

3087 if identity is not None:

3088 cdict["identity"] = identity

3089

3090 columns[(schema, table_name)].append(cdict)

3091

3092 # NOTE: default not needed since all tables have columns

3093 # default = ReflectionDefaults.columns

3094 # return (

3095 # (key, value if value else default())

3096 # for key, value in columns.items()

3097 # )

3098 return columns.items()

3099

3100 def _parse_identity_options(self, identity_options, default_on_null):

3101 # identity_options is a string that starts with 'ALWAYS,' or

3102 # 'BY DEFAULT,' and continues with

3103 # START WITH: 1, INCREMENT BY: 1, MAX_VALUE: 123, MIN_VALUE: 1,

3104 # CYCLE_FLAG: N, CACHE_SIZE: 1, ORDER_FLAG: N, SCALE_FLAG: N,

3105 # EXTEND_FLAG: N, SESSION_FLAG: N, KEEP_VALUE: N

3106 parts = [p.strip() for p in identity_options.split(",")]

3107 identity = {

3108 "always": parts[0] == "ALWAYS",

3109 "oracle_on_null": default_on_null == "YES",

3110 }

3111

3112 for part in parts[1:]:

3113 option, value = part.split(":")

3114 value = value.strip()

3115

3116 if "START WITH" in option:

3117 identity["start"] = int(value)

3118 elif "INCREMENT BY" in option:

3119 identity["increment"] = int(value)

3120 elif "MAX_VALUE" in option:

3121 identity["maxvalue"] = int(value)

3122 elif "MIN_VALUE" in option:

3123 identity["minvalue"] = int(value)

3124 elif "CYCLE_FLAG" in option:

3125 identity["cycle"] = value == "Y"

3126 elif "CACHE_SIZE" in option:

3127 identity["cache"] = int(value)

3128 elif "ORDER_FLAG" in option:

3129 identity["oracle_order"] = value == "Y"

3130 return identity

3131

3132 @lru_cache()

3133 def _comment_query(self, owner, scope, kind, has_filter_names):

3134 # NOTE: all_tab_comments / all_mview_comments have a row for all

3135 # object even if they don't have comments

3136 queries = []

3137 if ObjectKind.TABLE in kind or ObjectKind.VIEW in kind:

3138 # all_tab_comments returns also plain views

3139 tbl_view = select(

3140 dictionary.all_tab_comments.c.table_name,

3141 dictionary.all_tab_comments.c.comments,

3142 ).where(

3143 dictionary.all_tab_comments.c.owner == owner,

3144 dictionary.all_tab_comments.c.table_name.not_like("BIN$%"),

3145 )

3146 if ObjectKind.VIEW not in kind:

3147 tbl_view = tbl_view.where(

3148 dictionary.all_tab_comments.c.table_type == "TABLE"

3149 )

3150 elif ObjectKind.TABLE not in kind:

3151 tbl_view = tbl_view.where(

3152 dictionary.all_tab_comments.c.table_type == "VIEW"

3153 )

3154 queries.append(tbl_view)

3155 if ObjectKind.MATERIALIZED_VIEW in kind:

3156 mat_view = select(

3157 dictionary.all_mview_comments.c.mview_name.label("table_name"),

3158 dictionary.all_mview_comments.c.comments,

3159 ).where(

3160 dictionary.all_mview_comments.c.owner == owner,

3161 dictionary.all_mview_comments.c.mview_name.not_like("BIN$%"),

3162 )

3163 queries.append(mat_view)

3164 if len(queries) == 1:

3165 query = queries[0]

3166 else:

3167 union = sql.union_all(*queries).subquery("tables_and_views")

3168 query = select(union.c.table_name, union.c.comments)

3169

3170 name_col = query.selected_columns.table_name

3171

3172 if scope in (ObjectScope.DEFAULT, ObjectScope.TEMPORARY):

3173 temp = "Y" if scope is ObjectScope.TEMPORARY else "N"

3174 # need distinct since materialized view are listed also

3175 # as tables in all_objects

3176 query = query.distinct().join(

3177 dictionary.all_objects,

3178 and_(

3179 dictionary.all_objects.c.owner == owner,

3180 dictionary.all_objects.c.object_name == name_col,

3181 dictionary.all_objects.c.temporary == temp,

3182 ),

3183 )

3184 if has_filter_names:

3185 query = query.where(name_col.in_(bindparam("filter_names")))

3186 return query

3187

3188 @_handle_synonyms_decorator

3189 def get_multi_table_comment(

3190 self,

3191 connection,

3192 *,

3193 schema,

3194 filter_names,

3195 scope,

3196 kind,

3197 dblink=None,

3198 **kw,

3199 ):

3200 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3201 ``oracle_resolve_synonyms`` to resolve names to synonyms

3202 """

3203 owner = self.denormalize_schema_name(

3204 schema or self.default_schema_name

3205 )

3206 has_filter_names, params = self._prepare_filter_names(filter_names)

3207 query = self._comment_query(owner, scope, kind, has_filter_names)

3208

3209 result = self._execute_reflection(

3210 connection, query, dblink, returns_long=False, params=params

3211 )

3212 default = ReflectionDefaults.table_comment

3213 # materialized views by default seem to have a comment like

3214 # "snapshot table for snapshot owner.mat_view_name"

3215 ignore_mat_view = "snapshot table for snapshot "

3216 return (

3217 (

3218 (schema, self.normalize_name(table)),

3219 (

3220 {"text": comment}

3221 if comment is not None

3222 and not comment.startswith(ignore_mat_view)

3223 else default()

3224 ),

3225 )

3226 for table, comment in result

3227 )

3228

3229 @lru_cache()

3230 def _index_query(self, owner):

3231 return (

3232 select(

3233 dictionary.all_ind_columns.c.table_name,

3234 dictionary.all_ind_columns.c.index_name,

3235 dictionary.all_ind_columns.c.column_name,

3236 dictionary.all_indexes.c.index_type,

3237 dictionary.all_indexes.c.uniqueness,

3238 dictionary.all_indexes.c.compression,

3239 dictionary.all_indexes.c.prefix_length,

3240 dictionary.all_ind_columns.c.descend,

3241 dictionary.all_ind_expressions.c.column_expression,

3242 )

3243 .select_from(dictionary.all_ind_columns)

3244 .join(

3245 dictionary.all_indexes,

3246 sql.and_(

3247 dictionary.all_ind_columns.c.index_name

3248 == dictionary.all_indexes.c.index_name,

3249 dictionary.all_ind_columns.c.index_owner

3250 == dictionary.all_indexes.c.owner,

3251 ),

3252 )

3253 .outerjoin(

3254 # NOTE: this adds about 20% to the query time. Using a

3255 # case expression with a scalar subquery only when needed

3256 # with the assumption that most indexes are not expression

3257 # would be faster but oracle does not like that with

3258 # LONG datatype. It errors with:

3259 # ORA-00997: illegal use of LONG datatype

3260 dictionary.all_ind_expressions,

3261 sql.and_(

3262 dictionary.all_ind_expressions.c.index_name

3263 == dictionary.all_ind_columns.c.index_name,

3264 dictionary.all_ind_expressions.c.index_owner

3265 == dictionary.all_ind_columns.c.index_owner,

3266 dictionary.all_ind_expressions.c.column_position

3267 == dictionary.all_ind_columns.c.column_position,

3268 ),

3269 )

3270 .where(

3271 dictionary.all_indexes.c.table_owner == owner,

3272 dictionary.all_indexes.c.table_name.in_(

3273 bindparam("all_objects")

3274 ),

3275 )

3276 .order_by(

3277 dictionary.all_ind_columns.c.index_name,

3278 dictionary.all_ind_columns.c.column_position,

3279 )

3280 )

3281

3282 @reflection.flexi_cache(

3283 ("schema", InternalTraversal.dp_string),

3284 ("dblink", InternalTraversal.dp_string),

3285 ("all_objects", InternalTraversal.dp_string_list),

3286 )

3287 def _get_indexes_rows(self, connection, schema, dblink, all_objects, **kw):

3288 owner = self.denormalize_schema_name(

3289 schema or self.default_schema_name

3290 )

3291

3292 query = self._index_query(owner)

3293

3294 pks = {

3295 row_dict["constraint_name"]

3296 for row_dict in self._get_all_constraint_rows(

3297 connection, schema, dblink, all_objects, **kw

3298 )

3299 if row_dict["constraint_type"] == "P"

3300 }

3301

3302 # all_ind_expressions.column_expression is LONG

3303 result = self._run_batches(

3304 connection,

3305 query,

3306 dblink,

3307 returns_long=True,

3308 mappings=True,

3309 all_objects=all_objects,

3310 )

3311

3312 return [

3313 row_dict

3314 for row_dict in result

3315 if row_dict["index_name"] not in pks

3316 ]

3317

3318 @_handle_synonyms_decorator

3319 def get_multi_indexes(

3320 self,

3321 connection,

3322 *,

3323 schema,

3324 filter_names,

3325 scope,

3326 kind,

3327 dblink=None,

3328 **kw,

3329 ):

3330 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3331 ``oracle_resolve_synonyms`` to resolve names to synonyms

3332 """

3333 all_objects = self._get_all_objects(

3334 connection, schema, scope, kind, filter_names, dblink, **kw

3335 )

3336

3337 uniqueness = {"NONUNIQUE": False, "UNIQUE": True}

3338 enabled = {"DISABLED": False, "ENABLED": True}

3339 is_bitmap = {"BITMAP", "FUNCTION-BASED BITMAP"}

3340

3341 indexes = defaultdict(dict)

3342

3343 for row_dict in self._get_indexes_rows(

3344 connection, schema, dblink, all_objects, **kw

3345 ):

3346 index_name = self.normalize_name(row_dict["index_name"])

3347 table_name = self.normalize_name(row_dict["table_name"])

3348 table_indexes = indexes[(schema, table_name)]

3349

3350 if index_name not in table_indexes:

3351 table_indexes[index_name] = index_dict = {

3352 "name": index_name,

3353 "column_names": [],

3354 "dialect_options": {},

3355 "unique": uniqueness.get(row_dict["uniqueness"], False),

3356 }

3357 do = index_dict["dialect_options"]

3358 if row_dict["index_type"] in is_bitmap:

3359 do["oracle_bitmap"] = True

3360 if enabled.get(row_dict["compression"], False):

3361 do["oracle_compress"] = row_dict["prefix_length"]

3362

3363 else:

3364 index_dict = table_indexes[index_name]

3365

3366 expr = row_dict["column_expression"]

3367 if expr is not None:

3368 index_dict["column_names"].append(None)

3369 if "expressions" in index_dict:

3370 index_dict["expressions"].append(expr)

3371 else:

3372 index_dict["expressions"] = index_dict["column_names"][:-1]

3373 index_dict["expressions"].append(expr)

3374

3375 if row_dict["descend"].lower() != "asc":

3376 assert row_dict["descend"].lower() == "desc"

3377 cs = index_dict.setdefault("column_sorting", {})

3378 cs[expr] = ("desc",)

3379 else:

3380 assert row_dict["descend"].lower() == "asc"

3381 cn = self.normalize_name(row_dict["column_name"])

3382 index_dict["column_names"].append(cn)

3383 if "expressions" in index_dict:

3384 index_dict["expressions"].append(cn)

3385

3386 default = ReflectionDefaults.indexes

3387

3388 return (

3389 (key, list(indexes[key].values()) if key in indexes else default())

3390 for key in (

3391 (schema, self.normalize_name(obj_name))

3392 for obj_name in all_objects

3393 )

3394 )

3395

3396 @lru_cache()

3397 def _constraint_query(self, owner):

3398 local = dictionary.all_cons_columns.alias("local")

3399 remote = dictionary.all_cons_columns.alias("remote")

3400 return (

3401 select(

3402 dictionary.all_constraints.c.table_name,

3403 dictionary.all_constraints.c.constraint_type,

3404 dictionary.all_constraints.c.constraint_name,

3405 local.c.column_name.label("local_column"),

3406 remote.c.table_name.label("remote_table"),

3407 remote.c.column_name.label("remote_column"),

3408 remote.c.owner.label("remote_owner"),

3409 dictionary.all_constraints.c.search_condition,

3410 dictionary.all_constraints.c.delete_rule,

3411 )

3412 .select_from(dictionary.all_constraints)

3413 .join(

3414 local,

3415 and_(

3416 local.c.owner == dictionary.all_constraints.c.owner,

3417 dictionary.all_constraints.c.constraint_name

3418 == local.c.constraint_name,

3419 ),

3420 )

3421 .outerjoin(

3422 remote,

3423 and_(

3424 dictionary.all_constraints.c.r_owner == remote.c.owner,

3425 dictionary.all_constraints.c.r_constraint_name

3426 == remote.c.constraint_name,

3427 or_(

3428 remote.c.position.is_(sql.null()),

3429 local.c.position == remote.c.position,

3430 ),

3431 ),

3432 )

3433 .where(

3434 dictionary.all_constraints.c.owner == owner,

3435 dictionary.all_constraints.c.table_name.in_(

3436 bindparam("all_objects")

3437 ),

3438 dictionary.all_constraints.c.constraint_type.in_(

3439 ("R", "P", "U", "C")

3440 ),

3441 )

3442 .order_by(

3443 dictionary.all_constraints.c.constraint_name, local.c.position

3444 )

3445 )

3446

3447 @reflection.flexi_cache(

3448 ("schema", InternalTraversal.dp_string),

3449 ("dblink", InternalTraversal.dp_string),

3450 ("all_objects", InternalTraversal.dp_string_list),

3451 )

3452 def _get_all_constraint_rows(

3453 self, connection, schema, dblink, all_objects, **kw

3454 ):

3455 owner = self.denormalize_schema_name(

3456 schema or self.default_schema_name

3457 )

3458 query = self._constraint_query(owner)

3459

3460 # since the result is cached a list must be created

3461 values = list(

3462 self._run_batches(

3463 connection,

3464 query,

3465 dblink,

3466 returns_long=False,

3467 mappings=True,

3468 all_objects=all_objects,

3469 )

3470 )

3471 return values

3472

3473 @_handle_synonyms_decorator

3474 def get_multi_pk_constraint(

3475 self,

3476 connection,

3477 *,

3478 scope,

3479 schema,

3480 filter_names,

3481 kind,

3482 dblink=None,

3483 **kw,

3484 ):

3485 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3486 ``oracle_resolve_synonyms`` to resolve names to synonyms

3487 """

3488 all_objects = self._get_all_objects(

3489 connection, schema, scope, kind, filter_names, dblink, **kw

3490 )

3491

3492 primary_keys = defaultdict(dict)

3493 default = ReflectionDefaults.pk_constraint

3494

3495 for row_dict in self._get_all_constraint_rows(

3496 connection, schema, dblink, all_objects, **kw

3497 ):

3498 if row_dict["constraint_type"] != "P":

3499 continue

3500 table_name = self.normalize_name(row_dict["table_name"])

3501 constraint_name = self.normalize_name(row_dict["constraint_name"])

3502 column_name = self.normalize_name(row_dict["local_column"])

3503

3504 table_pk = primary_keys[(schema, table_name)]

3505 if not table_pk:

3506 table_pk["name"] = constraint_name

3507 table_pk["constrained_columns"] = [column_name]

3508 else:

3509 table_pk["constrained_columns"].append(column_name)

3510

3511 return (

3512 (key, primary_keys[key] if key in primary_keys else default())

3513 for key in (

3514 (schema, self.normalize_name(obj_name))

3515 for obj_name in all_objects

3516 )

3517 )

3518

3519 @_handle_synonyms_decorator

3520 def get_multi_foreign_keys(

3521 self,

3522 connection,

3523 *,

3524 scope,

3525 schema,

3526 filter_names,

3527 kind,

3528 dblink=None,

3529 **kw,

3530 ):

3531 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3532 ``oracle_resolve_synonyms`` to resolve names to synonyms

3533 """

3534 all_objects = self._get_all_objects(

3535 connection, schema, scope, kind, filter_names, dblink, **kw

3536 )

3537

3538 resolve_synonyms = kw.get("oracle_resolve_synonyms", False)

3539

3540 owner = self.denormalize_schema_name(

3541 schema or self.default_schema_name

3542 )

3543

3544 all_remote_owners = set()

3545 fkeys = defaultdict(dict)

3546

3547 for row_dict in self._get_all_constraint_rows(

3548 connection, schema, dblink, all_objects, **kw

3549 ):

3550 if row_dict["constraint_type"] != "R":

3551 continue

3552

3553 table_name = self.normalize_name(row_dict["table_name"])

3554 constraint_name = self.normalize_name(row_dict["constraint_name"])

3555 table_fkey = fkeys[(schema, table_name)]

3556

3557 assert constraint_name is not None

3558

3559 local_column = self.normalize_name(row_dict["local_column"])

3560 remote_table = self.normalize_name(row_dict["remote_table"])

3561 remote_column = self.normalize_name(row_dict["remote_column"])

3562 remote_owner_orig = row_dict["remote_owner"]

3563 remote_owner = self.normalize_name(remote_owner_orig)

3564 if remote_owner_orig is not None:

3565 all_remote_owners.add(remote_owner_orig)

3566

3567 if remote_table is None:

3568 # ticket 363

3569 if dblink and not dblink.startswith("@"):

3570 dblink = f"@{dblink}"

3571 util.warn(

3572 "Got 'None' querying 'table_name' from "

3573 f"all_cons_columns{dblink or ''} - does the user have "

3574 "proper rights to the table?"

3575 )

3576 continue

3577

3578 if constraint_name not in table_fkey:

3579 table_fkey[constraint_name] = fkey = {

3580 "name": constraint_name,

3581 "constrained_columns": [],

3582 "referred_schema": None,

3583 "referred_table": remote_table,

3584 "referred_columns": [],

3585 "options": {},

3586 }

3587

3588 if resolve_synonyms:

3589 # will be removed below

3590 fkey["_ref_schema"] = remote_owner

3591

3592 if schema is not None or remote_owner_orig != owner:

3593 fkey["referred_schema"] = remote_owner

3594

3595 delete_rule = row_dict["delete_rule"]

3596 if delete_rule != "NO ACTION":

3597 fkey["options"]["ondelete"] = delete_rule

3598

3599 else:

3600 fkey = table_fkey[constraint_name]

3601

3602 fkey["constrained_columns"].append(local_column)

3603 fkey["referred_columns"].append(remote_column)

3604

3605 if resolve_synonyms and all_remote_owners:

3606 query = select(

3607 dictionary.all_synonyms.c.owner,

3608 dictionary.all_synonyms.c.table_name,

3609 dictionary.all_synonyms.c.table_owner,

3610 dictionary.all_synonyms.c.synonym_name,

3611 ).where(dictionary.all_synonyms.c.owner.in_(all_remote_owners))

3612

3613 result = self._execute_reflection(

3614 connection, query, dblink, returns_long=False

3615 ).mappings()

3616

3617 remote_owners_lut = {}

3618 for row in result:

3619 synonym_owner = self.normalize_name(row["owner"])

3620 table_name = self.normalize_name(row["table_name"])

3621

3622 remote_owners_lut[(synonym_owner, table_name)] = (

3623 row["table_owner"],

3624 row["synonym_name"],

3625 )

3626

3627 empty = (None, None)

3628 for table_fkeys in fkeys.values():

3629 for table_fkey in table_fkeys.values():

3630 key = (

3631 table_fkey.pop("_ref_schema"),

3632 table_fkey["referred_table"],

3633 )

3634 remote_owner, syn_name = remote_owners_lut.get(key, empty)

3635 if syn_name:

3636 sn = self.normalize_name(syn_name)

3637 table_fkey["referred_table"] = sn

3638 if schema is not None or remote_owner != owner:

3639 ro = self.normalize_name(remote_owner)

3640 table_fkey["referred_schema"] = ro

3641 else:

3642 table_fkey["referred_schema"] = None

3643 default = ReflectionDefaults.foreign_keys

3644

3645 return (

3646 (key, list(fkeys[key].values()) if key in fkeys else default())

3647 for key in (

3648 (schema, self.normalize_name(obj_name))

3649 for obj_name in all_objects

3650 )

3651 )

3652

3653 @_handle_synonyms_decorator

3654 def get_multi_unique_constraints(

3655 self,

3656 connection,

3657 *,

3658 scope,

3659 schema,

3660 filter_names,

3661 kind,

3662 dblink=None,

3663 **kw,

3664 ):

3665 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3666 ``oracle_resolve_synonyms`` to resolve names to synonyms

3667 """

3668 all_objects = self._get_all_objects(

3669 connection, schema, scope, kind, filter_names, dblink, **kw

3670 )

3671

3672 unique_cons = defaultdict(dict)

3673

3674 index_names = {

3675 row_dict["index_name"]

3676 for row_dict in self._get_indexes_rows(

3677 connection, schema, dblink, all_objects, **kw

3678 )

3679 }

3680

3681 for row_dict in self._get_all_constraint_rows(

3682 connection, schema, dblink, all_objects, **kw

3683 ):

3684 if row_dict["constraint_type"] != "U":

3685 continue

3686 table_name = self.normalize_name(row_dict["table_name"])

3687 constraint_name_orig = row_dict["constraint_name"]

3688 constraint_name = self.normalize_name(constraint_name_orig)

3689 column_name = self.normalize_name(row_dict["local_column"])

3690 table_uc = unique_cons[(schema, table_name)]

3691

3692 assert constraint_name is not None

3693

3694 if constraint_name not in table_uc:

3695 table_uc[constraint_name] = uc = {

3696 "name": constraint_name,

3697 "column_names": [],

3698 "duplicates_index": (

3699 constraint_name

3700 if constraint_name_orig in index_names

3701 else None

3702 ),

3703 }

3704 else:

3705 uc = table_uc[constraint_name]

3706

3707 uc["column_names"].append(column_name)

3708

3709 default = ReflectionDefaults.unique_constraints

3710

3711 return (

3712 (

3713 key,

3714 (

3715 list(unique_cons[key].values())

3716 if key in unique_cons

3717 else default()

3718 ),

3719 )

3720 for key in (

3721 (schema, self.normalize_name(obj_name))

3722 for obj_name in all_objects

3723 )

3724 )

3725

3726 @reflection.cache

3727 def get_view_definition(

3728 self,

3729 connection,

3730 view_name,

3731 schema=None,

3732 dblink=None,

3733 **kw,

3734 ):

3735 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3736 ``oracle_resolve_synonyms`` to resolve names to synonyms

3737 """

3738 if kw.get("oracle_resolve_synonyms", False):

3739 synonyms = self._get_synonyms(

3740 connection, schema, filter_names=[view_name], dblink=dblink

3741 )

3742 if synonyms:

3743 assert len(synonyms) == 1

3744 row_dict = synonyms[0]

3745 dblink = self.normalize_name(row_dict["db_link"])

3746 schema = row_dict["table_owner"]

3747 view_name = row_dict["table_name"]

3748

3749 name = self.denormalize_name(view_name)

3750 owner = self.denormalize_schema_name(

3751 schema or self.default_schema_name

3752 )

3753 query = (

3754 select(dictionary.all_views.c.text)

3755 .where(

3756 dictionary.all_views.c.view_name == name,

3757 dictionary.all_views.c.owner == owner,

3758 )

3759 .union_all(

3760 select(dictionary.all_mviews.c.query).where(

3761 dictionary.all_mviews.c.mview_name == name,

3762 dictionary.all_mviews.c.owner == owner,

3763 )

3764 )

3765 )

3766

3767 rp = self._execute_reflection(

3768 connection, query, dblink, returns_long=False

3769 ).scalar()

3770 if rp is None:

3771 raise exc.NoSuchTableError(

3772 f"{schema}.{view_name}" if schema else view_name

3773 )

3774 else:

3775 return rp

3776

3777 @_handle_synonyms_decorator

3778 def get_multi_check_constraints(

3779 self,

3780 connection,

3781 *,

3782 schema,

3783 filter_names,

3784 dblink=None,

3785 scope,

3786 kind,

3787 include_all=False,

3788 **kw,

3789 ):

3790 """Supported kw arguments are: ``dblink`` to reflect via a db link;

3791 ``oracle_resolve_synonyms`` to resolve names to synonyms

3792 """

3793 all_objects = self._get_all_objects(

3794 connection, schema, scope, kind, filter_names, dblink, **kw

3795 )

3796

3797 not_null = re.compile(r"..+?. IS NOT NULL$")

3798

3799 check_constraints = defaultdict(list)

3800

3801 for row_dict in self._get_all_constraint_rows(

3802 connection, schema, dblink, all_objects, **kw

3803 ):

3804 if row_dict["constraint_type"] != "C":

3805 continue

3806 table_name = self.normalize_name(row_dict["table_name"])

3807 constraint_name = self.normalize_name(row_dict["constraint_name"])

3808 search_condition = row_dict["search_condition"]

3809

3810 table_checks = check_constraints[(schema, table_name)]

3811 if constraint_name is not None and (

3812 include_all or not not_null.match(search_condition)

3813 ):

3814 table_checks.append(

3815 {"name": constraint_name, "sqltext": search_condition}

3816 )

3817

3818 default = ReflectionDefaults.check_constraints

3819

3820 return (

3821 (

3822 key,

3823 (

3824 check_constraints[key]

3825 if key in check_constraints

3826 else default()

3827 ),

3828 )

3829 for key in (

3830 (schema, self.normalize_name(obj_name))

3831 for obj_name in all_objects

3832 )

3833 )

3834

3835 def _list_dblinks(self, connection, dblink=None):

3836 query = select(dictionary.all_db_links.c.db_link)

3837 links = self._execute_reflection(

3838 connection, query, dblink, returns_long=False

3839 ).scalars()

3840 return [self.normalize_name(link) for link in links]

3841

3842

3843class _OuterJoinColumn(sql.ClauseElement):

3844 __visit_name__ = "outer_join_column"

3845

3846 def __init__(self, column):

3847 self.column = column