如何使用 sqlalchemy 编写自己的聚合函数？答案

【问题标题】：How can I write my own aggregate functions with sqlalchemy?如何使用 sqlalchemy 编写自己的聚合函数？
【发布时间】：2010-11-03 01:45:10
【问题描述】：

如何使用 SQLAlchemy 编写自己的聚合函数？作为一个简单的例子，我想使用 numpy 来计算方差。使用 sqlite，它看起来像这样：

import sqlite3 as sqlite
import numpy as np

class self_written_SQLvar(object):
  def __init__(self):
    import numpy as np
    self.values = []
  def step(self, value):
    self.values.append(value)
  def finalize(self):
    return np.array(self.values).var()

cxn = sqlite.connect(':memory:')
cur = cxn.cursor()
cxn.create_aggregate("self_written_SQLvar", 1, self_written_SQLvar)
# Now - how to use it:
cur.execute("CREATE TABLE 'mytable' ('numbers' INTEGER)")
cur.execute("INSERT INTO 'mytable' VALUES (1)") 
cur.execute("INSERT INTO 'mytable' VALUES (2)") 
cur.execute("INSERT INTO 'mytable' VALUES (3)") 
cur.execute("INSERT INTO 'mytable' VALUES (4)")
a = cur.execute("SELECT avg(numbers), self_written_SQLvar(numbers) FROM mytable")
print a.fetchall()
>>> [(2.5, 1.25)]

【问题讨论】：

标签： python sqlite sqlalchemy aggregate-functions

【解决方案1】：

新聚合函数的创建依赖于后端，并且必须完成直接使用下划线连接的API。 SQLAlchemy 不提供用于创建这些的工具。

但是创建后你可以在 SQLAlchemy 中正常使用它们。

例子：

import sqlalchemy
from sqlalchemy import Column, Table, create_engine, MetaData, Integer
from sqlalchemy import func, select
from sqlalchemy.pool import StaticPool
from random import randrange
import numpy
import sqlite3

class NumpyVarAggregate(object):
  def __init__(self):
    self.values = []
  def step(self, value):
    self.values.append(value)
  def finalize(self):
    return numpy.array(self.values).var()

def sqlite_memory_engine_creator():
    con = sqlite3.connect(':memory:')
    con.create_aggregate("np_var", 1, NumpyVarAggregate)
    return con

e = create_engine('sqlite://', echo=True, poolclass=StaticPool,
                  creator=sqlite_memory_engine_creator)
m = MetaData(bind=e)
t = Table('mytable', m, 
            Column('id', Integer, primary_key=True),
            Column('number', Integer)
          )
m.create_all()

现在开始测试：

# insert 30 random-valued rows
t.insert().execute([{'number': randrange(100)} for x in xrange(30)])

for row in select([func.avg(t.c.number), func.np_var(t.c.number)]).execute():
    print 'RESULT ROW: ', row

打印（打开 SQLAlchemy 语句回显）：

2009-06-15 14:55:34,171 INFO sqlalchemy.engine.base.Engine.0x...d20c PRAGMA 
table_info("mytable")
2009-06-15 14:55:34,174 INFO sqlalchemy.engine.base.Engine.0x...d20c ()
2009-06-15 14:55:34,175 INFO sqlalchemy.engine.base.Engine.0x...d20c 
CREATE TABLE mytable (
    id INTEGER NOT NULL, 
    number INTEGER, 
    PRIMARY KEY (id)
)
2009-06-15 14:55:34,175 INFO sqlalchemy.engine.base.Engine.0x...d20c ()
2009-06-15 14:55:34,176 INFO sqlalchemy.engine.base.Engine.0x...d20c COMMIT
2009-06-15 14:55:34,177 INFO sqlalchemy.engine.base.Engine.0x...d20c INSERT
INTO mytable (number) VALUES (?)
2009-06-15 14:55:34,177 INFO sqlalchemy.engine.base.Engine.0x...d20c [[98], 
[94], [7], [1], [79], [77], [51], [28], [85], [26], [34], [68], [15], [43], 
[52], [97], [64], [82], [11], [71], [27], [75], [60], [85], [42], [40], 
[76], [12], [81], [69]]
2009-06-15 14:55:34,178 INFO sqlalchemy.engine.base.Engine.0x...d20c COMMIT
2009-06-15 14:55:34,180 INFO sqlalchemy.engine.base.Engine.0x...d20c SELECT
avg(mytable.number) AS avg_1, np_var(mytable.number) AS np_var_1 FROM mytable
2009-06-15 14:55:34,180 INFO sqlalchemy.engine.base.Engine.0x...d20c []
RESULT ROW: (55.0, 831.0)

请注意，我没有使用 SQLAlchemy 的 ORM（仅使用了 SQLAlchemy 的 sql 表达式部分），但您也可以使用 ORM。

【讨论】：

非常感谢。这是一个非常好的答案！
可以参考文档或其他内容吗？如果你有。 @nosklo

【解决方案2】：

首先你必须从 sqlalchemy 导入 func

你可以写

func.avg('fieldname')

或 func.avg('fieldname').label('user_deined')

【讨论】：

这适用于标准聚合函数，如 avg 或 count。但问题是如何对待自我实现的聚合函数。顺便说一句，教程似乎相当不错。谢谢你的链接。