Database Objects

class DbObject:
    """Represent Oracle database objects including object types and collections."""
    
    # Properties
    type: DbObjectType  # Object type metadata
    
    def asdict(self) -> dict:
        """
        Convert object to dictionary representation.
        
        Returns:
        dict: Object attributes as key-value pairs
        """
    
    def aslist(self) -> list:
        """
        Convert collection object to list representation.
        
        Returns:
        list: Collection elements as list
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def copy(self) -> DbObject:
        """
        Create a copy of the object.
        
        Returns:
        DbObject: Copy of the object
        """
    
    def append(self, element) -> None:
        """
        Append element to collection object.
        
        Parameters:
        - element: Element to append
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def extend(self, sequence) -> None:
        """
        Extend collection with sequence of elements.
        
        Parameters:
        - sequence: Sequence of elements to add
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def delete(self, index) -> None:
        """
        Delete element at specified index from collection.
        
        Parameters:
        - index (int): Index of element to delete
        
        Raises:
        TypeError: If object is not a collection
        IndexError: If index is out of range
        """
    
    def exists(self, index) -> bool:
        """
        Check if element exists at specified index in collection.
        
        Parameters:
        - index (int): Index to check
        
        Returns:
        bool: True if element exists at index
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def getelement(self, index):
        """
        Get element at specified index from collection.
        
        Parameters:
        - index (int): Index of element to retrieve
        
        Returns:
        Element at specified index
        
        Raises:
        TypeError: If object is not a collection
        IndexError: If index is out of range
        """
    
    def setelement(self, index, value) -> None:
        """
        Set element at specified index in collection.
        
        Parameters:
        - index (int): Index to set
        - value: Value to set
        
        Raises:
        TypeError: If object is not a collection
        IndexError: If index is out of range
        """
    
    def size(self) -> int:
        """
        Get size of collection object.
        
        Returns:
        int: Number of elements in collection
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def trim(self, size) -> None:
        """
        Trim collection to specified size.
        
        Parameters:
        - size (int): New size for collection
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def first(self) -> int:
        """
        Get index of first element in collection.
        
        Returns:
        int: Index of first element
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def last(self) -> int:
        """
        Get index of last element in collection.
        
        Returns:
        int: Index of last element
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def next(self, index) -> int:
        """
        Get index of next element after specified index.
        
        Parameters:
        - index (int): Current index
        
        Returns:
        int: Index of next element or None if no next element
        
        Raises:
        TypeError: If object is not a collection
        """
    
    def prev(self, index) -> int:
        """
        Get index of previous element before specified index.
        
        Parameters:
        - index (int): Current index
        
        Returns:
        int: Index of previous element or None if no previous element
        
        Raises:
        TypeError: If object is not a collection
        """

class DbObjectType:
    """Metadata for database object types."""
    
    # Properties
    name: str               # Type name
    schema: str            # Schema name
    package_name: str      # Package name (for package types)
    attributes: list       # List of DbObjectAttr objects
    iscollection: bool     # True if type is a collection
    element_type: DbObjectType  # Element type for collections
    
    def newobject(self, value=None) -> DbObject:
        """
        Create new object instance of this type.
        
        Parameters:
        - value: Initial value for object (dict, list, or compatible value)
        
        Returns:
        DbObject: New object instance
        """

class DbObjectAttr:
    """Attribute metadata for database objects."""
    
    # Properties
    name: str           # Attribute name
    type: type          # Attribute data type
    precision: int      # Numeric precision (for numeric types)
    scale: int          # Numeric scale (for numeric types)
    max_size: int       # Maximum size (for string/raw types)
    type_code: int      # Internal type code
    null_ok: bool       # True if attribute can be NULL

import oracledb

connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")

# Create object type in database
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE OR REPLACE TYPE address_type AS OBJECT (
            street VARCHAR2(100),
            city VARCHAR2(50),
            state VARCHAR2(20),
            zip_code VARCHAR2(10)
        )
    """)
    
    cursor.execute("""
        CREATE OR REPLACE TYPE person_type AS OBJECT (
            first_name VARCHAR2(50),
            last_name VARCHAR2(50),
            age NUMBER,
            address address_type
        )
    """)
    
    cursor.execute("""
        CREATE TABLE people_obj (
            id NUMBER PRIMARY KEY,
            person person_type
        )
    """)

# Get object type
person_type = connection.gettype("PERSON_TYPE")
address_type = connection.gettype("ADDRESS_TYPE")

print(f"Type: {person_type.name}")
print(f"Schema: {person_type.schema}")
print(f"Is collection: {person_type.iscollection}")

# Examine attributes
print("Attributes:")
for attr in person_type.attributes:
    print(f"  {attr.name}: {attr.type} (max_size: {attr.max_size})")

# Create object instances
address = address_type.newobject()
address.STREET = "123 Main St"
address.CITY = "Anytown"
address.STATE = "CA"
address.ZIP_CODE = "12345"

person = person_type.newobject()
person.FIRST_NAME = "John"
person.LAST_NAME = "Doe"
person.AGE = 30
person.ADDRESS = address

# Insert object into table
with connection.cursor() as cursor:
    cursor.execute("""
        INSERT INTO people_obj (id, person) VALUES (:1, :2)
    """, [1, person])
    
    connection.commit()

# Query and work with objects
with connection.cursor() as cursor:
    cursor.execute("SELECT person FROM people_obj WHERE id = :1", [1])
    result = cursor.fetchone()
    person_obj = result[0]
    
    print(f"Name: {person_obj.FIRST_NAME} {person_obj.LAST_NAME}")
    print(f"Age: {person_obj.AGE}")
    print(f"Address: {person_obj.ADDRESS.STREET}, {person_obj.ADDRESS.CITY}")
    
    # Convert to dictionary
    person_dict = person_obj.asdict()
    print(f"As dict: {person_dict}")

connection.close()

import oracledb

connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")

# Create collection types
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE OR REPLACE TYPE string_array AS VARRAY(100) OF VARCHAR2(100)
    """)
    
    cursor.execute("""
        CREATE OR REPLACE TYPE number_table AS TABLE OF NUMBER
    """)
    
    cursor.execute("""
        CREATE TABLE collections_demo (
            id NUMBER PRIMARY KEY,
            tags string_array,
            scores number_table
        ) NESTED TABLE scores STORE AS scores_tab
    """)

# Get collection types
string_array_type = connection.gettype("STRING_ARRAY")
number_table_type = connection.gettype("NUMBER_TABLE")

print(f"String array is collection: {string_array_type.iscollection}")
print(f"Element type: {string_array_type.element_type}")

# Create collection instances
tags = string_array_type.newobject()
tags.append("python")
tags.append("oracle")
tags.append("database")

scores = number_table_type.newobject()
scores.extend([95, 87, 92, 89, 94])

print(f"Tags size: {tags.size()}")
print(f"Scores size: {scores.size()}")

# Insert collections
with connection.cursor() as cursor:
    cursor.execute("""
        INSERT INTO collections_demo (id, tags, scores)
        VALUES (:1, :2, :3)
    """, [1, tags, scores])
    
    connection.commit()

# Query and manipulate collections
with connection.cursor() as cursor:
    cursor.execute("SELECT tags, scores FROM collections_demo WHERE id = :1", [1])
    result = cursor.fetchone()
    tags_obj, scores_obj = result
    
    # Work with VARRAY
    print("Tags:")
    for i in range(tags_obj.size()):
        print(f"  {tags_obj.getelement(i)}")
    
    # Convert to list
    tags_list = tags_obj.aslist()
    print(f"Tags as list: {tags_list}")
    
    # Work with nested table
    print("Scores:")
    for i in range(scores_obj.size()):
        print(f"  Score {i}: {scores_obj.getelement(i)}")
    
    # Add new score
    scores_obj.append(96)
    
    # Update in database
    cursor.execute("""
        UPDATE collections_demo SET scores = :1 WHERE id = :2
    """, [scores_obj, 1])
    
    connection.commit()

connection.close()

import oracledb

connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")

# Create complex object hierarchy
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE OR REPLACE TYPE phone_type AS OBJECT (
            area_code VARCHAR2(3),
            number VARCHAR2(8),
            type VARCHAR2(10)
        )
    """)
    
    cursor.execute("""
        CREATE OR REPLACE TYPE phone_list AS TABLE OF phone_type
    """)
    
    cursor.execute("""
        CREATE OR REPLACE TYPE contact_type AS OBJECT (
            name VARCHAR2(100),
            email VARCHAR2(100),
            phones phone_list
        )
    """)
    
    cursor.execute("""
        CREATE TABLE contacts (
            id NUMBER PRIMARY KEY,
            contact contact_type
        ) NESTED TABLE contact.phones STORE AS contact_phones_tab
    """)

# Get types
contact_type = connection.gettype("CONTACT_TYPE")
phone_type = connection.gettype("PHONE_TYPE")
phone_list_type = connection.gettype("PHONE_LIST")

# Create complex object
contact = contact_type.newobject()
contact.NAME = "Jane Smith"
contact.EMAIL = "jane.smith@example.com"

# Create phone list
phones = phone_list_type.newobject()

# Create individual phones
home_phone = phone_type.newobject()
home_phone.AREA_CODE = "555"
home_phone.NUMBER = "1234567"
home_phone.TYPE = "home"

work_phone = phone_type.newobject()
work_phone.AREA_CODE = "555"
work_phone.NUMBER = "7654321"
work_phone.TYPE = "work"

mobile_phone = phone_type.newobject()
mobile_phone.AREA_CODE = "555"
mobile_phone.NUMBER = "9876543"
mobile_phone.TYPE = "mobile"

# Add phones to collection
phones.append(home_phone)
phones.append(work_phone)
phones.append(mobile_phone)

# Assign phones to contact
contact.PHONES = phones

# Insert complex object
with connection.cursor() as cursor:
    cursor.execute("""
        INSERT INTO contacts (id, contact) VALUES (:1, :2)
    """, [1, contact])
    
    connection.commit()

# Query and navigate complex object
with connection.cursor() as cursor:
    cursor.execute("SELECT contact FROM contacts WHERE id = :1", [1])
    result = cursor.fetchone()
    contact_obj = result[0]
    
    print(f"Contact: {contact_obj.NAME}")
    print(f"Email: {contact_obj.EMAIL}")
    print("Phones:")
    
    phones_obj = contact_obj.PHONES
    for i in range(phones_obj.size()):
        phone = phones_obj.getelement(i)
        print(f"  {phone.TYPE}: ({phone.AREA_CODE}) {phone.NUMBER}")
    
    # Modify collection
    print(f"\nOriginal phone count: {phones_obj.size()}")
    
    # Remove work phone (index 1)
    phones_obj.delete(1)
    print(f"After deletion: {phones_obj.size()}")
    
    # Add new phone
    fax_phone = phone_type.newobject()
    fax_phone.AREA_CODE = "555"
    fax_phone.NUMBER = "1111111"
    fax_phone.TYPE = "fax"
    phones_obj.append(fax_phone)
    
    print(f"After addition: {phones_obj.size()}")
    
    # Update in database
    cursor.execute("""
        UPDATE contacts SET contact = :1 WHERE id = :2
    """, [contact_obj, 1])
    
    connection.commit()

connection.close()

import oracledb

connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")

def inspect_object_type(connection, type_name):
    """Inspect an object type and display its metadata."""
    
    try:
        obj_type = connection.gettype(type_name)
        
        print(f"Object Type: {obj_type.name}")
        print(f"Schema: {obj_type.schema}")
        print(f"Package: {obj_type.package_name or 'None'}")
        print(f"Is Collection: {obj_type.iscollection}")
        
        if obj_type.iscollection:
            print(f"Element Type: {obj_type.element_type.name if obj_type.element_type else 'Unknown'}")
        else:
            print("Attributes:")
            for attr in obj_type.attributes:
                nullable = "NULL" if attr.null_ok else "NOT NULL"
                size_info = f"({attr.max_size})" if attr.max_size else ""
                precision_info = f"({attr.precision},{attr.scale})" if attr.precision else ""
                
                print(f"  {attr.name}: {attr.type.__name__}{size_info}{precision_info} {nullable}")
    
    except Exception as e:
        print(f"Error inspecting {type_name}: {e}")

# Create sample types for inspection
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE OR REPLACE TYPE employee_type AS OBJECT (
            emp_id NUMBER(6) NOT NULL,
            first_name VARCHAR2(50),
            last_name VARCHAR2(50) NOT NULL,
            email VARCHAR2(100),
            hire_date DATE,
            salary NUMBER(8,2),
            commission_pct NUMBER(2,2)
        )
    """)
    
    cursor.execute("""
        CREATE OR REPLACE TYPE dept_employees AS TABLE OF employee_type
    """)

# Inspect different object types
inspect_object_type(connection, "EMPLOYEE_TYPE")
print()
inspect_object_type(connection, "DEPT_EMPLOYEES")

connection.close()

import oracledb
from datetime import date, datetime

connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")

# Create object type
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE OR REPLACE TYPE product_type AS OBJECT (
            product_id NUMBER,
            name VARCHAR2(100),
            price NUMBER(10,2),
            launch_date DATE,
            features SYS.ODCIVARCHAR2LIST
        )
    """)

product_type = connection.gettype("PRODUCT_TYPE")
string_list_type = connection.gettype("SYS.ODCIVARCHAR2LIST")

# Create from Python dictionary
product_data = {
    'PRODUCT_ID': 1001,
    'NAME': 'Laptop Computer',
    'PRICE': 1299.99,
    'LAUNCH_DATE': date(2024, 1, 15),
    'FEATURES': ['16GB RAM', '512GB SSD', 'Backlit Keyboard', 'Webcam']
}

# Method 1: Create object and set attributes
product = product_type.newobject()
product.PRODUCT_ID = product_data['PRODUCT_ID']
product.NAME = product_data['NAME']
product.PRICE = product_data['PRICE']
product.LAUNCH_DATE = product_data['LAUNCH_DATE']

# Create features collection
features = string_list_type.newobject()
features.extend(product_data['FEATURES'])
product.FEATURES = features

# Method 2: Create object with initial value (dictionary)
product2 = product_type.newobject(product_data)

print("Product 1:")
print(f"  ID: {product.PRODUCT_ID}")
print(f"  Name: {product.NAME}")
print(f"  Price: {product.PRICE}")
print(f"  Launch Date: {product.LAUNCH_DATE}")
print(f"  Features: {product.FEATURES.aslist()}")

print("\nProduct 2:")
product2_dict = product2.asdict()
print(f"  As dict: {product2_dict}")

# Use in SQL
with connection.cursor() as cursor:
    cursor.execute("""
        CREATE TABLE products_obj (
            id NUMBER PRIMARY KEY,
            product product_type
        ) NESTED TABLE product.features STORE AS product_features_tab
    """)
    
    cursor.execute("""
        INSERT INTO products_obj (id, product) VALUES (:1, :2)
    """, [1, product])
    
    connection.commit()

connection.close()