Api reference

components.global_account.public.account_entity ¶

AccountStatus ¶

Bases: AlanBaseEnum

churned `class-attribute` `instance-attribute` ¶

churned = 'churned'

live `class-attribute` `instance-attribute` ¶

live = 'live'

prospect `class-attribute` `instance-attribute` ¶

prospect = 'prospect'

delete_account_if_it_has_no_entity ¶

delete_account_if_it_has_no_entity(
    account_id, new_account_id
)

Source code in components/global_account/external/account.py

@tracer.wrap(service="global_account")
def delete_account_if_it_has_no_entity(account_id: UUID, new_account_id: UUID) -> None:
    from components.fr.internal.business_logic.account.actions.account import (  # noqa: ALN043
        delete_account,
        move_account_related_models_to_new_account,
    )

    account_sqla = get_or_raise_missing_resource(AccountSQLA, account_id)
    if len(account_sqla.companies) == 0:
        # Move related data to the new account
        move_account_related_models_to_new_account(new_account_id, account_id)

    account = get_account(account_id)
    if len(account.entities) == 0:
        current_logger.info(
            f"Deleting {account} because it has no entity left", account_id=account_id
        )
        delete_account(account_id)

get_account ¶

get_account(account_id)

Return full account, delegating to the app's account repository.

Source code in components/global_account/external/account.py

@tracer.wrap(service="global_account")
def get_account(account_id: UUID) -> Account:
    """Return full account, delegating to the app's account repository."""
    return get_account_repository().get_full_account(account_id)

get_account_entities ¶

get_account_entities(account_id)

Return entities for an account, delegating to the app's account repository.

Source code in components/global_account/external/account.py

@tracer.wrap(service="global_account")
def get_account_entities(account_id: UUID) -> list[Entity]:
    """Return entities for an account, delegating to the app's account repository."""
    return get_account_repository().get_full_account(account_id).entities

get_light_account ¶

get_light_account(account_ref)

Source code in components/global_account/external/account.py

def get_light_account(account_ref: str) -> LightAccount:
    account = get_or_raise_missing_resource(AccountSQLA, account_ref)
    return get_light_account_from_model(account)

components.global_account.public.account_repository ¶

get_account_repository ¶

get_account_repository()

Return the AccountRepository for the current app's country.

Source code in components/global_account/public/account_repository.py

def get_account_repository() -> AccountRepository:
    """Return the AccountRepository for the current app's country."""
    return get_app_dependency().get_account_repository()

components.global_account.public.actions ¶

bookmark_account ¶

bookmark_account(account_id, user_id)

Bookmark an account for a specific user. Only creates the bookmark if it doesn't already exist.

Source code in components/global_account/public/actions/bookmark_account.py

def bookmark_account(account_id: UUID, user_id: str) -> None:
    """
    Bookmark an account for a specific user.
    Only creates the bookmark if it doesn't already exist.
    """
    existing_bookmark = (
        current_session.query(BookmarkAccount)  # noqa: ALN085
        .filter(
            BookmarkAccount.account_id == account_id,
            BookmarkAccount.user_id == user_id,
        )
        .first()
    )
    if not existing_bookmark:
        bookmark = BookmarkAccount(account_id=account_id, user_id=user_id)
        current_session.add(bookmark)
        current_logger.info(
            f"Bookmark created for user {user_id} and account {account_id}"
        )
    else:
        current_logger.info(
            f"Bookmark already exists for user {user_id} and account {account_id}"
        )

unbookmark_account ¶

unbookmark_account(account_id, user_id)

Remove bookmark for an account for a specific user.

Source code in components/global_account/public/actions/bookmark_account.py

def unbookmark_account(account_id: UUID, user_id: str) -> None:
    """
    Remove bookmark for an account for a specific user.
    """
    bookmarks = (
        current_session.query(BookmarkAccount)  # noqa: ALN085
        .filter(
            BookmarkAccount.account_id == account_id,
            BookmarkAccount.user_id == user_id,
        )
        .all()
    )
    if bookmarks:
        for bookmark in bookmarks:
            current_session.delete(bookmark)

        current_logger.info(
            f"Bookmark removed for user {user_id} and account {account_id}"
        )

builder_prospect ¶

get_or_create_prospect_from_company `module-attribute` ¶

get_or_create_prospect_from_company = (
    get_or_create_prospect_from_company
)

update_builder_prospect `module-attribute` ¶

update_builder_prospect = update_builder_prospect

components.global_account.public.builder_prospect ¶

get_builder_prospect_by_ccn_and_siren ¶

get_builder_prospect_by_ccn_and_siren(
    ccn_code, siren, nic=None, is_from_self_serve=None
)

Return the CompanyProspect for the given ccn_code and SIREN + NIC if given, or None if not found.

Source code in components/global_account/public/builder_prospect.py

def get_builder_prospect_by_ccn_and_siren(
    ccn_code: str,
    siren: str,
    nic: str | None = None,
    is_from_self_serve: bool | None = None,
) -> CompanyProspect | None:
    """
    Return the CompanyProspect for the given ccn_code and SIREN + NIC if given, or None if not found.
    """
    from components.global_account.internal.models.builder_prospect import (
        BuilderProspect,
    )

    stmt = select(BuilderProspect).where(
        BuilderProspect.company_identifier == siren,
        BuilderProspect.industry_framework_agreement_code == ccn_code,
        BuilderProspect.specific_company_identifier == nic,
    )

    if is_from_self_serve is not None:
        stmt = stmt.where(BuilderProspect.is_from_self_serve == is_from_self_serve)

    builder_prospect_model = current_session.scalars(
        stmt.order_by(
            # Adding order_by to make the query deterministic (even though we should only get 1 result)
            BuilderProspect.created_at.desc()
        )
    ).first()

    return (
        get_company_prospect_or_none(builder_prospect_model.id)
        if builder_prospect_model
        else None
    )

get_company_prospect ¶

get_company_prospect(id)

Source code in components/global_account/internal/queries/builder_prospect.py

def get_company_prospect(id: UUID | str) -> CompanyProspect:
    return one(get_company_prospects([id]))

get_company_prospect_or_none ¶

get_company_prospect_or_none(id)

Source code in components/global_account/internal/queries/builder_prospect.py

def get_company_prospect_or_none(id: UUID | str) -> CompanyProspect | None:
    return one_or_none(get_company_prospects([id], raise_if_some_missing=False))

get_company_prospects ¶

get_company_prospects(ids, raise_if_some_missing=True)

Source code in components/global_account/internal/queries/builder_prospect.py

def get_company_prospects(
    ids: Sequence[UUID | str], raise_if_some_missing: bool = True
) -> list[CompanyProspect]:
    if not ids:
        return []

    builder_prospects = get_account_repository().search_builder_prospects(
        prospect_refs=[str(id_) for id_ in ids]
    )

    found_ids = {str(bp.id) for bp in builder_prospects}
    missing_ids = {str(id_) for id_ in ids} - found_ids
    if missing_ids and raise_if_some_missing:
        raise BaseErrorCode.missing_resource(
            message=f"BuilderProspects ids={missing_ids} not found"
        )
    return [CompanyProspect.from_builder_prospect(bp) for bp in builder_prospects]

get_or_create_builder_prospects_from_companies ¶

get_or_create_builder_prospects_from_companies(company_ids)

Return a mapping (company_ref -> prospect) for all the provided companies. If the company already has a builder prospect, the existing builder prospect is returned. Otherwise, a new builder prospect is created.

Source code in components/global_account/internal/actions/builder_prospect.py

def get_or_create_builder_prospects_from_companies(
    company_ids: list[int],
) -> dict[str, CompanyProspect]:
    """
    Return a mapping (company_ref -> prospect) for all the provided companies.
    If the company already has a builder prospect, the existing builder prospect
    is returned. Otherwise, a new builder prospect is created.
    """
    from components.fr.internal.models.company import (  # noqa: ALN069
        Company,
    )

    if not company_ids:
        return {}

    companies: list[Company] = (
        current_session.query(Company)  # noqa: ALN085
        .options(
            joinedload(Company.address),
            joinedload(Company.ccn),
            joinedload(Company.franchise),
        )
        .filter(Company.id.in_(company_ids))
        .all()
    )
    account_id = one(
        set(company.account_id for company in companies),
        "All companies must belong to the same account",
    )

    prospects: list[BuilderProspectSQLA] = (
        current_session.query(BuilderProspectSQLA)  # noqa: ALN085
        .filter(
            BuilderProspectSQLA.alan_account_id == str(account_id),
            BuilderProspectSQLA.company_ref.in_(
                str(company.id) for company in companies
            ),
        )
        .all()
    )

    companies_with_prospects = {prospect.company_ref for prospect in prospects}

    for company in companies:
        if str(company.id) not in companies_with_prospects:
            prospects.append(_create_builder_prospect_from_company(company))

    return {
        mandatory(company_prospect.company_ref): company_prospect
        for company_prospect in get_company_prospects(
            [str(prospect.id) for prospect in prospects]
        )
    }

get_or_create_prospects_from_account_id_and_entities ¶

get_or_create_prospects_from_account_id_and_entities(
    account_id, entities
)

Fetches all the builder prospects from the account (possibly creates prospects if only comnpanies exist) then returns all the prospects that are referenced by the entities list

Source code in components/global_account/internal/actions/builder_prospect.py

def get_or_create_prospects_from_account_id_and_entities(
    account_id: uuid.UUID,
    entities: list[EntityRef],
) -> list[CompanyProspect]:
    """Fetches all the builder prospects from the account (possibly creates prospects if only comnpanies exist)
    then returns all the prospects that are referenced by the entities list"""
    builder_prospects = get_account_repository().get_or_create_builder_prospects(
        account_id
    )

    relevant_prospects = [
        bp
        for bp in builder_prospects
        for entity in entities
        if str(bp.alan_account_id) == str(account_id)
        and (
            str(bp.id) == str(entity.prospect_ref)
            or (
                bp.company is not None
                and entity.company_ref is not None
                and str(bp.company.id) == str(entity.company_ref)
            )
        )
    ]

    if len(entities) != len(relevant_prospects):
        matched_prospect_ids = {str(bp.id) for bp in relevant_prospects}
        matched_company_ids = {
            str(bp.company.id) for bp in relevant_prospects if bp.company is not None
        }
        unmatched = [
            entity
            for entity in entities
            if (
                entity.prospect_ref is None
                or str(entity.prospect_ref) not in matched_prospect_ids
            )
            and (
                entity.company_ref is None
                or str(entity.company_ref) not in matched_company_ids
            )
        ]
        raise BaseErrorCode.invalid_arguments(
            description=(
                f"Could not resolve prospects on account {account_id} for entities: "
                + ", ".join(
                    f"(prospect_ref={e.prospect_ref}, company_ref={e.company_ref})"
                    for e in unmatched
                )
                + ". Please open an Eng on-call card with these IDs — the Company may live on a different account."
            )
        )

    return [CompanyProspect.from_builder_prospect(bp) for bp in relevant_prospects]

get_or_create_prospects_from_entities ¶

get_or_create_prospects_from_entities(entities)

Get or create builder prospects from entities.

If the entity has a prospect_ref, get the prospect by ID from the DB.
If the entity has a company_ref:
get the builder prospect for the company if it exists.
else create it.

Source code in components/global_account/internal/actions/builder_prospect.py

def get_or_create_prospects_from_entities(
    entities: list[EntityRef],
) -> list[CompanyProspect]:
    """
    Get or create builder prospects from entities.

    - If the entity has a prospect_ref, get the prospect by ID from the DB.
    - If the entity has a company_ref:
      - get the builder prospect for the company if it exists.
      - else create it.
    """
    company_prospects: list[CompanyProspect] = []
    with_prospect, without_prospect = partition(
        entities, lambda e: e.prospect_ref is not None
    )

    company_prospects += get_company_prospects(
        [mandatory(e.prospect_ref) for e in with_prospect]
    )

    company_prospects += get_or_create_builder_prospects_from_companies(
        [int(mandatory(e.company_ref)) for e in without_prospect]
    ).values()

    return company_prospects

get_self_serve_builder_prospect ¶

get_self_serve_builder_prospect(onboarding_id, country)

Return the most recent self-serve BuilderProspect for the given onboarding_id and country.

Source code in components/global_account/public/builder_prospect.py

def get_self_serve_builder_prospect(
    onboarding_id: str,
    country: GlobalAccountSupportedCountry,
) -> BuilderProspect | None:
    """Return the most recent self-serve BuilderProspect for the given onboarding_id and country."""
    from components.global_account.internal.mappers.builder_prospect_mapper import (
        BuilderProspectMapper,
    )
    from components.global_account.internal.models.builder_prospect import (
        BuilderProspect as BuilderProspectSQLA,
    )
    from components.global_account.internal.repositories.account_repository import (
        create_account_repository,
    )

    model = current_session.scalars(
        select(BuilderProspectSQLA)
        .where(
            BuilderProspectSQLA.onboarding_id == onboarding_id,
            BuilderProspectSQLA.is_from_self_serve.is_(True),
            BuilderProspectSQLA.country == country,
        )
        .order_by(BuilderProspectSQLA.created_at.desc())
    ).first()
    if not model:
        return None
    account_repository = create_account_repository(country=country)
    company = (
        account_repository.get_company(model.company_ref) if model.company_ref else None
    )
    return BuilderProspectMapper.to_entity(model, company)

components.global_account.public.country_specific ¶

be ¶

dependencies ¶

global_account_dependency `module-attribute` ¶

global_account_dependency = BeGlobalAccountDependency()

es ¶

dependencies ¶

global_account_dependency `module-attribute` ¶

global_account_dependency = EsGlobalAccountDependency()

fr ¶

dependencies ¶

global_account_dependency `module-attribute` ¶

global_account_dependency = FrGlobalAccountDependency()

components.global_account.public.dependencies ¶

COMPONENT_NAME `module-attribute` ¶

COMPONENT_NAME = 'global_account'

GlobalAccountDependency ¶

Bases: ABC

Abstract class defining the dependencies needed by the Global Account component.