dnrops
/
python-cx_Oracle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Modify samples to follow the PEP 8 style guide.

This commit is contained in:
Anthony Tuininga 2020-12-08 11:46:17 -07:00
parent 6b9a7b011b
commit 30979c6a57
50 changed files with 389 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/src/release_notes.rst
View File

@ -26,7 +26,7 @@ Version 8.1 (TBD)
#) Tests can now be run with tox in order to automate testing of the different
environments that are supported.
#) The value of prefetchrows for REF CURSOR variables is now honored.
#) Improved documentation and test suite.
#) Improved documentation, samples and test suite.
Version 8.0.1 (August 2020)

10
samples/AdvancedQueuingNotification.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2018, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -13,7 +13,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import threading
import time
@ -29,9 +29,11 @@ def ProcessMessages(message):
print("Queue name:", message.queueName)
print("Consumer name:", message.consumerName)
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString(), events = True)
connection = cx_Oracle.connect(sample_env.get_main_connect_string(),
events=True)
sub = connection.subscribe(namespace=cx_Oracle.SUBSCR_NAMESPACE_AQ,
name="DEMO_BOOK_QUEUE", callback=ProcessMessages, timeout=300)
name="DEMO_BOOK_QUEUE", callback=ProcessMessages,
timeout=300)
print("Subscription:", sub)
print("--> Connection:", sub.connection)
print("--> Callback:", sub.callback)

8
samples/AppContext.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,7 +17,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
# define constants used throughout the script; adjust as desired
APP_CTX_NAMESPACE = "CLIENTCONTEXT"
@ -27,8 +27,8 @@ APP_CTX_ENTRIES = [
( APP_CTX_NAMESPACE, "ATTR3", "VALUE3" )
]
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString(),
appcontext = APP_CTX_ENTRIES)
connection = cx_Oracle.connect(sample_env.get_main_connect_string(),
appcontext=APP_CTX_ENTRIES)
cursor = connection.cursor()
for namespace, name, value in APP_CTX_ENTRIES:
cursor.execute("select sys_context(:1, :2) from dual", (namespace, name))

22
samples/ArrayDMLRowCounts.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -14,36 +14,36 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# show the number of rows for each parent ID as a means of verifying the
# output from the delete statement
for parentId, count in cursor.execute("""
for parent_id, count in cursor.execute("""
select ParentId, count(*)
from ChildTable
group by ParentId
order by ParentId"""):
print("Parent ID:", parentId, "has", int(count), "rows.")
print("Parent ID:", parent_id, "has", int(count), "rows.")
print()
# delete the following parent IDs only
parentIdsToDelete = [20, 30, 50]
parent_ids_to_delete = [20, 30, 50]
print("Deleting Parent IDs:", parentIdsToDelete)
print("Deleting Parent IDs:", parent_ids_to_delete)
print()
# enable array DML row counts for each iteration executed in executemany()
cursor.executemany("""
delete from ChildTable
where ParentId = :1""",
[(i,) for i in parentIdsToDelete],
[(i,) for i in parent_ids_to_delete],
arraydmlrowcounts = True)
# display the number of rows deleted for each parent ID
rowCounts = cursor.getarraydmlrowcounts()
for parentId, count in zip(parentIdsToDelete, rowCounts):
print("Parent ID:", parentId, "deleted", count, "rows.")
row_counts = cursor.getarraydmlrowcounts()
for parent_id, count in zip(parent_ids_to_delete, row_counts):
print("Parent ID:", parent_id, "deleted", count, "rows.")

20
samples/BatchErrors.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -16,13 +16,13 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# define data to insert
dataToInsert = [
data_to_insert = [
(1016, 10, 'Child B of Parent 10'),
(1017, 10, 'Child C of Parent 10'),
(1018, 20, 'Child D of Parent 20'),
@ -39,14 +39,14 @@ cursor.execute("""
from ChildTable""")
count, = cursor.fetchone()
print("number of rows in child table:", int(count))
print("number of rows to insert:", len(dataToInsert))
print("number of rows to insert:", len(data_to_insert))
# old method: executemany() with data errors results in stoppage after the
# first error takes place; the row count is updated to show how many rows
# actually succeeded
try:
cursor.executemany("insert into ChildTable values (:1, :2, :3)",
dataToInsert)
data_to_insert)
except cx_Oracle.DatabaseError as e:
error, = e.args
print("FAILED with error:", error.message)
@ -64,12 +64,12 @@ connection.rollback()
# new method: executemany() with batch errors enabled (and array DML row counts
# also enabled) results in no immediate error being raised
cursor.executemany("insert into ChildTable values (:1, :2, :3)", dataToInsert,
batcherrors = True, arraydmlrowcounts = True)
cursor.executemany("insert into ChildTable values (:1, :2, :3)",
data_to_insert, batcherrors=True, arraydmlrowcounts=True)
# where errors have taken place, the row count is 0; otherwise it is 1
rowCounts = cursor.getarraydmlrowcounts()
print("Array DML row counts:", rowCounts)
row_counts = cursor.getarraydmlrowcounts()
print("Array DML row counts:", row_counts)
# display the errors that have taken place
errors = cursor.getbatcherrors()

6
samples/BindInsert.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -9,9 +9,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
rows = [ (1, "First" ),
(2, "Second" ),

6
samples/BindQuery.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -13,9 +13,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
sql = 'select * from SampleQueryTab where id = :bvid'

24
samples/BulkAQ.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2019, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,7 +17,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
QUEUE_NAME = "DEMO_RAW_QUEUE"
PAYLOAD_DATA = [
@ -36,7 +36,7 @@ PAYLOAD_DATA = [
]
# connect to database
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# create queue
@ -51,22 +51,22 @@ while queue.deqOne():
# enqueue a few messages
print("Enqueuing messages...")
batchSize = 6
dataToEnq = PAYLOAD_DATA
while dataToEnq:
batchData = dataToEnq[:batchSize]
dataToEnq = dataToEnq[batchSize:]
messages = [connection.msgproperties(payload=d) for d in batchData]
for data in batchData:
batch_size = 6
data_to_enqueue = PAYLOAD_DATA
while data_to_enqueue:
batch_data = data_to_enqueue[:batch_size]
data_to_enqueue = data_to_enqueue[batch_size:]
messages = [connection.msgproperties(payload=d) for d in batch_data]
for data in batch_data:
print(data)
queue.enqMany(messages)
connection.commit()
# dequeue the messages
print("\nDequeuing messages...")
batchSize = 8
batch_size = 8
while True:
messages = queue.deqMany(batchSize)
messages = queue.deqMany(batch_size)
if not messages:
break
for props in messages:

8
samples/CQN.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -18,7 +18,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import time
registered = True
@ -47,7 +47,8 @@ def callback(message):
print("-" * 60)
print("=" * 60)
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString(), events = True)
connection = cx_Oracle.connect(sample_env.get_main_connect_string(),
events=True)
sub = connection.subscribe(callback = callback, timeout = 1800,
qos = cx_Oracle.SUBSCR_QOS_QUERY | cx_Oracle.SUBSCR_QOS_ROWIDS)
print("Subscription:", sub)
@ -64,4 +65,3 @@ print("Registered query:", queryId)
while registered:
print("Waiting for notifications....")
time.sleep(5)

25
samples/CQN2.py
View File

@ -17,7 +17,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import time
registered = True
@ -34,11 +34,11 @@ def callback(message):
if table.rows is None:
print("Too many row changes detected in table", table.name)
continue
numRowsDeleted = 0
num_rows_deleted = 0
print(len(table.rows), "row changes detected in table", table.name)
for row in table.rows:
if row.operation & cx_Oracle.OPCODE_DELETE:
numRowsDeleted += 1
num_rows_deleted += 1
continue
ops = []
if row.operation & cx_Oracle.OPCODE_INSERT:
@ -51,21 +51,22 @@ def callback(message):
from TestTempTable
where rowid = :rid""",
rid=row.rowid)
intCol, = cursor.fetchone()
print(" Row with IntCol", intCol, "was", " and ".join(ops))
if numRowsDeleted > 0:
print(" ", numRowsDeleted, "rows deleted")
int_col, = cursor.fetchone()
print(" Row with IntCol", int_col, "was", " and ".join(ops))
if num_rows_deleted > 0:
print(" ", num_rows_deleted, "rows deleted")
print("=" * 60)
pool = cx_Oracle.SessionPool(SampleEnv.GetMainUser(),
SampleEnv.GetMainPassword(), SampleEnv.GetConnectString(), min=2,
max=5, increment=1, events=True, threaded=True)
pool = cx_Oracle.SessionPool(user=sample_env.get_main_user(),
password=sample_env.get_main_password(),
dsn=sample_env.get_connect_string(), min=2,
max=5, increment=1, events=True, threaded=True)
with pool.acquire() as connection:
sub = connection.subscribe(callback=callback, timeout=1800,
qos=cx_Oracle.SUBSCR_QOS_QUERY | cx_Oracle.SUBSCR_QOS_ROWIDS)
print("Subscription created with ID:", sub.id)
queryId = sub.registerquery("select * from TestTempTable")
print("Registered query with ID:", queryId)
query_id = sub.registerquery("select * from TestTempTable")
print("Registered query with ID:", query_id)
while registered:
print("Waiting for notifications....")

11
samples/CallTimeout.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2019, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -14,9 +14,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
connection.callTimeout = 2000
print("Call timeout set at", connection.callTimeout, "milliseconds...")
@ -26,17 +26,16 @@ today, = cursor.fetchone()
print("Fetch of current date before timeout:", today)
# dbms_session.sleep() replaces dbms_lock.sleep() from Oracle Database 18c
sleepProcName = "dbms_session.sleep" \
sleep_proc_name = "dbms_session.sleep" \
if int(connection.version.split(".")[0]) >= 18 \
else "dbms_lock.sleep"
print("Sleeping...should time out...")
try:
cursor.callproc(sleepProcName, (3,))
cursor.callproc(sleep_proc_name, (3,))
except cx_Oracle.DatabaseError as e:
print("ERROR:", e)
cursor.execute("select sysdate from dual")
today, = cursor.fetchone()
print("Fetch of current date after timeout:", today)

13
samples/ConnectionPool.py
View File

@ -21,17 +21,18 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import threading
# Create a Connection Pool
pool = cx_Oracle.SessionPool(SampleEnv.GetMainUser(),
SampleEnv.GetMainPassword(), SampleEnv.GetConnectString(), min=2,
max=5, increment=1, threaded=True)
pool = cx_Oracle.SessionPool(user=sample_env.get_main_user(),
password=sample_env.get_main_password(),
dsn=sample_env.get_connect_string(), min=2,
max=5, increment=1, threaded=True)
# dbms_session.sleep() replaces dbms_lock.sleep() from Oracle Database 18c
with pool.acquire() as conn:
sleepProcName = "dbms_session.sleep" \
sleep_proc_name = "dbms_session.sleep" \
if int(conn.version.split(".")[0]) >= 18 \
else "dbms_lock.sleep"
@ -62,7 +63,7 @@ def DoALock():
with pool.acquire() as conn:
cursor = conn.cursor()
print("DoALock(): beginning execute...")
cursor.callproc(sleepProcName, (5,))
cursor.callproc(sleep_proc_name, (5,))
print("DoALock(): done execute...")

21
samples/DMLReturningMultipleRows.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2017, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,10 +17,10 @@
import cx_Oracle
import datetime
import SampleEnv
import sample_env
# truncate table first so that script can be rerun
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
print("Truncating table...")
cursor.execute("truncate table TestTempTable")
@ -32,15 +32,14 @@ for i in range(5):
cursor.execute("insert into TestTempTable values (:1, :2)", data)
# now delete them and use DML returning to return the data that was inserted
intCol = cursor.var(int)
stringCol = cursor.var(str)
int_col = cursor.var(int)
string_col = cursor.var(str)
print("Deleting data with DML returning...")
cursor.execute("""
delete from TestTempTable
returning IntCol, StringCol into :intCol, :stringCol""",
intCol = intCol,
stringCol = stringCol)
returning IntCol, StringCol into :int_col, :string_col""",
int_col=int_col,
string_col=string_col)
print("Data returned:")
for intVal, stringVal in zip(intCol.getvalue(), stringCol.getvalue()):
print(tuple([intVal, stringVal]))
for int_val, string_val in zip(int_col.getvalue(), string_col.getvalue()):
print(tuple([int_val, string_val]))

7
samples/DRCP.py
View File

@ -33,12 +33,11 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
conn = cx_Oracle.connect(SampleEnv.GetDrcpConnectString(), cclass = "PYCLASS",
purity = cx_Oracle.ATTR_PURITY_SELF)
conn = cx_Oracle.connect(sample_env.get_drcp_connect_string(),
cclass="PYCLASS", purity=cx_Oracle.ATTR_PURITY_SELF)
cursor = conn.cursor()
print("Performing query using DRCP...")
for row in cursor.execute("select * from TestNumbers order by IntCol"):
print(row)

8
samples/DatabaseChangeNotification.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -18,7 +18,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import time
registered = True
@ -44,7 +44,8 @@ def callback(message):
print("-" * 60)
print("=" * 60)
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString(), events = True)
connection = cx_Oracle.connect(sample_env.get_main_connect_string(),
events=True)
sub = connection.subscribe(callback = callback, timeout = 1800,
qos = cx_Oracle.SUBSCR_QOS_ROWIDS)
print("Subscription:", sub)
@ -61,4 +62,3 @@ sub.registerquery("select * from TestTempTable")
while registered:
print("Waiting for notifications....")
time.sleep(5)

5
samples/DatabaseShutdown.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2017, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -19,7 +19,7 @@
import cx_Oracle
# need to connect as SYSDBA or SYSOPER
connection = cx_Oracle.connect("/", mode = cx_Oracle.SYSDBA)
connection = cx_Oracle.connect("/", mode=cx_Oracle.SYSDBA)
# first shutdown() call must specify the mode, if DBSHUTDOWN_ABORT is used,
# there is no need for any of the other steps
@ -32,4 +32,3 @@ cursor.execute("alter database dismount")
# perform the final shutdown call
connection.shutdown(mode = cx_Oracle.DBSHUTDOWN_FINAL)

7
samples/DatabaseStartup.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2017, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -20,12 +20,11 @@ import cx_Oracle
# the connection must be in PRELIM_AUTH mode
connection = cx_Oracle.connect("/",
mode = cx_Oracle.SYSDBA | cx_Oracle.PRELIM_AUTH)
mode=cx_Oracle.SYSDBA | cx_Oracle.PRELIM_AUTH)
connection.startup()
# the following statements must be issued in normal SYSDBA mode
connection = cx_Oracle.connect("/", mode = cx_Oracle.SYSDBA)
connection = cx_Oracle.connect("/", mode=cx_Oracle.SYSDBA)
cursor = connection.cursor()
cursor.execute("alter database mount")
cursor.execute("alter database open")

4
samples/DbmsOutput.py
View File

@ -9,9 +9,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# enable DBMS_OUTPUT

16
samples/DropSamples.py
View File

@ -9,16 +9,16 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
def DropSamples(conn):
def drop_samples(conn):
print("Dropping sample schemas and edition...")
SampleEnv.RunSqlScript(conn, "DropSamples",
main_user = SampleEnv.GetMainUser(),
edition_user = SampleEnv.GetEditionUser(),
edition_name = SampleEnv.GetEditionName())
sample_env.run_sql_script(conn, "DropSamples",
main_user=sample_env.get_main_user(),
edition_user=sample_env.get_edition_user(),
edition_name=sample_env.get_edition_name())
if __name__ == "__main__":
conn = cx_Oracle.connect(SampleEnv.GetAdminConnectString())
DropSamples(conn)
conn = cx_Oracle.connect(sample_env.get_admin_connect_string())
drop_samples(conn)
print("Done.")

41
samples/Editioning.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -18,14 +18,14 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import os
# connect to the editions user and create a procedure
editionConnectString = SampleEnv.GetEditionConnectString()
connection = cx_Oracle.connect(editionConnectString)
print("Edition should be None, actual value is:",
repr(connection.edition))
edition_connect_string = sample_env.get_edition_connect_string()
edition_name = sample_env.get_edition_name()
connection = cx_Oracle.connect(edition_connect_string)
print("Edition should be None, actual value is:", repr(connection.edition))
cursor = connection.cursor()
cursor.execute("""
create or replace function TestEditions return varchar2 as
@ -34,12 +34,12 @@ cursor.execute("""
end;""")
result = cursor.callfunc("TestEditions", str)
print("Function should return 'Base Procedure', actually returns:",
repr(result))
repr(result))
# next, change the edition and recreate the procedure in the new edition
cursor.execute("alter session set edition = %s" % SampleEnv.GetEditionName())
print("Edition should be", repr(SampleEnv.GetEditionName().upper()),
"actual value is:", repr(connection.edition))
cursor.execute("alter session set edition = %s" % edition_name)
print("Edition should be", repr(edition_name.upper()),
"actual value is:", repr(connection.edition))
cursor.execute("""
create or replace function TestEditions return varchar2 as
begin
@ -47,30 +47,29 @@ cursor.execute("""
end;""")
result = cursor.callfunc("TestEditions", str)
print("Function should return 'Edition 1 Procedure', actually returns:",
repr(result))
repr(result))
# next, change the edition back to the base edition and demonstrate that the
# original function is being called
cursor.execute("alter session set edition = ORA$BASE")
result = cursor.callfunc("TestEditions", str)
print("Function should return 'Base Procedure', actually returns:",
repr(result))
repr(result))
# the edition can be set upon connection
connection = cx_Oracle.connect(editionConnectString,
edition = SampleEnv.GetEditionName().upper())
connection = cx_Oracle.connect(edition_connect_string,
edition=edition_name.upper())
cursor = connection.cursor()
result = cursor.callfunc("TestEditions", str)
print("Function should return 'Edition 1 Procedure', actually returns:",
repr(result))
repr(result))
# it can also be set via the environment variable ORA_EDITION
os.environ["ORA_EDITION"] = SampleEnv.GetEditionName().upper()
connection = cx_Oracle.connect(editionConnectString)
print("Edition should be", repr(SampleEnv.GetEditionName().upper()),
"actual value is:", repr(connection.edition))
os.environ["ORA_EDITION"] = edition_name.upper()
connection = cx_Oracle.connect(edition_connect_string)
print("Edition should be", repr(edition_name.upper()),
"actual value is:", repr(connection.edition))
cursor = connection.cursor()
result = cursor.callfunc("TestEditions", str)
print("Function should return 'Edition 1 Procedure', actually returns:",
repr(result))
repr(result))

11
samples/GenericRowFactory.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -11,7 +11,7 @@
import collections
import cx_Oracle
import SampleEnv
import sample_env
class Connection(cx_Oracle.Connection):
@ -22,9 +22,9 @@ class Connection(cx_Oracle.Connection):
class Cursor(cx_Oracle.Cursor):
def execute(self, statement, args = None):
prepareNeeded = (self.statement != statement)
prepare_needed = (self.statement != statement)
result = super(Cursor, self).execute(statement, args or [])
if prepareNeeded:
if prepare_needed:
description = self.description
if description:
names = [d[0] for d in description]
@ -33,7 +33,7 @@ class Cursor(cx_Oracle.Cursor):
# create new subclassed connection and cursor
connection = Connection(SampleEnv.GetMainConnectString())
connection = Connection(sample_env.get_main_connect_string())
cursor = connection.cursor()
# the names are now available directly for each query executed
@ -44,4 +44,3 @@ print()
for row in cursor.execute("select ChildId, Description from ChildTable"):
print(row.CHILDID, "->", row.DESCRIPTION)
print()

11
samples/ImplicitResults.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,9 +17,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# use PL/SQL block to return two cursors
@ -42,9 +42,8 @@ cursor.execute("""
end;""")
# display results
for ix, resultSet in enumerate(cursor.getimplicitresults()):
for ix, result_set in enumerate(cursor.getimplicitresults()):
print("Result Set #" + str(ix + 1))
for row in resultSet:
for row in result_set:
print(row)
print()

21
samples/InsertGeometry.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -16,18 +16,18 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
# create and populate Oracle objects
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
typeObj = connection.gettype("MDSYS.SDO_GEOMETRY")
elementInfoTypeObj = connection.gettype("MDSYS.SDO_ELEM_INFO_ARRAY")
ordinateTypeObj = connection.gettype("MDSYS.SDO_ORDINATE_ARRAY")
obj = typeObj.newobject()
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
type_obj = connection.gettype("MDSYS.SDO_GEOMETRY")
element_info_type_obj = connection.gettype("MDSYS.SDO_ELEM_INFO_ARRAY")
ordinate_type_obj = connection.gettype("MDSYS.SDO_ORDINATE_ARRAY")
obj = type_obj.newobject()
obj.SDO_GTYPE = 2003
obj.SDO_ELEM_INFO = elementInfoTypeObj.newobject()
obj.SDO_ELEM_INFO = element_info_type_obj.newobject()
obj.SDO_ELEM_INFO.extend([1, 1003, 3])
obj.SDO_ORDINATES = ordinateTypeObj.newobject()
obj.SDO_ORDINATES = ordinate_type_obj.newobject()
obj.SDO_ORDINATES.extend([1, 1, 5, 7])
print("Created object", obj)
@ -50,7 +50,6 @@ if count == 0:
print("Removing any existing rows...")
cursor.execute("delete from TestGeometry")
print("Adding row to table...")
cursor.execute("insert into TestGeometry values (1, :obj)", obj = obj)
cursor.execute("insert into TestGeometry values (1, :obj)", obj=obj)
connection.commit()
print("Success!")

6
samples/LastRowid.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2019, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -10,9 +10,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
row1 = [1, "First"]
row2 = [2, "Second"]

4
samples/MultiConsumerAQ.py
View File

@ -17,7 +17,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
QUEUE_NAME = "DEMO_RAW_QUEUE_MULTI"
PAYLOAD_DATA = [
@ -28,7 +28,7 @@ PAYLOAD_DATA = [
]
# connect to database
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# create queue

12
samples/ObjectAQ.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,7 +17,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import decimal
BOOK_TYPE_NAME = "UDT_BOOK"
@ -30,12 +30,12 @@ BOOK_DATA = [
]
# connect to database
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# create queue
booksType = connection.gettype(BOOK_TYPE_NAME)
queue = connection.queue(QUEUE_NAME, booksType)
books_type = connection.gettype(BOOK_TYPE_NAME)
queue = connection.queue(QUEUE_NAME, books_type)
queue.deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
queue.deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
@ -47,7 +47,7 @@ while queue.deqOne():
# enqueue a few messages
print("Enqueuing messages...")
for title, authors, price in BOOK_DATA:
book = booksType.newobject()
book = books_type.newobject()
book.TITLE = title
book.AUTHORS = authors
book.PRICE = price

11
samples/PLSQLCollection.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -14,14 +14,14 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
# create new empty object of the correct type
# note the use of a PL/SQL type defined in a package
typeObj = connection.gettype("PKG_DEMO.UDT_STRINGLIST")
obj = typeObj.newobject()
type_obj = connection.gettype("PKG_DEMO.UDT_STRINGLIST")
obj = type_obj.newobject()
# call the stored procedure which will populate the object
cursor = connection.cursor()
@ -44,4 +44,3 @@ print()
print("Values of collection as dictionary:")
print(obj.asdict())
print()

7
samples/PLSQLFunction.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -9,11 +9,10 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
res = cursor.callfunc('myfunc', int, ('abc', 2))
print(res)

7
samples/PLSQLProcedure.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -10,12 +10,11 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
myvar = cursor.var(int)
cursor.callproc('myproc', (123, myvar))
print(myvar.getvalue())

11
samples/PLSQLRecord.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -12,16 +12,16 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import datetime
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
# create new object of the correct type
# note the use of a PL/SQL record defined in a package
# a table record identified by TABLE%ROWTYPE can also be used
typeObj = connection.gettype("PKG_DEMO.UDT_DEMORECORD")
obj = typeObj.newobject()
type_obj = connection.gettype("PKG_DEMO.UDT_DEMORECORD")
obj = type_obj.newobject()
obj.NUMBERVALUE = 6
obj.STRINGVALUE = "Test String"
obj.DATEVALUE = datetime.datetime(2016, 5, 28)
@ -44,4 +44,3 @@ print("STRINGVALUE ->", obj.STRINGVALUE)
print("DATEVALUE ->", obj.DATEVALUE)
print("BOOLEANVALUE ->", obj.BOOLEANVALUE)
print()

4
samples/Query.py
View File

@ -9,9 +9,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
sql = """
select * from SampleQueryTab

7
samples/QueryArraysize.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -12,9 +12,9 @@
import time
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
start = time.time()
@ -26,4 +26,3 @@ res = cursor.fetchall()
elapsed = (time.time() - start)
print("Retrieved", len(res), "rows in", elapsed, "seconds")

6
samples/RawAQ.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2019, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -16,7 +16,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
QUEUE_NAME = "DEMO_RAW_QUEUE"
PAYLOAD_DATA = [
@ -27,7 +27,7 @@ PAYLOAD_DATA = [
]
# connect to database
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# create queue

6
samples/RefCursor.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2018, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -8,9 +8,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
ref_cursor = connection.cursor()

35
samples/ReturnLobsAsStrings.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -19,34 +19,34 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
def OutputTypeHandler(cursor, name, defaultType, size, precision, scale):
if defaultType == cx_Oracle.CLOB:
return cursor.var(cx_Oracle.LONG_STRING, arraysize = cursor.arraysize)
if defaultType == cx_Oracle.BLOB:
return cursor.var(cx_Oracle.LONG_BINARY, arraysize = cursor.arraysize)
def output_type_handler(cursor, name, default_type, size, precision, scale):
if default_type == cx_Oracle.CLOB:
return cursor.var(cx_Oracle.LONG_STRING, arraysize=cursor.arraysize)
if default_type == cx_Oracle.BLOB:
return cursor.var(cx_Oracle.LONG_BINARY, arraysize=cursor.arraysize)
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection.outputtypehandler = OutputTypeHandler
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
connection.outputtypehandler = output_type_handler
cursor = connection.cursor()
# add some data to the tables
print("Populating tables with data...")
cursor.execute("truncate table TestClobs")
cursor.execute("truncate table TestBlobs")
longString = ""
long_string = ""
for i in range(10):
char = chr(ord('A') + i)
longString += char * 25000
long_string += char * 25000
# uncomment the line below for cx_Oracle 5.3 and earlier
# cursor.setinputsizes(None, cx_Oracle.LONG_STRING)
cursor.execute("insert into TestClobs values (:1, :2)",
(i + 1, "STRING " + longString))
(i + 1, "STRING " + long_string))
# uncomment the line below for cx_Oracle 5.3 and earlier
# cursor.setinputsizes(None, cx_Oracle.LONG_BINARY)
cursor.execute("insert into TestBlobs values (:1, :2)",
(i + 1, longString.encode("ascii")))
(i + 1, long_string.encode("ascii")))
connection.commit()
# fetch the data and show the results
@ -57,8 +57,8 @@ cursor.execute("""
ClobCol
from TestClobs
order by IntCol""")
for intCol, value in cursor:
print("Row:", intCol, "string of length", len(value))
for int_col, value in cursor:
print("Row:", int_col, "string of length", len(value))
print()
print("BLOBS returned as bytes")
cursor.execute("""
@ -67,6 +67,5 @@ cursor.execute("""
BlobCol
from TestBlobs
order by IntCol""")
for intCol, value in cursor:
print("Row:", intCol, "string of length", value and len(value) or 0)
for int_col, value in cursor:
print("Row:", int_col, "string of length", value and len(value) or 0)

11
samples/ReturnNumbersAsDecimals.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2017, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -16,16 +16,15 @@
import cx_Oracle
import decimal
import SampleEnv
import sample_env
def OutputTypeHandler(cursor, name, defaultType, size, precision, scale):
def output_type_handler(cursor, name, defaultType, size, precision, scale):
if defaultType == cx_Oracle.NUMBER:
return cursor.var(decimal.Decimal, arraysize = cursor.arraysize)
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection.outputtypehandler = OutputTypeHandler
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
connection.outputtypehandler = output_type_handler
cursor = connection.cursor()
cursor.execute("select * from TestNumbers")
for row in cursor:
print("Row:", row)

7
samples/RowsAsInstance.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,7 +17,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
class Test(object):
@ -26,7 +26,7 @@ class Test(object):
self.b = b
self.c = c
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# change this to False if you want to create the table yourself using SQL*Plus
@ -55,4 +55,3 @@ cursor.rowfactory = Test
print("Rows:")
for row in cursor:
print("a = %s, b = %s, c = %s" % (row.a, row.b, row.c))

7
samples/ScrollableCursors.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,9 +17,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
# show all of the rows available in the table
cursor = connection.cursor()
@ -68,4 +68,3 @@ cursor.scroll(-4)
print("SKIP BACK 4 ROWS")
print(cursor.fetchone())
print()

22
samples/SessionCallback.py
View File

@ -21,7 +21,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
# define a dictionary of NLS_DATE_FORMAT formats supported by this sample
SUPPORTED_FORMATS = {
@ -42,18 +42,18 @@ SUPPORTED_KEYS = {
}
# define session callback
def InitSession(conn, requestedTag):
def init_session(conn, requested_tag):
# display the requested and actual tags
print("InitSession(): requested tag=%r, actual tag=%r" % \
(requestedTag, conn.tag))
print("init_session(): requested tag=%r, actual tag=%r" % \
(requested_tag, conn.tag))
# tags are expected to be in the form "key1=value1;key2=value2"
# in this example, they are used to set NLS parameters and the tag is
# parsed to validate it
if requestedTag is not None:
if requested_tag is not None:
stateParts = []
for directive in requestedTag.split(";"):
for directive in requested_tag.split(";"):
parts = directive.split("=")
if len(parts) != 2:
raise ValueError("Tag must contain key=value pairs")
@ -74,13 +74,15 @@ def InitSession(conn, requestedTag):
# assign the requested tag to the connection so that when the connection
# is closed, it will automatically be retagged; note that if the requested
# tag is None (no tag was requested) this has no effect
conn.tag = requestedTag
conn.tag = requested_tag
# create pool with session callback defined
pool = cx_Oracle.SessionPool(SampleEnv.GetMainUser(),
SampleEnv.GetMainPassword(), SampleEnv.GetConnectString(), min=2,
max=5, increment=1, threaded=True, sessionCallback=InitSession)
pool = cx_Oracle.SessionPool(user=sample_env.get_main_user(),
password=sample_env.get_main_password(),
dsn=sample_env.get_connect_string(), min=2, max=5,
increment=1, threaded=True,
sessionCallback=init_session)
# acquire session without specifying a tag; since the session returned is
# newly created, the callback will be invoked but since there is no tag

11
samples/SessionCallbackPLSQL.py
View File

@ -23,13 +23,14 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
# create pool with session callback defined
pool = cx_Oracle.SessionPool(SampleEnv.GetMainUser(),
SampleEnv.GetMainPassword(), SampleEnv.GetConnectString(), min=2,
max=5, increment=1, threaded=True,
sessionCallback="pkg_SessionCallback.TheCallback")
pool = cx_Oracle.SessionPool(user=sample_env.get_main_user(),
password=sample_env.get_main_password(),
dsn=sample_env.get_connect_string(), min=2, max=5,
increment=1, threaded=True,
sessionCallback="pkg_SessionCallback.TheCallback")
# truncate table logging calls to PL/SQL session callback
with pool.acquire() as conn:

18
samples/SetupSamples.py
View File

@ -12,22 +12,22 @@
import cx_Oracle
import SampleEnv
import sample_env
import DropSamples
# connect as administrative user (usually SYSTEM or ADMIN)
conn = cx_Oracle.connect(SampleEnv.GetAdminConnectString())
conn = cx_Oracle.connect(sample_env.get_admin_connect_string())
# drop existing users and editions, if applicable
DropSamples.DropSamples(conn)
DropSamples.drop_samples(conn)
# create sample schema and edition
print("Creating sample schemas and edition...")
SampleEnv.RunSqlScript(conn, "SetupSamples",
main_user = SampleEnv.GetMainUser(),
main_password = SampleEnv.GetMainPassword(),
edition_user = SampleEnv.GetEditionUser(),
edition_password = SampleEnv.GetEditionPassword(),
edition_name = SampleEnv.GetEditionName())
sample_env.run_sql_script(conn, "SetupSamples",
main_user=sample_env.get_main_user(),
main_password=sample_env.get_main_password(),
edition_user=sample_env.get_edition_user(),
edition_password=sample_env.get_edition_password(),
edition_name=sample_env.get_edition_name())
print("Done.")

19
samples/ShardingNumberKey.py
View File

@ -16,19 +16,20 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
pool = cx_Oracle.SessionPool(SampleEnv.GetMainUser(),
SampleEnv.GetMainPassword(), SampleEnv.GetConnectString(), min=1,
max=5, increment=1)
pool = cx_Oracle.SessionPool(user=sample_env.get_main_user(),
password=sample_env.get_main_password(),
dsn=sample_env.get_connect_string(), min=1, max=5,
increment=1)
def ConnectAndDisplay(shardingKey):
print("Connecting with sharding key:", shardingKey)
with pool.acquire(shardingkey=[shardingKey]) as conn:
def connect_and_display(sharding_key):
print("Connecting with sharding key:", sharding_key)
with pool.acquire(shardingkey=[sharding_key]) as conn:
cursor = conn.cursor()
cursor.execute("select sys_context('userenv', 'db_name') from dual")
name, = cursor.fetchone()
print("--> connected to database", name)
ConnectAndDisplay(100)
ConnectAndDisplay(167)
connect_and_display(100)
connect_and_display(167)

24
samples/SodaBasic.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2018, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -13,9 +13,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
# The general recommendation for simple SODA usage is to enable autocommit
connection.autocommit = True
@ -27,12 +27,17 @@ soda = connection.getSodaDatabase()
# This will open an existing collection, if the name is already in use.
collection = soda.createCollection("mycollection")
indexSpec = { 'name': 'CITY_IDX',
'fields': [ {
'path': 'address.city',
'datatype': 'string',
'order': 'asc' } ] }
collection.createIndex(indexSpec)
index_spec = {
'name': 'CITY_IDX',
'fields': [
{
'path': 'address.city',
'datatype': 'string',
'order': 'asc'
}
]
}
collection.createIndex(index_spec)
# Insert a document.
# A system generated key is created by default.
@ -85,4 +90,3 @@ print('Collection has', c, 'documents')
# Drop the collection
if collection.drop():
print('Collection was dropped')

12
samples/SodaBulkInsert.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2019, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -13,9 +13,9 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
# the general recommendation for simple SODA usage is to enable autocommit
connection.autocommit = True
@ -30,7 +30,7 @@ collection = soda.createCollection("SodaBulkInsert")
collection.find().remove()
# define some documents that will be stored
inDocs = [
in_docs = [
dict(name="Sam", age=8),
dict(name="George", age=46),
dict(name="Bill", age=35),
@ -40,8 +40,8 @@ inDocs = [
]
# perform bulk insert
resultDocs = collection.insertManyAndGet(inDocs)
for doc in resultDocs:
result_docs = collection.insertManyAndGet(in_docs)
for doc in result_docs:
print("Inserted SODA document with key", doc.key)
print()

43
samples/SpatialToGeoPandas.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2018, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -22,13 +22,13 @@
# dependencies (see http://geopandas.org/install.html).
#------------------------------------------------------------------------------
import SampleEnv
import sample_env
import cx_Oracle
from shapely.wkb import loads
import geopandas as gpd
# create Oracle connection and cursor objects
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
# enable autocommit to avoid the additional round trip to the database to
@ -38,10 +38,10 @@ connection.autocommit = True
# define output type handler to fetch LOBs, avoiding the second round trip to
# the database to read the LOB contents
def OutputTypeHandler(cursor, name, defaultType, size, precision, scale):
if defaultType == cx_Oracle.BLOB:
return cursor.var(cx_Oracle.LONG_BINARY, arraysize = cursor.arraysize)
connection.outputtypehandler = OutputTypeHandler
def output_type_handler(cursor, name, default_type, size, precision, scale):
if default_type == cx_Oracle.BLOB:
return cursor.var(cx_Oracle.LONG_BINARY, arraysize=cursor.arraysize)
connection.outputtypehandler = output_type_handler
# drop and create table
print("Dropping and creating table...")
@ -60,23 +60,23 @@ cursor.execute("""
)""")
# acquire types used for creating SDO_GEOMETRY objects
typeObj = connection.gettype("MDSYS.SDO_GEOMETRY")
elementInfoTypeObj = connection.gettype("MDSYS.SDO_ELEM_INFO_ARRAY")
ordinateTypeObj = connection.gettype("MDSYS.SDO_ORDINATE_ARRAY")
type_obj = connection.gettype("MDSYS.SDO_GEOMETRY")
element_info_type_obj = connection.gettype("MDSYS.SDO_ELEM_INFO_ARRAY")
ordinate_type_obj = connection.gettype("MDSYS.SDO_ORDINATE_ARRAY")
# define function for creating an SDO_GEOMETRY object
def CreateGeometryObj(*ordinates):
geometry = typeObj.newobject()
def create_geometry_obj(*ordinates):
geometry = type_obj.newobject()
geometry.SDO_GTYPE = 2003
geometry.SDO_SRID = 8307
geometry.SDO_ELEM_INFO = elementInfoTypeObj.newobject()
geometry.SDO_ELEM_INFO = element_info_type_obj.newobject()
geometry.SDO_ELEM_INFO.extend([1, 1003, 1])
geometry.SDO_ORDINATES = ordinateTypeObj.newobject()
geometry.SDO_ORDINATES = ordinate_type_obj.newobject()
geometry.SDO_ORDINATES.extend(ordinates)
return geometry
# create SDO_GEOMETRY objects for three adjacent states in the USA
geometryNevada = CreateGeometryObj(-114.052025, 37.103989, -114.049797,
geometry_nevada = create_geometry_obj(-114.052025, 37.103989, -114.049797,
37.000423, -113.484375, 37, -112.898598, 37.000401,-112.539604,
37.000683, -112, 37.000977, -111.412048, 37.001514, -111.133018,
37.00079,-110.75, 37.003201, -110.5, 37.004265, -110.469505, 36.998001,
@ -96,7 +96,7 @@ geometryNevada = CreateGeometryObj(-114.052025, 37.103989, -114.049797,
-114.049339, 38.572968, -114.049095, 38.14864, -114.0476,
37.80946,-114.05098, 37.746284, -114.051666, 37.604805, -114.052025,
37.103989)
geometryWyoming = CreateGeometryObj(-111.045815, 41.251774, -111.045982,
geometry_wyoming = create_geometry_obj(-111.045815, 41.251774, -111.045982,
40.998013, -110.5, 40.994801, -110.047768, 40.997696, -110, 40.997398,
-109.534233, 40.998184, -109.2313, 41.002102, -109.0494, 41.000702,
-108.525368, 40.999634, -107.917793, 41.002071, -107.317177, 41.002956,
@ -120,7 +120,7 @@ geometryWyoming = CreateGeometryObj(-111.045815, 41.251774, -111.045982,
-111.043846, 43.3158, -111.043381, 43.02013, -111.042786, 42.719578,
-111.045967, 42.513187, -111.045944, 42.001633, -111.045097, 41.579899,
-111.045815, 41.251774)
geometryColorado = CreateGeometryObj(-109.045143, 37.375, -109.044571,
geometry_colorado = create_geometry_obj(-109.045143, 37.375, -109.044571,
36.999088, -108.378571, 36.999516, -107.481133, 37, -107.420311, 37,
-106.876701, 37.00013, -106.869209, 36.992416, -106.475639, 36.993748,
-106.006058, 36.995327, -105.717834, 36.995823, -105.220055, 36.995144,
@ -151,9 +151,9 @@ geometryColorado = CreateGeometryObj(-109.045143, 37.375, -109.044571,
# will skip the metadata and indexes.
print("Adding rows to table...")
data = [
('Nevada', geometryNevada),
('Colorado', geometryColorado),
('Wyoming', geometryWyoming)
('Nevada', geometry_nevada),
('Colorado', geometry_colorado),
('Wyoming', geometry_wyoming)
]
cursor.executemany('insert into TestStates values (:state, :obj)', data)
@ -168,7 +168,7 @@ cursor.executemany('insert into TestStates values (:state, :obj)', data)
cursor.execute("""
SELECT state, sdo_util.to_wkbgeometry(geometry)
FROM TestStates""")
gdf = gpd.GeoDataFrame(cursor.fetchall(), columns = ['state', 'wkbgeometry'])
gdf = gpd.GeoDataFrame(cursor.fetchall(), columns=['state', 'wkbgeometry'])
# create GeoSeries to replace the WKB geometry column
gdf['geometry'] = gpd.GeoSeries(gdf['wkbgeometry'].apply(lambda x: loads(x)))
@ -183,4 +183,3 @@ print(gdf)
print()
print("GeoPandas combining the 3 geometries into a single geometry...")
print(gdf.unary_union)

13
samples/Subclassing.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -11,7 +11,7 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
# sample subclassed connection which overrides the constructor (so no
# parameters are required) and the cursor() method (so that the subclassed
@ -19,9 +19,9 @@ import SampleEnv
class Connection(cx_Oracle.Connection):
def __init__(self):
connectString = SampleEnv.GetMainConnectString()
connect_string = sample_env.get_main_connect_string()
print("CONNECT to database")
return super(Connection, self).__init__(connectString)
return super(Connection, self).__init__(connect_string)
def cursor(self):
return Cursor(self)
@ -34,8 +34,8 @@ class Cursor(cx_Oracle.Cursor):
def execute(self, statement, args):
print("EXECUTE", statement)
print("ARGS:")
for argIndex, arg in enumerate(args):
print(" ", argIndex + 1, "=>", repr(arg))
for arg_index, arg in enumerate(args):
print(" ", arg_index + 1, "=>", repr(arg))
return super(Cursor, self).execute(statement, args)
def fetchone(self):
@ -51,4 +51,3 @@ cursor = connection.cursor()
cursor.execute("select count(*) from ChildTable where ParentId = :1", (30,))
count, = cursor.fetchone()
print("COUNT:", int(count))

16
samples/TransactionGuard.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -34,16 +34,16 @@
#------------------------------------------------------------------------------
import cx_Oracle
import SampleEnv
import sample_env
import sys
# constants
CONNECT_STRING = "localhost/orcl-tg"
# create transaction and generate a recoverable error
pool = cx_Oracle.SessionPool(SampleEnv.GetMainUser(),
SampleEnv.GetMainPassword(), CONNECT_STRING, min=1,
max=9, increment=2)
pool = cx_Oracle.SessionPool(user=sample_env.get_main_user(),
password=sample_env.get_main_password(),
dsn=CONNECT_STRING, min=1, max=9, increment=2)
connection = pool.acquire()
cursor = connection.cursor()
cursor.execute("""
@ -53,7 +53,7 @@ cursor.execute("""
insert into TestTempTable
values (1, null)""")
input("Please kill %s session now. Press ENTER when complete." % \
SampleEnv.GetMainUser())
sample_env.get_main_user())
try:
connection.commit() # this should fail
sys.exit("Session was not killed. Terminating.")
@ -69,8 +69,8 @@ pool.drop(connection)
# check if previous transaction completed
connection = pool.acquire()
cursor = connection.cursor()
args = (cx_Oracle.Binary(ltxid), cursor.var(bool), cursor.var(bool))
_, committed, completed = cursor.callproc("dbms_app_cont.get_ltxid_outcome",
(cx_Oracle.Binary(ltxid), cursor.var(bool), cursor.var(bool)))
args)
print("Failed transaction was committed:", committed)
print("Failed call was completed:", completed)

46
samples/TypeHandlers.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -19,46 +19,47 @@
import cx_Oracle
import datetime
import SampleEnv
import sample_env
con = cx_Oracle.connect(SampleEnv.GetMainConnectString())
objType = con.gettype("UDT_BUILDING")
con = cx_Oracle.connect(sample_env.get_main_connect_string())
obj_type = con.gettype("UDT_BUILDING")
class Building(object):
def __init__(self, buildingId, description, numFloors, dateBuilt):
self.buildingId = buildingId
def __init__(self, building_id, description, num_floors, dateBuilt):
self.building_id = building_id
self.description = description
self.numFloors = numFloors
self.num_floors = num_floors
self.dateBuilt = dateBuilt
def __repr__(self):
return "<Building %s: %s>" % (self.buildingId, self.description)
return "<Building %s: %s>" % (self.building_id, self.description)
def BuildingInConverter(value):
obj = objType.newobject()
obj.BUILDINGID = value.buildingId
def building_in_converter(value):
obj = obj_type.newobject()
obj.BUILDINGID = value.building_id
obj.DESCRIPTION = value.description
obj.NUMFLOORS = value.numFloors
obj.NUMFLOORS = value.num_floors
obj.DATEBUILT = value.dateBuilt
return obj
def BuildingOutConverter(obj):
def building_out_converter(obj):
return Building(int(obj.BUILDINGID), obj.DESCRIPTION, int(obj.NUMFLOORS),
obj.DATEBUILT)
def InputTypeHandler(cursor, value, numElements):
def input_type_handler(cursor, value, numElements):
if isinstance(value, Building):
return cursor.var(cx_Oracle.OBJECT, arraysize = numElements,
inconverter = BuildingInConverter, typename = objType.name)
inconverter = building_in_converter, typename = obj_type.name)
def OutputTypeHandler(cursor, name, defaultType, size, precision, scale):
if defaultType == cx_Oracle.OBJECT:
return cursor.var(cx_Oracle.OBJECT, arraysize = cursor.arraysize,
outconverter = BuildingOutConverter, typename = objType.name)
def output_type_handler(cursor, name, default_type, size, precision, scale):
if default_type == cx_Oracle.OBJECT:
return cursor.var(cx_Oracle.OBJECT, arraysize=cursor.arraysize,
outconverter=building_out_converter,
typename=obj_type.name)
buildings = [
Building(1, "The First Building", 5, datetime.date(2007, 5, 18)),
@ -67,11 +68,11 @@ buildings = [
]
cur = con.cursor()
cur.inputtypehandler = InputTypeHandler
cur.inputtypehandler = input_type_handler
for building in buildings:
try:
cur.execute("insert into TestBuildings values (:1, :2)",
(building.buildingId, building))
(building.building_id, building))
except cx_Oracle.DatabaseError as e:
error, = e.args
print("CONTEXT:", error.context)
@ -84,9 +85,8 @@ for row in cur.execute("select * from TestBuildings order by BuildingId"):
print()
cur = con.cursor()
cur.outputtypehandler = OutputTypeHandler
cur.outputtypehandler = output_type_handler
print("WITH OUTPUT TYPE HANDLER:")
for row in cur.execute("select * from TestBuildings order by BuildingId"):
print(row)
print()

13
samples/UniversalRowids.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2017, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
@ -17,7 +17,7 @@
import cx_Oracle
import datetime
import SampleEnv
import sample_env
DATA = [
(1, "String #1", datetime.datetime(2017, 4, 4)),
@ -26,7 +26,7 @@ DATA = [
]
# truncate table so sample can be rerun
connection = cx_Oracle.connect(SampleEnv.GetMainConnectString())
connection = cx_Oracle.connect(sample_env.get_main_connect_string())
cursor = connection.cursor()
print("Truncating table...")
cursor.execute("truncate table TestUniversalRowids")
@ -50,8 +50,7 @@ for rowid in rowids:
from TestUniversalRowids
where rowid = :rid""",
rid = rowid)
intCol, stringCol, dateCol = cursor.fetchone()
print("IntCol:", intCol)
print("StringCol:", stringCol)
int_col, string_col, dateCol = cursor.fetchone()
print("IntCol:", int_col)
print("StringCol:", string_col)
print("DateCol:", dateCol)

100
samples/SampleEnv.py → samples/sample_env.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved.
# Copyright (c) 2017, 2020, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
@ -57,93 +57,95 @@ DEFAULT_DRCP_CONNECT_STRING = "localhost/orclpdb1:pooled"
# directly) and then stored so that a value is not requested more than once
PARAMETERS = {}
def GetValue(name, label, defaultValue=""):
def get_value(name, label, default_value=""):
value = PARAMETERS.get(name)
if value is not None:
return value
envName = "CX_ORACLE_SAMPLES_" + name
value = os.environ.get(envName)
env_name = "CX_ORACLE_SAMPLES_" + name
value = os.environ.get(env_name)
if value is None:
if defaultValue:
label += " [%s]" % defaultValue
if default_value:
label += " [%s]" % default_value
label += ": "
if defaultValue:
if default_value:
value = input(label).strip()
else:
value = getpass.getpass(label)
if not value:
value = defaultValue
value = default_value
PARAMETERS[name] = value
return value
def GetMainUser():
return GetValue("MAIN_USER", "Main User Name", DEFAULT_MAIN_USER)
def get_main_user():
return get_value("MAIN_USER", "Main User Name", DEFAULT_MAIN_USER)
def GetMainPassword():
return GetValue("MAIN_PASSWORD", "Password for %s" % GetMainUser())
def get_main_password():
return get_value("MAIN_PASSWORD", "Password for %s" % get_main_user())
def GetEditionUser():
return GetValue("EDITION_USER", "Edition User Name", DEFAULT_EDITION_USER)
def get_edition_user():
return get_value("EDITION_USER", "Edition User Name", DEFAULT_EDITION_USER)
def GetEditionPassword():
return GetValue("EDITION_PASSWORD", "Password for %s" % GetEditionUser())
def get_edition_password():
return get_value("EDITION_PASSWORD",
"Password for %s" % get_edition_user())
def GetEditionName():
return GetValue("EDITION_NAME", "Edition Name", DEFAULT_EDITION_NAME)
def get_edition_name():
return get_value("EDITION_NAME", "Edition Name", DEFAULT_EDITION_NAME)
def GetConnectString():
return GetValue("CONNECT_STRING", "Connect String", DEFAULT_CONNECT_STRING)
def get_connect_string():
return get_value("CONNECT_STRING", "Connect String",
DEFAULT_CONNECT_STRING)
def GetMainConnectString(password=None):
def get_main_connect_string(password=None):
if password is None:
password = GetMainPassword()
return "%s/%s@%s" % (GetMainUser(), password, GetConnectString())
password = get_main_password()
return "%s/%s@%s" % (get_main_user(), password, get_connect_string())
def GetDrcpConnectString():
connectString = GetValue("DRCP_CONNECT_STRING", "DRCP Connect String",
def get_drcp_connect_string():
connectString = get_value("DRCP_CONNECT_STRING", "DRCP Connect String",
DEFAULT_DRCP_CONNECT_STRING)
return "%s/%s@%s" % (GetMainUser(), GetMainPassword(), connectString)
return "%s/%s@%s" % (get_main_user(), get_main_password(), connectString)
def GetEditionConnectString():
def get_edition_connect_string():
return "%s/%s@%s" % \
(GetEditionUser(), GetEditionPassword(), GetConnectString())
(get_edition_user(), get_edition_password(), get_connect_string())
def GetAdminConnectString():
adminUser = GetValue("ADMIN_USER", "Administrative user", "admin")
adminPassword = GetValue("ADMIN_PASSWORD", "Password for %s" % adminUser)
return "%s/%s@%s" % (adminUser, adminPassword, GetConnectString())
def get_admin_connect_string():
admin_user = get_value("ADMIN_USER", "Administrative user", "admin")
admin_password = get_value("ADMIN_PASSWORD", "Password for %s" % admin_user)
return "%s/%s@%s" % (admin_user, admin_password, get_connect_string())
def RunSqlScript(conn, scriptName, **kwargs):
statementParts = []
def run_sql_script(conn, script_name, **kwargs):
statement_parts = []
cursor = conn.cursor()
replaceValues = [("&" + k + ".", v) for k, v in kwargs.items()] + \
[("&" + k, v) for k, v in kwargs.items()]
scriptDir = os.path.dirname(os.path.abspath(sys.argv[0]))
fileName = os.path.join(scriptDir, "sql", scriptName + "Exec.sql")
for line in open(fileName):
replace_values = [("&" + k + ".", v) for k, v in kwargs.items()] + \
[("&" + k, v) for k, v in kwargs.items()]
script_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
file_name = os.path.join(script_dir, "sql", script_name + "Exec.sql")
for line in open(file_name):
if line.strip() == "/":
statement = "".join(statementParts).strip()
statement = "".join(statement_parts).strip()
if statement:
for searchValue, replaceValue in replaceValues:
statement = statement.replace(searchValue, replaceValue)
for search_value, replace_value in replace_values:
statement = statement.replace(search_value, replace_value)
try:
cursor.execute(statement)
except:
print("Failed to execute SQL:", statement)
raise
statementParts = []
statement_parts = []
else:
statementParts.append(line)
statement_parts.append(line)
cursor.execute("""
select name, type, line, position, text
from dba_errors
where owner = upper(:owner)
order by name, type, line, position""",
owner = GetMainUser())
prevName = prevObjType = None
owner = get_main_user())
prev_name = prev_obj_type = None
for name, objType, lineNum, position, text in cursor:
if name != prevName or objType != prevObjType:
if name != prev_name or objType != prev_obj_type:
print("%s (%s)" % (name, objType))
prevName = name
prevObjType = objType
prev_name = name
prev_obj_type = objType
print(" %s/%s %s" % (lineNum, position, text))