1
0

oracle23ai.py 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888
  1. """
  2. Oracle 23ai Vector Database Client - Fixed Version
  3. # .env
  4. VECTOR_DB = "oracle23ai"
  5. ## DBCS or oracle 23ai free
  6. ORACLE_DB_USE_WALLET = false
  7. ORACLE_DB_USER = "DEMOUSER"
  8. ORACLE_DB_PASSWORD = "Welcome123456"
  9. ORACLE_DB_DSN = "localhost:1521/FREEPDB1"
  10. ## ADW or ATP
  11. # ORACLE_DB_USE_WALLET = true
  12. # ORACLE_DB_USER = "DEMOUSER"
  13. # ORACLE_DB_PASSWORD = "Welcome123456"
  14. # ORACLE_DB_DSN = "medium"
  15. # ORACLE_DB_DSN = "(description= (retry_count=3)(retry_delay=3)(address=(protocol=tcps)(port=1522)(host=xx.oraclecloud.com))(connect_data=(service_name=yy.adb.oraclecloud.com))(security=(ssl_server_dn_match=no)))"
  16. # ORACLE_WALLET_DIR = "/home/opc/adb_wallet"
  17. # ORACLE_WALLET_PASSWORD = "Welcome1"
  18. ORACLE_VECTOR_LENGTH = 768
  19. ORACLE_DB_POOL_MIN = 2
  20. ORACLE_DB_POOL_MAX = 10
  21. ORACLE_DB_POOL_INCREMENT = 1
  22. """
  23. from typing import Optional, List, Dict, Any, Union
  24. from decimal import Decimal
  25. import logging
  26. import os
  27. import threading
  28. import time
  29. import json
  30. import array
  31. import oracledb
  32. from open_webui.retrieval.vector.main import (
  33. VectorDBBase,
  34. VectorItem,
  35. SearchResult,
  36. GetResult,
  37. )
  38. from open_webui.config import (
  39. ORACLE_DB_USE_WALLET,
  40. ORACLE_DB_USER,
  41. ORACLE_DB_PASSWORD,
  42. ORACLE_DB_DSN,
  43. ORACLE_WALLET_DIR,
  44. ORACLE_WALLET_PASSWORD,
  45. ORACLE_VECTOR_LENGTH,
  46. ORACLE_DB_POOL_MIN,
  47. ORACLE_DB_POOL_MAX,
  48. ORACLE_DB_POOL_INCREMENT,
  49. )
  50. from open_webui.env import SRC_LOG_LEVELS
  51. log = logging.getLogger(__name__)
  52. log.setLevel(SRC_LOG_LEVELS["RAG"])
  53. class Oracle23aiClient(VectorDBBase):
  54. """
  55. Oracle Vector Database Client for vector similarity search using Oracle Database 23ai.
  56. This client provides an interface to store, retrieve, and search vector embeddings
  57. in an Oracle database. It uses connection pooling for efficient database access
  58. and supports vector similarity search operations.
  59. Attributes:
  60. pool: Connection pool for Oracle database connections
  61. """
  62. def __init__(self) -> None:
  63. """
  64. Initialize the Oracle23aiClient with a connection pool.
  65. Creates a connection pool with configurable min/max connections, initializes
  66. the database schema if needed, and sets up necessary tables and indexes.
  67. Raises:
  68. ValueError: If required configuration parameters are missing
  69. Exception: If database initialization fails
  70. """
  71. self.pool = None
  72. try:
  73. # Create the appropriate connection pool based on DB type
  74. if ORACLE_DB_USE_WALLET:
  75. self._create_adb_pool()
  76. else: # DBCS
  77. self._create_dbcs_pool()
  78. dsn = ORACLE_DB_DSN
  79. log.info(f"Creating Connection Pool [{ORACLE_DB_USER}:**@{dsn}]")
  80. with self.get_connection() as connection:
  81. log.info(f"Connection version: {connection.version}")
  82. self._initialize_database(connection)
  83. log.info("Oracle Vector Search initialization complete.")
  84. except Exception as e:
  85. log.exception(f"Error during Oracle Vector Search initialization: {e}")
  86. raise
  87. def _create_adb_pool(self) -> None:
  88. """
  89. Create connection pool for Oracle Autonomous Database.
  90. Uses wallet-based authentication.
  91. """
  92. self.pool = oracledb.create_pool(
  93. user=ORACLE_DB_USER,
  94. password=ORACLE_DB_PASSWORD,
  95. dsn=ORACLE_DB_DSN,
  96. min=ORACLE_DB_POOL_MIN,
  97. max=ORACLE_DB_POOL_MAX,
  98. increment=ORACLE_DB_POOL_INCREMENT,
  99. config_dir=ORACLE_WALLET_DIR,
  100. wallet_location=ORACLE_WALLET_DIR,
  101. wallet_password=ORACLE_WALLET_PASSWORD
  102. )
  103. log.info("Created ADB connection pool with wallet authentication.")
  104. def _create_dbcs_pool(self) -> None:
  105. """
  106. Create connection pool for Oracle Database Cloud Service.
  107. Uses basic authentication without wallet.
  108. """
  109. self.pool = oracledb.create_pool(
  110. user=ORACLE_DB_USER,
  111. password=ORACLE_DB_PASSWORD,
  112. dsn=ORACLE_DB_DSN,
  113. min=ORACLE_DB_POOL_MIN,
  114. max=ORACLE_DB_POOL_MAX,
  115. increment=ORACLE_DB_POOL_INCREMENT
  116. )
  117. log.info("Created DB connection pool with basic authentication.")
  118. def get_connection(self):
  119. """
  120. Acquire a connection from the connection pool with retry logic.
  121. Returns:
  122. connection: A database connection with output type handler configured
  123. """
  124. max_retries = 3
  125. for attempt in range(max_retries):
  126. try:
  127. connection = self.pool.acquire()
  128. connection.outputtypehandler = self._output_type_handler
  129. return connection
  130. except oracledb.DatabaseError as e:
  131. error_obj, = e.args
  132. log.exception(f"Connection attempt {attempt + 1} failed: {error_obj.message}")
  133. if attempt < max_retries - 1:
  134. wait_time = 2 ** attempt
  135. log.info(f"Retrying in {wait_time} seconds...")
  136. time.sleep(wait_time)
  137. else:
  138. raise
  139. def start_health_monitor(self, interval_seconds: int = 60):
  140. """
  141. Start a background thread to periodically check the health of the connection pool.
  142. Args:
  143. interval_seconds (int): Number of seconds between health checks
  144. """
  145. def _monitor():
  146. while True:
  147. try:
  148. log.info("[HealthCheck] Running periodic DB health check...")
  149. self.ensure_connection()
  150. log.info("[HealthCheck] Connection is healthy.")
  151. except Exception as e:
  152. log.exception(f"[HealthCheck] Connection health check failed: {e}")
  153. time.sleep(interval_seconds)
  154. thread = threading.Thread(target=_monitor, daemon=True)
  155. thread.start()
  156. log.info(f"Started DB health monitor every {interval_seconds} seconds.")
  157. def _reconnect_pool(self):
  158. """
  159. Attempt to reinitialize the connection pool if it's been closed or broken.
  160. """
  161. try:
  162. log.info("Attempting to reinitialize the Oracle connection pool...")
  163. # Close existing pool if it exists
  164. if self.pool:
  165. try:
  166. self.pool.close()
  167. except Exception as close_error:
  168. log.warning(f"Error closing existing pool: {close_error}")
  169. # Re-create the appropriate connection pool based on DB type
  170. if ORACLE_DB_USE_WALLET:
  171. self._create_adb_pool()
  172. else: # DBCS
  173. self._create_dbcs_pool()
  174. log.info("Connection pool reinitialized.")
  175. except Exception as e:
  176. log.exception(f"Failed to reinitialize the connection pool: {e}")
  177. raise
  178. def ensure_connection(self):
  179. """
  180. Ensure the database connection is alive, reconnecting pool if needed.
  181. """
  182. try:
  183. with self.get_connection() as connection:
  184. with connection.cursor() as cursor:
  185. cursor.execute("SELECT 1 FROM dual")
  186. except Exception as e:
  187. log.exception(f"Connection check failed: {e}, attempting to reconnect pool...")
  188. self._reconnect_pool()
  189. def _output_type_handler(self, cursor, metadata):
  190. """
  191. Handle Oracle vector type conversion.
  192. Args:
  193. cursor: Oracle database cursor
  194. metadata: Metadata for the column
  195. Returns:
  196. A variable with appropriate conversion for vector types
  197. """
  198. if metadata.type_code is oracledb.DB_TYPE_VECTOR:
  199. return cursor.var(metadata.type_code, arraysize=cursor.arraysize,
  200. outconverter=list)
  201. def _initialize_database(self, connection) -> None:
  202. """
  203. Initialize database schema, tables and indexes.
  204. Creates the document_chunk table and necessary indexes if they don't exist.
  205. Args:
  206. connection: Oracle database connection
  207. Raises:
  208. Exception: If schema initialization fails
  209. """
  210. with connection.cursor() as cursor:
  211. try:
  212. log.info("Creating Table document_chunk")
  213. cursor.execute("""
  214. BEGIN
  215. EXECUTE IMMEDIATE '
  216. CREATE TABLE IF NOT EXISTS document_chunk (
  217. id VARCHAR2(255) PRIMARY KEY,
  218. collection_name VARCHAR2(255) NOT NULL,
  219. text CLOB,
  220. vmetadata JSON,
  221. vector vector(*, float32)
  222. )
  223. ';
  224. EXCEPTION
  225. WHEN OTHERS THEN
  226. IF SQLCODE != -955 THEN
  227. RAISE;
  228. END IF;
  229. END;
  230. """)
  231. log.info("Creating Index document_chunk_collection_name_idx")
  232. cursor.execute("""
  233. BEGIN
  234. EXECUTE IMMEDIATE '
  235. CREATE INDEX IF NOT EXISTS document_chunk_collection_name_idx
  236. ON document_chunk (collection_name)
  237. ';
  238. EXCEPTION
  239. WHEN OTHERS THEN
  240. IF SQLCODE != -955 THEN
  241. RAISE;
  242. END IF;
  243. END;
  244. """)
  245. log.info("Creating VECTOR INDEX document_chunk_vector_ivf_idx")
  246. cursor.execute("""
  247. BEGIN
  248. EXECUTE IMMEDIATE '
  249. CREATE VECTOR INDEX IF NOT EXISTS document_chunk_vector_ivf_idx
  250. ON document_chunk(vector)
  251. ORGANIZATION NEIGHBOR PARTITIONS
  252. DISTANCE COSINE
  253. WITH TARGET ACCURACY 95
  254. PARAMETERS (TYPE IVF, NEIGHBOR PARTITIONS 100)
  255. ';
  256. EXCEPTION
  257. WHEN OTHERS THEN
  258. IF SQLCODE != -955 THEN
  259. RAISE;
  260. END IF;
  261. END;
  262. """)
  263. connection.commit()
  264. log.info("Database initialization completed successfully.")
  265. except Exception as e:
  266. connection.rollback()
  267. log.exception(f"Error during database initialization: {e}")
  268. raise
  269. def check_vector_length(self) -> None:
  270. """
  271. Check vector length compatibility (placeholder).
  272. This method would check if the configured vector length matches the database schema.
  273. Currently implemented as a placeholder.
  274. """
  275. pass
  276. def _vector_to_blob(self, vector: List[float]) -> bytes:
  277. """
  278. Convert a vector to Oracle BLOB format.
  279. Args:
  280. vector (List[float]): The vector to convert
  281. Returns:
  282. bytes: The vector in Oracle BLOB format
  283. """
  284. return array.array("f", vector)
  285. def adjust_vector_length(self, vector: List[float]) -> List[float]:
  286. """
  287. Adjust vector to the expected length if needed.
  288. Args:
  289. vector (List[float]): The vector to adjust
  290. Returns:
  291. List[float]: The adjusted vector
  292. """
  293. return vector
  294. def _decimal_handler(self, obj):
  295. """
  296. Handle Decimal objects for JSON serialization.
  297. Args:
  298. obj: Object to serialize
  299. Returns:
  300. float: Converted decimal value
  301. Raises:
  302. TypeError: If object is not JSON serializable
  303. """
  304. if isinstance(obj, Decimal):
  305. return float(obj)
  306. raise TypeError(f"{obj} is not JSON serializable")
  307. def _metadata_to_json(self, metadata: Dict) -> str:
  308. """
  309. Convert metadata dictionary to JSON string.
  310. Args:
  311. metadata (Dict): Metadata dictionary
  312. Returns:
  313. str: JSON representation of metadata
  314. """
  315. return json.dumps(metadata, default=self._decimal_handler) if metadata else "{}"
  316. def _json_to_metadata(self, json_str: str) -> Dict:
  317. """
  318. Convert JSON string to metadata dictionary.
  319. Args:
  320. json_str (str): JSON string
  321. Returns:
  322. Dict: Metadata dictionary
  323. """
  324. return json.loads(json_str) if json_str else {}
  325. def insert(self, collection_name: str, items: List[VectorItem]) -> None:
  326. """
  327. Insert vector items into the database.
  328. Args:
  329. collection_name (str): Name of the collection
  330. items (List[VectorItem]): List of vector items to insert
  331. Raises:
  332. Exception: If insertion fails
  333. Example:
  334. >>> client = Oracle23aiClient()
  335. >>> items = [
  336. ... {"id": "1", "text": "Sample text", "vector": [0.1, 0.2, ...], "metadata": {"source": "doc1"}},
  337. ... {"id": "2", "text": "Another text", "vector": [0.3, 0.4, ...], "metadata": {"source": "doc2"}}
  338. ... ]
  339. >>> client.insert("my_collection", items)
  340. """
  341. log.info(f"Inserting {len(items)} items into collection '{collection_name}'.")
  342. with self.get_connection() as connection:
  343. try:
  344. with connection.cursor() as cursor:
  345. for item in items:
  346. vector_blob = self._vector_to_blob(item["vector"])
  347. metadata_json = self._metadata_to_json(item["metadata"])
  348. cursor.execute("""
  349. INSERT INTO document_chunk
  350. (id, collection_name, text, vmetadata, vector)
  351. VALUES (:id, :collection_name, :text, :metadata, :vector)
  352. """, {
  353. 'id': item["id"],
  354. 'collection_name': collection_name,
  355. 'text': item["text"],
  356. 'metadata': metadata_json,
  357. 'vector': vector_blob
  358. })
  359. connection.commit()
  360. log.info(f"Successfully inserted {len(items)} items into collection '{collection_name}'.")
  361. except Exception as e:
  362. connection.rollback()
  363. log.exception(f"Error during insert: {e}")
  364. raise
  365. def upsert(self, collection_name: str, items: List[VectorItem]) -> None:
  366. """
  367. Update or insert vector items into the database.
  368. If an item with the same ID exists, it will be updated;
  369. otherwise, it will be inserted.
  370. Args:
  371. collection_name (str): Name of the collection
  372. items (List[VectorItem]): List of vector items to upsert
  373. Raises:
  374. Exception: If upsert operation fails
  375. Example:
  376. >>> client = Oracle23aiClient()
  377. >>> items = [
  378. ... {"id": "1", "text": "Updated text", "vector": [0.1, 0.2, ...], "metadata": {"source": "doc1"}},
  379. ... {"id": "3", "text": "New item", "vector": [0.5, 0.6, ...], "metadata": {"source": "doc3"}}
  380. ... ]
  381. >>> client.upsert("my_collection", items)
  382. """
  383. log.info(f"Upserting {len(items)} items into collection '{collection_name}'.")
  384. with self.get_connection() as connection:
  385. try:
  386. with connection.cursor() as cursor:
  387. for item in items:
  388. vector_blob = self._vector_to_blob(item["vector"])
  389. metadata_json = self._metadata_to_json(item["metadata"])
  390. cursor.execute("""
  391. MERGE INTO document_chunk d
  392. USING (SELECT :merge_id as id FROM dual) s
  393. ON (d.id = s.id)
  394. WHEN MATCHED THEN
  395. UPDATE SET
  396. collection_name = :upd_collection_name,
  397. text = :upd_text,
  398. vmetadata = :upd_metadata,
  399. vector = :upd_vector
  400. WHEN NOT MATCHED THEN
  401. INSERT (id, collection_name, text, vmetadata, vector)
  402. VALUES (:ins_id, :ins_collection_name, :ins_text, :ins_metadata, :ins_vector)
  403. """, {
  404. 'merge_id': item["id"],
  405. 'upd_collection_name': collection_name,
  406. 'upd_text': item["text"],
  407. 'upd_metadata': metadata_json,
  408. 'upd_vector': vector_blob,
  409. 'ins_id': item["id"],
  410. 'ins_collection_name': collection_name,
  411. 'ins_text': item["text"],
  412. 'ins_metadata': metadata_json,
  413. 'ins_vector': vector_blob
  414. })
  415. connection.commit()
  416. log.info(f"Successfully upserted {len(items)} items into collection '{collection_name}'.")
  417. except Exception as e:
  418. connection.rollback()
  419. log.exception(f"Error during upsert: {e}")
  420. raise
  421. def search(
  422. self,
  423. collection_name: str,
  424. vectors: List[List[Union[float, int]]],
  425. limit: int
  426. ) -> Optional[SearchResult]:
  427. """
  428. Search for similar vectors in the database.
  429. Performs vector similarity search using cosine distance.
  430. Args:
  431. collection_name (str): Name of the collection to search
  432. vectors (List[List[Union[float, int]]]): Query vectors to find similar items for
  433. limit (int): Maximum number of results to return per query
  434. Returns:
  435. Optional[SearchResult]: Search results containing ids, distances, documents, and metadata
  436. Example:
  437. >>> client = Oracle23aiClient()
  438. >>> query_vector = [0.1, 0.2, 0.3, ...] # Must match VECTOR_LENGTH
  439. >>> results = client.search("my_collection", [query_vector], limit=5)
  440. >>> if results:
  441. ... log.info(f"Found {len(results.ids[0])} matches")
  442. ... for i, (id, dist) in enumerate(zip(results.ids[0], results.distances[0])):
  443. ... log.info(f"Match {i+1}: id={id}, distance={dist}")
  444. """
  445. log.info(f"Searching items from collection '{collection_name}' with limit {limit}.")
  446. try:
  447. if not vectors:
  448. log.warning("No vectors provided for search.")
  449. return None
  450. num_queries = len(vectors)
  451. ids = [[] for _ in range(num_queries)]
  452. distances = [[] for _ in range(num_queries)]
  453. documents = [[] for _ in range(num_queries)]
  454. metadatas = [[] for _ in range(num_queries)]
  455. with self.get_connection() as connection:
  456. with connection.cursor() as cursor:
  457. for qid, vector in enumerate(vectors):
  458. vector_blob = self._vector_to_blob(vector)
  459. cursor.execute("""
  460. SELECT dc.id, dc.text,
  461. JSON_SERIALIZE(dc.vmetadata RETURNING VARCHAR2(4096)) as vmetadata,
  462. VECTOR_DISTANCE(dc.vector, :query_vector, COSINE) as distance
  463. FROM document_chunk dc
  464. WHERE dc.collection_name = :collection_name
  465. ORDER BY VECTOR_DISTANCE(dc.vector, :query_vector, COSINE)
  466. FETCH APPROX FIRST :limit ROWS ONLY
  467. """, {
  468. 'query_vector': vector_blob,
  469. 'collection_name': collection_name,
  470. 'limit': limit
  471. })
  472. results = cursor.fetchall()
  473. for row in results:
  474. ids[qid].append(row[0])
  475. documents[qid].append(row[1].read() if isinstance(row[1], oracledb.LOB) else str(row[1]))
  476. # 🔧 FIXED: Parse JSON metadata properly
  477. metadata_str = row[2].read() if isinstance(row[2], oracledb.LOB) else row[2]
  478. metadatas[qid].append(self._json_to_metadata(metadata_str))
  479. distances[qid].append(float(row[3]))
  480. log.info(f"Search completed. Found {sum(len(ids[i]) for i in range(num_queries))} total results.")
  481. return SearchResult(
  482. ids=ids,
  483. distances=distances,
  484. documents=documents,
  485. metadatas=metadatas
  486. )
  487. except Exception as e:
  488. log.exception(f"Error during search: {e}")
  489. return None
  490. def query(
  491. self,
  492. collection_name: str,
  493. filter: Dict,
  494. limit: Optional[int] = None
  495. ) -> Optional[GetResult]:
  496. """
  497. Query items based on metadata filters.
  498. Retrieves items that match specified metadata criteria.
  499. Args:
  500. collection_name (str): Name of the collection to query
  501. filter (Dict[str, Any]): Metadata filters to apply
  502. limit (Optional[int]): Maximum number of results to return
  503. Returns:
  504. Optional[GetResult]: Query results containing ids, documents, and metadata
  505. Example:
  506. >>> client = Oracle23aiClient()
  507. >>> filter = {"source": "doc1", "category": "finance"}
  508. >>> results = client.query("my_collection", filter, limit=20)
  509. >>> if results:
  510. ... print(f"Found {len(results.ids[0])} matching documents")
  511. """
  512. log.info(f"Querying items from collection '{collection_name}' with filters.")
  513. try:
  514. limit = limit or 100
  515. query = """
  516. SELECT id, text, JSON_SERIALIZE(vmetadata RETURNING VARCHAR2(4096)) as vmetadata
  517. FROM document_chunk
  518. WHERE collection_name = :collection_name
  519. """
  520. params = {'collection_name': collection_name}
  521. for i, (key, value) in enumerate(filter.items()):
  522. param_name = f"value_{i}"
  523. query += f" AND JSON_VALUE(vmetadata, '$.{key}' RETURNING VARCHAR2(4096)) = :{param_name}"
  524. params[param_name] = str(value)
  525. query += " FETCH FIRST :limit ROWS ONLY"
  526. params['limit'] = limit
  527. with self.get_connection() as connection:
  528. with connection.cursor() as cursor:
  529. cursor.execute(query, params)
  530. results = cursor.fetchall()
  531. if not results:
  532. log.info("No results found for query.")
  533. return None
  534. ids = [[row[0] for row in results]]
  535. documents = [[row[1].read() if isinstance(row[1], oracledb.LOB) else str(row[1]) for row in results]]
  536. # 🔧 FIXED: Parse JSON metadata properly
  537. metadatas = [[self._json_to_metadata(row[2].read() if isinstance(row[2], oracledb.LOB) else row[2]) for row in results]]
  538. log.info(f"Query completed. Found {len(results)} results.")
  539. return GetResult(
  540. ids=ids,
  541. documents=documents,
  542. metadatas=metadatas
  543. )
  544. except Exception as e:
  545. log.exception(f"Error during query: {e}")
  546. return None
  547. def get(
  548. self,
  549. collection_name: str
  550. ) -> Optional[GetResult]:
  551. """
  552. Get all items in a collection.
  553. Retrieves items from a specified collection up to the limit.
  554. Args:
  555. collection_name (str): Name of the collection to retrieve
  556. limit (Optional[int]): Maximum number of items to retrieve
  557. Returns:
  558. Optional[GetResult]: Result containing ids, documents, and metadata
  559. Example:
  560. >>> client = Oracle23aiClient()
  561. >>> results = client.get("my_collection", limit=50)
  562. >>> if results:
  563. ... print(f"Retrieved {len(results.ids[0])} documents from collection")
  564. """
  565. log.info(f"Getting items from collection '{collection_name}' with limit {limit}.")
  566. try:
  567. limit = limit or 1000
  568. with self.get_connection() as connection:
  569. with connection.cursor() as cursor:
  570. cursor.execute("""
  571. SELECT /*+ MONITOR */ id, text, JSON_SERIALIZE(vmetadata RETURNING VARCHAR2(4096)) as vmetadata
  572. FROM document_chunk
  573. WHERE collection_name = :collection_name
  574. FETCH FIRST :limit ROWS ONLY
  575. """, {
  576. 'collection_name': collection_name,
  577. 'limit': limit
  578. })
  579. results = cursor.fetchall()
  580. if not results:
  581. log.info("No results found.")
  582. return None
  583. ids = [[row[0] for row in results]]
  584. documents = [[row[1].read() if isinstance(row[1], oracledb.LOB) else str(row[1]) for row in results]]
  585. # 🔧 FIXED: Parse JSON metadata properly
  586. metadatas = [[self._json_to_metadata(row[2].read() if isinstance(row[2], oracledb.LOB) else row[2]) for row in results]]
  587. return GetResult(
  588. ids=ids,
  589. documents=documents,
  590. metadatas=metadatas
  591. )
  592. except Exception as e:
  593. log.exception(f"Error during get: {e}")
  594. return None
  595. def delete(
  596. self,
  597. collection_name: str,
  598. ids: Optional[List[str]] = None,
  599. filter: Optional[Dict[str, Any]] = None,
  600. ) -> None:
  601. """
  602. Delete items from the database.
  603. Deletes items from a collection based on IDs or metadata filters.
  604. Args:
  605. collection_name (str): Name of the collection to delete from
  606. ids (Optional[List[str]]): Specific item IDs to delete
  607. filter (Optional[Dict[str, Any]]): Metadata filters for deletion
  608. Raises:
  609. Exception: If deletion fails
  610. Example:
  611. >>> client = Oracle23aiClient()
  612. >>> # Delete specific items by ID
  613. >>> client.delete("my_collection", ids=["1", "3", "5"])
  614. >>> # Or delete by metadata filter
  615. >>> client.delete("my_collection", filter={"source": "deprecated_source"})
  616. """
  617. log.info(f"Deleting items from collection '{collection_name}'.")
  618. try:
  619. query = "DELETE FROM document_chunk WHERE collection_name = :collection_name"
  620. params = {'collection_name': collection_name}
  621. if ids:
  622. # 🔧 FIXED: Use proper parameterized query to prevent SQL injection
  623. placeholders = ','.join([f':id_{i}' for i in range(len(ids))])
  624. query += f" AND id IN ({placeholders})"
  625. for i, id_val in enumerate(ids):
  626. params[f'id_{i}'] = id_val
  627. if filter:
  628. for i, (key, value) in enumerate(filter.items()):
  629. param_name = f"value_{i}"
  630. query += f" AND JSON_VALUE(vmetadata, '$.{key}' RETURNING VARCHAR2(4096)) = :{param_name}"
  631. params[param_name] = str(value)
  632. with self.get_connection() as connection:
  633. with connection.cursor() as cursor:
  634. cursor.execute(query, params)
  635. deleted = cursor.rowcount
  636. connection.commit()
  637. log.info(f"Deleted {deleted} items from collection '{collection_name}'.")
  638. except Exception as e:
  639. log.exception(f"Error during delete: {e}")
  640. raise
  641. def reset(self) -> None:
  642. """
  643. Reset the database by deleting all items.
  644. Deletes all items from the document_chunk table.
  645. Raises:
  646. Exception: If reset fails
  647. Example:
  648. >>> client = Oracle23aiClient()
  649. >>> client.reset() # Warning: Removes all data!
  650. """
  651. log.info("Resetting database - deleting all items.")
  652. try:
  653. with self.get_connection() as connection:
  654. with connection.cursor() as cursor:
  655. cursor.execute("DELETE FROM document_chunk")
  656. deleted = cursor.rowcount
  657. connection.commit()
  658. log.info(f"Reset complete. Deleted {deleted} items from 'document_chunk' table.")
  659. except Exception as e:
  660. log.exception(f"Error during reset: {e}")
  661. raise
  662. def close(self) -> None:
  663. """
  664. Close the database connection pool.
  665. Properly closes the connection pool and releases all resources.
  666. Example:
  667. >>> client = Oracle23aiClient()
  668. >>> # After finishing all operations
  669. >>> client.close()
  670. """
  671. try:
  672. if hasattr(self, 'pool') and self.pool:
  673. self.pool.close()
  674. log.info("Oracle Vector Search connection pool closed.")
  675. except Exception as e:
  676. log.exception(f"Error closing connection pool: {e}")
  677. def has_collection(self, collection_name: str) -> bool:
  678. """
  679. Check if a collection exists.
  680. Args:
  681. collection_name (str): Name of the collection to check
  682. Returns:
  683. bool: True if the collection exists, False otherwise
  684. Example:
  685. >>> client = Oracle23aiClient()
  686. >>> if client.has_collection("my_collection"):
  687. ... print("Collection exists!")
  688. ... else:
  689. ... print("Collection does not exist.")
  690. """
  691. try:
  692. with self.get_connection() as connection:
  693. with connection.cursor() as cursor:
  694. cursor.execute("""
  695. SELECT COUNT(*)
  696. FROM document_chunk
  697. WHERE collection_name = :collection_name
  698. FETCH FIRST 1 ROWS ONLY
  699. """, {'collection_name': collection_name})
  700. count = cursor.fetchone()[0]
  701. return count > 0
  702. except Exception as e:
  703. log.exception(f"Error checking collection existence: {e}")
  704. return False
  705. def delete_collection(self, collection_name: str) -> None:
  706. """
  707. Delete an entire collection.
  708. Removes all items belonging to the specified collection.
  709. Args:
  710. collection_name (str): Name of the collection to delete
  711. Example:
  712. >>> client = Oracle23aiClient()
  713. >>> client.delete_collection("obsolete_collection")
  714. """
  715. log.info(f"Deleting collection '{collection_name}'.")
  716. try:
  717. with self.get_connection() as connection:
  718. with connection.cursor() as cursor:
  719. cursor.execute("""
  720. DELETE FROM document_chunk
  721. WHERE collection_name = :collection_name
  722. """, {'collection_name': collection_name})
  723. deleted = cursor.rowcount
  724. connection.commit()
  725. log.info(f"Collection '{collection_name}' deleted. Removed {deleted} items.")
  726. except Exception as e:
  727. log.exception(f"Error deleting collection '{collection_name}': {e}")
  728. raise