"""Scheduled update CRUD operation.""" from datetime import datetime, timedelta, timezone from dateutil import parser from oto import response from oto import status as http_status from sqlalchemy import ( Column, Date, DateTime, Enum, ForeignKey, Integer, String, ) from sqlalchemy.orm import relationship from timed_release.connectors import sql from timed_release.constants.error import ERROR_MESSAGE_SCHEDULE_NOT_FOUND from timed_release.constants.timed_release import ( OA, SALES_DATE_DAYS_OFFSET, SOURCE, WORKSTATION, ) class ScheduledProductUpdateDms(sql.base_model): """Table definition for scheduled_product_update_dms table.""" __tablename__ = 'scheduled_product_update_dms' scheduled_product_update_dms_id = Column( Integer, primary_key=True, nullable=False, autoincrement=True ) scheduled_product_update_id = Column( 'scheduled_product_update_id', Integer, ForeignKey('scheduled_product_update.scheduled_product_update_id'), nullable=False) scheduled_product_update = relationship( 'ScheduledProductUpdate', back_populates='dsps' ) customer_master_master_id = Column(Integer, nullable=False) class ScheduledProductUpdateValues(sql.base_model): """Table definition for scheduled_product_update_values table.""" __tablename__ = 'scheduled_product_update_values' scheduled_product_update_id = Column( 'scheduled_product_update_id', Integer, ForeignKey('scheduled_product_update.scheduled_product_update_id'), primary_key=True, nullable=False ) scheduled_product_update = relationship( 'ScheduledProductUpdate', back_populates='update' ) sale_start_date = Column(Date, nullable=True) def to_dict(self): """Return dictionary of a scheduled_product_update_values.""" return { 'sale_start_date': self.sale_start_date.strftime('%Y-%m-%d') } class ScheduledProductUpdate(sql.base_model): """Table definition for scheduled_product_update table.""" __tablename__ = 'scheduled_product_update' scheduled_product_update_id = Column( Integer, primary_key=True, nullable=False ) product_id = Column(Integer, nullable=False) sales_date_time = Column( 'scheduled_product_update_sales_date_time', DateTime, nullable=False ) process_at = Column( 'scheduled_product_update_process_at', DateTime, nullable=False ) process_at_offset = Column( 'scheduled_product_update_process_at_offset', String, nullable=False, default='00:00' ) scheduled_by = Column( 'scheduled_product_update_scheduled_by', String, nullable=False ) dsps = relationship( 'ScheduledProductUpdateDms', back_populates='scheduled_product_update', cascade='save-update, merge, delete-orphan, delete' ) update = relationship( ScheduledProductUpdateValues, uselist=False, cascade='save-update, delete-orphan, delete', backref='scheduled_update' ) created_on = Column(DateTime, nullable=False) created_by = Column(String(127)) updated_on = Column(DateTime, nullable=False) updated_by = Column(String(127)) source = Column(Enum(*SOURCE), nullable=False) user_timezone = Column(String(50)) def to_dict(self): """Return dictionary of a scheduled_product_update.""" return { 'product_id': self.product_id, 'schedule_datetime': self.sales_date_time.strftime('%Y-%m-%dT%H:%M:%SZ'), 'schedule_time_offset': self.process_at_offset, 'process_at_datetime': self.process_at.strftime('%Y-%m-%dT%H:%M:%SZ'), 'delivery_store_ids': list(map( lambda dsp: dsp.customer_master_master_id, self.dsps )), 'scheduled_by': self.scheduled_by, 'update': self.update.to_dict(), 'source': self.source, 'user_timezone': self.user_timezone } @sql.wrap_db_errors def create(product_id, data, orchard_identity_id, session=None): """Create scheduled product update.""" scheduled_update = ScheduledProductUpdate( product_id=product_id, sales_date_time=parser.isoparse(data['schedule_datetime']), process_at=parser.isoparse(data['process_at_datetime']), process_at_offset=data['schedule_time_offset'], created_on=datetime.now(timezone.utc), created_by=orchard_identity_id, updated_by=orchard_identity_id, updated_on=datetime.now(timezone.utc), scheduled_by=orchard_identity_id, dsps=list(map( lambda x: ScheduledProductUpdateDms(customer_master_master_id=x), data['delivery_store_ids'] )), update=ScheduledProductUpdateValues( sale_start_date=parser.isoparse( data['update']['sale_start_date']).date() ), source=OA ) session.add(scheduled_update) return response.Response( status=201, message=_get(product_id, session=session).to_dict() ) def _get(product_id, session=None): """Retrieve scheduled product update by product_id.""" result = session.query(ScheduledProductUpdate)\ .filter(ScheduledProductUpdate.product_id == product_id)\ .first() return result if result else None @sql.db_session_wrap def get(product_id, session=None): """Retrieve scheduled product update by product_id. Return: http payload """ result = _get(product_id, session=session) if not result: return response.Response( status=404, message=ERROR_MESSAGE_SCHEDULE_NOT_FOUND.format(product_id) ) return response.Response(status=200, message=result.to_dict()) @sql.wrap_db_errors def update(product_id, existing, data, orchard_identity_id, session=None): """Update existing product.""" existing.sales_date_time = parser.isoparse(data['schedule_datetime']) existing.process_at = parser.isoparse(data['process_at_datetime']) existing.process_at_offset = data['schedule_time_offset'] existing.scheduled_by = orchard_identity_id existing.update = ScheduledProductUpdateValues( sale_start_date=parser.isoparse( data['update']['sale_start_date']).date() ) existing.dsps = list(map( lambda x: ScheduledProductUpdateDms(customer_master_master_id=x), data['delivery_store_ids'] )) existing.updated_by = orchard_identity_id existing.updated_on = datetime.now(timezone.utc) existing.source = OA session.add(existing) return response.Response( status=200, message=_get(product_id, session=session).to_dict() ) @sql.db_session_wrap def upsert(product_id, data, orchard_identity_id, session=None): """Check if scheduled update data exists for the product. If data exists, update the data. If data does not exist, create new records. Args: product_id (int): Product id for which scheduled update data has to be updated. data (dict): payload for update orchard_identity_id (string): users identity id session: database session Return: response.Response: timed release data on successful update or error response. """ existing = _get(product_id, session) if existing: return update( product_id, existing, data, orchard_identity_id, session=session ) else: return create( product_id, data, orchard_identity_id, session=session ) @sql.db_session_wrap @sql.wrap_db_errors def delete(product_id, session=None): """Delete scheduled update data for a product.""" scheduled_updates = session.query(ScheduledProductUpdate) \ .filter(ScheduledProductUpdate.product_id == product_id)\ .all() if not scheduled_updates: return response.Response(status=404) for scheduled_update in scheduled_updates: session.delete(scheduled_update) return response.Response( status=202 ) @sql.wrap_db_errors def get_unsupported_stores_timed_release_ws(product_id, session): """Get product's unsupported stores timed release set in workstation. Args: product_id (int): Product id to fetch product timed release details. session (object): database session Return: response (object): unsupported stores sales_date_time and user_timezone or 404 or error. """ result = session.query(ScheduledProductUpdate).filter( ScheduledProductUpdate.product_id == product_id, ScheduledProductUpdate.source == WORKSTATION ).first() if not result: return response.Response(status=404) result_dict = result.to_dict() return response.Response( status=http_status.OK, message={ 'sales_date_time': result_dict['schedule_datetime'], 'user_timezone': result_dict['user_timezone'] } ) @sql.wrap_db_errors def check_if_unsupported_stores_schedule_oa_exists(product_id, session): """Check if unsupported stores scheduled update set in OA exists. Args: product_id (int): Product id to fetch product scheduled update details. session (object): database session Return: result (boolean): True or False. """ result = session.query(ScheduledProductUpdate).filter( ScheduledProductUpdate.product_id == product_id, ScheduledProductUpdate.source == OA ).first() if not result: return False return True @sql.wrap_db_errors def upsert_unsupported_stores_schedule_ws( product_id, data, identity_id, session): """Upsert unsupported stores schedule ws by product id. If data exists, update the data. If data does not exist, create new records. Args: product_id (int): Product id for which scheduled update data has to be upserted. data (dict): payload for update identity_id (string): users identity id session: database session Return: response.Response: timed release data on successful update or error response. """ existing_schedule = session.query(ScheduledProductUpdate).filter( ScheduledProductUpdate.product_id == product_id, ScheduledProductUpdate.source == WORKSTATION ).first() if existing_schedule: return update_unsupported_stores_schedule_ws( existing_schedule, data, identity_id, session=session ) else: return add_unsupported_stores_schedule_ws( product_id, data, identity_id, session=session ) @sql.wrap_db_errors def update_unsupported_stores_schedule_ws( existing_schedule, data, identity_id, session): """Update existing workstation set unsupported stores product schedule.""" existing_schedule.sales_date_time = data['schedule_datetime'] existing_schedule.process_at = data['process_at_datetime'] existing_schedule.process_at_offset = data['process_at_offset'] existing_schedule.scheduled_by = identity_id existing_schedule.update = ScheduledProductUpdateValues( sale_start_date=data['update']['sale_start_date'] ) existing_schedule.dsps = list(map( lambda x: ScheduledProductUpdateDms(customer_master_master_id=x), data['delivery_store_ids'] )) existing_schedule.updated_by = identity_id existing_schedule.updated_on = datetime.now(timezone.utc) existing_schedule.user_timezone = data.get('user_timezone') session.add(existing_schedule) return response.Response( status=http_status.OK, message={'unsupported_releases': existing_schedule.to_dict()} ) @sql.wrap_db_errors def add_unsupported_stores_schedule_ws( product_id, data, identity_id, session): """Add workstation set unsupported stores product schedule.""" schedule = ScheduledProductUpdate( product_id=product_id, sales_date_time=data['schedule_datetime'], process_at=data['process_at_datetime'], process_at_offset=data['process_at_offset'], created_on=datetime.now(timezone.utc), created_by=identity_id, updated_by=identity_id, updated_on=datetime.now(timezone.utc), scheduled_by=identity_id, dsps=list(map( lambda x: ScheduledProductUpdateDms(customer_master_master_id=x), data['delivery_store_ids'] )), update=ScheduledProductUpdateValues( sale_start_date=data['update']['sale_start_date'] ), source=WORKSTATION, user_timezone=data.get('user_timezone') ) session.add(schedule) return response.Response( status=http_status.OK, message={'unsupported_releases': schedule.to_dict()} ) @sql.wrap_db_errors def get_product_scheduled_update_by_store_id(product_id, store_id, session): """ Retrieve the scheduled update details for a specific product and store. It checks whether the event schedule process time has already passed. If so, return 404 response. Otherwise, the sales date time is padded by 2 days and returned in the response. Args: product_id (int): The unique identifier of the product. store_id (int): The DMS identifier of the store. session (Session): Database session. Returns: response.Response: - 404 NOT_FOUND if no schedule found or process time has expired. - 200 OK with payload containing padded sales_date_time. """ result = session.query(ScheduledProductUpdate).join( ScheduledProductUpdate.dsps).filter( ScheduledProductUpdate.product_id == product_id, ScheduledProductUpdateDms.customer_master_master_id == store_id ).first() if not result: return response.Response(status=http_status.NOT_FOUND) schedule = result.to_dict() sales_date_time = parser.isoparse(schedule['schedule_datetime']) process_at_datetime = parser.isoparse(schedule['process_at_datetime']) current_datetime = datetime.now(timezone.utc) is_after_event_schedule_processed = ( current_datetime > process_at_datetime ) if is_after_event_schedule_processed: earliest_tz = timezone(timedelta(hours=-12)) earliest_sales_date_time = ( sales_date_time.astimezone(earliest_tz).strftime( '%Y-%m-%dT%H:%M:%SZ') ) return response.Response( status=http_status.OK, message={ 'timed': { 'sales_date_time': earliest_sales_date_time }, 'staggered': None } ) padded_sales_date_time = ( sales_date_time + timedelta(days=SALES_DATE_DAYS_OFFSET) ).strftime('%Y-%m-%dT%H:%M:%SZ') return response.Response( status=http_status.OK, message={ 'timed': { 'sales_date_time': padded_sales_date_time }, 'staggered': None } ) @sql.wrap_db_errors def delete_scheduled_update_ws(product_id, session=None): """Delete workstation set scheduled update data for the product_id. If no schedule update found then return 404. Args: product_id (int): Product id for which timed release data has to be updated. session (object): database session """ scheduled_updates = session.query(ScheduledProductUpdate) \ .filter( ScheduledProductUpdate.product_id == product_id, ScheduledProductUpdate.source == WORKSTATION ).all() if not scheduled_updates: return response.Response(status=http_status.NOT_FOUND) for scheduled_update in scheduled_updates: session.delete(scheduled_update) return response.Response(status=http_status.OK) @sql.wrap_db_errors def get_scheduled_update_ws(product_id, session): """Get workstation set scheduled product update by product_id. If no schedule update found then return None. Args: product_id (int): Product id for which scheduled update data has to be fetched. session (object): database session """ result = session.query(ScheduledProductUpdate)\ .filter(ScheduledProductUpdate.product_id == product_id)\ .filter(ScheduledProductUpdate.source == WORKSTATION)\ .first() if not result: return response.Response(status=http_status.NOT_FOUND) return response.Response(status=http_status.OK, message=result.to_dict())