# ----------------------------------------------------------------------------- # Copyright (c) 2021, 2023, Oracle and/or its affiliates. # # This software is dual-licensed to you under the Universal Permissive License # (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License # 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose # either license. # # If you elect to accept the software under the Apache License, Version 2.0, # the following applies: # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- # ----------------------------------------------------------------------------- # dbobject.py # # Contains the classes used for managing database objects and the database # object type metadata: DbObject, DbObjectType and DbObjectAttr. # ----------------------------------------------------------------------------- from typing import Any, Sequence, Union from . import errors from . import __name__ as MODULE_NAME from .base_impl import DbType class DbObject: __module__ = MODULE_NAME def __getattr__(self, name): try: attr_impl = self._impl.type.attrs_by_name[name] except KeyError: return super().__getattribute__(name) return self._impl.get_attr_value(attr_impl) def __repr__(self): return ( f"" ) def __setattr__(self, name, value): if name == "_impl" or name == "_type": super().__setattr__(name, value) else: attr_impl = self._impl.type.attrs_by_name[name] self._impl.set_attr_value(attr_impl, value) def _ensure_is_collection(self): """ Ensures that the object refers to a collection. If not, an exception is raised. """ if not self.type.iscollection: errors._raise_err( errors.ERR_OBJECT_IS_NOT_A_COLLECTION, name=self.type._get_full_name(), ) @classmethod def _from_impl(cls, impl): obj = cls.__new__(cls) obj._impl = impl obj._type = None return obj def append(self, element: Any) -> None: """ Append an element to the collection object. If no elements exist in the collection, this creates an element at index 0; otherwise, it creates an element immediately following the highest index available in the collection. """ self._impl.append(element) def asdict(self) -> dict: """ Return a dictionary where the collection’s indexes are the keys and the elements are its values. """ self._ensure_is_collection() result = {} ix = self._impl.get_first_index() while ix is not None: result[ix] = self._impl.get_element_by_index(ix) ix = self._impl.get_next_index(ix) return result def aslist(self) -> list: """ Return a list of each of the collection’s elements in index order. """ self._ensure_is_collection() result = [] ix = self._impl.get_first_index() while ix is not None: result.append(self._impl.get_element_by_index(ix)) ix = self._impl.get_next_index(ix) return result def copy(self) -> "DbObject": """ Create a copy of the object and return it. """ copied_impl = self._impl.copy() return DbObject._from_impl(copied_impl) def delete(self, index: int) -> None: """ Delete the element at the specified index of the collection. If the element does not exist or is otherwise invalid, an error is raised. Note that the indices of the remaining elements in the collection are not changed. In other words, the delete operation creates holes in the collection. """ self._ensure_is_collection() self._impl.delete_by_index(index) def exists(self, index: int) -> bool: """ Return True or False indicating if an element exists in the collection at the specified index. """ self._ensure_is_collection() return self._impl.exists_by_index(index) def extend(self, seq: list) -> None: """ Append all of the elements in the sequence to the collection. This is the equivalent of performing append() for each element found in the sequence. """ self._ensure_is_collection() for value in seq: self.append(value) def first(self) -> int: """ Return the index of the first element in the collection. If the collection is empty, None is returned. """ self._ensure_is_collection() return self._impl.get_first_index() def getelement(self, index: int) -> Any: """ Return the element at the specified index of the collection. If no element exists at that index, an exception is raised. """ self._ensure_is_collection() return self._impl.get_element_by_index(index) def last(self) -> int: """ Return the index of the last element in the collection. If the collection is empty, None is returned. """ self._ensure_is_collection() return self._impl.get_last_index() def next(self, index: int) -> int: """ Return the index of the next element in the collection following the specified index. If there are no elements in the collection following the specified index, None is returned. """ self._ensure_is_collection() return self._impl.get_next_index(index) def prev(self, index: int) -> int: """ Return the index of the element in the collection preceding the specified index. If there are no elements in the collection preceding the specified index, None is returned. """ self._ensure_is_collection() return self._impl.get_prev_index(index) def setelement(self, index: int, value: Any) -> None: """ Set the value in the collection at the specified index to the given value. """ self._ensure_is_collection() self._impl.set_element_by_index(index, value) def size(self) -> int: """ Return the number of elements in the collection. """ self._ensure_is_collection() return self._impl.get_size() def trim(self, num: int) -> None: """ Remove the specified number of elements from the end of the collection. """ self._ensure_is_collection() self._impl.trim(num) @property def type(self) -> "DbObjectType": """ Returns an ObjectType corresponding to the type of the object. """ if self._type is None: self._type = DbObjectType._from_impl(self._impl.type) return self._type class DbObjectAttr: __module__ = MODULE_NAME def __repr__(self): return f"" @classmethod def _from_impl(cls, impl): attr = cls.__new__(cls) attr._impl = impl attr._type = None return attr @property def name(self) -> str: """ This read-only attribute returns the name of the attribute. """ return self._impl.name @property def type(self) -> Union["DbObjectType", DbType]: """ This read-only attribute returns the type of the attribute. This will be an Oracle Object Type if the variable binds Oracle objects; otherwise, it will be one of the database type constants. """ if self._type is None: if self._impl.objtype is not None: self._type = DbObjectType._from_impl(self._impl.objtype) else: self._type = self._impl.dbtype return self._type class DbObjectType: __module__ = MODULE_NAME def __call__(self, value=None): return self.newobject(value) def __eq__(self, other): if isinstance(other, DbObjectType): return other._impl == self._impl return NotImplemented def __repr__(self): return f"" @classmethod def _from_impl(cls, impl): typ = cls.__new__(cls) typ._impl = impl typ._attributes = None typ._element_type = None return typ def _get_full_name(self): """ Returns the full name of the type. """ return self._impl._get_fqn() @property def attributes(self) -> list: """ This read-only attribute returns a list of the attributes that make up the object type. """ if self._attributes is None: self._attributes = [ DbObjectAttr._from_impl(i) for i in self._impl.attrs ] return self._attributes @property def iscollection(self) -> bool: """ This read-only attribute returns a boolean indicating if the object type refers to a collection or not. """ return self._impl.is_collection @property def name(self) -> str: """ This read-only attribute returns the name of the type. """ return self._impl.name @property def element_type(self) -> Union["DbObjectType", DbType]: """ This read-only attribute returns the type of elements found in collections of this type, if iscollection is True; otherwise, it returns None. If the collection contains objects, this will be another object type; otherwise, it will be one of the database type constants. """ if self._element_type is None: if self._impl.element_objtype is not None: typ_impl = self._impl.element_objtype self._element_type = DbObjectType._from_impl(typ_impl) else: self._element_type = self._impl.element_dbtype return self._element_type def newobject(self, value: Sequence = None) -> DbObject: """ Return a new Oracle object of the given type. This object can then be modified by setting its attributes and then bound to a cursor for interaction with Oracle. If the object type refers to a collection, a sequence may be passed and the collection will be initialized with the items in that sequence. """ obj_impl = self._impl.create_new_object() obj = DbObject._from_impl(obj_impl) if value is not None: obj.extend(value) return obj @property def package_name(self) -> str: """ This read-only attribute returns the name of the package containing the PL/SQL type or None if the type is not a PL/SQL type. """ return self._impl.package_name @property def schema(self) -> str: """ This read-only attribute returns the name of the schema that owns the type. """ return self._impl.schema