50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
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 |