"""Blacklist Words Model.""" import csv import datetime from owsresponse import response from sentry_sdk import capture_exception from sqlalchemy import Column from sqlalchemy import DateTime from sqlalchemy import ForeignKey from sqlalchemy import func from sqlalchemy import Integer from sqlalchemy import VARCHAR from sqlalchemy.exc import SQLAlchemyError from blacklist_manager.connectors import mysql from blacklist_manager.constants import error from blacklist_manager.constants import header from blacklist_manager.models.blacklist_reasons import BlacklistReasons from blacklist_manager.models.orchadmin_users import OrchadminUsers class BlacklistWords(mysql.BaseModel): """Blacklist Words model.""" __tablename__ = 'blacklist_words' blacklist_word_id = Column( 'id', Integer, primary_key=True, autoincrement=True, nullable=False) word = Column(VARCHAR(240), nullable=False) notes = Column(VARCHAR(256), nullable=True) blacklist_reason_id = Column( Integer, ForeignKey('blacklist_reasons.id'), nullable=False) vendor_id = Column(Integer, nullable=True) added_by = Column( Integer, ForeignKey('orchadmin_users.id'), nullable=False) date_added = Column(DateTime, nullable=True) added_by_identity = Column(VARCHAR(36), nullable=True) def to_dict(self): """Representation of a BlacklistWords row.""" return { 'id': self.blacklist_word_id, 'word': self.word, 'notes': self.notes, 'blacklist_reason_id': self.blacklist_reason_id, 'vendor_id': self.vendor_id, 'added_by': self.added_by, 'date_added': self.date_added and self.date_added.strftime('%Y-%m-%d'), 'added_by_identity': self.added_by_identity } def get_blacklist_words(page_offset=None, page_limit=None, term=None): """Get all the records of blacklist words. Args: page_offset (int): page offset. page_limit (int): total number of records per page. term (string): Blacklist term to match. Returns: Response: A response object with existing data. """ with mysql.ar_db_session() as session: try: filters = [ (BlacklistWords.blacklist_reason_id == BlacklistReasons.blacklist_reason_id), (BlacklistWords.added_by == OrchadminUsers.orchadmin_users_id) ] if term: filters.append((BlacklistWords.word.like('%{}%'.format(term)))) result = session.query( BlacklistWords, BlacklistReasons.reason, OrchadminUsers.f_name, OrchadminUsers.l_name).filter( *filters).order_by(BlacklistWords.word).offset( page_offset).limit(page_limit).all() data = [] for items in result: blacklist_word_dict = items.BlacklistWords.to_dict() blacklist_reason_dict = { 'blacklist_reason': { 'id': blacklist_word_dict['blacklist_reason_id'], 'reason': items[1] } } orchadmin_user_dict = { 'added_by': { 'id': blacklist_word_dict['added_by'], 'name': items[2] + ' ' + items[3] } } del (blacklist_word_dict['blacklist_reason_id']) del (blacklist_word_dict['added_by']) data.append( { **blacklist_word_dict, **blacklist_reason_dict, **orchadmin_user_dict } ) words = { 'items': data, 'pagination': dict( type='standard', page_offset=page_offset, page_limit=page_limit) } return response.Response(message=words, status=200) except SQLAlchemyError: capture_exception() return response.create_fatal_response() def write_blacklist_words_to_file(file_path): """Write all blacklist words to a csv file. Args: file_path (string): Path to file on disk. Returns: Response: A response object with existing data. """ with mysql.ar_db_session() as session: try: filters = [ (BlacklistWords.blacklist_reason_id == BlacklistReasons.blacklist_reason_id), (BlacklistWords.added_by == OrchadminUsers.orchadmin_users_id) ] rows = session.query( BlacklistWords.blacklist_word_id, BlacklistWords.word, BlacklistWords.notes, BlacklistReasons.reason, BlacklistWords.date_added, func.concat(OrchadminUsers.f_name, ' ', OrchadminUsers.l_name) ).filter(*filters).all() if not rows or len(rows) < 1: return response.create_not_found_response( 'no blacklist records found') with open(file_path, 'w', newline='\n', encoding='utf-8') as file: writer = csv.writer( file, delimiter=',', quotechar='"', quoting=csv.QUOTE_NONNUMERIC) writer.writerow(header.EXPORT_HEADER) for data in rows: writer.writerow(data) return response.Response(message={'count': len(rows)}) except Exception as err: capture_exception(err) return response.create_fatal_response(str(err)) def get_blacklist_words_count(term=None): """Get count of blacklisted words. Args: term (string): Blacklist term to match. Returns: response.Response: contains number of blacklisted words """ try: with mysql.ar_db_session() as session: filters = [ (BlacklistWords.blacklist_reason_id == BlacklistReasons.blacklist_reason_id), (BlacklistWords.added_by == OrchadminUsers.orchadmin_users_id) ] if term: filters.append((BlacklistWords.word.like('%{}%'.format(term)))) count = session.query(func.count( BlacklistWords.blacklist_word_id)).filter( *filters).one_or_none() if count: message = count[0] return response.Response(message) except SQLAlchemyError: capture_exception() return response.create_fatal_response() def delete_blacklist_word(blacklist_id): """Delete word by blacklist id. Args: blacklist_id (int): Blacklist id. Returns: response.Response: success response if object exists else 404. """ try: with mysql.ar_db_session() as session: count = session.query(BlacklistWords).filter_by( blacklist_word_id=blacklist_id).delete() if not count: return response.create_not_found_response( error.BLACKLIST_WORD_NOT_FOUND.format(blacklist_id)) return response.Response() except SQLAlchemyError as e: capture_exception(e) return response.create_fatal_response(e.args) def create_blacklist_word(blacklist_word_data): """Save data in blacklist_words table. Args: blacklist_word_data (dict): properties to assign to the new blacklist word. Returns: response.Response: wrapper containing the created blacklist word id or errors. """ try: with mysql.ar_db_session() as session: blacklist_word_data['date_added'] = datetime.datetime.now() blacklist_data = BlacklistWords(**blacklist_word_data) session.add(blacklist_data) session.flush() return response.Response( message={'id': blacklist_data.blacklist_word_id}, status=201) except SQLAlchemyError as e: capture_exception(e) return response.create_fatal_response(e.args) def update_blacklist_word(blacklist_id, blacklist_word_data): """Update a blacklist word. Args: blacklist_id (int): unique identifier of blacklist_word. blacklist_word_data (dict): properties to update of the existing blacklist word. Returns: Response: A response object with updated blacklist id. """ try: with mysql.ar_db_session() as session: count = session.query(BlacklistWords).\ filter(BlacklistWords.blacklist_word_id == blacklist_id).\ update(blacklist_word_data) if not count: return response.create_not_found_response( error.BLACKLIST_WORD_NOT_FOUND.format(blacklist_id)) return response.Response(message={'id': blacklist_id}, status=200) except SQLAlchemyError as e: capture_exception(e) return response.create_fatal_response(e.args) def get_blacklist_words_for_validation(session): """Get all the records of blacklist words. Returns: Response: A response object with existing data. """ try: result = session.query( BlacklistWords.word, BlacklistWords.blacklist_reason_id ).all() words = { 'items': [{'word': item[0], 'reason_id': item[1]} for item in result], } return response.Response(message=words, status=200) except SQLAlchemyError: capture_exception() return response.create_fatal_response() def get_blacklist_words_by_reason_ids(session, reason_ids): """Get all the records of the blacklist words filtered by reason IDs. Args: reason_ids (Iterable[int]): Iterable of blacklist reason IDs to filter by. Returns: Response: A response object with existing data. """ try: result = session.query( BlacklistWords.word, BlacklistWords.blacklist_reason_id ).filter( BlacklistWords.blacklist_reason_id.in_(reason_ids) ).all() words = { 'items': [{'word': item[0], 'reason_id': item[1]} for item in result], } return response.Response(message=words, status=200) except SQLAlchemyError: capture_exception() return response.create_fatal_response()