from sqlalchemy import Table, Column, Integer, String,MetaData,select,insert,NVARCHAR,TIMESTAMP,and_ from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import Session import datetime Base = declarative_base() class LabelHandler: def __init__(self,engine,engineSource) -> None: self.engine=engine self.engineSource=engineSource def getNewReleases(self): class LabelsChecked(Base): __tablename__ = 'LabelsChecked' id = Column(Integer, primary_key=True) DW_ID_Label = Column(Integer) DW_ID_Project = Column(Integer) timestamp = Column(TIMESTAMP) #LabelsChecked. (self.engine) session = Session(self.engine, future=True) sessionSource = Session(self.engineSource, future=True) meta = MetaData() meta.reflect(bind=self.engineSource,schema='dpl') t_label=meta.tables['dpl.Label'] class Label(Base): #biadmin.dpl.[Label] __table__= t_label __mapper_args__ = { 'primary_key':[t_label.c.DW_ID_Label] } # statement=select(ListDeployedLabels).join(PropertyNamesChecked,PropertyNamesChecked.Project==ListDeployedLabels.Project) listOfUncheckedPropertyNames=sessionSource.scalars(select(Label).where(Label.isCurrent==True).where(Label.DateDeployed>=datetime.datetime.today()-datetime.timedelta(days=1))).all() return listOfUncheckedPropertyNames def markAsChecked(self): pass def getLabel(self,listDeployedLabels:Base) -> str: strLabel="" strLabel=listDeployedLabels.Label.split(':')[2].strip() return strLabel