调用数据库存储过程及其他感兴趣的高级 Python 编程功能。

2010 年 3 月发布

对于涉及数据库的软件开发来说，有两种主流开发方法：一种是在应用程序中（对于三层体系结构，也可以是在中间件中）实现所有业务逻辑，另一种是在数据库内部实现所有业务逻辑。本教程不讨论这两种解决方案的优缺点；不过，使用 Oracle 数据库方法还是会为面向数据库的应用程序带来某些好处。

用 PL/SQL 嵌入所有业务逻辑可大大减少应用程序与数据库之间的往返次数，从而此使处理都在服务器端进行。PL/SQL 与 SQL 紧密集成，并与 Python 类似，提供了大量的标准程序包库：从安排数据库作业时间 (DBMS_SCHEDULER)，到自动查询调优 (DBMS_SQLTUNE) 和闪回 (DBMS_FLASHBACK)，再到线性代数 (UTL_NLA) 和 LDAP 访问 (DBMS_LDAP)。

本教程介绍使用 cx_Oracle 模块在 Python 中调用 Oracle 数据库内部的 PL/SQL 存储过程和函数的各种方法，同时还介绍一些使用 PL/SQL 无法实现或者实现起来非常复杂的编程功能。在本文的最后，我们将简单介绍 Oracle Berkeley DB，它作为即取即用插件内置在 Python 中。

IN-OUT 方式

Oracle 过程和函数是将 SQL 功能与编程语言功能相结合一些数据库对象。过程（从现在开始也称其为函数）的参数可以是以下三种类型之一：

• IN：传递给过程，但不能写入到过程内部
• OUT：从过程返回，在过程主体内部可写
• IN OUT：传递给过程，在过程内部完全可写

默认情况下，参数都是 IN 类型。

为了说明 Python 和 Oracle 过程之间的交互，我们考虑将以下程序包安装在 Oracle Database XE 实例的 HR 模式中。

 CREATE OR REPLACE PACKAGE pkg_hr AS 

  PROCEDURE add_department(

    p_department_id OUT NUMBER,

    p_department_name IN VARCHAR2,

    p_manager_id IN NUMBER,

    p_location_id IN NUMBER

  ); 

  FUNCTION get_employee_count(

    p_department_id IN NUMBER

  ) RETURN NUMBER;

  PROCEDURE find_employees(

    p_query IN VARCHAR2,

    p_results OUT SYS_REFCURSOR

  ); 

END pkg_hr;

/ 

CREATE OR REPLACE PACKAGE BODY pkg_hr AS 

  PROCEDURE add_department(

    p_department_id OUT NUMBER,

    p_department_name IN VARCHAR2,

    p_manager_id IN NUMBER,

    p_location_id IN NUMBER

  ) AS

  BEGIN

    INSERT INTO departments(department_id, department_name, manager_id, location_id)

    VALUES (departments_seq.nextval, p_department_name, p_manager_id, p_location_id)

    RETURNING department_id

    INTO p_department_id;

    COMMIT;

  END add_department; 

  FUNCTION get_employee_count(

    p_department_id IN NUMBER

  ) RETURN NUMBER AS

    l_count NUMBER;

  BEGIN

    SELECT COUNT(*)

    INTO l_count

    FROM employees

    WHERE department_id= p_department_id;

    RETURN l_count;

  END get_employee_count; 

 PROCEDURE find_employees(

    p_query IN VARCHAR2,

    p_results OUT SYS_REFCURSOR

  ) AS

  BEGIN

    OPEN p_results FOR

      SELECT *

      FROM employees

      WHERE UPPER(first_name||' '||last_name||' '||email) LIKE '%'||UPPER(p_query)||'%';

  END find_employees; 

END pkg_hr;

/

上面的示例中引入了三种不同的访问方法：一个带有 IN 参数和 OUT 参数的过程、一个返回数字的函数以及一个带有 OUT REF CURSOR 参数的过程。每个过程需要不同的调用方法，如下所示。

import cx_Oracle

class HR:

  def __enter__(self):

    self.__db = cx_Oracle.Connection("hr/hrpwd@//localhost:1521/XE")

    self.__cursor = self.__db.cursor()

    return self 

  def __exit__(self, type, value, traceback):

    self.__cursor.close()

    self.__db.close()

  def add_department(self, p_department_name, p_manager_id, p_location_id):

    l_department_id = self.__cursor.var(cx_Oracle.NUMBER)

    self.__cursor.callproc("PKG_HR.ADD_DEPARTMENT",

      [l_department_id, p_department_name, p_manager_id, p_location_id]) 

    # there are no OUT parameters in Python, regular return here

    return l_department_id    

  def get_employee_count(self, p_department_id):

    l_count = self.__cursor.callfunc("PKG_HR.GET_EMPLOYEE_COUNT",

      cx_Oracle.NUMBER, [p_department_id])

    return l_count 

  def find_employees(self, p_query):

    # as it comes to all complex types we need to tell Oracle Client

    # what type to expect from an OUT parameter

    l_cur = self.__cursor.var(cx_Oracle.CURSOR)

    l_query, l_emp = self.__cursor.callproc("PKG_HR.FIND_EMPLOYEES", [p_query, l_cur])

    return list(l_emp)

从上面的示例可以看到，通过一些基本规则规定了从 Python 调用存储过程的方法：

• 使用 cx_Oracle.Cursor.callproc(proc, [params]) 调用过程，使用 cx_Oracle.Cursor.callfunc(proc, returnType, [params]) 调用函数。需要预先定义函数的返回类型 — get_employee_count() 方法声明了从 PKG_HR.GET_EMPLOYEE_COUNT 返回的类型为 cx_Oracle.NUMBER。
• 使用 cx_Oracle 变量对象作为 callproc/callfunc 调用的参数可以返回类似 REF CURSOR 这样的复杂类型。

使用 arrayvar 传递数组

cx_Oracle 中 DB API 2.0 的另一个扩展允许在存储过程调用中使用数组作为参数。当前支持使用 INDEX BY 子句的 PL/SQL 数组。作为一个使用 arrayvar 对象的示例，要确保下面的 DDL 找到自己进入数据库的道路。

CREATE OR REPLACE PACKAGE pkg_arrayvar AS

  TYPE num_array IS TABLE OF NUMBER INDEX BY PLS_INTEGER;

  FUNCTION sum(p_list IN NUM_ARRAY) RETURN NUMBER;

END pkg_arrayvar;

/

CREATE OR REPLACE PACKAGE BODY pkg_arrayvar AS

FUNCTION sum(p_list IN NUM_ARRAY) RETURN NUMBER AS

    l_sum NUMBER := 0;

  BEGIN

    FOR i IN 1..p_list.COUNT LOOP

      l_sum := l_sum+p_list(i);

    END LOOP i;

    RETURN l_sum;

  END sum;

END pkg_arrayvar;

/

现在，对 Python 对象的声明和对函数的实际调用如下（后跟一个断言来验证结果）：

>>> db = cx_Oracle.connect("hr/hrpwd@//localhost:1521/XE")

>>> cursor = db.cursor()

>>> L = cursor.arrayvar(cx_Oracle.NUMBER, [1, 2, 3])

>>> sum_result = cursor.callfunc("pkg_arrayvar.sum", cx_Oracle.NUMBER, [L])

>>> assert sum_result==6

上面提到过，在 Python 中调用函数需要显式声明返回类型，这可能会造成混淆，因为 callproc() 只需要两个参数，但是这是必经之路。

转到 Python

PL/SQL 是一种功能强大的语言，它尽可能地结合了 Oracle 数据库的功能，可显著减少开发工作量，并使您可以利用数据库的大多数特性。但某些编程功能无法通过 PL/SQL 的数据库中固有特性表达，或者无法通过 PL/SQL 的数据库中固有特性使用。因此，需要用其他编程语言对其进行补充时，Python 是一个不错的选择，它可以缩短开发时间，加快开发完成。

Multiprocessing

从 2.6 版开始，Python 中的并行处理不再受 GIL（全局解释器锁）的限制。随标准库一起提供的 threading 模块被限定为一次只运行一个操作。通过用操作系统进程替换线程，现在可将所有 CPU 提供给应用程序，因此能够真正执行并行计算。Multiprocessing 模块让应用程序生成新的进程、锁定对象、在内存*享对象，而且，所有这些既可本地进行，也可远程（不同的计算机上）进行。

下面是一个简单数据库基准测试实用程序的示例。

import cx_Oracle

import os

import time

from multiprocessing import Pool

from optparse import OptionParser

def benchmark(options):

  params = eval(options.bind) if options.bind else {}

  with cx_Oracle.connect(options.db) as db:

    try:

      cursor = db.cursor()

      before = time.clock()

      for i in xrange(options.requests):

        cursor.execute(options.sql, params)

      return (time.clock()-before)/options.requests

    except KeyboardInterrupt:

      pass

    finally:

      cursor.close()

class Orabench:

  def __init__(self, options):

    self.options = options

    print "Requests=%d, Concurrency=%d" % (self.options.requests,

                                           self.options.concurrency)  

  def run(self):

    pool = Pool(processes=self.options.concurrency)

    result = pool.map_async(benchmark, [self.options]*self.options.concurrency)

    L = result.get()

    avg = sum(L)/len(L)

    print "Average=%.4f (%.4f requests per second)" % (avg, 1/avg)

if __name__ == "__main__":

  opt = OptionParser()

  opt.add_option("-d", "--database", help="EZCONNECT string", action="store",

                 type="string", dest="db")

  opt.add_option("-n", "--requests", help="number of requests", action="store",

                 type="int", dest="requests", default=10)

  opt.add_option("-c", "--concurrency", help="number of concurrent connections",

                 action="store", type="int", dest="concurrency", default=1)

  opt.add_option("-s", "--sql", help="SQL query or PL/SQL block",

                 action="store", type="string", dest="sql")

  opt.add_option("-b", "--bind", help="dictionary of bind parameters",

                 action="store", type="string", dest="bind")

  (options, args) = opt.parse_args()

  bench = Orabench(options)

  bench.run()

利用 optparse 模块（该模块可以很好地解析命令行参数），该工具在使用“--help”开关参数时会自动生成使用说明。

pp@oel:~$ python26 orabench.py --help

Usage: orabench.py [options]

Options:

  -h, --help            show this help message and exit

  -d DB, --database=DB  EZCONNECT string

  -n REQUESTS, --requests=REQUESTS

                        number of requests

  -c CONCURRENCY, --concurrency=CONCURRENCY

                        number of concurrent connections

  -s SQL, --sql=SQL     SQL query or PL/SQL block

  -b BIND, --bind=BIND  dictionary of bind parameters

然后，使用 HR 模式在 10 个进程中执行 1000 次查询的基准测试：

pp@oel:~$ python26 orabench.py -d hr/hrpwd@//localhost:1521/XE -n 1000 -c 10 -s "select count(*) from employees"

Requests=1000, Concurrency=10

Average=0.0006 (1667.7460 requests per second)

数据库外部的 GROUP BY（函数式编程）

对各种函数式编程功能，很少有模块能优于 itertools。它包含许多可以生成自定义、优化迭代器的遍历函数。简单提示一下，迭代器是一些对象，其中 __iter__() 方法返回迭代器本身，next() 方法步进到后续元素，或者引发结束迭代的 StopIteration 异常。通过遍历大型数据集，您可以发现使用迭代器与使用列表或字节组的差别，因为迭代器避免了在内存中提交整个集合。

import cx_Oracle

import itertools

from operator import itemgetter

with cx_Oracle.connect("hr/hrpwd@//localhost:1521/XE") as db:

  cursor = db.cursor()

  # fetch all employee data into local variable, no aggregation here

  employees = cursor.execute("select * from employees").fetchall()

D = {}

for dept, emp in itertools.groupby(employees, itemgetter(10)):

  D[dept] = len(list(emp))

operator 模块包括原生对象使用的所有核心运算符，这意味着无论您何时运行 2+2，operator.add() 方法都会处理这一计算。因为 itertools.groupby() 方法接受两个参数：可迭代的变量和主函数，我们需要使用 itemgetter(10) 从所有行中提取 department_id，itemgetter(10) 仅返回一个集合中的第 10 个元素。对 itertools 的结果进行遍历与您对列表、字节组和字典的遍历非常类似。我们为每个部门生成部门 ID 及该部门的所有员工数 (SELECT department_id, COUNT(*) FROM employees GROUP BY department_id)。

序列化数据

在 Python 中，用 pickle 模块及其 C 中对应的 cPickle（比原生 Python 的 pickle 实现最多快 1000 倍）处理数据的序列化和反序列化。“序列化”对象意味着将对象转换为可逆的字节表示：

>>> import pickle

>>> A = {'a':1, 'b':2, 'c':3}

>>> B = pickle.dumps(A)

>>> print B

"(dp0\nS'a'\np1\nI1\nsS'c'\np2\nI3\nsS'b'\np3\nI2\ns."

>>> C = pickle.loads(B)

>>> assert A==C

紧邻关系数据存储复杂结构时，序列化显得尤为有用，这样，我们就可以将常规 Python 对象当作数据库本来就支持的对象进行读写操作。

至于 pickle 支持的类型，只有很少的限制，因为它能处理的内容非常广泛，从字典和字节组，到数据集和函数，再到类和实例。其中一个不能序列化的对象是 cx_Oracle.Connection 对象，原因显而易见。

惰性化缓存 (Sleepy Cache)

Oracle Berkeley DB 是一个事务性键值存储解决方案，具有细粒度锁定、高可用性和复制功能。它可以应对需要极端效率及完整关系数据库开销过高情况下的所有问题。（直到 2.6 版，Python 才以 bsddb 模块的形式包括了针对 Oracle Berkeley DB 的内置接口。Python 的新版本（从 3.0 开始）使用一个需要单独安装的外部模块 PyBSDDB。 )

下面，我们将利用 Python 2.6 附带的内置驱动程序将值从 Oracle 数据库缓存到 Oracle Berkeley DB 中：

import bsddb

import cx_Oracle

import pickle

class Cache:

  def __init__(self, tab):

    self.__db = cx_Oracle.connect("hr/hrpwd@//localhost:1521/XE")

    self.__cursor = self.__db.cursor()

    self.__bdb = bsddb.hashopen(None)

    self.__cursor.execute("select * from employees")

    d = self.__cursor.description

    for row in self.__cursor:

      rowdict = dict((d[i][0].lower(), row[i]) for i in xrange(len(d)))

      self.__bdb[str(row[0])] = pickle.dumps(rowdict)

  def __del__(self):

    self.__cursor.close()

    self.__db.close()

  def __getitem__(self, name):

    try:

      return pickle.loads(self.__bdb[str(name)])

    except KeyError:

      raise Warning, "No such employee with ID %s" % name

if __name__ == "__main__":

  emp = Cache("EMPLOYEES")

现在可以像使用“emp[100]”一样方便地访问员工了，emp[100] 可以访问快速的内存中散列表，并对序列化的员工数据进行反序列化。您可以使用任一内置服务器（SimpleHTTPServer、SimpleXMLRPCServer、wsgiref.simple_server）轻松包装这样的缓存，或者使用 Twisted 框架使缓存变得更强健。

总结

这次我们介绍了 Oracle 和 Python 结合使用时的几个核心领域，包括 PL/SQL 存储过程调用以及如何处理 PL/SQL 函数结果。谈到 Python，现在您应对 multiprocessing 模块要点有所了解，它是该语言新添加的最重要的模块。最后，介绍了 Oracle Berkeley DB 在概念验证的内存缓存中的使用情况。