26420 - Affiliation Pagination via Entity Mapping (bcgov#3399) (bcgov#3418)

Co-authored-by: Rajandeep Kaur <144159721+Rajandeep98@users.noreply.github.com>

Author: seeker25

auth-api/flags.json

@@ -3,6 +3,7 @@
         "string-flag": "a string value",
         "bool-flag": true,
         "integer-flag": 10,
-        "enable-publish-account-events": true
+        "enable-publish-account-events": true,
+        "enable-entity-mapping": true
     }
 }

auth-api/migrations/versions/2025_05_22_6f2c09061fd3_entity_mapping.py

@@ -0,0 +1,42 @@
+"""Add in entity mapping table.
+
+Revision ID: 6f2c09061fd3
+Revises: bddf9fe7468c
+Create Date: 2025-05-22 17:21:47.908470
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '6f2c09061fd3'
+down_revision = 'bddf9fe7468c'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('entities', sa.Column('is_loaded_lear', sa.Boolean(), nullable=False, server_default=sa.true()))
+
+    op.create_table('entity_mapping',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('business_identifier', sa.String(length=75), nullable=True),
+    sa.Column('bootstrap_identifier', sa.String(length=75), nullable=True),
+    sa.Column('nr_identifier', sa.String(length=75), nullable=True),
+    sa.PrimaryKeyConstraint('id')
+    )
+    with op.batch_alter_table('entity_mapping', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_entity_mapping_bootstrap_identifier'), ['bootstrap_identifier'], unique=False)
+        batch_op.create_index(batch_op.f('ix_entity_mapping_business_identifier'), ['business_identifier'], unique=False)
+        batch_op.create_index(batch_op.f('ix_entity_mapping_nr_identifier'), ['nr_identifier'], unique=False)
+
+
+def downgrade():
+    with op.batch_alter_table('entity_mapping', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_entity_mapping_nr_identifier'))
+        batch_op.drop_index(batch_op.f('ix_entity_mapping_business_identifier'))
+        batch_op.drop_index(batch_op.f('ix_entity_mapping_bootstrap_identifier'))
+
+    op.drop_table('entity_mapping')
+    op.drop_column('entities', 'is_loaded_lear')

auth-api/src/auth_api/__init__.py

@@ -53,6 +53,7 @@ def create_app(run_mode=os.getenv("DEPLOYMENT_ENV", "production")):
         with app.app_context():
             execute_migrations(app)
         logger.info("Finished migration upgrade.")
+        logger.info("Note: endpoints will 404 until the DEPLOYMENT_ENV is switched off of migration.")
     else:
         flags.init_app(app)
         ma.init_app(app)

auth-api/src/auth_api/models/__init__.py

@@ -33,6 +33,7 @@
 from .db import db, ma
 from .documents import Documents
 from .entity import Entity
+from .entity_mapping import EntityMapping
 from .invitation import Invitation
 from .invitation_membership import InvitationMembership
 from .invitation_type import InvitationType

auth-api/src/auth_api/models/dataclass.py

@@ -14,6 +14,7 @@
 """This module holds data classes."""
 
 from dataclasses import dataclass, field
+from datetime import datetime
 from typing import List, Optional, Self
 
 from requests import Request
@@ -53,12 +54,12 @@ class AffiliationInvitationSearch:  # pylint: disable=too-many-instance-attribut
 class AffiliationSearchDetails:  # pylint: disable=too-many-instance-attributes
     """Used for filtering Affiliations based on filters passed."""
 
-    identifier: Optional[str]
-    status: Optional[str]
-    name: Optional[str]
-    type: Optional[str]
     page: int
     limit: int
+    identifier: Optional[str] = None
+    status: Optional[str] = None
+    name: Optional[str] = None
+    type: Optional[str] = None
 
     @classmethod
     def from_request_args(cls, req: Request) -> Self:
@@ -201,3 +202,11 @@ class ProductReviewTask:
     product_subscription_id: int
     user_id: str
     external_source_id: Optional[str] = None
+
+
+@dataclass
+class AffiliationBase:
+    """Small class for searching in Names and LEAR."""
+
+    identifier: str
+    created: datetime

auth-api/src/auth_api/models/entity.py

@@ -43,6 +43,7 @@ class Entity(BaseModel):  # pylint: disable=too-few-public-methods, too-many-ins
     status = Column(String(), nullable=True)
     last_modified_by = Column(String(), nullable=True)
     last_modified = Column(DateTime, default=None, nullable=True)
+    is_loaded_lear = Column(Boolean(), default=True, nullable=False)
 
     contacts = relationship("ContactLink", back_populates="entity")
     corp_type = relationship("CorpType", foreign_keys=[corp_type_code], lazy="joined", innerjoin=True)

auth-api/src/auth_api/models/entity_mapping.py

@@ -0,0 +1,29 @@
+"""Entity Mapping model which enables us to be able to do pagination on affiliations."""
+
+from sqlalchemy import Column, Integer, String
+
+from .db import db
+
+
+class EntityMapping(db.Model):  # pylint: disable=too-few-public-methods, too-many-instance-attributes
+    """This is the Entity model for the Auth service."""
+
+    __tablename__ = "entity_mapping"
+
+    id = Column(Integer, primary_key=True)
+    business_identifier = Column("business_identifier", String(75), index=True, unique=False, nullable=True)
+    bootstrap_identifier = Column("bootstrap_identifier", String(75), index=True, unique=False, nullable=True)
+    nr_identifier = Column("nr_identifier", String(75), index=True, unique=False, nullable=True)
+
+    def flush(self):
+        """Save and flush."""
+        db.session.add(self)
+        db.session.flush()
+        return self
+
+    def save(self):
+        """Save and commit."""
+        db.session.add(self)
+        db.session.flush()
+        db.session.commit()
+        return self

auth-api/src/auth_api/resources/v1/org.py

@@ -37,6 +37,8 @@
 from auth_api.services import SimpleOrg as SimpleOrgService
 from auth_api.services import User as UserService
 from auth_api.services.authorization import Authorization as AuthorizationService
+from auth_api.services.entity_mapping import EntityMappingService
+from auth_api.services.flags import flags
 from auth_api.utils.auth import jwt as _jwt
 from auth_api.utils.endpoints_enums import EndpointEnum
 from auth_api.utils.enums import AccessType, NotificationType, OrgStatus, OrgType, PatchActions, Status
@@ -351,7 +353,7 @@ def delete_organzization_contact(org_id):
 def get_organization_affiliations_search(org_id):
     """Get all affiliated entities for the given org, this works with pagination."""
     try:
-        response, status = new_affiliation_search(org_id)
+        response, status = affiliation_search(org_id, use_entity_mapping=True)
     except BusinessException as exception:
         response, status = {"code": exception.code, "message": exception.message}, exception.status_code
     except ServiceUnavailableException as exception:
@@ -371,7 +373,7 @@ def get_organization_affiliations(org_id):
                 HTTPStatus.OK,
             )
         # Remove below after UI is pointing at new route.
-        response, status = new_affiliation_search(org_id)
+        response, status = affiliation_search(org_id)
     except BusinessException as exception:
         response, status = {"code": exception.code, "message": exception.message}, exception.status_code
     except ServiceUnavailableException as exception:
@@ -380,16 +382,19 @@ def get_organization_affiliations(org_id):
     return response, status
 
 
-def new_affiliation_search(org_id):
+def affiliation_search(org_id, use_entity_mapping=False):
     """Get all affiliated entities for the given org by calling into Names and LEAR."""
-    # get affiliation identifiers and the urls for the source data
     org = OrgService.find_by_org_id(org_id, allowed_roles=ALL_ALLOWED_ROLES)
     if org is None:
         raise BusinessException(Error.DATA_NOT_FOUND, None)
-    affiliations = AffiliationModel.find_affiliations_by_org_id(org_id)
     search_details = AffiliationSearchDetails.from_request_args(request)
+    if use_entity_mapping:
+        affiliation_bases = EntityMappingService.populate_affiliation_base(org_id, search_details)
+    else:
+        affiliations = AffiliationModel.find_affiliations_by_org_id(org_id)
+        affiliation_bases = AffiliationService.affiliation_to_affiliation_base(affiliations)
     affiliations_details_list = asyncio.run(
-        AffiliationService.get_affiliation_details(affiliations, search_details, org_id)
+        AffiliationService.get_affiliation_details(affiliation_bases, search_details, org_id)
     )
     # Use orjson serializer here, it's quite a bit faster.
     response, status = (
@@ -428,7 +433,6 @@ def post_organization_affiliation(org_id):
                 phone=request_json.get("phone"),
                 certified_by_name=request_json.get("certifiedByName"),
             )
-
             response, status = (
                 AffiliationService.create_new_business_affiliation(affiliation_data).as_dict(),
                 HTTPStatus.CREATED,
@@ -444,10 +448,15 @@ def post_organization_affiliation(org_id):
                 ).as_dict(),
                 HTTPStatus.CREATED,
             )
-
         entity_details = request_json.get("entityDetails", None)
         if entity_details:
+            if flags.is_on("enable-entity-mapping", default=False) is True:
+                EntityMappingService.from_entity_details(entity_details)
             AffiliationService.fix_stale_affiliations(org_id, entity_details)
+        # Auth-queue handles the row creation for new business (NR only), this handles passcode missing info
+        elif is_new_business is False:
+            if flags.is_on("enable-entity-mapping", default=False) is True:
+                EntityMappingService.populate_entity_mapping_for_identifier(business_identifier)
     except BusinessException as exception:
         response, status = {"code": exception.code, "message": exception.message}, exception.status_code
 

auth-api/src/auth_api/services/affiliation.py

@@ -31,7 +31,7 @@
 from auth_api.models.contact_link import ContactLink
 from auth_api.models.dataclass import Activity
 from auth_api.models.dataclass import Affiliation as AffiliationData
-from auth_api.models.dataclass import AffiliationSearchDetails, DeleteAffiliationRequest
+from auth_api.models.dataclass import AffiliationBase, AffiliationSearchDetails, DeleteAffiliationRequest
 from auth_api.models.entity import Entity
 from auth_api.models.membership import Membership as MembershipModel
 from auth_api.schemas import AffiliationSchema
@@ -452,23 +452,34 @@ def fix_stale_affiliations(org_id: int, entity_details: Dict, **kwargs):
         logger.debug(">fix_stale_affiliations")
 
     @staticmethod
-    def _affiliation_details_url(affiliation: AffiliationModel) -> str:
+    def _affiliation_details_url(identifier: str) -> str:
         """Determine url to call for affiliation details."""
         # only have LEAR and NAMEX affiliations
-        if affiliation.entity.corp_type_code == CorpType.NR.value:
+        if identifier.startswith("NR"):
             return current_app.config.get("NAMEX_AFFILIATION_DETAILS_URL")
         return current_app.config.get("LEAR_AFFILIATION_DETAILS_URL")
 
+    @staticmethod
+    def affiliation_to_affiliation_base(affiliations: List[AffiliationModel]) -> List[AffiliationBase]:
+        """Convert affiliations to a common data class."""
+        return [
+            AffiliationBase(identifier=affiliation.entity.business_identifier, created=affiliation.created)
+            for affiliation in affiliations
+        ]
+
     @staticmethod
     async def get_affiliation_details(
-        affiliations: List[AffiliationModel], search_details: AffiliationSearchDetails, org_id
+        affiliation_bases: List[AffiliationBase], search_details: AffiliationSearchDetails, org_id
     ) -> List:
         """Return affiliation details by calling the source api."""
         url_identifiers = {}  # i.e. turns into { url: [identifiers...] }
+        # Our pagination is already handled at the auth level when not doing a search.
+        if not (search_details.status and search_details.name and search_details.type and search_details.identifier):
+            search_details.page = 1
         search_dict = asdict(search_details)
-        for affiliation in affiliations:
-            url = Affiliation._affiliation_details_url(affiliation)
-            url_identifiers.setdefault(url, []).append(affiliation.entity.business_identifier)
+        for affiliation_base in affiliation_bases:
+            url = Affiliation._affiliation_details_url(affiliation_base.identifier)
+            url_identifiers.setdefault(url, []).append(affiliation_base.identifier)
         call_info = [
             {
                 "url": url,
@@ -486,12 +497,8 @@ async def get_affiliation_details(
         try:
             responses = await RestService.call_posts_in_parallel(call_info, token, org_id)
             combined = Affiliation._combine_affiliation_details(responses)
-            # Should provide us with ascending order
-            affiliations_sorted = sorted(affiliations, key=lambda x: x.created, reverse=True)
-            # Provide us with a dict with the max created date.
-            ordered = {
-                affiliation.entity.business_identifier: affiliation.created for affiliation in affiliations_sorted
-            }
+            affiliations_bases_sorted = sorted(affiliation_bases, key=lambda x: x.created, reverse=True)
+            ordered = {affiliation.identifier: affiliation.created for affiliation in affiliations_bases_sorted}
 
             def sort_key(item):
                 identifier = item.get("identifier", item.get("nameRequest", {}).get("nrNum", ""))
@@ -502,7 +509,7 @@ def sort_key(item):
             return combined
         except ServiceUnavailableException as err:
             logger.debug(err)
-            logger.debug("Failed to get affiliations details:  %s", affiliations)
+            logger.debug("Failed to get affiliations details:  %s", affiliation_bases)
             raise ServiceUnavailableException("Failed to get affiliation details") from err
 
     @staticmethod

auth-api/src/auth_api/services/affiliation_invitation.py

@@ -38,6 +38,7 @@
 from auth_api.models.org import Org as OrgModel
 from auth_api.schemas import AffiliationInvitationSchema
 from auth_api.services.entity import Entity as EntityService
+from auth_api.services.entity_mapping import EntityMappingService
 from auth_api.services.flags import flags
 from auth_api.services.org import Org as OrgService
 from auth_api.services.user import User as UserService
@@ -503,32 +504,20 @@ def find_affiliation_invitation_by_id(invitation_id):
     @staticmethod
     def _get_token_confirm_path(org_name, token, query_params=None):
         """Get the config for different email types."""
-        if flags.is_on("enable-new-magic-link-formatting-with-query-params", default=False):
-            # New query parameter based URL structure
-            params = {
-                "token": token,
-                "orgName": escape_wam_friendly_url(org_name),
-            }
-
-            # Add any additional query params
-            if query_params:
-                params.update(query_params)
-
-            # Point to new business registry project
-            app_url = current_app.config.get("BUSINESS_REGISTRY_URL") + "/"
-            # Build the URL with query parameters
-            token_confirm_url = f"{app_url}/affiliationInvitation/acceptToken?{urlencode(params)}"
-        else:
-            # Point to auth-web
-            app_url = current_app.config.get("WEB_APP_URL") + "/"
-            # Original URL structure
-            escape_url = escape_wam_friendly_url(org_name)
-            path = f"{escape_url}/affiliationInvitation/acceptToken"
-            token_confirm_url = f"{app_url}/{path}/{token}"
+        # New query parameter based URL structure
+        params = {
+            "token": token,
+            "orgName": escape_wam_friendly_url(org_name),
+        }
 
-            if query_params:
-                token_confirm_url += f"?{urlencode(query_params)}"
+        # Add any additional query params
+        if query_params:
+            params.update(query_params)
 
+        # Point to new business registry project
+        app_url = current_app.config.get("BUSINESS_REGISTRY_URL") + "/"
+        # Build the URL with query parameters
+        token_confirm_url = f"{app_url}/affiliationInvitation/acceptToken?{urlencode(params)}"
         return token_confirm_url
 
     @staticmethod
@@ -697,6 +686,8 @@ def accept_affiliation_invitation(
             affiliation_model = AffiliationModel(org_id=org_id, entity_id=entity_id, certified_by_name=None)
             affiliation_model.save()
             entity_model = EntityModel.find_by_entity_id(entity_id)
+            if flags.is_on("enable-entity-mapping", default=False) is True:
+                EntityMappingService.populate_entity_mapping_for_identifier(entity_model.business_identifier)
             publish_affiliation_event(
                 QueueMessageTypes.BUSINESS_AFFILIATED.value, org_id, entity_model.business_identifier
             )