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)

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

@tracer.wrap(service="global_account")
def get_account(account_id: UUID) -> Account:
    account = _get_sqla_account(account_id)

    entities = get_account_entities(account_id)
    ccns = get_ccns({entity.ccn_code for entity in entities if entity.ccn_code})
    account_history = get_account_settings_history(
        account_id=account_id, created_at=account.created_at
    )

    return Account(
        id=account.id,
        name=account.name,
        created_at=account.created_at,
        entities=entities,
        ccns=ccns,
        periods=account_history,
    )

get_account_entities ¶

get_account_entities(account_id)

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

@tracer.wrap(service="global_account")
def get_account_entities(account_id: UUID) -> list[Entity]:
    account_repository = CompositeAccountRepository()
    account = _get_sqla_account(account_id)

    prospects = account_repository.search_builder_prospects(
        alan_account_id=str(account_id)
    )
    prospect_by_company_ref = {
        mandatory(prospect.company).id: prospect
        for prospect in prospects
        if prospect.company is not None
    }

    # Building entities from Builder Prospects
    entities: list[Entity] = [
        from_builder_prospect_to_entity(prospect) for prospect in prospects
    ]

    companies_without_builder_prospect = []
    entity_by_company_ref = {entity.company_ref: entity for entity in entities}

    for company in account.companies:
        # Aligning builder prospect & companies names for coherence
        if str(company.id) in prospect_by_company_ref:
            entity = entity_by_company_ref[str(company.id)]
            entity.name = company.name
            entity.formatted_name = company.display_name

        # Building entities from Companies not linked to any BuilderProspect
        else:
            companies_without_builder_prospect.append(
                CompanyEntityAPISchema(company_ref=str(company.id), prospect=None),
            )

    return entities + get_entities_from_companies(companies_without_builder_prospect)

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.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_builder_prospects_from_contract_versions `module-attribute` ¶

get_or_create_builder_prospects_from_contract_versions = (
    get_or_create_builder_prospects_from_contract_versions
)

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,
    )
    from components.global_account.internal.queries.builder_prospect import (
        get_company_prospect_from_model,
    )

    query = current_session.query(BuilderProspect).filter(  # noqa: ALN085
        BuilderProspect.company_identifier == siren,
        BuilderProspect.industry_framework_agreement_code == ccn_code,
        BuilderProspect.specific_company_identifier == nic,
    )

    if is_from_self_serve is not None:
        query = query.filter(BuilderProspect.is_from_self_serve == is_from_self_serve)

    builder_prospect_model = query.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_from_model(builder_prospect_model)
        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 get_company_prospect_from_model(
        get_or_raise_missing_resource(BuilderProspectSQLA, 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:
    builder_prospect = get_resource_or_none(BuilderProspectSQLA, id)
    return (
        get_company_prospect_from_model(builder_prospect) if builder_prospect else None
    )

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: list[BuilderProspectSQLA] = list(
        current_session.scalars(
            select(BuilderProspectSQLA).where(BuilderProspectSQLA.id.in_(ids))
        ).all()
    )

    found_ids = {str(builder_prospect.id) for builder_prospect in builder_prospects}
    missing_ids = set(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 get_company_prospects_from_models(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_from_models(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 = CompositeAccountRepository().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):
        raise ValueError(f"Missing/extra prospects? {entities=} {relevant_prospects=}")

    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.
    """
    from components.global_account.internal.queries.builder_prospect import (
        get_company_prospects,
    )

    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

components.global_account.public.composite_account_repository ¶

CompositeAccountRepository ¶

Bases: AccountRepository

Single AccountRepository that will query across countries.