"""Logic module for get list of applications and roles enabled for a Collaborator.""" from typing import Callable from flask import g from owsrequest import error_response from owsresponse import response from owsresponse.adaptors.flask import flaskify from sqlalchemy.orm import Session from account.connectors.mysql import db_read_only_session_wrap from account.constants.features import FEATURES from account.models import ( application as app_model, feature as feature_model, neo4j_collaborator as collaborator_model, neo4j_identity as neo4j_identity_model, ows_collaborator as ows_collaborator_model, types, ) from account.models.types import TenantApplication CollaboratorApplicationChecker = Callable[ [types.CollaboratorAttributes], types.TenantApplication | None ] ENABLED_APPLICATIONS = [app_model.CUSTOMER_ACCOUNTING_APP, app_model.BANKING_TAX_APP] def generate_collaborator_app_check_map( collaborator_attributes: types.CollaboratorAttributes, ) -> dict[str, Callable[[], types.TenantApplication | None]]: """ Generates a dictionary of functions that can be used to check if a particular application is enabled for a given collaborator. The list of applications to check can become dynamic in the future. Args: collaborator_attributes (CollaboratorAttributes): The collaborator object used to check enabled applications. Returns: dict[str, Callable[[], TenantApplication | None]]: A dictionary where keys are application IDs and values are functions that check the enabled status of those applications. """ app_checkers: dict[str, CollaboratorApplicationChecker] = { app_model.CUSTOMER_ACCOUNTING_APP: get_customer_accounting_app_for_collaborator, app_model.BANKING_TAX_APP: get_banking_tax_app_for_collaborator, } return { app: (lambda app_id=app: app_checkers[app_id](collaborator_attributes)) for app in ENABLED_APPLICATIONS } def get_enabled_collaborator_applications( collaborator_attributes: types.CollaboratorAttributes, ) -> list[types.TenantApplication]: """ Get all applications for a collaborator. Args: collaborator_attributes (CollaboratorAttributes): The collaborator object used to check enabled applications. Return: list[TenantApplication]: list of collaborator applications. """ applications_map = generate_collaborator_app_check_map(collaborator_attributes) applications = [func() for app_id, func in applications_map.items()] return [app for app in applications if app is not None] @db_read_only_session_wrap def get_collaborator_applications(collaborator_uuid: str, session: Session) -> response.Response: """ Get all applications for a collaborator. Args: collaborator_uuid (str): Collaborator identifier. session (Session): SQLAlchemy session. Return: Response: containing dict of applications and roles """ # Getting an AdminContext to pass it to the ows-collaborator request identity_id = g.request_context.jwt_identity_id settings_profile = neo4j_identity_model.get_identity_settings_profile(identity_id) if not settings_profile: return flaskify(error_response.create_error_forbidden_user()) # Getting basic collaborator properties collaborator, vendor, company_brand = collaborator_model.get_collaborator_by_uuid( collaborator_uuid ) if collaborator is None: return response.create_not_found_response() # Getting available features fc = feature_model.get_enabled_features_for_vendor_with_session( vendor['vendor_id'], session=session ) vendor_features = feature_model.convert_features_to_list(fc) # Setting all required collaborator attributes collaborator_attributes = types.CollaboratorAttributes( collaborator_id=collaborator['collaborator_id'], uuid=collaborator['uuid'], dp_enabled_date=collaborator['dp_enabled_date'], vendor_attributes=types.VendorAttributes( vendor_id=vendor['vendor_id'], uuid=vendor['uuid'], company_brand=company_brand, features=vendor_features, ), ) applications = get_enabled_collaborator_applications(collaborator_attributes) return response.Response({'applications': applications}) def get_customer_accounting_app_for_collaborator( collaborator_attributes: types.CollaboratorAttributes, ) -> types.TenantApplication | None: """ Get Customer Accounting application for a Collaborator, if enabled, based on the CompanyBrand of the parent Vendor. Args: collaborator_attributes (CollaboratorAttributes): CollaboratorAttributes. Return: TenantApplication | None: Customer accounting application with a base role. """ return TenantApplication( application_id=app_model.CUSTOMER_ACCOUNTING_APP, roles=[app_model.CUSTOMER_ACCOUNTING_BASE_ROLE], url=app_model.get_application_url( app_model.CUSTOMER_ACCOUNTING_APP, collaborator_attributes['vendor_attributes']['company_brand']['name'], ), ) def get_banking_tax_app_for_collaborator( collaborator_attributes: types.CollaboratorAttributes, ) -> types.TenantApplication | None: """ Get Banking and Taxes application for a Collaborator, if enabled, based on Direct Payments FC and dpEnabledDate property. Args: collaborator_attributes (CollaboratorAttributes): CollaboratorAttributes. Return: TenantApplication | None: Banking and Taxes application with a base role. """ required_features = [ FEATURES.COLLABORATORS_DIRECT_PAYMENTS, ] if not any( feature in collaborator_attributes['vendor_attributes']['features'] for feature in required_features ): return None if not collaborator_attributes['dp_enabled_date']: return None terms_and_conditions = ows_collaborator_model.get_vendor_terms_and_conditions( collaborator_attributes['vendor_attributes']['vendor_id'] ) # If no terms and conditions data is available, skip processing if terms_and_conditions is None: return None # Check if terms and conditions agreement is needed latest_version = terms_and_conditions.latest_version agreed_version = terms_and_conditions.agreed_version has_no_agreement_data = not latest_version or not agreed_version needs_agreement = has_no_agreement_data or latest_version > agreed_version if needs_agreement: return None return TenantApplication( application_id=app_model.BANKING_TAX_APP, roles=[app_model.BANKING_TAX_BASE_ROLE], url=app_model.get_application_url( app_model.BANKING_TAX_APP, collaborator_attributes['vendor_attributes']['company_brand']['name'], ), )