Api reference

components.asset_freeze.public.commands

app_group

asset_freeze_commands module-attribute

asset_freeze_commands = AppGroup('asset_freeze')

potential_hits

retrieve_potential_hits

retrieve_potential_hits()

Retrieve hits from Turing, and store them (or update them) if needed in the backend

Source code in components/asset_freeze/public/commands/potential_hits.py

@asset_freeze_commands.command(requires_authentication=False)
def retrieve_potential_hits() -> None:
    """
    Retrieve hits from Turing, and store them (or update them) if needed in the backend
    """
    from shared.helpers.turing.fetch import fetch_all_from_turing

    token = json_secret_from_config(
        config_key="ASSET_FREEZE_AIRFLOW_DECRYPT_TOKEN",
        default_secret_value=current_config.get("ASSET_FREEZE_AIRFLOW_DECRYPT_TOKEN"),
    )

    if not token or "decryption_token" not in token:
        current_logger.error(
            "ASSET_FREEZE_AIRFLOW_DECRYPT_TOKEN is missing or not well formatted"
        )
        return

    decryption_token = token["decryption_token"]

    potential_hits_from_turing = fetch_all_from_turing(
        """
        SELECT
            company_id AS ENTITY_REF
            , company_type AS ENTITY_TYPE
            , original_name AS ORIGINAL_NAME
            , country_id AS COUNTRY
            , asset_freeze_payloads AS PAYLOADS
        FROM asset_freeze.potential_hits_companies_and_accounts
    UNION
        SELECT
            user_id AS ENTITY_REF
            , 'user' AS ENTITY_TYPE
            , func.decrypt(original_name, %(decryption_token)s) AS ORIGINAL_NAME
            , country_id AS COUNTRY
            , asset_freeze_payloads AS PAYLOADS
        FROM asset_freeze.potential_hits_users
    """,
        {"decryption_token": decryption_token},
    )

    all_existing_potential_hits: list[PotentialHit] = (
        current_session.execute(select(PotentialHit)).scalars().all()  # type: ignore[assignment]
    )
    all_existing_potential_hits_by_type_and_ref: dict[
        tuple[str, str, str], PotentialHit
    ] = {
        (r.country, r.entity_type, r.entity_ref): r for r in all_existing_potential_hits
    }

    # We're going to track the number of new hits so we can warn Ops in case there are
    new_hit_count = 0

    # tuple: country, type, id
    found_hits: set[tuple[str, str, str]] = set()
    for potential_hit in potential_hits_from_turing:
        country: AssetFreezeCountry = mandatory(potential_hit.get("COUNTRY"))
        entity_type: PotentialHitEntityType = mandatory(
            potential_hit.get("ENTITY_TYPE")
        )
        entity_ref: str = mandatory(potential_hit.get("ENTITY_REF"))
        original_name: str = mandatory(potential_hit.get("ORIGINAL_NAME"))
        payloads: str = mandatory(potential_hit.get("PAYLOADS"))

        # NOTE:
        # There's a slight simplification here, as we're only looking for hits that matched the same
        # user, but we're actually not looking at what we matched against in the official list.
        # What does that means exactly? Let's take the following example:
        # - Official list has "Jean Dupont" in it, with id ABCD and we have user #123 "Jean Dupont" in our base
        # - We match - it creates a potential hit, in a pending state
        # - An Ops looks at it — they're not the same person — and dismisses it for a good reason
        # - a few months later, the official has a new entry: "Jean Dupont", with id EFGH (a different person than the first entry)
        # - We match again with our #123 user — but when we arrive here, the hit exists already, and is dismissed, and stays
        #   dismissed in that case
        # - if _ever_ this match was actually a legit one (user #123 was in fact the same person as id EFGH), then we would miss
        #   it.
        # All in all, I'm setting this edge case aside as unlikely to happen in real life. If it ever does happen, we could change the model
        # to store the ids of the official list in PotentialHit, and find the existing hits with this dimension too.
        existing_hit = all_existing_potential_hits_by_type_and_ref.get(
            (
                country,
                entity_type,
                entity_ref,
            ),
            None,
        )

        if existing_hit:
            # In case the hit already exists, update its "last_seen_at" flag to indicate
            # we still surface this entity as a potential hit
            existing_hit.last_seen_at = utcnow()
        else:
            # We don't have a potential hit already, we need to create it
            new_hit = PotentialHit(
                country=country,
                entity_type=entity_type,
                entity_ref=entity_ref,
                original_name=original_name,
                payloads=json.loads(payloads),
                last_seen_at=utcnow(),
            )
            current_session.add(new_hit)
            new_hit_count += 1

        # Store the hit so we can then identify the ones we miss
        found_hits.add((country, entity_type, entity_ref))

    # Now, we need to dismiss the hits in the backend if they don't appear in the Turing export
    # Two cases:
    # - the hit is already confirmed / dismissed --> we don't do anything
    # - the hit is still pending verification --> we can dismiss it as "not a hit anymore" (in the comment)
    dismissed_hits = 0
    for existing_hit in all_existing_potential_hits:
        if (
            not existing_hit.dismissed_at
            and not existing_hit.confirmed_at
            and (
                existing_hit.country,
                existing_hit.entity_type,
                existing_hit.entity_ref,
            )
            not in found_hits
        ):
            dismiss_potential_hit(
                id=existing_hit.id,
                comment=f"Automatically dismissed as not present in the asset freeze official list as of {utcnow()}",
            )
            dismissed_hits += 1

    current_session.commit()

    current_logger.info(f"Added {new_hit_count} potential hits")
    current_logger.info(f"Dismissed {dismissed_hits} potential hits")

run_delete_expired_potential_hits

run_delete_expired_potential_hits(dry_run)

Delete expired asset freeze potential hits. There is a 5-years data retention for the data collected for anti-money laundering / counter financing of terrorism purposes, as from the end of the contract/membership.

Source code in components/asset_freeze/public/commands/potential_hits.py

@asset_freeze_commands.command(requires_authentication=False)
@command_with_dry_run
def run_delete_expired_potential_hits(dry_run: bool) -> None:
    """
    Delete expired asset freeze potential hits.
    There is a 5-years data retention for the data collected for anti-money laundering / counter financing of terrorism purposes, as from the end of the contract/membership.
    """
    from components.asset_freeze.internal.business_logic.actions.potential_hits import (
        delete_expired_potential_hits,
    )

    delete_expired_potential_hits(commit=not dry_run)

components.asset_freeze.public.consts

ASSET_FREEZE_COMPONENT_NAME module-attribute

ASSET_FREEZE_COMPONENT_NAME = 'asset_freeze'

components.asset_freeze.public.dependencies

AssetFreezeDependency

AssetFreezeDependency(country_dependencies)

AssetFreezeDependency defines the interface that apps using the asset_freeze component need to implement

Source code in components/asset_freeze/public/dependencies.py

def __init__(
    self,
    country_dependencies: dict[AssetFreezeCountry, CountryAssetFreezeDependency],
):
    self.country_dependencies = country_dependencies

country_dependencies instance-attribute

country_dependencies = country_dependencies

CountryAssetFreezeDependency

CountryAssetFreezeDependency defines the interface that apps using the asset_freeze component need to implement

account_has_live_contract_since

account_has_live_contract_since(account_id, since)

The method returns True if a given account has a prevoyance or health contract active since the given date

Source code in components/asset_freeze/public/dependencies.py

def account_has_live_contract_since(
    self, account_id: str, since: datetime.date
) -> bool:
    """The method returns True if a given account has a prevoyance or health contract active since the given date"""
    raise NotImplementedError()

company_has_live_contract_since

company_has_live_contract_since(company_id, since)

The method returns True if a given company has a prevoyance or health contract active since the given date

Source code in components/asset_freeze/public/dependencies.py

def company_has_live_contract_since(
    self, company_id: str, since: datetime.date
) -> bool:
    """The method returns True if a given company has a prevoyance or health contract active since the given date"""
    raise NotImplementedError()

user_has_live_enrollment_since

user_has_live_enrollment_since(user_id, since)

The method returns True if a given user has an enrollment active since the given date

Source code in components/asset_freeze/public/dependencies.py

def user_has_live_enrollment_since(
    self, user_id: str, since: datetime.date
) -> bool:
    """The method returns True if a given user has an enrollment active since the given date"""
    raise NotImplementedError()

get_app_dependency

get_app_dependency()

Retrieves at runtime the asset_freeze dependency set by set_app_dependency

Source code in components/asset_freeze/public/dependencies.py

def get_app_dependency() -> AssetFreezeDependency:
    """Retrieves at runtime the asset_freeze dependency set by set_app_dependency"""
    from flask import current_app

    app = cast("CustomFlask", current_app)

    return cast(
        "AssetFreezeDependency",
        app.get_component_dependency(ASSET_FREEZE_COMPONENT_NAME),
    )

set_app_dependencies

set_app_dependencies(dependency)

Sets the asset_freeze dependency to the app so it can be accessed within this component at runtime

Source code in components/asset_freeze/public/dependencies.py

def set_app_dependencies(dependency: AssetFreezeDependency) -> None:
    """Sets the asset_freeze dependency to the app so it can be accessed within this component at runtime"""
    from flask import current_app

    cast("CustomFlask", current_app).add_component_dependency(
        ASSET_FREEZE_COMPONENT_NAME, dependency
    )

components.asset_freeze.public.entities

AssetFreezeHit dataclass

AssetFreezeHit(
    id,
    entity_ref,
    entity_type,
    original_name,
    country,
    first_seen_at,
    last_seen_at,
    person,
    confirmed_at=None,
    dismissed_at=None,
    updated_by=None,
    comment=None,
)

Bases: DataClassJsonMixin

Holds the asset freeze potential hit data to send to the dashboard

comment class-attribute instance-attribute

comment = None

confirmed_at class-attribute instance-attribute

confirmed_at = None

country instance-attribute

country

dismissed_at class-attribute instance-attribute

dismissed_at = None

entity_ref instance-attribute

entity_ref

entity_type instance-attribute

entity_type

first_seen_at instance-attribute

first_seen_at

id instance-attribute

id

last_seen_at instance-attribute

last_seen_at

original_name instance-attribute

original_name

person instance-attribute

person

updated_by class-attribute instance-attribute

updated_by = None

LegalPerson dataclass

LegalPerson(payloads, name, display_name)

Bases: DataClassJsonMixin

A legal person, in terms of asset freeze official lists

display_name instance-attribute

display_name

name instance-attribute

name

payloads instance-attribute

payloads

NaturalPerson dataclass

NaturalPerson(
    payloads,
    last_name,
    first_name,
    birth_year,
    birth_month,
    birth_day,
    display_name,
)

Bases: DataClassJsonMixin

A natural (physique) person, in terms of asset freeze official lists

birth_day instance-attribute

birth_day

birth_month instance-attribute

birth_month

birth_year instance-attribute

birth_year

display_name instance-attribute

display_name

first_name instance-attribute

first_name

last_name instance-attribute

last_name

payloads instance-attribute

payloads

components.asset_freeze.public.enums

asset_freeze_country

AssetFreezeCountry

Bases: AlanBaseEnum

Enumerates the countries for which asset freeze is implemented

BE class-attribute instance-attribute

BE = 'be'

FR class-attribute instance-attribute

FR = 'fr'

potential_hit_entity_type

PotentialHitEntityType

Bases: AlanBaseEnum

Enumerates the different types of hits we can find

ACCOUNT class-attribute instance-attribute

ACCOUNT = 'account'

COMPANY class-attribute instance-attribute

COMPANY = 'company'

USER class-attribute instance-attribute

USER = 'user'