"""Model for art_relations.delivery_history.""" from sqlalchemy import Column from sqlalchemy import Date from sqlalchemy import Integer from sqlalchemy import String from deliveryhistory.connectors import mysql class DeliveryHistory(mysql.BaseModel): """Model class for delivery_history.""" __tablename__ = 'delivery_history' delivery_id = Column(Integer, primary_key=True) upc = Column(String(20), nullable=False) encoder_id = Column(Integer, nullable=False) date_delivered = Column(Date, nullable=False) customer_master_master_id = Column(Integer, nullable=False) package_size = Column(Integer, nullable=False) def __init__( self, upc, encoder_id, date_delivered, customer_master_master_id, package_size): """Init. Args: upc (string): Universal Product Code. encoder_id (int): Encoder ID. date_delivered (str): Delivery date. customer_master_master_id (int): Store ID. package_size (int): Package size in kilobytes. """ self.upc = upc self.encoder_id = encoder_id self.date_delivered = date_delivered self.customer_master_master_id = customer_master_master_id self.package_size = package_size def insert_delivery_history(records): """Bulk insert delivery history records. Args: records (list): List of DeliveryHistory objects Returns: bool """ with mysql.ar_session_scope() as sess: sess.bulk_save_objects(records) sess.commit() return True def check_for_sme_delivery_history(upc, store_id, encoder_id, date_delivered): """Check for SME delivery history for given UPC and store ID. Args: upc (str): UPC store_id (int): Store ID encoder_id (int): Encoder ID date_delivered (str): Date delivered Returns: list """ with mysql.ar_session_scope() as sess: return sess.query(DeliveryHistory).filter( DeliveryHistory.upc == upc, DeliveryHistory.customer_master_master_id == store_id, DeliveryHistory.encoder_id == encoder_id, DeliveryHistory.date_delivered == date_delivered ).all()