forked from oracle/python-cx_Oracle
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtype_input.py
More file actions
85 lines (73 loc) · 2.99 KB
/
type_input.py
File metadata and controls
85 lines (73 loc) · 2.99 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
#------------------------------------------------------------------------------
# type_input.py (Section 6.3)
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
# Copyright (c) 2017, 2021, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------
import cx_Oracle
import db_config
con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)
cur = con.cursor()
# Create table
cur.execute("""begin
execute immediate 'drop table testgeometry';
exception when others then
if sqlcode <> -942 then
raise;
end if;
end;""")
cur.execute("""create table testgeometry (
id number(9) not null,
geometry MDSYS.SDO_GEOMETRY not null)""")
# Create a Python class for an SDO
class mySDO(object):
def __init__(self, gtype, elemInfo, ordinates):
self.gtype = gtype
self.elemInfo = elemInfo
self.ordinates = ordinates
# Get Oracle type information
objType = con.gettype("MDSYS.SDO_GEOMETRY")
elementInfoTypeObj = con.gettype("MDSYS.SDO_ELEM_INFO_ARRAY")
ordinateTypeObj = con.gettype("MDSYS.SDO_ORDINATE_ARRAY")
# Convert a Python object to MDSYS.SDO_GEOMETRY
def SDOInConverter(value):
obj = objType.newobject()
obj.SDO_GTYPE = value.gtype
obj.SDO_ELEM_INFO = elementInfoTypeObj.newobject()
obj.SDO_ELEM_INFO.extend(value.elemInfo)
obj.SDO_ORDINATES = ordinateTypeObj.newobject()
obj.SDO_ORDINATES.extend(value.ordinates)
return obj
def SDOInputTypeHandler(cursor, value, numElements):
if isinstance(value, mySDO):
return cursor.var(cx_Oracle.OBJECT, arraysize = numElements,
inconverter = SDOInConverter, typename = objType.name)
sdo = mySDO(2003, [1, 1003, 3], [1, 1, 5, 7]) # Python object
cur.inputtypehandler = SDOInputTypeHandler
cur.execute("insert into testgeometry values (:1, :2)", (1, sdo))
# Define a function to dump the contents of an Oracle object
def dumpobject(obj, prefix = " "):
if obj.type.iscollection:
print(prefix, "[")
for value in obj.aslist():
if isinstance(value, cx_Oracle.Object):
dumpobject(value, prefix + " ")
else:
print(prefix + " ", repr(value))
print(prefix, "]")
else:
print(prefix, "{")
for attr in obj.type.attributes:
value = getattr(obj, attr.name)
if isinstance(value, cx_Oracle.Object):
print(prefix + " " + attr.name + " :")
dumpobject(value, prefix + " ")
else:
print(prefix + " " + attr.name + " :", repr(value))
print(prefix, "}")
# Query the row
print("Querying row just inserted...")
cur.execute("select id, geometry from testgeometry")
for (id, obj) in cur:
print("Id: ", id)
dumpobject(obj)