Hi people.
I'm having trouble with a test query.
As the subject line says, I'm getting:
sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (1054, "Unknown
column 'tb_br.id' in 'on clause'")
But it seems like tb_br exists, and has an id column - tb_br being an empty
table, but still, existent:
mysql> show create table tb_br;
+-------+---------------------------------------------------------------------------------------------------------------------------------------------------------+
| Table | Create Table
|
+-------+---------------------------------------------------------------------------------------------------------------------------------------------------------+
| tb_br | CREATE TABLE `tb_br` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(45) NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 |
+-------+---------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.04 sec)
mysql> select * from tb_br;
Empty set (0.03 sec)
The query, along with sample models, looks like:
#!/usr/bin/env python3
"""
A little test program.
Environment variables:
DBU Your database user
DBP Your database password
DBH Your database host
IDB Your initial database
"""
import os
import pprint
from sqlalchemy import create_engine, select
from sqlalchemy.orm import aliased, sessionmaker, declarative_base
from sqlalchemy.sql.expression import func
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
Base = declarative_base()
class NV(Base):
__tablename__ = "tb_nv"
__bind_key__ = "testdb"
__table_args__ = (
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8",
"mysql_collate": "utf8_general_ci",
},
)
id = db.Column("id", db.Integer, primary_key=True, autoincrement=True)
builds = db.relationship("Bld", primaryjoin="(NV.id == Bld.variant_id)")
class Vers(Base):
__tablename__ = "tb_vers"
__bind_key__ = "testdb"
__table_args__ = (
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8",
"mysql_collate": "utf8_general_ci",
},
)
id = db.Column("id", db.Integer, primary_key=True, autoincrement=True)
class St(Base):
__tablename__ = "tb_brst"
__bind_key__ = "testdb"
__table_args__ = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8"},)
id = db.Column("id", db.Integer, primary_key=True, autoincrement=True)
version_id = db.Column(
"version_id",
db.Integer,
db.ForeignKey(
"tb_vers.id",
name="fk_tb_brst_version_id",
onupdate="CASCADE",
ondelete="RESTRICT",
),
nullable=False,
)
branch_id = db.Column(
"branch_id",
db.Integer,
db.ForeignKey(
"tb_br.id",
name="fk_tb_brst_branch_id",
onupdate="CASCADE",
ondelete="RESTRICT",
),
nullable=False,
)
build_id = db.Column(
"build_id",
db.Integer,
db.ForeignKey(
"tb_bld.id",
name="fk_tb_brst_build_id",
onupdate="CASCADE",
ondelete="RESTRICT",
),
nullable=False,
)
version = db.relationship(
"Vers", innerjoin=True, primaryjoin="(St.version_id == Vers.id)"
)
branch = db.relationship(
"Br", innerjoin=True, primaryjoin="(St.branch_id == Br.id)"
)
build = db.relationship(
"Bld", innerjoin=True, primaryjoin="(St.build_id == Bld.id)"
)
class Br(Base):
__tablename__ = "tb_br"
__bind_key__ = "testdb"
__table_args__ = (
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8",
"mysql_collate": "utf8_general_ci",
},
)
id = db.Column("id", db.Integer, primary_key=True, autoincrement=True)
name = db.Column("name", db.String(45), nullable=False)
class Bld(Base):
__tablename__ = "tb_bld"
__bind_key__ = "testdb"
__table_args__ = (
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8",
"mysql_collate": "utf8_general_ci",
},
)
id = db.Column("id", db.Integer, primary_key=True, autoincrement=True)
name = db.Column("name", db.String(100), nullable=False)
variant_id = db.Column(
"variant_id",
db.Integer,
db.ForeignKey(
"tb_nv.id",
name="fk_tb_bld_variant_id",
onupdate="CASCADE",
ondelete="RESTRICT",
),
nullable=False,
)
variant = db.relationship("NV")
def display(values):
"""Display values in a decent way."""
pprint.pprint(values)
def connect():
"""
Connect to Staging for testing.
This is based on
https://medium.com/analytics-vidhya/translating-sql-queries-to-sqlalchemy-orm-a8603085762b
...and
./game-publishing/services/api/deploy/celery/config/staging-base.j2
"""
conn_str = "mysql://{}:{}@{}/{}".format(
os.environ["DBU"],
os.environ["DBP"],
os.environ["DBH"],
os.environ["IDB"],
)
engine = create_engine(conn_str)
# Added 2023-03-17
Base.metadata.create_all(engine)
session = sessionmaker(bind=engine)
sess = session()
return (engine, sess)
def main():
"""A minimal query that exhibits the problem."""
(engine, session) = connect()
Base.metadata.create_all(engine)
v_2 = aliased(Vers, name="v_2")
bs = aliased(St, name="bs")
bs_2 = aliased(St, name="bs_2")
bs_3 = aliased(St, name="bs_3")
# sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (1054,
"Unknown column 'tb_br.id' in 'on clause'")
# [SQL: SELECT tb_nv.id, min(bs_3.build_id) AS min_1
# FROM tb_nv, tb_brst AS bs_3, tb_brst AS bs
# INNER JOIN tb_vers AS v_2 ON bs.version_id = v_2.id
# INNER JOIN tb_brst AS bs_2 ON tb_br.id = bs_2.branch_id]
query = (
select(NV.id, func.min(bs_3.build_id))
.select_from(bs)
.join(v_2, onclause=(bs.version_id == v_2.id))
.join(bs_2, onclause=(Br.id == bs_2.branch_id))
)
result = session.execute(query)
display(result.scalar().all())
main()
I'm using:
$ python3 -m pip list -v | grep -i sqlalchemy
Flask-SQLAlchemy 2.5.1
/data/home/dstromberg/.local/lib/python3.10/site-packages pip SQLAlchemy
1.4.36 /data/home/dstromberg/.local/lib/python3.10/site-packages pip $
python3 -m pip list -v | grep -i mysql mysqlclient 2.1.1
/data/home/dstromberg/.local/lib/python3.10/site-packages pip PyMySQL 0.8.0
/data/home/dstromberg/.local/lib/python3.10/site-packages pip bash-4.2#
mysql --version mysql Ver 14.14 Distrib 5.7.41, for Linux (x86_64) using
EditLine wrapper
Any hints?
Thanks!
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/1576c1a1-e108-4bb0-83eb-57ab4125c457n%40googlegroups.com.
