我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.union()。
def test_plain_union(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2) s2 = select([table]).where(table.c.id == 3) u1 = union(s1, s2) self._assert_result( u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] )
def test_select_from_plain_union(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2) s2 = select([table]).where(table.c.id == 3) u1 = union(s1, s2).alias().select() self._assert_result( u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] )
def test_limit_offset_selectable_in_unions(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).\ limit(1).order_by(table.c.id) s2 = select([table]).where(table.c.id == 3).\ limit(1).order_by(table.c.id) u1 = union(s1, s2).limit(2) self._assert_result( u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] )
def test_order_by_selectable_in_unions(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).\ order_by(table.c.id) s2 = select([table]).where(table.c.id == 3).\ order_by(table.c.id) u1 = union(s1, s2).limit(2) self._assert_result( u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] )
def test_distinct_selectable_in_unions(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).\ distinct() s2 = select([table]).where(table.c.id == 3).\ distinct() u1 = union(s1, s2).limit(2) self._assert_result( u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] )
def test_limit_offset_aliased_selectable_in_unions(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).\ limit(1).order_by(table.c.id).alias().select() s2 = select([table]).where(table.c.id == 3).\ limit(1).order_by(table.c.id).alias().select() u1 = union(s1, s2).limit(2) self._assert_result( u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] )
def _get_history_result_mapper(self, session, resource_type): mappers = self._resource_type_to_mappers(session, resource_type) resource_cls = mappers['resource'] history_cls = mappers['history'] resource_cols = {} history_cols = {} for col in sqlalchemy.inspect(history_cls).columns: history_cols[col.name] = col if col.name in ["revision", "revision_end"]: value = None if col.name == "revision_end" else -1 resource_cols[col.name] = sqlalchemy.bindparam( col.name, value, col.type).label(col.name) else: resource_cols[col.name] = getattr(resource_cls, col.name) s1 = sqlalchemy.select(history_cols.values()) s2 = sqlalchemy.select(resource_cols.values()) if resource_type != "generic": s1 = s1.where(history_cls.revision == ResourceHistory.revision) s2 = s2.where(resource_cls.id == Resource.id) union_stmt = sqlalchemy.union(s1, s2) stmt = union_stmt.alias("result") class Result(base.ResourceJsonifier, base.GnocchiBase): def __iter__(self): return iter((key, getattr(self, key)) for key in stmt.c.keys()) sqlalchemy.orm.mapper( Result, stmt, primary_key=[stmt.c.id, stmt.c.revision], properties={ 'metrics': sqlalchemy.orm.relationship( Metric, primaryjoin=sqlalchemy.and_( Metric.resource_id == stmt.c.id, Metric.status == 'active'), foreign_keys=Metric.resource_id) }) return Result
