كيف يمكنني كتابة وظائف مجموعتي الخاصة مع Sqlalchemy؟
-
13-09-2019 - |
سؤال
كيف يمكنني كتابة وظائف مجموعتي الخاصة مع 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)]
المحلول
إن إنشاء وظائف إجمالية جديدة يعتمد على التراجع، ويجب القيام به مباشرة مع واجهة برمجة التطبيقات من الاتصال الأسطحي. 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 (مجرد جزء تعبير SQL من Sqlalchemy) ولكن يمكنك استخدام Orm أيضا أيضا.
نصائح أخرى
في البداية عليك استيراد Func من Sqlalchemy
يمكنك كتابة
funct.avg ('اسم الحقل')
أو funct.avg ('اسم الحقل'). تسمية ('user_deined')
أو يمكنك الذهاب من خلال معلومات MRE
http://www.sqlalchemy.org/docs/05/ormtutorial.html#Using-subqueries.