Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/dialects/oracle»

Browse Source
This commit is contained in:
Polina
2026-07-02 20:22:44 +00:00
parent b5e6d92654
commit c984ad3a5a
3 changed files with 976 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

297
venv/Lib/site-packages/sqlalchemy/dialects/oracle/provision.py Normal file
View File

@@ -0,0 +1,297 @@
# dialects/oracle/provision.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
import time
from ... import create_engine
from ... import exc
from ... import inspect
from ...engine import url as sa_url
from ...testing.provision import configure_follower
from ...testing.provision import create_db
from ...testing.provision import drop_all_schema_objects_post_tables
from ...testing.provision import drop_all_schema_objects_pre_tables
from ...testing.provision import drop_db
from ...testing.provision import follower_url_from_main
from ...testing.provision import generate_driver_url
from ...testing.provision import is_preferred_driver
from ...testing.provision import log
from ...testing.provision import post_configure_engine
from ...testing.provision import post_configure_testing_engine
from ...testing.provision import run_reap_dbs
from ...testing.provision import set_default_schema_on_connection
from ...testing.provision import stop_test_class_outside_fixtures
from ...testing.provision import temp_table_keyword_args
from ...testing.provision import update_db_opts
from ...testing.warnings import warn_test_suite
@generate_driver_url.for_db("oracle")
def _oracle_generate_driver_url(url, driver, query_str):
backend = url.get_backend_name()
new_url = url.set(
drivername="%s+%s" % (backend, driver),
)
# use oracledb's retry feature, which is essential for oracle 23c
# which otherwise frequently rejects connections under load
# for cx_oracle we have a connect event instead
if driver in ("oracledb", "oracledb_async"):
# oracledb is even nice enough to convert from string to int
# for these opts, apparently
new_url = new_url.update_query_pairs(
[("retry_count", "5"), ("retry_delay", "2")]
)
else:
# remove these params for cx_oracle if we received an
# already-modified URL
new_url = new_url.difference_update_query(
["retry_count", "retry_delay"]
)
try:
new_url.get_dialect()
except exc.NoSuchModuleError:
return None
else:
return new_url
@create_db.for_db("oracle")
def _oracle_create_db(cfg, eng, ident):
# NOTE: make sure you've run "ALTER DATABASE default tablespace users" or
# similar, so that the default tablespace is not "system"; reflection will
# fail otherwise
with eng.begin() as conn:
conn.exec_driver_sql("create user %s identified by xe" % ident)
conn.exec_driver_sql("create user %s_ts1 identified by xe" % ident)
conn.exec_driver_sql("create user %s_ts2 identified by xe" % ident)
conn.exec_driver_sql("grant dba to %s" % (ident,))
conn.exec_driver_sql("grant unlimited tablespace to %s" % ident)
conn.exec_driver_sql("grant unlimited tablespace to %s_ts1" % ident)
conn.exec_driver_sql("grant unlimited tablespace to %s_ts2" % ident)
# these are needed to create materialized views
conn.exec_driver_sql("grant create table to %s" % ident)
conn.exec_driver_sql("grant create table to %s_ts1" % ident)
conn.exec_driver_sql("grant create table to %s_ts2" % ident)
@configure_follower.for_db("oracle")
def _oracle_configure_follower(config, ident):
config.test_schema = "%s_ts1" % ident
config.test_schema_2 = "%s_ts2" % ident
def _ora_drop_ignore(conn, dbname):
try:
conn.exec_driver_sql("drop user %s cascade" % dbname)
log.info("Reaped db: %s", dbname)
return True
except exc.DatabaseError as err:
log.warning("couldn't drop db: %s", err)
return False
@drop_all_schema_objects_pre_tables.for_db("oracle")
def _ora_drop_all_schema_objects_pre_tables(cfg, eng):
_purge_recyclebin(eng)
_purge_recyclebin(eng, cfg.test_schema)
@drop_all_schema_objects_post_tables.for_db("oracle")
def _ora_drop_all_schema_objects_post_tables(cfg, eng):
with eng.begin() as conn:
for syn in conn.dialect._get_synonyms(conn, None, None, None):
conn.exec_driver_sql(f"drop synonym {syn['synonym_name']}")
for syn in conn.dialect._get_synonyms(
conn, cfg.test_schema, None, None
):
conn.exec_driver_sql(
f"drop synonym {cfg.test_schema}.{syn['synonym_name']}"
)
for tmp_table in inspect(conn).get_temp_table_names():
conn.exec_driver_sql(f"drop table {tmp_table}")
@drop_db.for_db("oracle")
def _oracle_drop_db(cfg, eng, ident):
with eng.begin() as conn:
# cx_Oracle seems to occasionally leak open connections when a large
# suite it run, even if we confirm we have zero references to
# connection objects.
# while there is a "kill session" command in Oracle Database,
# it unfortunately does not release the connection sufficiently.
_ora_drop_ignore(conn, ident)
_ora_drop_ignore(conn, "%s_ts1" % ident)
_ora_drop_ignore(conn, "%s_ts2" % ident)
@stop_test_class_outside_fixtures.for_db("oracle")
def _ora_stop_test_class_outside_fixtures(config, db, cls):
try:
_purge_recyclebin(db)
except exc.DatabaseError as err:
log.warning("purge recyclebin command failed: %s", err)
def _purge_recyclebin(eng, schema=None):
with eng.begin() as conn:
if schema is None:
# run magic command to get rid of identity sequences
# https://floo.bar/2019/11/29/drop-the-underlying-sequence-of-an-identity-column/ # noqa: E501
conn.exec_driver_sql("purge recyclebin")
else:
# per user: https://community.oracle.com/tech/developers/discussion/2255402/how-to-clear-dba-recyclebin-for-a-particular-user # noqa: E501
for owner, object_name, type_ in conn.exec_driver_sql(
"select owner, object_name,type from "
"dba_recyclebin where owner=:schema and type='TABLE'",
{"schema": conn.dialect.denormalize_name(schema)},
).all():
conn.exec_driver_sql(f'purge {type_} {owner}."{object_name}"')
@is_preferred_driver.for_db("oracle")
def _oracle_is_preferred_driver(cfg, engine):
"""establish oracledb as the preferred driver to use for tests, even
though cx_Oracle is still the "default" driver"""
return engine.dialect.driver == "oracledb" and not engine.dialect.is_async
def _connect_with_retry(dialect, conn_rec, cargs, cparams):
assert dialect.driver == "cx_oracle"
def _is_couldnt_connect(err):
return "DPY-6005" in str(err) or "ORA-12516" in str(err)
err_ = None
for _ in range(5):
try:
return dialect.loaded_dbapi.connect(*cargs, **cparams)
except (
dialect.loaded_dbapi.DatabaseError,
dialect.loaded_dbapi.OperationalError,
) as err:
err_ = err
if _is_couldnt_connect(err):
warn_test_suite("Oracle database reconnecting...")
time.sleep(2)
continue
else:
raise
if err_ is not None:
raise Exception("connect failed after five attempts") from err_
@post_configure_testing_engine.for_db("oracle")
def _oracle_post_configure_testing_engine(url, engine, options, scope):
from ... import event
if engine.dialect.driver == "cx_oracle":
event.listen(engine, "do_connect", _connect_with_retry)
@post_configure_engine.for_db("oracle")
def _oracle_post_configure_engine(url, engine, follower_ident):
from ... import event
@event.listens_for(engine, "checkin")
def checkin(dbapi_connection, connection_record):
# this was meant to work around this issue:
# https://github.com/oracle/python-cx_Oracle/issues/530
# invalidate oracle connections that had 2pc set up
# however things are too complex with some of the 2pc tests,
# so just block cx_oracle from being used in 2pc tests (use oracledb
# instead)
# if "cx_oracle_xid" in connection_record.info:
# connection_record.invalidate()
# clear statement cache on all connections that were used
# https://github.com/oracle/python-cx_Oracle/issues/519
# TODO: oracledb claims to have this feature built in somehow,
# see if that's in use and/or if it needs to be enabled
# (or if this doesn't even apply to the newer oracle's we're using)
try:
sc = dbapi_connection.stmtcachesize
except:
# connection closed
pass
else:
dbapi_connection.stmtcachesize = 0
dbapi_connection.stmtcachesize = sc
@run_reap_dbs.for_db("oracle")
def _reap_oracle_dbs(url, idents):
log.info("db reaper connecting to %r", url)
eng = create_engine(url)
with eng.begin() as conn:
log.info("identifiers in file: %s", ", ".join(idents))
to_reap = conn.exec_driver_sql(
"select u.username from all_users u where username "
"like 'TEST_%' and not exists (select username "
"from v$session where username=u.username)"
)
all_names = {username.lower() for (username,) in to_reap}
to_drop = set()
for name in all_names:
if name.endswith("_ts1") or name.endswith("_ts2"):
continue
elif name in idents:
to_drop.add(name)
if "%s_ts1" % name in all_names:
to_drop.add("%s_ts1" % name)
if "%s_ts2" % name in all_names:
to_drop.add("%s_ts2" % name)
dropped = total = 0
for total, username in enumerate(to_drop, 1):
if _ora_drop_ignore(conn, username):
dropped += 1
log.info(
"Dropped %d out of %d stale databases detected", dropped, total
)
@follower_url_from_main.for_db("oracle")
def _oracle_follower_url_from_main(url, ident):
url = sa_url.make_url(url)
return url.set(username=ident, password="xe")
@temp_table_keyword_args.for_db("oracle")
def _oracle_temp_table_keyword_args(cfg, eng):
return {
"prefixes": ["GLOBAL TEMPORARY"],
"oracle_on_commit": "PRESERVE ROWS",
}
@set_default_schema_on_connection.for_db("oracle")
def _oracle_set_default_schema_on_connection(
cfg, dbapi_connection, schema_name
):
cursor = dbapi_connection.cursor()
cursor.execute("ALTER SESSION SET CURRENT_SCHEMA=%s" % schema_name)
cursor.close()
@update_db_opts.for_db("oracle")
def _update_db_opts(db_url, db_opts, options):
"""Set database options (db_opts) for a test database that we created."""
if (
options.oracledb_thick_mode
and sa_url.make_url(db_url).get_driver_name() == "oracledb"
):
db_opts["thick_mode"] = True

316
venv/Lib/site-packages/sqlalchemy/dialects/oracle/types.py Normal file
View File

@@ -0,0 +1,316 @@
# dialects/oracle/types.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
from __future__ import annotations
import datetime as dt
from typing import Optional
from typing import Type
from typing import TYPE_CHECKING
from ... import exc
from ...sql import sqltypes
from ...types import NVARCHAR
from ...types import VARCHAR
if TYPE_CHECKING:
from ...engine.interfaces import Dialect
from ...sql.type_api import _LiteralProcessorType
class RAW(sqltypes._Binary):
__visit_name__ = "RAW"
OracleRaw = RAW
class NCLOB(sqltypes.Text):
__visit_name__ = "NCLOB"
class VARCHAR2(VARCHAR):
__visit_name__ = "VARCHAR2"
NVARCHAR2 = NVARCHAR
class NUMBER(sqltypes.Numeric, sqltypes.Integer):
__visit_name__ = "NUMBER"
def __init__(self, precision=None, scale=None, asdecimal=None):
if asdecimal is None:
asdecimal = bool(scale and scale > 0)
super().__init__(precision=precision, scale=scale, asdecimal=asdecimal)
def adapt(self, impltype):
ret = super().adapt(impltype)
# leave a hint for the DBAPI handler
ret._is_oracle_number = True
return ret
@property
def _type_affinity(self):
if bool(self.scale and self.scale > 0):
return sqltypes.Numeric
else:
return sqltypes.Integer
class FLOAT(sqltypes.FLOAT):
"""Oracle Database FLOAT.
This is the same as :class:`_sqltypes.FLOAT` except that
an Oracle Database -specific :paramref:`_oracle.FLOAT.binary_precision`
parameter is accepted, and
the :paramref:`_sqltypes.Float.precision` parameter is not accepted.
Oracle Database FLOAT types indicate precision in terms of "binary
precision", which defaults to 126. For a REAL type, the value is 63. This
parameter does not cleanly map to a specific number of decimal places but
is roughly equivalent to the desired number of decimal places divided by
0.3103.
.. versionadded:: 2.0
"""
__visit_name__ = "FLOAT"
def __init__(
self,
binary_precision=None,
asdecimal=False,
decimal_return_scale=None,
):
r"""
Construct a FLOAT
:param binary_precision: Oracle Database binary precision value to be
rendered in DDL. This may be approximated to the number of decimal
characters using the formula "decimal precision = 0.30103 * binary
precision". The default value used by Oracle Database for FLOAT /
DOUBLE PRECISION is 126.
:param asdecimal: See :paramref:`_sqltypes.Float.asdecimal`
:param decimal_return_scale: See
:paramref:`_sqltypes.Float.decimal_return_scale`
"""
super().__init__(
asdecimal=asdecimal, decimal_return_scale=decimal_return_scale
)
self.binary_precision = binary_precision
class BINARY_DOUBLE(sqltypes.Double):
"""Implement the Oracle ``BINARY_DOUBLE`` datatype.
This datatype differs from the Oracle ``DOUBLE`` datatype in that it
delivers a true 8-byte FP value. The datatype may be combined with a
generic :class:`.Double` datatype using :meth:`.TypeEngine.with_variant`.
.. seealso::
:ref:`oracle_float_support`
"""
__visit_name__ = "BINARY_DOUBLE"
class BINARY_FLOAT(sqltypes.Float):
"""Implement the Oracle ``BINARY_FLOAT`` datatype.
This datatype differs from the Oracle ``FLOAT`` datatype in that it
delivers a true 4-byte FP value. The datatype may be combined with a
generic :class:`.Float` datatype using :meth:`.TypeEngine.with_variant`.
.. seealso::
:ref:`oracle_float_support`
"""
__visit_name__ = "BINARY_FLOAT"
class BFILE(sqltypes.LargeBinary):
__visit_name__ = "BFILE"
class LONG(sqltypes.Text):
__visit_name__ = "LONG"
class _OracleDateLiteralRender:
def _literal_processor_datetime(self, dialect):
def process(value):
if getattr(value, "microsecond", None):
value = (
f"""TO_TIMESTAMP"""
f"""('{value.isoformat().replace("T", " ")}', """
"""'YYYY-MM-DD HH24:MI:SS.FF')"""
)
else:
value = (
f"""TO_DATE"""
f"""('{value.isoformat().replace("T", " ")}', """
"""'YYYY-MM-DD HH24:MI:SS')"""
)
return value
return process
def _literal_processor_date(self, dialect):
def process(value):
if getattr(value, "microsecond", None):
value = (
f"""TO_TIMESTAMP"""
f"""('{value.isoformat().split("T")[0]}', """
"""'YYYY-MM-DD')"""
)
else:
value = (
f"""TO_DATE"""
f"""('{value.isoformat().split("T")[0]}', """
"""'YYYY-MM-DD')"""
)
return value
return process
class DATE(_OracleDateLiteralRender, sqltypes.DateTime):
"""Provide the Oracle Database DATE type.
This type has no special Python behavior, except that it subclasses
:class:`_types.DateTime`; this is to suit the fact that the Oracle Database
``DATE`` type supports a time value.
"""
__visit_name__ = "DATE"
def literal_processor(self, dialect):
return self._literal_processor_datetime(dialect)
def _compare_type_affinity(self, other):
return other._type_affinity in (sqltypes.DateTime, sqltypes.Date)
class _OracleDate(_OracleDateLiteralRender, sqltypes.Date):
def literal_processor(self, dialect):
return self._literal_processor_date(dialect)
class INTERVAL(sqltypes.NativeForEmulated, sqltypes._AbstractInterval):
__visit_name__ = "INTERVAL"
def __init__(self, day_precision=None, second_precision=None):
"""Construct an INTERVAL.
Note that only DAY TO SECOND intervals are currently supported.
This is due to a lack of support for YEAR TO MONTH intervals
within available DBAPIs.
:param day_precision: the day precision value. this is the number of
digits to store for the day field. Defaults to "2"
:param second_precision: the second precision value. this is the
number of digits to store for the fractional seconds field.
Defaults to "6".
"""
self.day_precision = day_precision
self.second_precision = second_precision
@classmethod
def _adapt_from_generic_interval(cls, interval):
return INTERVAL(
day_precision=interval.day_precision,
second_precision=interval.second_precision,
)
@classmethod
def adapt_emulated_to_native(
cls, interval: sqltypes.Interval, **kw # type: ignore[override]
):
return INTERVAL(
day_precision=interval.day_precision,
second_precision=interval.second_precision,
)
@property
def _type_affinity(self):
return sqltypes.Interval
def as_generic(self, allow_nulltype=False):
return sqltypes.Interval(
native=True,
second_precision=self.second_precision,
day_precision=self.day_precision,
)
@property
def python_type(self) -> Type[dt.timedelta]:
return dt.timedelta
def literal_processor(
self, dialect: Dialect
) -> Optional[_LiteralProcessorType[dt.timedelta]]:
def process(value: dt.timedelta) -> str:
return f"NUMTODSINTERVAL({value.total_seconds()}, 'SECOND')"
return process
class TIMESTAMP(sqltypes.TIMESTAMP):
"""Oracle Database implementation of ``TIMESTAMP``, which supports
additional Oracle Database-specific modes
.. versionadded:: 2.0
"""
def __init__(self, timezone: bool = False, local_timezone: bool = False):
"""Construct a new :class:`_oracle.TIMESTAMP`.
:param timezone: boolean. Indicates that the TIMESTAMP type should
use Oracle Database's ``TIMESTAMP WITH TIME ZONE`` datatype.
:param local_timezone: boolean. Indicates that the TIMESTAMP type
should use Oracle Database's ``TIMESTAMP WITH LOCAL TIME ZONE``
datatype.
"""
if timezone and local_timezone:
raise exc.ArgumentError(
"timezone and local_timezone are mutually exclusive"
)
super().__init__(timezone=timezone)
self.local_timezone = local_timezone
class ROWID(sqltypes.TypeEngine):
"""Oracle Database ROWID type.
When used in a cast() or similar, generates ROWID.
"""
__visit_name__ = "ROWID"
class _OracleBoolean(sqltypes.Boolean):
def get_dbapi_type(self, dbapi):
return dbapi.NUMBER

363
venv/Lib/site-packages/sqlalchemy/dialects/oracle/vector.py Normal file
View File

@@ -0,0 +1,363 @@
# dialects/oracle/vector.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
from __future__ import annotations
import array
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from typing import Union
from ... import types
from ...types import Float
class VectorIndexType(Enum):
"""Enum representing different types of VECTOR index structures.
See :ref:`oracle_vector_datatype` for background.
.. versionadded:: 2.0.41
"""
HNSW = "HNSW"
"""
The HNSW (Hierarchical Navigable Small World) index type.
"""
IVF = "IVF"
"""
The IVF (Inverted File Index) index type
"""
class VectorDistanceType(Enum):
"""Enum representing different types of vector distance metrics.
See :ref:`oracle_vector_datatype` for background.
.. versionadded:: 2.0.41
"""
EUCLIDEAN = "EUCLIDEAN"
"""Euclidean distance (L2 norm).
Measures the straight-line distance between two vectors in space.
"""
DOT = "DOT"
"""Dot product similarity.
Measures the algebraic similarity between two vectors.
"""
COSINE = "COSINE"
"""Cosine similarity.
Measures the cosine of the angle between two vectors.
"""
MANHATTAN = "MANHATTAN"
"""Manhattan distance (L1 norm).
Calculates the sum of absolute differences across dimensions.
"""
class VectorStorageFormat(Enum):
"""Enum representing the data format used to store vector components.
See :ref:`oracle_vector_datatype` for background.
.. versionadded:: 2.0.41
"""
INT8 = "INT8"
"""
8-bit integer format.
"""
BINARY = "BINARY"
"""
Binary format.
"""
FLOAT32 = "FLOAT32"
"""
32-bit floating-point format.
"""
FLOAT64 = "FLOAT64"
"""
64-bit floating-point format.
"""
class VectorStorageType(Enum):
"""Enum representing the vector type,
See :ref:`oracle_vector_datatype` for background.
.. versionadded:: 2.0.43
"""
SPARSE = "SPARSE"
"""
A Sparse vector is a vector which has zero value for
most of its dimensions.
"""
DENSE = "DENSE"
"""
A Dense vector is a vector where most, if not all, elements
hold meaningful values.
"""
@dataclass
class VectorIndexConfig:
"""Define the configuration for Oracle VECTOR Index.
See :ref:`oracle_vector_datatype` for background.
.. versionadded:: 2.0.41
:param index_type: Enum value from :class:`.VectorIndexType`
Specifies the indexing method. For HNSW, this must be
:attr:`.VectorIndexType.HNSW`.
:param distance: Enum value from :class:`.VectorDistanceType`
specifies the metric for calculating distance between VECTORS.
:param accuracy: integer. Should be in the range 0 to 100
Specifies the accuracy of the nearest neighbor search during
query execution.
:param parallel: integer. Specifies degree of parallelism.
:param hnsw_neighbors: integer. Should be in the range 0 to
2048. Specifies the number of nearest neighbors considered
during the search. The attribute :attr:`.VectorIndexConfig.hnsw_neighbors`
is HNSW index specific.
:param hnsw_efconstruction: integer. Should be in the range 0
to 65535. Controls the trade-off between indexing speed and
recall quality during index construction. The attribute
:attr:`.VectorIndexConfig.hnsw_efconstruction` is HNSW index
specific.
:param ivf_neighbor_partitions: integer. Should be in the range
0 to 10,000,000. Specifies the number of partitions used to
divide the dataset. The attribute
:attr:`.VectorIndexConfig.ivf_neighbor_partitions` is IVF index
specific.
:param ivf_sample_per_partition: integer. Should be between 1
and ``num_vectors / neighbor partitions``. Specifies the
number of samples used per partition. The attribute
:attr:`.VectorIndexConfig.ivf_sample_per_partition` is IVF index
specific.
:param ivf_min_vectors_per_partition: integer. From 0 (no trimming)
to the total number of vectors (results in 1 partition). Specifies
the minimum number of vectors per partition. The attribute
:attr:`.VectorIndexConfig.ivf_min_vectors_per_partition`
is IVF index specific.
"""
index_type: VectorIndexType = VectorIndexType.HNSW
distance: Optional[VectorDistanceType] = None
accuracy: Optional[int] = None
hnsw_neighbors: Optional[int] = None
hnsw_efconstruction: Optional[int] = None
ivf_neighbor_partitions: Optional[int] = None
ivf_sample_per_partition: Optional[int] = None
ivf_min_vectors_per_partition: Optional[int] = None
parallel: Optional[int] = None
def __post_init__(self):
self.index_type = VectorIndexType(self.index_type)
for field in [
"hnsw_neighbors",
"hnsw_efconstruction",
"ivf_neighbor_partitions",
"ivf_sample_per_partition",
"ivf_min_vectors_per_partition",
"parallel",
"accuracy",
]:
value = getattr(self, field)
if value is not None and not isinstance(value, int):
raise TypeError(
f"{field} must be an integer if"
f"provided, got {type(value).__name__}"
)
class SparseVector:
"""
Lightweight SQLAlchemy-side version of SparseVector.
This mimics oracledb.SparseVector.
.. versionadded:: 2.0.43
"""
def __init__(
self,
num_dimensions: int,
indices: Union[list, array.array],
values: Union[list, array.array],
):
if not isinstance(indices, array.array) or indices.typecode != "I":
indices = array.array("I", indices)
if not isinstance(values, array.array):
values = array.array("d", values)
if len(indices) != len(values):
raise TypeError("indices and values must be of the same length!")
self.num_dimensions = num_dimensions
self.indices = indices
self.values = values
def __str__(self):
return (
f"SparseVector(num_dimensions={self.num_dimensions}, "
f"size={len(self.indices)}, typecode={self.values.typecode})"
)
class VECTOR(types.TypeEngine):
"""Oracle VECTOR datatype.
For complete background on using this type, see
:ref:`oracle_vector_datatype`.
.. versionadded:: 2.0.41
"""
cache_ok = True
__visit_name__ = "VECTOR"
_typecode_map = {
VectorStorageFormat.INT8: "b", # Signed int
VectorStorageFormat.BINARY: "B", # Unsigned int
VectorStorageFormat.FLOAT32: "f", # Float
VectorStorageFormat.FLOAT64: "d", # Double
}
def __init__(self, dim=None, storage_format=None, storage_type=None):
"""Construct a VECTOR.
:param dim: integer. The dimension of the VECTOR datatype. This
should be an integer value.
:param storage_format: VectorStorageFormat. The VECTOR storage
type format. This should be Enum values form
:class:`.VectorStorageFormat` INT8, BINARY, FLOAT32, or FLOAT64.
:param storage_type: VectorStorageType. The Vector storage type. This
should be Enum values from :class:`.VectorStorageType` SPARSE or
DENSE.
"""
if dim is not None and not isinstance(dim, int):
raise TypeError("dim must be an integer")
if storage_format is not None and not isinstance(
storage_format, VectorStorageFormat
):
raise TypeError(
"storage_format must be an enum of type VectorStorageFormat"
)
if storage_type is not None and not isinstance(
storage_type, VectorStorageType
):
raise TypeError(
"storage_type must be an enum of type VectorStorageType"
)
self.dim = dim
self.storage_format = storage_format
self.storage_type = storage_type
def _cached_bind_processor(self, dialect):
"""
Converts a Python-side SparseVector instance into an
oracledb.SparseVectormor a compatible array format before
binding it to the database.
"""
def process(value):
if value is None or isinstance(value, array.array):
return value
# Convert list to a array.array
elif isinstance(value, list):
typecode = self._array_typecode(self.storage_format)
value = array.array(typecode, value)
return value
# Convert SqlAlchemy SparseVector to oracledb SparseVector object
elif isinstance(value, SparseVector):
return dialect.dbapi.SparseVector(
value.num_dimensions,
value.indices,
value.values,
)
else:
raise TypeError("""
Invalid input for VECTOR: expected a list, an array.array,
or a SparseVector object.
""")
return process
def _cached_result_processor(self, dialect, coltype):
"""
Converts database-returned values into Python-native representations.
If the value is an oracledb.SparseVector, it is converted into the
SQLAlchemy-side SparseVector class.
If the value is a array.array, it is converted to a plain Python list.
"""
def process(value):
if value is None:
return None
elif isinstance(value, array.array):
return list(value)
# Convert Oracledb SparseVector to SqlAlchemy SparseVector object
elif isinstance(value, dialect.dbapi.SparseVector):
return SparseVector(
num_dimensions=value.num_dimensions,
indices=value.indices,
values=value.values,
)
return process
def _array_typecode(self, typecode):
"""
Map storage format to array typecode.
"""
return self._typecode_map.get(typecode, "d")
class comparator_factory(types.TypeEngine.Comparator):
def l2_distance(self, other):
return self.op("<->", return_type=Float)(other)
def inner_product(self, other):
return self.op("<#>", return_type=Float)(other)
def cosine_distance(self, other):
return self.op("<=>", return_type=Float)(other)