"""Model layer for graph based subscriptions.""" from typing import Any import stringcase from connector_neo4j import get_session from ddtrace import tracer from flask import g from neo4j.time import DateTime from owsresponse import response, status as ows_status from notifications.utils.neo4j import strip_query from notifications.validation.relationship import ( RELATIONSHIP_FROM_IDENTITY, RELATIONSHIP_FROM_PROFILE, ) DELETED_PREFIX = 'DELETED_' def create_subscription( # noqa: PLR0913 profile_type: str, profile_id: int, entity_type: str, entity_id: str, relationship: str, automatic: bool, ) -> response.Response: """Create given relationship beteen a Profile and another node. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node to follow (GlobalParticipant, etc) entity_id (str): The id of node to follow. relationship (str): The relationship type automatic (bool): If the subscription was created via automated process Returns: Response: 201 created. """ session = get_session() subscribe_query = strip_query(f"""MATCH (p:Profile),(pa:{entity_type}) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND pa.id = $entity_id CREATE (p)-[r:{relationship} {{dateCreated: localdatetime(), automatic: $automatic}}]->(pa) RETURN r.dateCreated""") result = session.run( subscribe_query, profile_type=profile_type, profile_id=profile_id, entity_id=entity_id, relationship=relationship, automatic=automatic, ) if not result.peek(): missing_query = strip_query(f""" OPTIONAL MATCH (p:Profile {{profileId: $profile_id, profileType: $profile_type}}) OPTIONAL MATCH (pa:{entity_type} {{id: $entity_id}}) RETURN p,pa""") result = session.run( missing_query, profile_type=profile_type, profile_id=profile_id, entity_id=entity_id ) nodes = result.single() message = 'Unknown error' if not nodes['pa']: message = f'{entity_type} not found with id {entity_id}' elif not nodes['p']: message = f'Profile not found with id {profile_id} and type {profile_type}' g.log.info(message) return response.create_not_found_response(message) return response.Response(status=201) def soft_undelete_subscription( # noqa: PLR0913 profile_type: str, profile_id: int, entity_type: str, entity_id: str, relationship: str, automatic: bool, ) -> response.Response: """Soft undelete relationship for profile and entity. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node to follow (GlobalParticipant, etc) entity_id (str): The id of node to follow. relationship (str): The relationship to undelete automatic (bool): Set attr on relationship if modified automatically Returns: Response: the relationship. """ return _update_subscription_relationship( profile_type, profile_id, entity_type, entity_id, f'{DELETED_PREFIX}{relationship}', relationship, 'dateCreated', automatic, ) def soft_delete_subscription( profile_type: str, profile_id: int, entity_type: str, entity_id: str, relationship: str ) -> response.Response: """Soft delete relationship for profile and entity. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node to follow (GlobalParticipant, etc) entity_id (str): The id of node to follow. relationship (str): The relationship to delete Returns: Response: the relationship. """ return _update_subscription_relationship( profile_type, profile_id, entity_type, entity_id, relationship, f'{DELETED_PREFIX}{relationship}', 'dateDeleted', ) def _update_subscription_relationship( # noqa: PLR0913 profile_type: str, profile_id: int, entity_type: str, entity_id: str, old_relationship: str, new_relationship: str, timestamp_label: str, automatic: bool | None = None, ) -> response.Response: """Change relationship type from old to new for profile and entity. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node to follow (GlobalParticipant, etc) entity_id (str): The id of node to follow. old_relationship (str): The relationship type to match new_relationship (str): The relationship to overwrite timestamp_label (str): label to give current timestamp on relationship automatic (bool): Set attr on relationship if modified automatically Returns: Response: the relationship. """ session = get_session() unsubscribe_query = strip_query(f"""MATCH (p:Profile)-[r:{old_relationship}]->(pa:{entity_type}) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND pa.id = $entity_id SET r.{timestamp_label} = localdatetime(), r.automatic = CASE WHEN $automatic IS NOT NULL THEN $automatic ELSE r.automatic END WITH r CALL apoc.refactor.setType(r, '{new_relationship}') YIELD input, output RETURN input, output""") result = session.run( unsubscribe_query, profile_type=profile_type, profile_id=profile_id, entity_id=entity_id, automatic=automatic, ) # neo4j client will execute but won't return errors unless results read [x for x in result] return response.Response(status=204) def fetch_ids(node_type: str, attr: str, attr_values: list[str]) -> list[str]: """Get true ids of nodes using another attribute. Args: node_type (str): Label of node to query attr (str): attribute of node to examine attr_values (list): values of attr to match Response: list: values of id attributes on matching nodes """ session = get_session() get_query = strip_query(f"""MATCH (n:{node_type}) WHERE n.{attr} IN $attr_values RETURN n """) result = session.run(get_query, attr_values=attr_values) return [x['n']['id'] for x in result] if result else [] def get_subscriptions( profile_type: str, profile_id: int, entity_type: str, relationship: str, logic_args: dict[str, Any], ) -> response.Response: """Get all the Participants that the Profile has given relationship to. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node being followed relationship (str): The relationship type logic_args (dict): Query modifiers Returns: Response: Participants that the Profile has given relationship to """ ids = logic_args['ids'] sub_type = logic_args['sub_type'] skip = logic_args['offset'] order_dir = logic_args['order_dir'].upper() order_by = logic_args['order_by'] if order_by == 'created_at': order_by = 'r.dateCreated' # convert to neo4j syntax limit = logic_args['limit'] limit_clause = f'LIMIT {limit}' if limit else '' and_clause = '' if entity_type.lower() == 'subaccount': and_clause = 'AND pa.isDeleted = false' elif entity_type.lower() == 'product:orchard': and_clause = """AND ( (p.fullCatalogAccess IS NOT null AND p.fullCatalogAccess = true) OR EXISTS { (p)-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->() <-[:BELONGS_TO|CREATED_FOR_PARTICIPANT]-(:Project)-[:INCLUDES]->(pa) } ) """ # handle pulling back deleted edges deleted_state = logic_args['state'] deleted_relationship = f'{DELETED_PREFIX}{relationship}' if deleted_state == 'undeleted': edge = relationship if deleted_state == 'deleted': edge = deleted_relationship elif deleted_state == 'all': edge = f'{deleted_relationship}|{relationship}' session = get_session() list_query = strip_query(f""" MATCH (p:Profile)-[r:{edge}]->(pa:{entity_type}) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND ( SIZE($ids) = 0 OR pa.id IN $ids ) AND ( $sub_type IS NULL OR $sub_type IN pa.types ) {and_clause} RETURN pa, r.dateCreated AS rel_created, TYPE(r) AS state, CASE TYPE(r) STARTS WITH '{DELETED_PREFIX}' WHEN true THEN r.dateDeleted ELSE r.dateCreated END AS last_modified ORDER BY {order_by} {order_dir} SKIP $skip {limit_clause} """) result = session.run( list_query, profile_type=profile_type, profile_id=profile_id, entity_type=entity_type, relationship=relationship, ids=ids, sub_type=sub_type, skip=skip, ) participants = [] for each in result: pa = each.get('pa') allow_list = ['id', 'name', 'spotifyId', 'chartmetricId', 'isrc', 'upc', 'channelId'] data = { 'created_at': str(each.get('rel_created')), 'last_modified': str(each.get('last_modified')), 'deleted': each.get('state').startswith(DELETED_PREFIX), } for k, v in pa.items(): v_ = v if isinstance(v, DateTime): v_ = str(v) if k in allow_list: data[stringcase.snakecase(k)] = v_ participants.append(data) return response.Response(participants) def has_deleted_subscription( profile_type: str, profile_id: int, entity_type: str, entity_id: str, relationship: str ) -> response.Response: """Check to see if a deleted relationship exists for Profile to entity. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node to follow (GlobalParticipant, etc) entity_id (str): The id of node to follow. relationship (str): The relationship type Returns: Response: 200 exists, 404 not exists. """ return has_subscription( profile_type, profile_id, entity_type, entity_id, f'{DELETED_PREFIX}{relationship}' ) def has_subscription( profile_type: str, profile_id: int, entity_type: str, entity_id: str, relationship: str ) -> response.Response: """Check to see if a relationship exists for Profile to entity. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. entity_type (str): The type of node to follow (GlobalParticipant, etc) entity_id (str): The id of node to follow. relationship (str): The relationship type Returns: Response: 200 exists, 404 not exists. """ session = get_session() unsubscribe_query = strip_query(f"""MATCH (p:Profile)-[r:{relationship}]->(pa:{entity_type}) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND pa.id = $entity_id RETURN r, pa""") check_result = session.run( unsubscribe_query, profile_type=profile_type, profile_id=profile_id, entity_id=entity_id, relationship=relationship, ) status = ows_status.OK if check_result.peek() else ows_status.NOT_FOUND return response.Response(status=status, message={'exists': status == ows_status.OK}) def soft_undelete_notification_subscription( profile_type: str, profile_id: int, notification_type: str, feed_type: str, followed_entity: str | None, ) -> response.Response: """Soft undelete notification subscriptions. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. notification_type (str): The type of notification feed_type (str): The name of feed followed_entity (str): Optional dimension on node type event source """ relationship = 'HAS_SUBSCRIPTION' return _update_notification_subscription_relationship( profile_type, profile_id, notification_type, feed_type, followed_entity, f'{DELETED_PREFIX}{relationship}', relationship, 'dateCreated', ) def soft_delete_notification_subscription( profile_type: str, profile_id: int, notification_type: str, feed_type: str, followed_entity: str | None, ) -> response.Response: """Soft delete notification subscriptions. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. notification_type (str): The type of notification feed_type (str): The name of feed followed_entity (str): Optional dimension on node type event source """ relationship = 'HAS_SUBSCRIPTION' return _update_notification_subscription_relationship( profile_type, profile_id, notification_type, feed_type, followed_entity, relationship, f'{DELETED_PREFIX}{relationship}', 'dateDeleted', ) def _update_notification_subscription_relationship( # noqa: PLR0913 profile_type: str, profile_id: int, notification_type: str, feed_type: str, followed_entity: str | None, old_relationship: str, new_relationship: str, timestamp_label: str, ) -> response.Response: notification_type = stringcase.camelcase(notification_type) feed_type = stringcase.camelcase(feed_type) session = get_session() query = strip_query(f"""MATCH (p:Profile)-[r:{old_relationship}]->(s:Subscription) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND s.notificationType = $notification_type AND s.feedType = $feed_type AND ( s.followedEntity = $followed_entity OR $followed_entity_empty ) SET r.{timestamp_label} = localdatetime() WITH r CALL apoc.refactor.setType(r, '{new_relationship}') YIELD input, output RETURN input, output""") result = session.run( query, profile_type=profile_type, profile_id=profile_id, notification_type=notification_type, feed_type=feed_type, followed_entity=followed_entity, followed_entity_empty=(followed_entity is None), ) # neo4j client will execute but won't return errors unless results read [x for x in result] return response.Response(status=204) @tracer.wrap() def get_subscribed_profiles( # noqa: PLR0913 profile_types: list[str], entity_type: str, entity_id: str | int, relationship: str, relationship_from: str = RELATIONSHIP_FROM_PROFILE, subscription_name: str | None = None, ) -> response.Response: """Get profiles that have relationship with entity. Args: profile_types (list): Allow list of profile types entity_type (str): Entity node type to match profile entity_id (str): ID of node entity to match profile relationship (str): Edge name to match profile and entity relationship_from (str): Node where the relationship starts. It can be Identity or profile. subscription_name (str): Subscription name. """ if tracer.enabled: span = tracer.current_span() if span: span.set_tag('profile_types', ', '.join(profile_types)) span.set_tag('entity_type', entity_type) span.set_tag('entity_id', str(entity_id)) span.set_tag('relationship_from', relationship_from) span.set_tag('subscription_name', subscription_name) session = get_session() if relationship_from == RELATIONSHIP_FROM_IDENTITY: # Identity level subscriptions are auto follows when a user # subscribes to their entire catalog. get_query = strip_query(f""" MATCH (s:Subscription), (g:{entity_type}) WHERE s.followedEntity = $entity_type AND s.name = $subscription_name AND g.id = $entity_id WITH s, g OPTIONAL MATCH (s)<-[:{relationship}]-(i:Identity) -[:HAS_PROFILE]->(p:Profile)-[:HAS_ACCESS_TO]->(g) WHERE p.profileType IN $profile_types AND i.active = 'Y' OPTIONAL MATCH (s)<-[:{relationship}]-(i2:Identity) -[:HAS_PROFILE]->(p2:Profile)-[:HAS_ACCESS_TO]->(v:Vendor)-[:OWNS]->(g) WHERE p2.profileType IN $profile_types AND i2.active = 'Y' WITH collect(p) + collect(p2) as listProfiles UNWIND listProfiles as profile RETURN DISTINCT profile """) else: # Profile level subscriptions deduplicated by identity. # When one identity has multiple profiles subscribed to the same entity, # prefer the vendor-level profile. get_query = strip_query(f"""MATCH (g:{entity_type}) WHERE g.id = $entity_id OPTIONAL MATCH (g)<-[:{relationship}]-(p:Profile)<-[:HAS_PROFILE]-(i:Identity) WHERE p.profileType IN $profile_types AND i.active = 'Y' OPTIONAL MATCH (g)<-[:OWNS]-(v:Vendor)<-[:{relationship}]-(p2:Profile)<-[:HAS_PROFILE]-(i2:Identity) WHERE p2.profileType IN $profile_types AND i2.active = 'Y' WITH collect({{profile: p, identity: i}}) + collect({{profile: p2, identity: i2}}) AS pairs UNWIND pairs AS profile_identity WITH profile_identity WHERE profile_identity.identity IS NOT NULL AND profile_identity.profile IS NOT NULL WITH profile_identity.identity.id AS identity_id, profile_identity.profile AS profile WITH identity_id, profile, EXISTS {{ (profile)-[:HAS_ACCESS_TO]->(:Subaccount) }} AS has_sub ORDER BY has_sub ASC WITH identity_id, collect(profile)[0] AS selected RETURN DISTINCT selected AS profile""") # noqa: E501 result = session.run( get_query, entity_id=entity_id, profile_types=profile_types, subscription_name=subscription_name, entity_type=entity_type, ) profiles = [] for each in result: node = each['profile'] profiles.append( { stringcase.snakecase(k): v for k, v in node.items() if k in ['profileId', 'profileType'] } ) return response.Response(profiles) @tracer.wrap() def get_identities_with_entities_subscriptions( profile_types: list[str], entity_types: list[str], entity_ids: list[int], relationships: list[str], subscription_names: list[str], ) -> list[dict[str, Any]]: """Get identities that subscribed to a given entity. Args: profile_types (list): Allow list of profile types entity_types (list): Entity node types to match profile entity_ids (list): ID's of node entity to match profile relationships (list): Edge names to match profile and entity subscription_names (list): Subscription names. """ if tracer.enabled: span = tracer.current_span() if span: span.set_tag('profile_types', ', '.join(profile_types)) span.set_tag('relationships', ', '.join(relationships)) span.set_tag('entity_ids', ', '.join(map(str, entity_ids))) span.set_tag('entity_types', ', '.join(entity_types)) span.set_tag('subscription_names', ', '.join(subscription_names)) session = get_session() entity_type = entity_types[0] get_query = strip_query(f"""MATCH (g:{entity_type}) WHERE g.id IN $entity_ids OPTIONAL MATCH (g)<-[r]-(p1:Profile)<-[:HAS_PROFILE]-(i1:Identity) WHERE type(r) IN $relationships AND p1.profileType IN $profile_types OPTIONAL MATCH (g)<-[:OWNS]-(v:Vendor)<-[r2]-(p2:Profile)<-[:HAS_PROFILE]-(i2:Identity) WHERE type(r2) IN $relationships AND p2.profileType IN $profile_types OPTIONAL MATCH (s:Subscription)<-[:HAS_AUTO_FOLLOWED]-(i3:Identity)-[:HAS_PROFILE]-> (p3:Profile)-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(g) WHERE p3.profileType IN $profile_types AND s.name IN $subscription_names WITH collect(i1) + collect(i2) + collect(i3) as listIdentities UNWIND listIdentities as identity RETURN DISTINCT identity""") result = session.run( get_query, entity_ids=entity_ids, profile_types=profile_types, subscription_names=subscription_names, relationships=relationships, ) identities = [] for each in result: node = each['identity'] identities.append({stringcase.snakecase(k): v for k, v in node.items()}) return identities def has_deleted_notification_subscription( profile_type: str, profile_id: int, notification_type: str, feed_type: str, followed_entity: str | None, ) -> response.Response: """Check to see if a deleted relationship exists for Profile->Subscription. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. notification_type (str): The type of notification feed_type (str): The name of feed relationship (str): Name of connection between Subscription-Profile followed_entity (str): Optional dimension on node type event source Returns: Response: 200 exists, 404 not exists. """ return has_notification_subscription( profile_type, profile_id, notification_type, feed_type, followed_entity, f'{DELETED_PREFIX}HAS_SUBSCRIPTION', ) def has_notification_subscription( # noqa: PLR0913 profile_type: str, profile_id: int, notification_type: str, feed_type: str, followed_entity: str | None, relationship: str = 'HAS_SUBSCRIPTION', ) -> response.Response: """Check to see if a relationship exists for Profile->Subscription. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. notification_type (str): The type of notification feed_type (str): The name of feed relationship (str): Name of connection between Subscription-Profile followed_entity (str): Optional dimension on node type event source Returns: Response: 200 exists, 404 not exists. """ notification_type = stringcase.camelcase(notification_type) feed_type = stringcase.camelcase(feed_type) session = get_session() exists_query = strip_query(f"""MATCH (p:Profile)-[r:{relationship}]->(s:Subscription) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND s.notificationType = $notification_type AND s.feedType = $feed_type AND ( s.followedEntity = $followed_entity OR $followed_entity_empty ) RETURN r, s, p""") check_result = session.run( exists_query, profile_type=profile_type, profile_id=profile_id, notification_type=notification_type, feed_type=feed_type, followed_entity=followed_entity, followed_entity_empty=(followed_entity is None), relationship=relationship, ) status = ows_status.OK if check_result.peek() else ows_status.NOT_FOUND return response.Response(status=status, message={'exists': status == ows_status.OK}) def get_all_notifications_for_profile(profile_type: str, profile_id: int) -> list[dict[str, Any]]: """Get all enabled notifications for a given profile. Args: profile_type (str): Profile type (e.g. LabelProfile). profile_id (int): Profile id. Returns: Response: a list of all active notifications. """ session = get_session() list_query = strip_query(""" MATCH (p:Profile)-[r:HAS_SUBSCRIPTION]->(s:Subscription) WHERE p.profileId = $profile_id AND p.profileType = $profile_type RETURN s """) result = session.run(list_query, profile_type=profile_type, profile_id=profile_id) notifications = [] for each in result: s = each.get('s') notifications.append( { 'notification_type': stringcase.snakecase(s['notificationType']), 'feed_type': stringcase.snakecase(s['feedType']), 'followed_entity': s['followedEntity'], } ) return notifications def get_all_subscriptions_for_identity( identity_id: str, profile_types: list[str] | None = None, app_ids: list[str] | None = None ) -> response.Response: """Get all enabled subscriptions for a given identity. Args: identity_id (str): identity id. profile_types (list): List of profile types to filter. app_ids (list): List of application names to filter only selected subscriptions. Returns: Response: a list of all active subscriptions with list of followedEntity. """ where_clause = ['true'] if app_ids: where_clause.append('s.appId IN $appIds') if profile_types: where_clause.append('p.profileType IN $profileTypes') session = get_session() # collect all the profile level follows before looking for AUTO_FOLLOW so we get unique # Subscriptions in result even when we have multiple profiles. list_query = strip_query( f"""Match (i:Identity {{id: $identityId}})-[:HAS_PROFILE]->(p:Profile) -[:HAS_SUBSCRIPTION]->(s:Subscription) WHERE {' AND '.join(where_clause)} OPTIONAL MATCH (p)-[rel_name]->(r) WHERE apoc.rel.type(rel_name) = "HAS_FOLLOWED_" + toUpper(s.name) WITH i, s, apoc.coll.toSet(collect( {{labels: labels(r), uuid: coalesce(r.uuid, r.id), name: r.name}} )) as resources OPTIONAL MATCH (i)-[auto:HAS_AUTO_FOLLOWED]->(s) WITH *, size(collect(auto)) as sizeOfAuto RETURN s.name as name, s.notificationType as notificationType, s.appId as appId, sizeOfAuto > 0 as followAllResources, CASE WHEN sizeOfAuto > 0 or resources IS NULL THEN null ELSE [r in resources WHERE r.uuid is not null] END as followResources """ ) result = session.run( list_query, identityId=identity_id, appIds=app_ids, profileTypes=profile_types ) subscriptions = [dict(x) for x in result] return response.Response(subscriptions) def get_all_subscriptions( app_ids: list[str] | None = None, notification_type: str | None = None ) -> response.Response: """Get all subscriptions. Args: app_ids (list): Application names to filter only selected subscriptions. notification_type (str): Type of notification ie. Email only or Push. Returns: Response: a list of all subscriptions. """ session = get_session() where_clause = ['true'] if app_ids: where_clause.append('s.appId IN $appIds') if notification_type: where_clause.append('s.notificationType = $notificationType') list_query = strip_query( f"""MATCH (s:Subscription) WHERE {' AND '.join(where_clause)} RETURN s.name as name, s.notificationType as notificationType, s.appId as appId, s.feedType as feedType, collect(s.followedEntity) as followedEntityTypes """ ) result = session.run(list_query, appIds=app_ids, notificationType=notification_type) subscription = [] for each in result: subscription.append(dict(each)) return response.Response(subscription) def create_notification_subscription( profile_type: str, profile_id: int, notification_type: str, feed_type: str, followed_entity: str | None, ) -> response.Response: """Create given relationship beteen a Profile and a Subscription. Args: profile_type (str): The type of profile (ArtistInsights, etc). profile_id (int): The id of the profile. notification_type (str): The type of notification feed_type (str): The name of feed followed_entity (str): Optional dimension on node type event source Returns: Response: 201 created. """ notification_type = stringcase.camelcase(notification_type) feed_type = stringcase.camelcase(feed_type) session = get_session() subscribe_query = strip_query("""MATCH (p:Profile),(s:Subscription) WHERE p.profileId = $profile_id AND p.profileType = $profile_type AND s.notificationType = $notification_type AND s.feedType = $feed_type AND ( s.followedEntity = $followed_entity OR $followed_entity_empty ) CREATE (p)-[r:HAS_SUBSCRIPTION]->(s) SET r.dateCreated = localdatetime() RETURN r.dateCreated""") followed_entity_empty = followed_entity is None result = session.run( subscribe_query, profile_type=profile_type, profile_id=profile_id, notification_type=notification_type, feed_type=feed_type, followed_entity=followed_entity, followed_entity_empty=followed_entity_empty, ) if not result.peek(): missing_query = strip_query(""" OPTIONAL MATCH (p:Profile {profileId: $profile_id, profileType: $profile_type}) OPTIONAL MATCH (s:Subscription {notificationType: $notification_type, feedType: $feed_type} ) WHERE ( s.followedEntity = $followed_entity OR $followed_entity_empty ) RETURN p,s""") result = session.run( missing_query, profile_type=profile_type, profile_id=profile_id, notification_type=notification_type, followed_entity=followed_entity, followed_entity_empty=followed_entity_empty, feed_type=feed_type, ) nodes = result.single() message = 'Unknown error' if not nodes or not nodes['s']: message = f'Subscription not found with type {notification_type} and feed {feed_type}' if not followed_entity_empty: message += f' and followed entity {followed_entity}' elif not nodes['p']: message = f'Profile not found with id {profile_id} and type {profile_type}' g.log.info(message) return response.create_not_found_response(message) return response.Response(status=201) def get_subscription_by_param( subscription_name: str | None = None, feed_type: str | None = None, followed_entity: str | None = None, ) -> response.Response: """Get subscription by name and feed_type and followed_entity. Args: subscription_name (str): name of the subscription. feed_type (str): feed type eg: productApproval. followed_entity (str): Vendor or Subaccount. Returns: Response: updated subscription object. """ where_clause = ['true'] if subscription_name: where_clause.append('s.name = $subscriptionName') if feed_type: where_clause.append('s.feedType = $feedType') if followed_entity: where_clause.append('s.followedEntity = $followedEntity') session = get_session() subscribe_query = strip_query( f"""MATCH (s:Subscription) WHERE {' AND '.join(where_clause)} RETURN s.name as name, s.notificationType as notificationType, s.appId as appId, s.feedType as feedType, collect(s.followedEntity) as followedEntityTypes""" ) subscription_result = session.run( subscribe_query, subscriptionName=subscription_name, feedType=feed_type, followedEntity=followed_entity, ).single() if not subscription_result: return response.create_not_found_response('No subscription found with these parameters.') return response.Response(subscription_result) def edit_subscription_for_identity( # noqa: PLR0913 identity_id: str, subscription_name: str, profile_type: str, follow_all_resources: bool, follow_resources: list[str], automatic: bool = False, ) -> response.Response: """Update subscriptions for a given identity and subscription_name. Args: identity_id (str): Identity UUID. subscription_name (str): name of the subscription. profile_type (str): Profile_type that will get the subscribe and follow. follow_all_resources (bool): indicate if it is ON for all labels. follow_resources (list): List of uuids if it is not follow_all_resources. automatic (bool): If this due to automatic process or explicitly by user. Returns: Response: updated subscription object. """ session = get_session() rel_name = f'HAS_FOLLOWED_{subscription_name.upper()}' # Add HAS_SUBSCRIPTION to all subscription with this name. ie vendor and subaccount both. # note: we are also creating new HAS_FOLLOWED_. if follow_resources: # profiles who have access to those resource will follow that selected resources. follow_query = strip_query(f""" MATCH (s:Subscription) WHERE s.name = $subscriptionName WITH s MATCH (r) WHERE (r:Vendor OR r:Subaccount OR r:LabelParticipants OR r:Collaborator) AND r.uuid IN $followResources WITH s, r MATCH (i:Identity)-[:HAS_PROFILE]->(p:Profile) WHERE i.id = $identityId AND p.profileType = $profileType AND ( EXISTS((p)-[:HAS_ACCESS_TO]->(r)) OR EXISTS((p)-[:HAS_ACCESS_TO]->(:Vendor)-[:OWNS]->(r:Subaccount)) ) MERGE (p)-[sub:HAS_SUBSCRIPTION]->(s) SET sub.dateCreated = localdatetime() MERGE (p)-[newrel:{rel_name}]->(r) SET newrel.dateCreated = localdatetime(), newrel.automatic = $automatic WITH p, r, s, newrel OPTIONAL MATCH (p)-[oldfoll:DELETED_HAS_FOLLOWED|DELETED_{rel_name}]->(r) DELETE oldfoll WITH s, newrel OPTIONAL MATCH (p)-[oldsub:DELETED_HAS_SUBSCRIPTION]->(s) DELETE oldsub RETURN distinct(newrel) """) result = session.run( follow_query, subscriptionName=subscription_name, followResources=follow_resources, identityId=identity_id, profileType=profile_type, automatic=automatic, relName=rel_name, ).data() if not result or len(result) < len(follow_resources): return response.create_not_found_response( f'Failed to follow resources for user {identity_id}', ) if follow_all_resources: # Add HAS_AUTO_FOLLOWED to this identity and HAS_SUBSCRIPTION to profiles. follow_query = strip_query(""" MATCH (s:Subscription) WHERE s.name = $subscriptionName WITH s MATCH (i:Identity) WHERE i.id = $identityId MERGE (i)-[auto:HAS_AUTO_FOLLOWED]->(s) SET auto.dateCreated = localdatetime() WITH i, s OPTIONAL MATCH (i)-[oldauto:DELETED_HAS_AUTO_FOLLOWED]->(s) DELETE oldauto WITH i, s MATCH (i)-[:HAS_PROFILE]->(p:Profile) WHERE p.profileType = $profileType MERGE (p)-[newrel:HAS_SUBSCRIPTION]->(s) SET newrel.dateCreated = localdatetime() WITH p, s, newrel OPTIONAL MATCH (p)-[oldsub:DELETED_HAS_SUBSCRIPTION]->(s) DELETE oldsub RETURN newrel """) result = session.run( follow_query, subscriptionName=subscription_name, followResources=follow_resources, identityId=identity_id, profileType=profile_type, ).data() if not result: return response.create_not_found_response( f'Failed to follow all resources for user {identity_id}', ) return response.Response( {'edit_subscription': f'Successfully edited {subscription_name} for identity'} ) def delete_subscription_for_identity( identity_id: str, subscription_name: str, profile_type: str ) -> response.Response: """Delete subscriptions for a given identity and subscription_name. Args: identity_id (str): Identity UUID. subscription_name (str): name of the subscription. profile_type (str): Profile_type that will get the subscribe and follow. Returns: Response: updated subscription object. """ session = get_session() rel_name = f'HAS_FOLLOWED_{subscription_name.upper()}' # delete existing HAS_AUTO_FOLLOWED relationship. auto_query = strip_query( """MATCH (i:Identity)-[auto:HAS_AUTO_FOLLOWED]->(s:Subscription {name: $subscriptionName} ) WHERE i.id = $identityId CALL apoc.refactor.setType(auto, 'DELETED_HAS_AUTO_FOLLOWED') YIELD input, output RETURN input, output""" ) session.run(auto_query, subscriptionName=subscription_name, identityId=identity_id).data() # delete existing HAS_SUBSCRIPTION relationship. subscription_query = strip_query( """MATCH (i:Identity)-[:HAS_PROFILE]->(p:Profile)-[sub:HAS_SUBSCRIPTION]->(s:Subscription) WHERE i.id = $identityId AND p.profileType = $profileType AND s.name = $subscriptionName CALL apoc.refactor.setType(sub, 'DELETED_HAS_SUBSCRIPTION') YIELD input, output RETURN input, output""" ) session.run( subscription_query, subscriptionName=subscription_name, identityId=identity_id, profileType=profile_type, ).data() # note: we won't delete HAS_FOLLOWED directly as that is common. # instead we remove the new FOLLOW relationships follow_query = strip_query( f"""MATCH (i:Identity)-[:HAS_PROFILE]->(p:Profile)-[foll:{rel_name}]->(r) WHERE i.id = $identityId AND p.profileType = $profileType AND (r:Vendor OR r:Subaccount OR r:LabelParticipants OR r:Collaborator) CALL apoc.refactor.setType(foll, 'DELETED_{rel_name}') YIELD input, output RETURN input, output""" ) session.run( follow_query, subscriptionName=subscription_name, identityId=identity_id, profileType=profile_type, ).data() return response.Response( {'edit_subscription': f'Successfully deleted {subscription_name} for identity'} ) def delete_selected_subscription_for_identity( identity_id: str, subscription_name: str, profile_type: str, follow_resources: list[str] ) -> response.Response: """Delete subscriptions to follow_resources for a given identity and subscription_name. Args: identity_id (str): Identity UUID. subscription_name (str): name of the subscription. profile_type (str): Profile_type that will get the subscribe and follow. follow_resources (list): List of resource uuids. Returns: Response: result object. """ session = get_session() rel_name = f'HAS_FOLLOWED_{subscription_name.upper()}' if follow_resources: # profiles who have access to those resource will get unfollowed. follow_query = strip_query(f""" MATCH (s:Subscription) WHERE s.name = $subscriptionName WITH s MATCH (r) WHERE (r:Vendor OR r:Subaccount OR r:LabelParticipants OR r:Collaborator) AND r.uuid IN $followResources WITH s, r MATCH (i:Identity)-[:HAS_PROFILE]->(p:Profile) WHERE i.id = $identityId AND p.profileType = $profileType AND ( EXISTS((p)-[:HAS_ACCESS_TO]->(r)) OR EXISTS((p)-[:HAS_ACCESS_TO]->(:Vendor)-[:OWNS]->(r:Subaccount)) ) WITH p, s, r MATCH (p)-[sub:HAS_SUBSCRIPTION]->(s) CALL apoc.refactor.setType(sub, 'DELETED_HAS_SUBSCRIPTION') YIELD input, output WITH p, r MATCH (p)-[rel:{rel_name}]->(r) CALL apoc.refactor.setType(rel, 'DELETED_{rel_name}') YIELD input, output RETURN distinct(r) """) result = session.run( follow_query, subscriptionName=subscription_name, followResources=follow_resources, identityId=identity_id, profileType=profile_type, relName=rel_name, ).data() if not result or len(result) < len(follow_resources): response.create_not_found_response( f'Failed to delete subscription all resources for user {identity_id}' ) return response.Response( {'edit_subscription': f'Successfully deleted {subscription_name} for identity'} )