GtDataStore/app/cleansing/models/data_center/models.py

from models.data_center.Base import Base
from sqlalchemy import text

class ItemData(Base):
    _table_name = 'dc_item_data'
    
    def multi_insert(self, datas):
        if len(datas) == 0:
            return
        query = "INSERT INTO `dc_item_data` (`project_id`, `item`, `val`, `htime`, `tag`)VALUES {}".format(','.join(datas))
        print(query, "\n")
        super().execute(query)

    def __init__(self, db):
        super().__init__(db)

class CleanBind(Base):
    _table_name = 'dc_clean_bind'
    
    def get(self, project_id, target_item):
        session = Base.get_session()

        query = 'select * from {} where project_id={} and `item` = \'{}\' limit 1'.format(self._table_name, project_id, target_item)

        try:
            data = session.execute(text(query)).one()
        except Exception:
            return None
        finally:
            session.close()

        return data

    def __init__(self, db):
        super().__init__(db)

class CleanQuality(Base):
    _table_name = 'dc_clean_quality'
    
    def insert(self, data):
        # query = "INSERT INTO `{}` (`project_id`, `tag`, `item`, `count`, `unique`, `min`, `max`, `mean`, `std`, `p25`, `p50`, `p75`, `top`, `freq`) VALUES({project_id},{tag},{item},{count},{unique},'{min}','{max}','{mean}',{std},{25%},{50%},{75%},'{top}','{freq}')".format(self._table_name, **data)
        query = "INSERT INTO `{}` (`project_id`, `tag`, `item`, `describe`) VALUES({project_id},'{tag}','{item}','{describe}')".format(self._table_name, **data)
        super().execute(query)

    def __init__(self, db):
        super().__init__(db)

class ItemMap(Base):

    engine = None
    _session = None
    _table_name = 'dc_item_map'

    def get(self, project_id, source_item, fields=['project_id','device_code','source_item','target_item']):
        session = Base.get_session()

        query = 'select {} from {} where project_id={} and `item_name` = \'{}\' limit 1'.format(','.join(fields), self._table_name, project_id, item_name)

        try:
            data = session.execute(text(query)).one()
        except Exception:
            return None
        finally:
            session.close()

        return data

    def find_all(self):
        session = Base.get_session()

        query = 'select project_id,device_code,source_item,target_item from {}'.format(self._table_name)
        try:
            data = session.execute(text(query)).all()
        finally:
            session.close()

        return data

    def __init__(self, db):
        super().__init__(db)