bad_sql_results.txt

Given these 2 SQL tables

CREATE TABLE labels (
	id serial4 NOT NULL,
	"path" int4 NOT NULL,
	labeling int4 NOT NULL,
	"label" int4 NOT NULL,
	"position" int4 NOT NULL,
	"index" int4 NOT NULL,
	CONSTRAINT labels_pkey PRIMARY KEY (id),
	CONSTRAINT labels_path_f FOREIGN KEY ("path") REFERENCES public.reaction_paths(id)
);

CREATE TABLE reaction_paths (
	id serial4 NOT NULL,
	pathways int4 NOT NULL,
	start_metabolite varchar NOT NULL,
	end_metabolite varchar NOT NULL,
	CONSTRAINT reaction_paths_id_key UNIQUE (id),
	CONSTRAINT reaction_paths_pkey PRIMARY KEY (id),
);

I have this query:

select count(distinct pos.path), max(positions) as labeled_positons_in_gauge
from (
  select max(l."path") as "path", array_agg(l."position" order by l.position asc) as positions from labels l 
  join reaction_paths rp ON l."path" = rp.id 
  inner join (
  	select l."path" , l.labeling, max(l."index") as idx from labels l
  	join reaction_paths rp on l."path" = rp.id
  	where l.metabolite = rp.end_metabolite and rp.start_metabolite = 'ALPHA-GLUCOSE' and rp.end_metabolite = 'L-LACTATE'
  	group by l."path", l.labeling
  ) as "ti" on l."path" = ti."path" and l.labeling = ti.labeling 
  where l."index" = ti.idx
  group by l."path", l.labeling
  having count(l."label") = 2
)as pos
group by pos.positions

Can you translate this into accurate and correct Python code using the SQLAlchemy 2.0 library?

From ChatGPT:
=================================================
from sqlalchemy import create_engine, select, func, and_, text, Integer, String, ForeignKey, Column, Table, MetaData, ARRAY
from sqlalchemy.orm import declarative_base, aliased

# Define database connection
engine = create_engine("your_database_url")

# Define metadata and models
Base = declarative_base()

class Labels(Base):
    __tablename__ = "labels"
    id = Column(Integer, primary_key=True)
    path = Column(Integer, ForeignKey("reaction_paths.id"), nullable=False)
    labeling = Column(Integer, nullable=False)
    label = Column(Integer, nullable=False)
    position = Column(Integer, nullable=False)
    index = Column(Integer, nullable=False)

class ReactionPaths(Base):
    __tablename__ = "reaction_paths"
    id = Column(Integer, primary_key=True, unique=True)
    pathways = Column(Integer, nullable=False)
    start_metabolite = Column(String, nullable=False)
    end_metabolite = Column(String, nullable=False)

# Create aliased tables for subqueries
LabelsAlias = aliased(Labels)

# Subquery "ti"
ti_subquery = (
    select(
        Labels.path,
        Labels.labeling,
        func.max(Labels.index).label("idx")
    )
    .join(ReactionPaths, Labels.path == ReactionPaths.id)
    .where(
        and_(
            Labels.label == ReactionPaths.end_metabolite,
            ReactionPaths.start_metabolite == "ALPHA-GLUCOSE",
            ReactionPaths.end_metabolite == "L-LACTATE"
        )
    )
    .group_by(Labels.path, Labels.labeling)
    .subquery(name="ti")
)

# Subquery "positions"
positions_subquery = (
    select(
        func.max(Labels.path).label("path"),
        func.array_agg(Labels.position).order_by(Labels.position).label("positions")
    )
    .join(ReactionPaths, Labels.path == ReactionPaths.id)
    .join(ti_subquery, and_(
        Labels.path == ti_subquery.c.path,
        Labels.labeling == ti_subquery.c.labeling
    ))
    .where(Labels.index == ti_subquery.c.idx)
    .group_by(Labels.path, Labels.labeling)
    .having(func.count(Labels.label) == 2)
    .subquery(name="positions")
)

# Main query
main_query = (
    select(
        func.count(func.distinct(positions_subquery.c.path)).label("distinct_paths_count"),
        func.max(positions_subquery.c.positions).label("labeled_positions_in_gauge")
    )
    .select_from(positions_subquery)
    .group_by(positions_subquery.c.positions)
)

# Execute the query
with engine.connect() as connection:
    result = connection.execute(main_query).fetchall()
    print(result)
=====================================================
Notes

From Codium:
=====================================================
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()

class Labels(Base):
    __tablename__ = 'labels'
    
    id = Column(Integer, primary_key=True)
    path = Column(Integer, ForeignKey('reaction_paths.id'), nullable=False)
    labeling = Column(Integer, nullable=False)
    label = Column(Integer, nullable=False)
    position = Column(Integer, nullable=False)
    index = Column(Integer, nullable=False)

class ReactionPaths(Base):
    __tablename__ = 'reaction_paths'
    
    id = Column(Integer, primary_key=True)
    pathways = Column(Integer, nullable=False)
    start_metabolite = Column(String, nullable=False)
    end_metabolite = Column(String, nullable=False)
    
    labels = relationship("Labels")

# Assuming you have already created the engine
engine = create_engine('your_database_connection_string')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

from sqlalchemy import func

subquery = session.query(Labels.path, func.max(Labels.position).label('positions')).\
    group_by(Labels.path, Labels.labeling).\
    having(func.count(Labels.label) == 2).subquery()

query = session.query(func.count(subquery.c.path), func.max(subquery.c.positions).label('labeled_positions_in_gauge')).\
    group_by(subquery.c.positions)

results = query.all()

for result in results:
    print(result)
=====================================================

Correct code:
=====================================================
gauge_labels = (
    select(Label.path, Label.labeling, func.max(Label.index).label("idx"))
    .join(ReactionPath)
    .where(
        and_(
            Label.metabolite == ReactionPath.end_metabolite,
            ReactionPath.start_metabolite == tracer,
            ReactionPath.end_metabolite == gauge,
        )
    )
    .group_by(Label.path, Label.labeling)
).subquery()

possible_positions = (
    select(
        func.max(Label.path).label("path"),
        func.array_agg(
            aggregate_order_by(Label.position, Label.position.asc())
        ).label("positions"),
    )
    .join(ReactionPath)
    .join(
        gauge_labels,
        and_(
            Label.path == gauge_labels.c.path,
            Label.labeling == gauge_labels.c.labeling,
        ),
        isouter=False,
    )
    .where(and_(Label.index == gauge_labels.c.idx, Label.label.in_(labeling)))
    .group_by(Label.path, Label.labeling)
    .having(func.count(Label.label) == label_count)
).subquery()

query = select(
    func.count(distinct(possible_positions.c.path)),
    func.max(possible_positions.c.positions),
).group_by(possible_positions.c.positions)

return session.execute(query).all()