python-SQLAlchemy,array_agg和匹配输入列表
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了python-SQLAlchemy,array_agg和匹配输入列表，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含6009字，纯文字阅读大概需要9分钟。
内容图文
我试图更充分地使用SQLAlchemy,而不是仅仅在遇到麻烦的最初迹象时才转而使用纯SQL.在这种情况下,我在Postgres数据库(9.5)中有一个表,该表通过将单个项atom_id与组标识符group_id关联来将一组整数存储为一个组.
给定一个atom_id的列表,我希望能够找出那个atom_id集合属于哪个group_id(如果有的话).仅使用group_id和atom_id列即可解决此问题.
现在,我试图进行概括,以使“组”不仅由atom_id列表组成,而且还由其他上下文组成.在下面的示例中,列表通过包括序列列来进行排序,但是从概念上讲,可以使用其他列,例如权重列,该列为每个atom_id提供一个[0,1]浮点值,表示该原子的“份额”.组.
以下是展示我的问题的大部分单元测试.
首先,进行一些设置：
def test_multi_column_grouping(self):
class MultiColumnGroups(base.Base):
__tablename__ = 'multi_groups'
group_id = Column(Integer)
atom_id = Column(Integer)
sequence = Column(Integer) # arbitrary 'other' column. In this case, an integer, but it could be a float (e.g. weighting factor)
base.Base.metadata.create_all(self.engine)
# Insert 6 rows representing 2 different 'groups' of values
vals = [
# Group 1
{'group_id': 1, 'atom_id': 1, 'sequence': 1},
{'group_id': 1, 'atom_id': 2, 'sequence': 2},
{'group_id': 1, 'atom_id': 3, 'sequence': 3},
# Group 2
{'group_id': 2, 'atom_id': 1, 'sequence': 3},
{'group_id': 2, 'atom_id': 2, 'sequence': 2},
{'group_id': 2, 'atom_id': 3, 'sequence': 1},
]
self.session.bulk_save_objects(
[MultiColumnGroups(**x) for x in vals])
self.session.flush()
self.assertEqual(6, len(self.session.query(MultiColumnGroups).all()))
现在,我想查询上表以查找一组特定输入所属的组.我正在使用(命名的)元组列表来表示查询参数.
from collections import namedtuple
Entity = namedtuple('Entity', ['atom_id', 'sequence'])
values_to_match = [
# (atom_id, sequence)
Entity(1, 3),
Entity(2, 2),
Entity(3, 1),
]
# The above list _should_ match with `group_id == 2`
原始SQL解决方案.我宁愿不退缩,因为本练习的一部分是学习更多SQLAlchemy.
r = self.session.execute('''
select group_id
from multi_groups
group by group_id
having array_agg((atom_id, sequence)) = :query_tuples
''', {'query_tuples': values_to_match}).fetchone()
print(r) # > (2,)
self.assertEqual(2, r[0])
这是上面的原始SQL解决方案,可以直接转换为
损坏的SQLAlchemy查询.运行此命令将产生psycopg2错误：(psycopg2.ProgrammingError)运算符不存在：record [] = integer [].我相信我需要将array_agg转换为int []吗？只要分组列都是整数(如果需要,这是一个可接受的限制),那将起作用,但是理想情况下,它将适用于混合类型输入元组/表列.
from sqlalchemy import tuple_
from sqlalchemy.dialects.postgresql import array_agg
existing_group = self.session.query(MultiColumnGroups).\
with_entities(MultiColumnGroups.group_id).\
group_by(MultiColumnGroups.group_id).\
having(array_agg(tuple_(MultiColumnGroups.atom_id, MultiColumnGroups.sequence)) == values_to_match).\
one_or_none()
self.assertIsNotNone(existing_group)
print('|{}|'.format(existing_group))
上面的session.query()是否关闭？我是否在这里蒙蔽了自己的眼光,并且错过了一些可以通过其他方式解决此问题的显而易见的东西？
解决方法:
我认为您的解决方案将产生不确定的结果,因为组中的行是未指定的顺序,因此基于此,数组聚合与给定数组之间的比较可能会产生true或false：
[local]:5432 u@sopython*=> select group_id
[local] u@sopython- > from multi_groups
[local] u@sopython- > group by group_id
[local] u@sopython- > having array_agg((atom_id, sequence)) = ARRAY[(1,3),(2,2),(3,1)];
group_id
----------
2
(1 row)
[local]:5432 u@sopython*=> update multi_groups set atom_id = atom_id where atom_id = 2;
UPDATE 2
[local]:5432 u@sopython*=> select group_id
from multi_groups
group by group_id
having array_agg((atom_id, sequence)) = ARRAY[(1,3),(2,2),(3,1)];
group_id
----------
(0 rows)
您可以对两者都应用排序,也可以尝试完全不同的方法：可以使用relational division代替数组比较.
为了进行划分,您必须从实体记录列表中形成一个临时关系.同样,有很多方法可以解决这个问题.这是一个使用非嵌套数组的数组：
In [112]: vtm = select([
...: func.unnest(postgresql.array([
...: getattr(e, f) for e in values_to_match
...: ])).label(f)
...: for f in Entity._fields
...: ]).alias()
另一个使用联合：
In [114]: vtm = union_all(*[
...: select([literal(e.atom_id).label('atom_id'),
...: literal(e.sequence).label('sequence')])
...: for e in values_to_match
...: ]).alias()
临时表也可以.
有了新的关系,您将找到答案“找到那些没有存在不属于该组的实体的multi_groups”.这句话很可怕,但是很有意义：
In [117]: mg = aliased(MultiColumnGroups)
In [119]: session.query(MultiColumnGroups.group_id).\
...: filter(~exists().
...: select_from(vtm).
...: where(~exists().
...: where(MultiColumnGroups.group_id == mg.group_id).
...: where(tuple_(vtm.c.atom_id, vtm.c.sequence) ==
...: tuple_(mg.atom_id, mg.sequence)).
...: correlate_except(mg))).\
...: distinct().\
...: all()
...:
Out[119]: [(2)]
另一方面,您也可以选择与给定实体的组的交集：
In [19]: gs = intersect(*[
...: session.query(MultiColumnGroups.group_id).
...: filter(MultiColumnGroups.atom_id == vtm.atom_id,
...: MultiColumnGroups.sequence == vtm.sequence)
...: for vtm in values_to_match
...: ])
In [20]: session.execute(gs).fetchall()
Out[20]: [(2,)]
错误
ProgrammingError: (psycopg2.ProgrammingError) operator does not exist: record[] = integer[]
LINE 3: ...gg((multi_groups.atom_id, multi_groups.sequence)) = ARRAY[AR...
^
HINT: No operator matches the given name and argument type(s). You might need to add explicit type casts.
[SQL: 'SELECT multi_groups.group_id AS multi_groups_group_id \nFROM multi_groups GROUP BY multi_groups.group_id \nHAVING array_agg((multi_groups.atom_id, multi_groups.sequence)) = %(array_agg_1)s'] [parameters: {'array_agg_1': [[1, 3], [2, 2], [3, 1]]}] (Background on this error at: http://sqlalche.me/e/f405)
是您的values_to_match如何首先转换为列表列表(由于未知原因),然后转换为converted to an array by your DB-API driver的结果.它导致一个整数数组数组,而不是一个记录数组(int,int).使用raw DB-API connection和游标,传递元组列表可以按预期工作.
在SQLAlchemy中,如果将列表values_to_match与sqlalchemy.dialects.postgresql.array()包装在一起,则它会按您的预期效果工作,尽管请记住结果是不确定的.
内容总结
以上是互联网集市为您收集整理的python-SQLAlchemy,array_agg和匹配输入列表全部内容，希望文章能够帮你解决python-SQLAlchemy,array_agg和匹配输入列表所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。