Estoy tratando de usar SQLAlchemy más completamente, en lugar de simplemente recurrir a SQL puro a la primera señal de angustia. En este caso, tengo una tabla en una base de datos de Postgres (9.5) que almacena un conjunto de enteros como grupo asociando elementos individualesatom_id con un identificador de grupogroup_id.

Dada una lista deatom_ids, Me gustaría poder averiguar quégroup_id, si alguno, ese conjunto deatom_ids pertenece a. Resolviendo esto solo con elgroup_id yatom_id Las columnas eran sencillas.

Ahora estoy tratando de generalizar de tal manera que un 'grupo' esté compuesto no solo por una lista deatom_ids, pero también otro contexto. En el siguiente ejemplo, la lista se ordena incluyendo unsequence columna, pero conceptualmente se podrían usar otras columnas en su lugar, como unweight columna que da a cadaatom_id un valor de punto flotante [0,1] que representa la "parte" del grupo de ese átomo.

A continuación se muestra la mayoría de las pruebas unitarias que demuestran mi problema.

Primero, alguna configuración:

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()))

Ahora, quiero consultar la tabla anterior para encontrar a qué grupo pertenece un conjunto específico de entradas. Estoy usando una lista de tuplas (con nombre) para representar los parámetros de consulta.

    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`

Solución SQL sin procesar. Carné de identidadpreferir no retroceder en esto, como parte de este ejercicio es aprender más 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])

Aquí está la solución de SQL sin procesar anterior convertida bastante directamente en una consulta SQLAlchemy rota. Ejecutar esto produce un error psycopg2:(psycopg2.ProgrammingError) operator does not exist: record[] = integer[]. Creo que necesito lanzar elarray_agg en unaint[]? Eso funcionaría siempre que las columnas de agrupación sean todas enteras (lo que, si es necesario, es una limitación aceptable), pero idealmente esto funcionaría con tuplas de entrada de tipo mixto / columnas de tabla.

    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))

Es lo anteriorsession.query() ¿cerrar? ¿Me he cegado aquí y me falta algo súper obvio que resolvería este problema de alguna otra manera?