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 += self.define_constraint_ondelete_cascade(constraint)

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 Exception:

2283 # Exception only; a bare except here also caught

2284 # asyncio.CancelledError and KeyboardInterrupt, returning the

2285 # default in place of propagating them

2286 return "READ COMMITTED"

2287

2288 def _execute_reflection(

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

2290 ):

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

2292 dblink = f"@{dblink}"

2293 execution_options = {

2294 # handle db links

2295 "_oracle_dblink": dblink or "",

2296 # override any schema translate map

2297 "schema_translate_map": None,

2298 }

2299

2300 if dblink and returns_long:

2301 # Oracle seems to error with

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

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

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

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

2306 def visit_bindparam(bindparam):

2307 bindparam.literal_execute = True

2308

2309 query = visitors.cloned_traverse(

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

2311 )

2312 return connection.execute(

2313 query, params, execution_options=execution_options

2314 )

2315

2316 @util.memoized_property

2317 def _has_multi_table_query(self):

2318 # materialized views are returned by all_tables

2319 tables = (

2320 select(

2321 dictionary.all_tables.c.table_name,

2322 dictionary.all_tables.c.owner,

2323 )

2324 .union_all(

2325 select(

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

2327 dictionary.all_views.c.owner,

2328 )

2329 )

2330 .subquery("tables_and_views")

2331 )

2332

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

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

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

2336 )

2337 return query

2338

2339 def has_multi_table(

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

2341 ):

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

2343

2344 owner = schema or self.default_schema_name

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

2346 params = {

2347 "table_names": db_tns,

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

2349 }

2350 cursor = self._execute_reflection(

2351 connection,

2352 self._has_multi_table_query,

2353 dblink,

2354 returns_long=False,

2355 params=params,

2356 )

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

2358 retval = {

2359 (schema, table): db_tn in existing

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

2361 }

2362 return retval.items()

2363

2364 @reflection.cache

2365 def has_sequence(

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

2367 ):

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

2369 if not schema:

2370 schema = self.default_schema_name

2371

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

2373 dictionary.all_sequences.c.sequence_name

2374 == self.denormalize_schema_name(sequence_name),

2375 dictionary.all_sequences.c.sequence_owner

2376 == self.denormalize_schema_name(schema),

2377 )

2378

2379 cursor = self._execute_reflection(

2380 connection, query, dblink, returns_long=False

2381 )

2382 return bool(cursor.scalar())

2383

2384 def _get_default_schema_name(self, connection):

2385 return self.normalize_name(

2386 connection.exec_driver_sql(

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

2388 ).scalar()

2389 )

2390

2391 def denormalize_schema_name(self, name):

2392 # look for quoted_name

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

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

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

2396 return "PUBLIC"

2397 return super().denormalize_name(name)

2398

2399 @reflection.flexi_cache(

2400 ("schema", InternalTraversal.dp_string),

2401 ("filter_names", InternalTraversal.dp_string_list),

2402 ("dblink", InternalTraversal.dp_string),

2403 )

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

2405 owner = self.denormalize_schema_name(

2406 schema or self.default_schema_name

2407 )

2408

2409 has_filter_names, params = self._prepare_filter_names(filter_names)

2410 query = select(

2411 dictionary.all_synonyms.c.synonym_name,

2412 dictionary.all_synonyms.c.table_name,

2413 dictionary.all_synonyms.c.table_owner,

2414 dictionary.all_synonyms.c.db_link,

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

2416 if has_filter_names:

2417 query = query.where(

2418 dictionary.all_synonyms.c.synonym_name.in_(

2419 params["filter_names"]

2420 )

2421 )

2422 result = self._execute_reflection(

2423 connection, query, dblink, returns_long=False

2424 ).mappings()

2425 return result.all()

2426

2427 @lru_cache()

2428 def _all_objects_query(

2429 self, owner, scope, kind, has_filter_names, has_mat_views

2430 ):

2431 query = (

2432 select(dictionary.all_objects.c.object_name)

2433 .select_from(dictionary.all_objects)

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

2435 )

2436

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

2438 # once as MATERIALIZE VIEW and once as TABLE

2439 if kind is ObjectKind.ANY:

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

2441 # need to add them to the in_.

2442 query = query.where(

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

2444 )

2445 else:

2446 object_type = []

2447 if ObjectKind.VIEW in kind:

2448 object_type.append("VIEW")

2449 if (

2450 ObjectKind.MATERIALIZED_VIEW in kind

2451 and ObjectKind.TABLE not in kind

2452 ):

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

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

2455 object_type.append("MATERIALIZED VIEW")

2456 if ObjectKind.TABLE in kind:

2457 object_type.append("TABLE")

2458 if has_mat_views and ObjectKind.MATERIALIZED_VIEW not in kind:

2459 # materialized view are listed also as tables,

2460 # so they need to be filtered out

2461 # EXCEPT ALL / MINUS profiles as faster than using

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

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

2464 # them only when needed

2465 query = query.where(

2466 dictionary.all_objects.c.object_name.not_in(

2467 bindparam("mat_views")

2468 )

2469 )

2470 query = query.where(

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

2472 )

2473

2474 # handles scope

2475 if scope is ObjectScope.DEFAULT:

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

2477 elif scope is ObjectScope.TEMPORARY:

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

2479

2480 if has_filter_names:

2481 query = query.where(

2482 dictionary.all_objects.c.object_name.in_(

2483 bindparam("filter_names")

2484 )

2485 )

2486 return query

2487

2488 @reflection.flexi_cache(

2489 ("schema", InternalTraversal.dp_string),

2490 ("scope", InternalTraversal.dp_plain_obj),

2491 ("kind", InternalTraversal.dp_plain_obj),

2492 ("filter_names", InternalTraversal.dp_string_list),

2493 ("dblink", InternalTraversal.dp_string),

2494 )

2495 def _get_all_objects(

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

2497 ):

2498 owner = self.denormalize_schema_name(

2499 schema or self.default_schema_name

2500 )

2501

2502 has_filter_names, params = self._prepare_filter_names(filter_names)

2503 has_mat_views = False

2504 if (

2505 ObjectKind.TABLE in kind

2506 and ObjectKind.MATERIALIZED_VIEW not in kind

2507 ):

2508 # see note in _all_objects_query

2509 mat_views = self.get_materialized_view_names(

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

2511 )

2512 if mat_views:

2513 params["mat_views"] = mat_views

2514 has_mat_views = True

2515

2516 query = self._all_objects_query(

2517 owner, scope, kind, has_filter_names, has_mat_views

2518 )

2519

2520 result = self._execute_reflection(

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

2522 ).scalars()

2523

2524 return result.all()

2525

2526 def _handle_synonyms_decorator(fn):

2527 @wraps(fn)

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

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

2530

2531 return wrapper

2532

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

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

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

2536

2537 original_kw = kwargs.copy()

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

2539 result = self._get_synonyms(

2540 connection,

2541 schema=schema,

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

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

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

2545 )

2546

2547 dblinks_owners = defaultdict(dict)

2548 for row in result:

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

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

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

2552

2553 if not dblinks_owners:

2554 # No synonym, do the plain thing

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

2556

2557 data = {}

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

2559 call_kw = {

2560 **original_kw,

2561 "schema": table_owner,

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

2563 "filter_names": mapping.keys(),

2564 }

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

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

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

2568 data[(schema, synonym_name)] = value

2569 return data.items()

2570

2571 @reflection.cache

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

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

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

2575 dictionary.all_users.c.username

2576 )

2577 result = self._execute_reflection(

2578 connection, query, dblink, returns_long=False

2579 ).scalars()

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

2581

2582 @reflection.cache

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

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

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

2586 if schema is None:

2587 schema = self.default_schema_name

2588

2589 den_schema = self.denormalize_schema_name(schema)

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

2591 tables = (

2592 select(

2593 dictionary.all_tables.c.table_name,

2594 dictionary.all_tables.c.owner,

2595 dictionary.all_tables.c.iot_name,

2596 dictionary.all_tables.c.duration,

2597 dictionary.all_tables.c.tablespace_name,

2598 )

2599 .union_all(

2600 select(

2601 dictionary.all_synonyms.c.synonym_name.label(

2602 "table_name"

2603 ),

2604 dictionary.all_synonyms.c.owner,

2605 dictionary.all_tables.c.iot_name,

2606 dictionary.all_tables.c.duration,

2607 dictionary.all_tables.c.tablespace_name,

2608 )

2609 .select_from(dictionary.all_tables)

2610 .join(

2611 dictionary.all_synonyms,

2612 and_(

2613 dictionary.all_tables.c.table_name

2614 == dictionary.all_synonyms.c.table_name,

2615 dictionary.all_tables.c.owner

2616 == func.coalesce(

2617 dictionary.all_synonyms.c.table_owner,

2618 dictionary.all_synonyms.c.owner,

2619 ),

2620 ),

2621 )

2622 )

2623 .subquery("available_tables")

2624 )

2625 else:

2626 tables = dictionary.all_tables

2627

2628 query = select(tables.c.table_name)

2629 if self.exclude_tablespaces:

2630 query = query.where(

2631 func.coalesce(

2632 tables.c.tablespace_name, "no tablespace"

2633 ).not_in(self.exclude_tablespaces)

2634 )

2635 query = query.where(

2636 tables.c.owner == den_schema,

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

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

2639 )

2640

2641 # remove materialized views

2642 mat_query = select(

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

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

2645

2646 query = (

2647 query.except_all(mat_query)

2648 if self._supports_except_all

2649 else query.except_(mat_query)

2650 )

2651

2652 result = self._execute_reflection(

2653 connection, query, dblink, returns_long=False

2654 ).scalars()

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

2656

2657 @reflection.cache

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

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

2660 schema = self.denormalize_schema_name(self.default_schema_name)

2661

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

2663 if self.exclude_tablespaces:

2664 query = query.where(

2665 func.coalesce(

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

2667 ).not_in(self.exclude_tablespaces)

2668 )

2669 query = query.where(

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

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

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

2673 )

2674

2675 result = self._execute_reflection(

2676 connection, query, dblink, returns_long=False

2677 ).scalars()

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

2679

2680 @reflection.cache

2681 def get_materialized_view_names(

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

2683 ):

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

2685 if not schema:

2686 schema = self.default_schema_name

2687

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

2689 dictionary.all_mviews.c.owner

2690 == self.denormalize_schema_name(schema)

2691 )

2692 result = self._execute_reflection(

2693 connection, query, dblink, returns_long=False

2694 ).scalars()

2695 if _normalize:

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

2697 else:

2698 return result.all()

2699

2700 @reflection.cache

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

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

2703 if not schema:

2704 schema = self.default_schema_name

2705

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

2707 dictionary.all_views.c.owner

2708 == self.denormalize_schema_name(schema)

2709 )

2710 result = self._execute_reflection(

2711 connection, query, dblink, returns_long=False

2712 ).scalars()

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

2714

2715 @reflection.cache

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

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

2718 if not schema:

2719 schema = self.default_schema_name

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

2721 dictionary.all_sequences.c.sequence_owner

2722 == self.denormalize_schema_name(schema)

2723 )

2724

2725 result = self._execute_reflection(

2726 connection, query, dblink, returns_long=False

2727 ).scalars()

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

2729

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

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

2732 try:

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

2734 except KeyError:

2735 raise exc.NoSuchTableError(

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

2737 ) from None

2738

2739 def _prepare_filter_names(self, filter_names):

2740 if filter_names:

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

2742 return True, {"filter_names": fn}

2743 else:

2744 return False, {}

2745

2746 @lru_cache()

2747 def _table_options_query(

2748 self, owner, scope, kind, has_filter_names, has_mat_views

2749 ):

2750 query = select(

2751 dictionary.all_tables.c.table_name,

2752 (

2753 dictionary.all_tables.c.compression

2754 if self._supports_table_compression

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

2756 ),

2757 (

2758 dictionary.all_tables.c.compress_for

2759 if self._supports_table_compress_for

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

2761 ),

2762 dictionary.all_tables.c.tablespace_name,

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

2764 if has_filter_names:

2765 query = query.where(

2766 dictionary.all_tables.c.table_name.in_(

2767 bindparam("filter_names")

2768 )

2769 )

2770 if scope is ObjectScope.DEFAULT:

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

2772 elif scope is ObjectScope.TEMPORARY:

2773 query = query.where(

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

2775 )

2776

2777 if (

2778 has_mat_views

2779 and ObjectKind.TABLE in kind

2780 and ObjectKind.MATERIALIZED_VIEW not in kind

2781 ):

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

2783 # excludable row vs. the query above

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

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

2786

2787 query = query.where(

2788 dictionary.all_tables.c.table_name.not_in(

2789 bindparam("mat_views")

2790 )

2791 )

2792 elif (

2793 ObjectKind.TABLE not in kind

2794 and ObjectKind.MATERIALIZED_VIEW in kind

2795 ):

2796 query = query.where(

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

2798 )

2799 return query

2800

2801 @_handle_synonyms_decorator

2802 def get_multi_table_options(

2803 self,

2804 connection,

2805 *,

2806 schema,

2807 filter_names,

2808 scope,

2809 kind,

2810 dblink=None,

2811 **kw,

2812 ):

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

2814 ``oracle_resolve_synonyms`` to resolve names to synonyms

2815 """

2816 owner = self.denormalize_schema_name(

2817 schema or self.default_schema_name

2818 )

2819

2820 has_filter_names, params = self._prepare_filter_names(filter_names)

2821 has_mat_views = False

2822

2823 if (

2824 ObjectKind.TABLE in kind

2825 and ObjectKind.MATERIALIZED_VIEW not in kind

2826 ):

2827 # see note in _table_options_query

2828 mat_views = self.get_materialized_view_names(

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

2830 )

2831 if mat_views:

2832 params["mat_views"] = mat_views

2833 has_mat_views = True

2834 elif (

2835 ObjectKind.TABLE not in kind

2836 and ObjectKind.MATERIALIZED_VIEW in kind

2837 ):

2838 mat_views = self.get_materialized_view_names(

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

2840 )

2841 params["mat_views"] = mat_views

2842

2843 options = {}

2844 default = ReflectionDefaults.table_options

2845

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

2847 query = self._table_options_query(

2848 owner, scope, kind, has_filter_names, has_mat_views

2849 )

2850 result = self._execute_reflection(

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

2852 )

2853

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

2855 data = default()

2856 if compression == "ENABLED":

2857 data["oracle_compress"] = compress_for

2858 if tablespace:

2859 data["oracle_tablespace"] = tablespace

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

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

2862 # add the views (no temporary views)

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

2864 if not filter_names or view in filter_names:

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

2866

2867 return options.items()

2868

2869 def _run_batches(

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

2871 ):

2872 each_batch = 500

2873 batches = list(all_objects)

2874 while batches:

2875 batch = batches[0:each_batch]

2876 batches[0:each_batch] = []

2877

2878 result = self._execute_reflection(

2879 connection,

2880 query,

2881 dblink,

2882 returns_long=returns_long,

2883 params={"all_objects": batch},

2884 )

2885 if mappings:

2886 yield from result.mappings()

2887 else:

2888 yield from result

2889

2890 @lru_cache()

2891 def _column_query(self, owner):

2892 all_cols = dictionary.all_tab_cols

2893 all_comments = dictionary.all_col_comments

2894 all_ids = dictionary.all_tab_identity_cols

2895

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

2897 add_cols = (

2898 all_cols.c.default_on_null,

2899 sql.case(

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

2901 else_=all_ids.c.generation_type

2902 + ","

2903 + all_ids.c.identity_options,

2904 ).label("identity_options"),

2905 )

2906 join_identity_cols = True

2907 else:

2908 add_cols = (

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

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

2911 )

2912 join_identity_cols = False

2913

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

2915 # a table cannot have all column hidden:

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

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

2918 # all_tab_cols does not return recycled tables

2919

2920 query = (

2921 select(

2922 all_cols.c.table_name,

2923 all_cols.c.column_name,

2924 all_cols.c.data_type,

2925 all_cols.c.char_length,

2926 all_cols.c.data_length,

2927 all_cols.c.data_precision,

2928 all_cols.c.data_scale,

2929 all_cols.c.nullable,

2930 all_cols.c.data_default,

2931 all_comments.c.comments,

2932 all_cols.c.virtual_column,

2933 *add_cols,

2934 ).select_from(all_cols)

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

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

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

2938 .outerjoin(

2939 all_comments,

2940 and_(

2941 all_cols.c.table_name == all_comments.c.table_name,

2942 all_cols.c.column_name == all_comments.c.column_name,

2943 all_cols.c.owner == all_comments.c.owner,

2944 ),

2945 )

2946 )

2947 if join_identity_cols:

2948 query = query.outerjoin(

2949 all_ids,

2950 and_(

2951 all_cols.c.table_name == all_ids.c.table_name,

2952 all_cols.c.column_name == all_ids.c.column_name,

2953 all_cols.c.owner == all_ids.c.owner,

2954 ),

2955 )

2956

2957 query = query.where(

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

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

2960 all_cols.c.owner == owner,

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

2962 return query

2963

2964 @_handle_synonyms_decorator

2965 def get_multi_columns(

2966 self,

2967 connection,

2968 *,

2969 schema,

2970 filter_names,

2971 scope,

2972 kind,

2973 dblink=None,

2974 **kw,

2975 ):

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

2977 ``oracle_resolve_synonyms`` to resolve names to synonyms

2978 """

2979 owner = self.denormalize_schema_name(

2980 schema or self.default_schema_name

2981 )

2982 query = self._column_query(owner)

2983

2984 if (

2985 filter_names

2986 and kind is ObjectKind.ANY

2987 and scope is ObjectScope.ANY

2988 ):

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

2990 else:

2991 all_objects = self._get_all_objects(

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

2993 )

2994

2995 columns = defaultdict(list)

2996

2997 # all_tab_cols.data_default is LONG

2998 result = self._run_batches(

2999 connection,

3000 query,

3001 dblink,

3002 returns_long=True,

3003 mappings=True,

3004 all_objects=all_objects,

3005 )

3006

3007 def maybe_int(value):

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

3009 return int(value)

3010 else:

3011 return value

3012

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

3014

3015 for row_dict in result:

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

3017 orig_colname = row_dict["column_name"]

3018 colname = self.normalize_name(orig_colname)

3019 coltype = row_dict["data_type"]

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

3021

3022 if coltype == "NUMBER":

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

3024 if precision is None and scale == 0:

3025 coltype = INTEGER()

3026 else:

3027 coltype = NUMBER(precision, scale)

3028 elif coltype == "FLOAT":

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

3030 if precision == 126:

3031 # The DOUBLE PRECISION datatype is a floating-point

3032 # number with binary precision 126.

3033 coltype = DOUBLE_PRECISION()

3034 elif precision == 63:

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

3036 # binary precision of 63, or 18 decimal.

3037 coltype = REAL()

3038 else:

3039 # non standard precision

3040 coltype = FLOAT(binary_precision=precision)

3041

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

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

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

3045 elif coltype == "RAW":

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

3047 coltype = RAW(data_length)

3048 elif "WITH TIME ZONE" in coltype:

3049 coltype = TIMESTAMP(timezone=True)

3050 elif "WITH LOCAL TIME ZONE" in coltype:

3051 coltype = TIMESTAMP(local_timezone=True)

3052 else:

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

3054 try:

3055 coltype = self.ischema_names[coltype]

3056 except KeyError:

3057 util.warn(

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

3059 % (coltype, colname)

3060 )

3061 coltype = sqltypes.NULLTYPE

3062

3063 default = row_dict["data_default"]

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

3065 computed = dict(sqltext=default)

3066 default = None

3067 else:

3068 computed = None

3069

3070 identity_options = row_dict["identity_options"]

3071 if identity_options is not None:

3072 identity = self._parse_identity_options(

3073 identity_options, row_dict["default_on_null"]

3074 )

3075 default = None

3076 else:

3077 identity = None

3078

3079 cdict = {

3080 "name": colname,

3081 "type": coltype,

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

3083 "default": default,

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

3085 }

3086 if orig_colname.lower() == orig_colname:

3087 cdict["quote"] = True

3088 if computed is not None:

3089 cdict["computed"] = computed

3090 if identity is not None:

3091 cdict["identity"] = identity

3092

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

3094

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

3096 # default = ReflectionDefaults.columns

3097 # return (

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

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

3100 # )

3101 return columns.items()

3102

3103 def _parse_identity_options(self, identity_options, default_on_null):

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

3105 # 'BY DEFAULT,' and continues with

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

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

3108 # EXTEND_FLAG: N, SESSION_FLAG: N, KEEP_VALUE: N

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

3110 identity = {

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

3112 "oracle_on_null": default_on_null == "YES",

3113 }

3114

3115 for part in parts[1:]:

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

3117 value = value.strip()

3118

3119 if "START WITH" in option:

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

3121 elif "INCREMENT BY" in option:

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

3123 elif "MAX_VALUE" in option:

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

3125 elif "MIN_VALUE" in option:

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

3127 elif "CYCLE_FLAG" in option:

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

3129 elif "CACHE_SIZE" in option:

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

3131 elif "ORDER_FLAG" in option:

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

3133 return identity

3134

3135 @lru_cache()

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

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

3138 # object even if they don't have comments

3139 queries = []

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

3141 # all_tab_comments returns also plain views

3142 tbl_view = select(

3143 dictionary.all_tab_comments.c.table_name,

3144 dictionary.all_tab_comments.c.comments,

3145 ).where(

3146 dictionary.all_tab_comments.c.owner == owner,

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

3148 )

3149 if ObjectKind.VIEW not in kind:

3150 tbl_view = tbl_view.where(

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

3152 )

3153 elif ObjectKind.TABLE not in kind:

3154 tbl_view = tbl_view.where(

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

3156 )

3157 queries.append(tbl_view)

3158 if ObjectKind.MATERIALIZED_VIEW in kind:

3159 mat_view = select(

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

3161 dictionary.all_mview_comments.c.comments,

3162 ).where(

3163 dictionary.all_mview_comments.c.owner == owner,

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

3165 )

3166 queries.append(mat_view)

3167 if len(queries) == 1:

3168 query = queries[0]

3169 else:

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

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

3172

3173 name_col = query.selected_columns.table_name

3174

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

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

3177 # need distinct since materialized view are listed also

3178 # as tables in all_objects

3179 query = query.distinct().join(

3180 dictionary.all_objects,

3181 and_(

3182 dictionary.all_objects.c.owner == owner,

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

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

3185 ),

3186 )

3187 if has_filter_names:

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

3189 return query

3190

3191 @_handle_synonyms_decorator

3192 def get_multi_table_comment(

3193 self,

3194 connection,

3195 *,

3196 schema,

3197 filter_names,

3198 scope,

3199 kind,

3200 dblink=None,

3201 **kw,

3202 ):

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

3204 ``oracle_resolve_synonyms`` to resolve names to synonyms

3205 """

3206 owner = self.denormalize_schema_name(

3207 schema or self.default_schema_name

3208 )

3209 has_filter_names, params = self._prepare_filter_names(filter_names)

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

3211

3212 result = self._execute_reflection(

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

3214 )

3215 default = ReflectionDefaults.table_comment

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

3217 # "snapshot table for snapshot owner.mat_view_name"

3218 ignore_mat_view = "snapshot table for snapshot "

3219 return (

3220 (

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

3222 (

3223 {"text": comment}

3224 if comment is not None

3225 and not comment.startswith(ignore_mat_view)

3226 else default()

3227 ),

3228 )

3229 for table, comment in result

3230 )

3231

3232 @lru_cache()

3233 def _index_query(self, owner):

3234 return (

3235 select(

3236 dictionary.all_ind_columns.c.table_name,

3237 dictionary.all_ind_columns.c.index_name,

3238 dictionary.all_ind_columns.c.column_name,

3239 dictionary.all_indexes.c.index_type,

3240 dictionary.all_indexes.c.uniqueness,

3241 dictionary.all_indexes.c.compression,

3242 dictionary.all_indexes.c.prefix_length,

3243 dictionary.all_ind_columns.c.descend,

3244 dictionary.all_ind_expressions.c.column_expression,

3245 )

3246 .select_from(dictionary.all_ind_columns)

3247 .join(

3248 dictionary.all_indexes,

3249 sql.and_(

3250 dictionary.all_ind_columns.c.index_name

3251 == dictionary.all_indexes.c.index_name,

3252 dictionary.all_ind_columns.c.index_owner

3253 == dictionary.all_indexes.c.owner,

3254 ),

3255 )

3256 .outerjoin(

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

3258 # case expression with a scalar subquery only when needed

3259 # with the assumption that most indexes are not expression

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

3261 # LONG datatype. It errors with:

3262 # ORA-00997: illegal use of LONG datatype

3263 dictionary.all_ind_expressions,

3264 sql.and_(

3265 dictionary.all_ind_expressions.c.index_name

3266 == dictionary.all_ind_columns.c.index_name,

3267 dictionary.all_ind_expressions.c.index_owner

3268 == dictionary.all_ind_columns.c.index_owner,

3269 dictionary.all_ind_expressions.c.column_position

3270 == dictionary.all_ind_columns.c.column_position,

3271 ),

3272 )

3273 .where(

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

3275 dictionary.all_indexes.c.table_name.in_(

3276 bindparam("all_objects")

3277 ),

3278 )

3279 .order_by(

3280 dictionary.all_ind_columns.c.index_name,

3281 dictionary.all_ind_columns.c.column_position,

3282 )

3283 )

3284

3285 @reflection.flexi_cache(

3286 ("schema", InternalTraversal.dp_string),

3287 ("dblink", InternalTraversal.dp_string),

3288 ("all_objects", InternalTraversal.dp_string_list),

3289 )

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

3291 owner = self.denormalize_schema_name(

3292 schema or self.default_schema_name

3293 )

3294

3295 query = self._index_query(owner)

3296

3297 pks = {

3298 row_dict["constraint_name"]

3299 for row_dict in self._get_all_constraint_rows(

3300 connection, schema, dblink, all_objects, **kw

3301 )

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

3303 }

3304

3305 # all_ind_expressions.column_expression is LONG

3306 result = self._run_batches(

3307 connection,

3308 query,

3309 dblink,

3310 returns_long=True,

3311 mappings=True,

3312 all_objects=all_objects,

3313 )

3314

3315 return [

3316 row_dict

3317 for row_dict in result

3318 if row_dict["index_name"] not in pks

3319 ]

3320

3321 @_handle_synonyms_decorator

3322 def get_multi_indexes(

3323 self,

3324 connection,

3325 *,

3326 schema,

3327 filter_names,

3328 scope,

3329 kind,

3330 dblink=None,

3331 **kw,

3332 ):

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

3334 ``oracle_resolve_synonyms`` to resolve names to synonyms

3335 """

3336 all_objects = self._get_all_objects(

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

3338 )

3339

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

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

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

3343

3344 indexes = defaultdict(dict)

3345

3346 for row_dict in self._get_indexes_rows(

3347 connection, schema, dblink, all_objects, **kw

3348 ):

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

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

3351 table_indexes = indexes[(schema, table_name)]

3352

3353 if index_name not in table_indexes:

3354 table_indexes[index_name] = index_dict = {

3355 "name": index_name,

3356 "column_names": [],

3357 "dialect_options": {},

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

3359 }

3360 do = index_dict["dialect_options"]

3361 if row_dict["index_type"] in is_bitmap:

3362 do["oracle_bitmap"] = True

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

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

3365

3366 else:

3367 index_dict = table_indexes[index_name]

3368

3369 expr = row_dict["column_expression"]

3370 if expr is not None:

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

3372 if "expressions" in index_dict:

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

3374 else:

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

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

3377

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

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

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

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

3382 else:

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

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

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

3386 if "expressions" in index_dict:

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

3388

3389 default = ReflectionDefaults.indexes

3390

3391 return (

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

3393 for key in (

3394 (schema, self.normalize_name(obj_name))

3395 for obj_name in all_objects

3396 )

3397 )

3398

3399 @lru_cache()

3400 def _constraint_query(self, owner):

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

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

3403 return (

3404 select(

3405 dictionary.all_constraints.c.table_name,

3406 dictionary.all_constraints.c.constraint_type,

3407 dictionary.all_constraints.c.constraint_name,

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

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

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

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

3412 dictionary.all_constraints.c.search_condition,

3413 dictionary.all_constraints.c.delete_rule,

3414 )

3415 .select_from(dictionary.all_constraints)

3416 .join(

3417 local,

3418 and_(

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

3420 dictionary.all_constraints.c.constraint_name

3421 == local.c.constraint_name,

3422 ),

3423 )

3424 .outerjoin(

3425 remote,

3426 and_(

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

3428 dictionary.all_constraints.c.r_constraint_name

3429 == remote.c.constraint_name,

3430 or_(

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

3432 local.c.position == remote.c.position,

3433 ),

3434 ),

3435 )

3436 .where(

3437 dictionary.all_constraints.c.owner == owner,

3438 dictionary.all_constraints.c.table_name.in_(

3439 bindparam("all_objects")

3440 ),

3441 dictionary.all_constraints.c.constraint_type.in_(

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

3443 ),

3444 )

3445 .order_by(

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

3447 )

3448 )

3449

3450 @reflection.flexi_cache(

3451 ("schema", InternalTraversal.dp_string),

3452 ("dblink", InternalTraversal.dp_string),

3453 ("all_objects", InternalTraversal.dp_string_list),

3454 )

3455 def _get_all_constraint_rows(

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

3457 ):

3458 owner = self.denormalize_schema_name(

3459 schema or self.default_schema_name

3460 )

3461 query = self._constraint_query(owner)

3462

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

3464 values = list(

3465 self._run_batches(

3466 connection,

3467 query,

3468 dblink,

3469 returns_long=False,

3470 mappings=True,

3471 all_objects=all_objects,

3472 )

3473 )

3474 return values

3475

3476 @_handle_synonyms_decorator

3477 def get_multi_pk_constraint(

3478 self,

3479 connection,

3480 *,

3481 scope,

3482 schema,

3483 filter_names,

3484 kind,

3485 dblink=None,

3486 **kw,

3487 ):

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

3489 ``oracle_resolve_synonyms`` to resolve names to synonyms

3490 """

3491 all_objects = self._get_all_objects(

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

3493 )

3494

3495 primary_keys = defaultdict(dict)

3496 default = ReflectionDefaults.pk_constraint

3497

3498 for row_dict in self._get_all_constraint_rows(

3499 connection, schema, dblink, all_objects, **kw

3500 ):

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

3502 continue

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

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

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

3506

3507 table_pk = primary_keys[(schema, table_name)]

3508 if not table_pk:

3509 table_pk["name"] = constraint_name

3510 table_pk["constrained_columns"] = [column_name]

3511 else:

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

3513

3514 return (

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

3516 for key in (

3517 (schema, self.normalize_name(obj_name))

3518 for obj_name in all_objects

3519 )

3520 )

3521

3522 @_handle_synonyms_decorator

3523 def get_multi_foreign_keys(

3524 self,

3525 connection,

3526 *,

3527 scope,

3528 schema,

3529 filter_names,

3530 kind,

3531 dblink=None,

3532 **kw,

3533 ):

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

3535 ``oracle_resolve_synonyms`` to resolve names to synonyms

3536 """

3537 all_objects = self._get_all_objects(

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

3539 )

3540

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

3542

3543 owner = self.denormalize_schema_name(

3544 schema or self.default_schema_name

3545 )

3546

3547 all_remote_owners = set()

3548 fkeys = defaultdict(dict)

3549

3550 for row_dict in self._get_all_constraint_rows(

3551 connection, schema, dblink, all_objects, **kw

3552 ):

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

3554 continue

3555

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

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

3558 table_fkey = fkeys[(schema, table_name)]

3559

3560 assert constraint_name is not None

3561

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

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

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

3565 remote_owner_orig = row_dict["remote_owner"]

3566 remote_owner = self.normalize_name(remote_owner_orig)

3567 if remote_owner_orig is not None:

3568 all_remote_owners.add(remote_owner_orig)

3569

3570 if remote_table is None:

3571 # ticket 363

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

3573 dblink = f"@{dblink}"

3574 util.warn(

3575 "Got 'None' querying 'table_name' from "

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

3577 "proper rights to the table?"

3578 )

3579 continue

3580

3581 if constraint_name not in table_fkey:

3582 table_fkey[constraint_name] = fkey = {

3583 "name": constraint_name,

3584 "constrained_columns": [],

3585 "referred_schema": None,

3586 "referred_table": remote_table,

3587 "referred_columns": [],

3588 "options": {},

3589 }

3590

3591 if resolve_synonyms:

3592 # will be removed below

3593 fkey["_ref_schema"] = remote_owner

3594

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

3596 fkey["referred_schema"] = remote_owner

3597

3598 delete_rule = row_dict["delete_rule"]

3599 if delete_rule != "NO ACTION":

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

3601

3602 else:

3603 fkey = table_fkey[constraint_name]

3604

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

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

3607

3608 if resolve_synonyms and all_remote_owners:

3609 query = select(

3610 dictionary.all_synonyms.c.owner,

3611 dictionary.all_synonyms.c.table_name,

3612 dictionary.all_synonyms.c.table_owner,

3613 dictionary.all_synonyms.c.synonym_name,

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

3615

3616 result = self._execute_reflection(

3617 connection, query, dblink, returns_long=False

3618 ).mappings()

3619

3620 remote_owners_lut = {}

3621 for row in result:

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

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

3624

3625 remote_owners_lut[(synonym_owner, table_name)] = (

3626 row["table_owner"],

3627 row["synonym_name"],

3628 )

3629

3630 empty = (None, None)

3631 for table_fkeys in fkeys.values():

3632 for table_fkey in table_fkeys.values():

3633 key = (

3634 table_fkey.pop("_ref_schema"),

3635 table_fkey["referred_table"],

3636 )

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

3638 if syn_name:

3639 sn = self.normalize_name(syn_name)

3640 table_fkey["referred_table"] = sn

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

3642 ro = self.normalize_name(remote_owner)

3643 table_fkey["referred_schema"] = ro

3644 else:

3645 table_fkey["referred_schema"] = None

3646 default = ReflectionDefaults.foreign_keys

3647

3648 return (

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

3650 for key in (

3651 (schema, self.normalize_name(obj_name))

3652 for obj_name in all_objects

3653 )

3654 )

3655

3656 @_handle_synonyms_decorator

3657 def get_multi_unique_constraints(

3658 self,

3659 connection,

3660 *,

3661 scope,

3662 schema,

3663 filter_names,

3664 kind,

3665 dblink=None,

3666 **kw,

3667 ):

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

3669 ``oracle_resolve_synonyms`` to resolve names to synonyms

3670 """

3671 all_objects = self._get_all_objects(

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

3673 )

3674

3675 unique_cons = defaultdict(dict)

3676

3677 index_names = {

3678 row_dict["index_name"]

3679 for row_dict in self._get_indexes_rows(

3680 connection, schema, dblink, all_objects, **kw

3681 )

3682 }

3683

3684 for row_dict in self._get_all_constraint_rows(

3685 connection, schema, dblink, all_objects, **kw

3686 ):

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

3688 continue

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

3690 constraint_name_orig = row_dict["constraint_name"]

3691 constraint_name = self.normalize_name(constraint_name_orig)

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

3693 table_uc = unique_cons[(schema, table_name)]

3694

3695 assert constraint_name is not None

3696

3697 if constraint_name not in table_uc:

3698 table_uc[constraint_name] = uc = {

3699 "name": constraint_name,

3700 "column_names": [],

3701 "duplicates_index": (

3702 constraint_name

3703 if constraint_name_orig in index_names

3704 else None

3705 ),

3706 }

3707 else:

3708 uc = table_uc[constraint_name]

3709

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

3711

3712 default = ReflectionDefaults.unique_constraints

3713

3714 return (

3715 (

3716 key,

3717 (

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

3719 if key in unique_cons

3720 else default()

3721 ),

3722 )

3723 for key in (

3724 (schema, self.normalize_name(obj_name))

3725 for obj_name in all_objects

3726 )

3727 )

3728

3729 @reflection.cache

3730 def get_view_definition(

3731 self,

3732 connection,

3733 view_name,

3734 schema=None,

3735 dblink=None,

3736 **kw,

3737 ):

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

3739 ``oracle_resolve_synonyms`` to resolve names to synonyms

3740 """

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

3742 synonyms = self._get_synonyms(

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

3744 )

3745 if synonyms:

3746 assert len(synonyms) == 1

3747 row_dict = synonyms[0]

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

3749 schema = row_dict["table_owner"]

3750 view_name = row_dict["table_name"]

3751

3752 name = self.denormalize_name(view_name)

3753 owner = self.denormalize_schema_name(

3754 schema or self.default_schema_name

3755 )

3756 query = (

3757 select(dictionary.all_views.c.text)

3758 .where(

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

3760 dictionary.all_views.c.owner == owner,

3761 )

3762 .union_all(

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

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

3765 dictionary.all_mviews.c.owner == owner,

3766 )

3767 )

3768 )

3769

3770 rp = self._execute_reflection(

3771 connection, query, dblink, returns_long=False

3772 ).scalar()

3773 if rp is None:

3774 raise exc.NoSuchTableError(

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

3776 )

3777 else:

3778 return rp

3779

3780 @_handle_synonyms_decorator

3781 def get_multi_check_constraints(

3782 self,

3783 connection,

3784 *,

3785 schema,

3786 filter_names,

3787 dblink=None,

3788 scope,

3789 kind,

3790 include_all=False,

3791 **kw,

3792 ):

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

3794 ``oracle_resolve_synonyms`` to resolve names to synonyms

3795 """

3796 all_objects = self._get_all_objects(

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

3798 )

3799

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

3801

3802 check_constraints = defaultdict(list)

3803

3804 for row_dict in self._get_all_constraint_rows(

3805 connection, schema, dblink, all_objects, **kw

3806 ):

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

3808 continue

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

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

3811 search_condition = row_dict["search_condition"]

3812

3813 table_checks = check_constraints[(schema, table_name)]

3814 if constraint_name is not None and (

3815 include_all or not not_null.match(search_condition)

3816 ):

3817 table_checks.append(

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

3819 )

3820

3821 default = ReflectionDefaults.check_constraints

3822

3823 return (

3824 (

3825 key,

3826 (

3827 check_constraints[key]

3828 if key in check_constraints

3829 else default()

3830 ),

3831 )

3832 for key in (

3833 (schema, self.normalize_name(obj_name))

3834 for obj_name in all_objects

3835 )

3836 )

3837

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

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

3840 links = self._execute_reflection(

3841 connection, query, dblink, returns_long=False

3842 ).scalars()

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

3844

3845

3846class _OuterJoinColumn(sql.ClauseElement):

3847 __visit_name__ = "outer_join_column"

3848

3849 def __init__(self, column):

3850 self.column = column