"""LabelDeliveryHistory Model. This LabelDeliveryHistory model uses sqlalchemy. """ from oto import response from sqlalchemy import Column from sqlalchemy import DateTime from sqlalchemy import func from sqlalchemy.dialects.mysql import SMALLINT from sqlalchemy.dialects.mysql import MEDIUMINT from prs.connectors import mysql class LabelDeliveryHistory(mysql.ppb_model): """Table definition for label_delivery_history table.""" __tablename__ = 'label_delivery_history' label_delivery_history_id = Column( 'id', MEDIUMINT(unsigned=True), autoincrement=True, primary_key=True, nullable=False) label_id = Column('label_id', SMALLINT(unsigned=True)) society_id = Column('society_id', SMALLINT(unsigned=True)) last_delivery = Column('last_delivery', DateTime) @mysql.wrap_db_errors def get_last_delivery_by_vendor_and_society_id(society_id, vendor_ids): """Get last delivery data by vendor id(s) and society id. Get list of label ID(s) and delivery dates which reflect the last delivery date of a given label for a given society. Args: society_id (int): Society id to fetch delivery data. vendor_ids (list): List of Vendor id(s) to fetch delivery data. Returns: response.Response: Response contains list describing delivery data for corresponding label id/ids. """ filters = [ LabelDeliveryHistory.society_id == society_id, LabelDeliveryHistory.label_id.in_(vendor_ids) ] order_by_list = [ LabelDeliveryHistory.label_id, LabelDeliveryHistory.last_delivery.desc() ] with mysql.ppb_db_session() as session: result = session.query( LabelDeliveryHistory.label_id, func.Date(LabelDeliveryHistory.last_delivery) ).filter(*filters).order_by(*order_by_list).all() result = [tuple(map(str, lh)) for lh in result] return response.Response(message={'label_delivery_history': result})