I'm not exactly sure how to what is going on, but for our testing we rename the tables before we generate everything (ie schema.table_name is renamed to table_name_schema due to issues with how we have to test). I have a file of generated queries to be able to pull from as constants, and for a good while everything was working fine and as expected. I don't know what happened or why, but as of Monday, my tests are erroring out stating that a table doesn't exist. It labels the table as the old name and not the new name, but the other table within that select statement is using the new name, so I'm confused as to why one is correct and the other isn't.
When I run the tests, I have importlib reload the queries so that the new table names are there after our test database updates everything:
def setUp(self):
self.now = TestConstants.TEST_TIME_WINDOW_CENTER
self.database = TestDatabase()
importlib.reload(queries)
Then my test runs this statement:
get_qc_sod_data = (
select(QcSodTable)
.join(
LocalTimeTable,
and_(
QcSodTable.platformid == LocalTimeTable.platformid,
QcSodTable.networktype == LocalTimeTable.networktype
),
isouter=True,
)
.where(
or_(QcSodTable.qc_flag.not_in(("P", "F")), QcSodTable.qc_flag == None),
(QcSodTable.datetime + cast(literal(1.5) + ' DAYS', Interval)) - (cast(func.coalesce(LocalTimeTable.time_conv, 0) / 24 + ' HOURS', Interval)) < func.now(),
)
)
The error comes at the LocalTimeTable where it states the table does not exist but lists the incorrect name (schema.table_name) though the Select table lists the proper name as expected, so I'm not sure why the other is not correct. If I change LocalTimeTable to LocalTimeTable.__ table__ then it passes fine
Is there something I'm missing here? Do I need to reload that table as well within my constants queries file?
EDIT:
SQLAlchemy version is 2.0.43
Here is the local time table:
class LocalTimeTableBase(Base):
__abstract__ = True
PLATFORMID_COLUMN_NAME = "platformid"
NETWORKTYPE_COLUMN_NAME = "networktype"
TIME_CONV_COLUMN_NAME = "time_conv"
DISTRIBUTIONCODE_COLUMN_NAME = "distributioncode"
SECURITYID_COLUMN_NAME = "securityid"
platformid = Column(String, name=PLATFORMID_COLUMN_NAME, primary_key=True)
networktype = Column(
String, name=NETWORKTYPE_COLUMN_NAME, nullable=False, primary_key=True
)
time_conv = Column(Numeric, name=TIME_CONV_COLUMN_NAME)
distributioncode = Column(String, name=DISTRIBUTIONCODE_COLUMN_NAME)
securityid = Column(Integer, name=SECURITYID_COLUMN_NAME)
class LocalTimeTable(LocalTimeTableBase):
__tablename__ = "local_time"
__table_args__ = {"schema": SCHEMA_NAME}