"""Rule model.""" import json from owsresponse import response from redis import exceptions as redis_exceptions from sentry_sdk import capture_exception from sqlalchemy import Column from sqlalchemy import Enum from sqlalchemy import func from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy.dialects.mysql import TIMESTAMP from vector_job_rules.connectors import mysql from vector_job_rules.connectors.redis import redis_client from vector_job_rules.constants.mysql import DATE_RESPONSE_FORMAT from vector_job_rules.models.rule_condition import DEFAULT_FIELDS as RC_DEFAULT_FIELDS from vector_job_rules.models.rule_condition import RuleCondition from vector_job_rules.models.rule_condition_field import RuleConditionField class Rule(mysql.BaseModel): """Rule Model.""" __tablename__ = 'rule' rule_id = Column('rule_id', Integer, primary_key=True) status = Column('status', Enum('inactive', 'active'), nullable=True) priority = Column('priority', Integer, nullable=True) description = Column('description', String, nullable=True) last_updated = Column( 'last_updated', TIMESTAMP, default=func.current_timestamp(), onupdate=func.current_timestamp(), ) updated_by = Column('updated_by', Integer, nullable=True) DEFAULT_FIELDS = { Rule.rule_id, Rule.status, Rule.priority, Rule.description, Rule.last_updated, Rule.updated_by, } def get_rules(status=None, rule_ids=None): """Get all rules. Args: status (str): inactive or active are acceptable status values. rule_ids (tuple): a list of rule ids to filter on. Returns: response.Response """ with mysql.db_session() as session: rows = ( session.query( *DEFAULT_FIELDS, *RC_DEFAULT_FIELDS, RuleConditionField.rule_condition_field_id, RuleConditionField.name, RuleConditionField.rule_condition_field_type, RuleConditionField.description.label('rcf_description') ) .join(RuleCondition, RuleCondition.rule_id == Rule.rule_id) .join( RuleConditionField, RuleConditionField.rule_condition_field_id == RuleCondition.rule_condition_field_id, # noqa ) ) if status: rows = rows.filter(Rule.status == status) if rule_ids: rows = rows.filter(Rule.rule_id.in_(rule_ids)) rows = rows.all() response_data = [row._asdict() for row in rows] return response.Response(response_data) def get_rules_from_cache(status=None, rule_ids=None): """Get rules from cache. We allow filters on cached data so we look inside the hgetall to see if we want to append to list when we respond. We decode each json string for each rule. Args: status (str): inactive or active are acceptable status values. rule_ids (tuple): a list of rule ids to filter on. Returns: response.Response """ all_rules = [] rules = redis_client.hgetall('ows_vector_job_rules') for rule in rules: rule_data = json.loads(rules.get(rule)) if rule_ids and rule_data['id'] not in rule_ids: continue if status and rule_data['status'] != status: continue all_rules.append(rule_data) return response.Response(all_rules) def push_rules_into_cache(formatted_rules): """Push rules into cache. Each rule gets loaded to a hash key. We call the format helper here because we do not ever want to store to redis without making sure its formatted in a final format which is json safe. We encode to json so that we store as a string. We decode on retrieval or client access cache directly will decode on retrieval. Args: formatted_rules (response.Response): Takes the rules pre formatted. Returns: response.Response """ try: for rule in formatted_rules.message: rule_data = json.dumps(rule) redis_client.hset('ows_vector_job_rules', rule['id'], rule_data) redis_client.set( 'ows_vector_job_rules_rule_{}'.format(rule['id']), rule_data ) except (redis_exceptions.ConnectionError, redis_exceptions.TimeoutError) as err: print('**ERROR==', err) capture_exception(err) return response.Response() def format_rules_into_response_format(rules_from_db): """Format rules into response format. This will take in a get_rules db model response and format it in a json safe way so it can be access directly in logic or if we need to store in redis in a safe format to decode. Args: rules_from_db (response.Response): The model response for all rules. Returns: response.Response """ tmp = {} for x in rules_from_db.message: rule_id = x['rule_id'] if rule_id not in tmp: last_updated = x['last_updated'].strftime(DATE_RESPONSE_FORMAT) tmp[rule_id] = { 'id': rule_id, 'conditions': [], 'status': x['status'], 'priority': x['priority'], 'description': x['description'], 'last_updated': last_updated, 'updated_by': x['updated_by'], } tmp_cond = { 'rule_condition_field_id': x['rule_condition_field_id'], 'rule_condition_field_type': x['rule_condition_field_type'], 'name': x['name'], 'rule_condition_field_description': x['rcf_description'], 'rule_condition_value': x['rule_condition_value'], } tmp[rule_id]['conditions'].append(tmp_cond) return response.Response(list(tmp.values())) def filter_formatted_rules(formatted_rules, status=None, rule_ids=None): """Filter formatted rules. Args: formatted_rules (response.Response): The formatted rules. status (str): active or inactive rule_ids (tuple): Takes a tuple of rule ids Returns: response.Response """ all_rules = [] for rule in formatted_rules.message: if rule_ids and rule['id'] not in rule_ids: continue if status and rule['status'] != status: continue all_rules.append(rule) return response.Response(all_rules) def update_rule(rule_id, priority=None, description=None, updated_by=None): """Update a single rule. Args: rule_id (int): A single rule id. priority (int): A single unique priority value. description (str): Describes the rule. updated_by (int): OA user_id of the person who updated the rules Returns: response.Response Raises: RuleNotFound: If rule is not found this is raised. """ with mysql.db_session() as session: update = {} if priority: update['priority'] = priority if description: update['description'] = description if updated_by: update['updated_by'] = updated_by session.query(Rule).filter(Rule.rule_id == rule_id).update(update) row = session.query(*DEFAULT_FIELDS).filter(Rule.rule_id == rule_id).one() return response.Response(message=row._asdict())