Added sample and test case to demonstrate use of universal rowids.

anthony-tuininga · anthony-tuininga · commit 3bc9bd2303ed · 2017-04-11T16:12:06.000-06:00
diff --git a/samples/UniversalRowids.py b/samples/UniversalRowids.py
@@ -0,0 +1,58 @@
+#------------------------------------------------------------------------------
+# Copyright 2017, Oracle and/or its affiliates. All rights reserved.
+#
+# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
+#
+# Portions Copyright 2001-2007, Computronix (Canada) Ltd., Edmonton, Alberta,
+# Canada. All rights reserved.
+#------------------------------------------------------------------------------
+
+#------------------------------------------------------------------------------
+# UniversalRowids.py
+#   This script demonstrates the use of universal rowids. Universal rowids are
+# used to identify rows in index organized tables.
+#
+# This script requires cx_Oracle 6.0 and higher.
+#------------------------------------------------------------------------------
+
+from __future__ import print_function
+
+import cx_Oracle
+import datetime
+
+DATA = [
+    (1, "String #1", datetime.datetime(2017, 4, 4)),
+    (2, "String #2", datetime.datetime(2017, 4, 5)),
+    (3, "A" * 250, datetime.datetime(2017, 4, 6))
+]
+
+# truncate table so sample can be rerun
+connection = cx_Oracle.Connection("cx_Oracle/dev@localhost/orcl")
+cursor = connection.cursor()
+print("Truncating table...")
+cursor.execute("truncate table TestUniversalRowids")
+
+# populate table with a few rows
+print("Populating table...")
+for row in DATA:
+    print("Inserting", row)
+    cursor.execute("insert into TestUniversalRowids values (:1, :2, :3)", row)
+connection.commit()
+
+# fetch the rowids from the table
+rowids = [r for r, in cursor.execute("select rowid from TestUniversalRowids")]
+
+# fetch each of the rows given the rowid
+for rowid in rowids:
+    print("-" * 79)
+    print("Rowid:", rowid)
+    cursor.execute("""
+            select IntCol, StringCol, DateCol
+            from TestUniversalRowids
+            where rowid = :rid""",
+            rid = rowid)
+    intCol, stringCol, dateCol = cursor.fetchone()
+    print("IntCol:", intCol)
+    print("StringCol:", stringCol)
+    print("DateCol:", dateCol)
+
diff --git a/test/SetupTest.sql b/test/SetupTest.sql
@@ -168,6 +168,13 @@ create table cx_Oracle.TestIntervals (
   NullableCol           interval day to second
 );
 
+create table cx_Oracle.TestUniversalRowids (
+  IntCol                number(9) not null,
+  StringCol             varchar2(250) not null,
+  DateCol               date not null,
+  constraint TestUniversalRowids_pk primary key (IntCol, StringCol, DateCol)
+) organization index;
+
 create table cx_Oracle.TestBuildings (
     BuildingId number(9) not null,
     BuildingObj cx_Oracle.udt_Building not null
diff --git a/test/StringVar.py b/test/StringVar.py
@@ -10,6 +10,8 @@
 
 """Module for testing string variables."""
 
+import datetime
+
 class TestStringVar(BaseTestCase):
 
     def setUp(self):
@@ -157,6 +159,33 @@ def testBindAndFetchRowid(self):
                 value = rowid)
         self.assertEqual(self.cursor.fetchall(), [self.dataByKey[3]])
 
+    def testBindAndFetchUniversalRowids(self):
+        "test binding (and fetching) universal rowids"
+        self.cursor.execute("truncate table TestUniversalRowids")
+        data = [
+            (1, "ABC" * 75, datetime.datetime(2017, 4, 11)),
+            (2, "DEF" * 80, datetime.datetime(2017, 4, 12))
+        ]
+        for row in data:
+            self.cursor.execute("""
+                    insert into TestUniversalRowids
+                    values (:1, :2, :3)""", row)
+        self.connection.commit()
+        self.cursor.execute("""
+                select rowid
+                from TestUniversalRowIds
+                order by IntCol""")
+        rowids = [r for r, in self.cursor]
+        fetchedData = []
+        for rowid in rowids:
+            self.cursor.execute("""
+                    select *
+                    from TestUniversalRowids
+                    where rowid = :rid""",
+                    rid = rowid)
+            fetchedData.extend(self.cursor.fetchall())
+        self.assertEqual(fetchedData, data)
+
     def testBindNull(self):
         "test binding in a null"
         self.cursor.execute("""
