File: subclassing.py

package info (click to toggle)
python-cx-oracle 8.3.0-3
  • links: PTS, VCS
  • area: contrib
  • in suites: bookworm, sid
  • size: 3,276 kB
  • sloc: ansic: 10,406; python: 9,358; sql: 1,724; makefile: 31
file content (53 lines) | stat: -rw-r--r-- 1,925 bytes parent folder | download
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
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2021, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------

#------------------------------------------------------------------------------
# subclassing.py
#
#   Demonstrate how to subclass connections and cursors in order to add
# additional functionality (like logging) or create specialized interfaces for
# paticular applications.
#------------------------------------------------------------------------------

import cx_Oracle as oracledb
import sample_env

# sample subclassed connection which overrides the constructor (so no
# parameters are required) and the cursor() method (so that the subclassed
# cursor is returned instead of the default cursor implementation)
class Connection(oracledb.Connection):

    def __init__(self):
        connect_string = sample_env.get_main_connect_string()
        print("CONNECT to database")
        super().__init__(connect_string)

    def cursor(self):
        return Cursor(self)


# sample subclassed cursor which overrides the execute() and fetchone()
# methods in order to perform some simple logging
class Cursor(oracledb.Cursor):

    def execute(self, statement, args):
        print("EXECUTE", statement)
        print("ARGS:")
        for arg_index, arg in enumerate(args):
            print("   ", arg_index + 1, "=>", repr(arg))
        return super().execute(statement, args)

    def fetchone(self):
        print("FETCH ONE")
        return super().fetchone()


# create instances of the subclassed connection and cursor
connection = Connection()
cursor = connection.cursor()

# demonstrate that the subclassed connection and cursor are being used
cursor.execute("select count(*) from ChildTable where ParentId = :1", (30,))
count, = cursor.fetchone()
print("COUNT:", int(count))