Api reference

components.growth.public.api

Growth component public API.

This module consolidates all public exports from the growth component. External code should import from this module rather than internal modules.

Accountant dataclass

Accountant(
    email,
    name,
    accounting_firm_partner_code,
    accounting_firm_name,
)

Bases: DataClassJsonMixin

accounting_firm_name instance-attribute

accounting_firm_name

accounting_firm_partner_code instance-attribute

accounting_firm_partner_code

email instance-attribute

email

name instance-attribute

name

CustomerIOEmailCaptureEventName

Bases: AlanBaseEnum

BE_EMAIL_CAPTURE class-attribute instance-attribute

BE_EMAIL_CAPTURE = 'be_email_captured'

FR_COMPANY_EMAIL_CAPTURE class-attribute instance-attribute

FR_COMPANY_EMAIL_CAPTURE = 'company_email_captured'

FR_INDIVIDUAL_EMAIL_CAPTURE class-attribute instance-attribute

FR_INDIVIDUAL_EMAIL_CAPTURE = 'individual_email_captured'

DocumentReviewIndividualParams dataclass

DocumentReviewIndividualParams(
    proposal_id,
    primary_age,
    plan_id,
    health_start_date,
    legacy_health_contract,
)

health_start_date instance-attribute

health_start_date

legacy_health_contract instance-attribute

legacy_health_contract

plan_id instance-attribute

plan_id

primary_age instance-attribute

primary_age

proposal_id instance-attribute

proposal_id

DocumentReviewParams dataclass

DocumentReviewParams(
    proposal_id,
    ccn_code,
    company_siren,
    company_name,
    plan_id,
    health_start_date,
    participation,
    contract_cover_option,
    legacy_health_contract,
    choose_prevoyance,
    legacy_prevoyance_contract,
)

ccn_code instance-attribute

ccn_code

choose_prevoyance instance-attribute

choose_prevoyance

company_name instance-attribute

company_name

company_siren instance-attribute

company_siren

contract_cover_option instance-attribute

contract_cover_option

health_start_date instance-attribute

health_start_date

legacy_health_contract instance-attribute

legacy_health_contract

legacy_prevoyance_contract instance-attribute

legacy_prevoyance_contract

participation instance-attribute

participation

plan_id instance-attribute

plan_id

proposal_id instance-attribute

proposal_id

EmailCaptureData dataclass

EmailCaptureData(
    *, customer_io_event_name, customer_io_event_attributes
)

Bases: DataClassJsonMixin

Event data to be sent to Customer.io for company email capture event

customer_io_event_attributes instance-attribute

customer_io_event_attributes

customer_io_event_name instance-attribute

customer_io_event_name

GrowthIntercomHandlerService

Service to handle Intercom webhook events for Growth tracking. Only handles user_replied and admin_replied topics.

handle_topic staticmethod

handle_topic(
    intercom_workspace_id, intercom_conversation, topic
)

Handle Intercom topics for Growth tracking.

Source code in components/growth/internal/services/growth_intercom_handler_service.py

@staticmethod
def handle_topic(
    intercom_workspace_id: str,
    intercom_conversation: IntercomConversation,
    topic: IntercomWebhookTopic,
) -> None:
    """
    Handle Intercom topics for Growth tracking.
    """
    # not supported for other countries as we don't have an Intercom client for them
    if get_current_app_country_code() not in ["FR", "BE"]:
        return

    current_logger.info(
        "GrowthIntercomHandlerService.handle_topic",
        conversation_id=intercom_conversation.id,
        workspace_id=intercom_workspace_id,
        topic=topic,
    )

    if topic == IntercomWebhookTopic.conversation_user_replied:
        GrowthIntercomHandlerService._handle_conversation_user_replied(  # noqa: ALN027
            intercom_conversation=intercom_conversation
        )
    elif topic == IntercomWebhookTopic.conversation_admin_replied:
        GrowthIntercomHandlerService._handle_conversation_admin_replied(  # noqa: ALN027
            intercom_conversation=intercom_conversation
        )

GrowthUser dataclass

GrowthUser(*, id, email, pro_email, alan_email, full_name)

Bases: DataClassJsonMixin

Represents a user from in the growth component's scope

alan_email instance-attribute

alan_email

email instance-attribute

email

full_name instance-attribute

full_name

id instance-attribute

id

pro_email instance-attribute

pro_email

InboundSalesTeam dataclass

InboundSalesTeam(
    *,
    crew_name,
    crew_slack_channel,
    won_contracts_slack_channel,
    crew_lead_user
)

Bases: DataClassJsonMixin

Represents an inbound sales team / crew for a specific country

crew_lead_user instance-attribute

crew_lead_user

crew_name instance-attribute

crew_name

crew_slack_channel instance-attribute

crew_slack_channel

won_contracts_slack_channel instance-attribute

won_contracts_slack_channel

IndividualDocumentReviewEvent dataclass

IndividualDocumentReviewEvent(
    proposal_id,
    health_start_date_timestamp,
    primary_price,
    primary_age,
    has_legacy_health_contract,
    document_review_version="1",
)

Bases: DataClassJsonMixin

document_review_version class-attribute instance-attribute

document_review_version = '1'

has_legacy_health_contract instance-attribute

has_legacy_health_contract

health_start_date_timestamp instance-attribute

health_start_date_timestamp

primary_age instance-attribute

primary_age

primary_price instance-attribute

primary_price

proposal_id instance-attribute

proposal_id

MINIMUM_NUMBER_OF_EMPLOYEES_FOR_SALES module-attribute

MINIMUM_NUMBER_OF_EMPLOYEES_FOR_SALES = 6

MarketingEventParameters dataclass

MarketingEventParameters(
    utms,
    all_utms,
    clids,
    facebook,
    alan_partner_code,
    referring_user_token,
    referring_partner,
    cookie_status,
    version,
    landing_referrer_url,
    landing_page_url,
    device_type=None,
)

Bases: DataClassJsonMixin, DataClassJsonAlanMixin

__post_init__

__post_init__()

Truncate all string fields to appropriate max lengths.

Source code in components/growth/internal/entities/prospect.py

def __post_init__(self) -> None:
    """Truncate all string fields to appropriate max lengths."""
    self.alan_partner_code = _truncate_field(
        self.alan_partner_code, "alan_partner_code", ""
    )
    self.referring_user_token = _truncate_field(
        self.referring_user_token, "referring_user_token", ""
    )
    self.referring_partner = _truncate_field(
        self.referring_partner, "referring_partner", ""
    )
    self.cookie_status = _truncate_field(self.cookie_status, "cookie_status", "")
    self.landing_referrer_url = _truncate_field(
        self.landing_referrer_url, "landing_referrer_url", "", MAX_URL_FIELD_LENGTH
    )
    self.landing_page_url = _truncate_field(
        self.landing_page_url, "landing_page_url", "", MAX_URL_FIELD_LENGTH
    )

alan_partner_code instance-attribute

alan_partner_code

all_utms instance-attribute

all_utms

clids instance-attribute

clids

cookie_status instance-attribute

cookie_status

device_type class-attribute instance-attribute

device_type = None

facebook instance-attribute

facebook

landing_page_url instance-attribute

landing_page_url

landing_referrer_url instance-attribute

landing_referrer_url

referring_partner instance-attribute

referring_partner

referring_user_token instance-attribute

referring_user_token

utms instance-attribute

utms

version instance-attribute

version

PROSPECT_SOURCE module-attribute

PROSPECT_SOURCE = 'prospect'

PersonaType

Bases: AlanBaseEnum

Marketing persona types (mirrors frontend PersonaType).

BelgiumSmc class-attribute instance-attribute

BelgiumSmc = 'BelgiumSMC'

CivilServant class-attribute instance-attribute

CivilServant = 'CivilServant'

Company class-attribute instance-attribute

Company = 'Company'

FranceTns class-attribute instance-attribute

FranceTns = 'FranceTNS'

Individual class-attribute instance-attribute

Individual = 'Individual'

Retiree class-attribute instance-attribute

Retiree = 'Retiree'

SpainAutonomo class-attribute instance-attribute

SpainAutonomo = 'SpainAutonomo'

ProspectEventContractSignedProperties dataclass

ProspectEventContractSignedProperties(*, contract_id)

Bases: DataClassJsonMixin

Properties for contract signed events.

contract_id instance-attribute

contract_id

ProspectEventManager

Static class for creating prospect events in the database.

create_prospect_event staticmethod

create_prospect_event(
    email,
    event_name,
    country_code,
    persona_type,
    marketing_parameters,
    event_properties,
    create_prospect_if_not_exists=True,
    commit=False,
)

Create a prospect event in the database.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def create_prospect_event(
    email: str,
    event_name: ProspectEventType,
    country_code: CountryCode | None,
    persona_type: PersonaType | None,
    marketing_parameters: MarketingEventParameters | None,
    event_properties: dict[str, Any] | None,
    create_prospect_if_not_exists: bool = True,
    commit: bool = False,
) -> None:
    """Create a prospect event in the database."""
    from components.growth.internal.models.prospect_event import ProspectEvent

    prospect = get_prospect_by_email(email)
    if prospect is None:
        if create_prospect_if_not_exists:
            current_logger.info(
                "Prospect not found, creating new prospect",
            )
            prospect = register_new_prospect(email)
        else:
            # Some Prospect Events should not create a prospect because they should normally already exists in our database
            # In that case, we would prefer not to create a prospect and log a warning (for now), once the feature will
            # be fully up and running we will switch this to error to better track unknown prospects.
            current_logger.warn(
                "Prospect not found while we would expect the prospect to exists",
                event_name=event_name,
            )
            return None

    prospect_event = ProspectEvent(
        prospect_id=prospect.id,
        country_code=country_code,
        event_name=event_name,
        persona_type=persona_type,
        marketing_parameters=serialize_marketing_parameters(marketing_parameters),
        event_properties=event_properties,
    )

    current_session.add(prospect_event)
    if commit:
        current_session.commit()

on_account_creation staticmethod

on_account_creation(
    email,
    country_code,
    persona_type,
    marketing_parameters,
    commit=True,
)

Dispatch account creation event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_account_creation(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType,
    marketing_parameters: MarketingEventParameters | None,
    commit: bool = True,
) -> None:
    """Dispatch account creation event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.FLOW_ACCOUNT_CREATION,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        event_properties=None,
        commit=commit,
    )

on_call_hungup staticmethod

on_call_hungup(email, country_code, commit=True)

Dispatch sales inbound call hung up event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_call_hungup(
    email: str,
    country_code: CountryCode,
    commit: bool = True,
) -> None:
    """Dispatch sales inbound call hung up event."""
    prospect = get_prospect_by_email(email)
    if prospect is None:
        return

    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.SALES_INBOUND_CALL_HUNG_UP,
        country_code=country_code,
        persona_type=None,
        marketing_parameters=None,
        event_properties=None,
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_contact_sales staticmethod

on_contact_sales(
    email,
    country_code,
    persona_type,
    properties,
    marketing_parameters,
    commit=True,
)

Dispatch meeting request event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_contact_sales(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType | None,
    properties: ProspectEventContactSalesProperties,
    marketing_parameters: MarketingEventParameters | None,
    commit: bool = True,
) -> None:
    """Dispatch meeting request event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.MEETING_REQUEST,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        event_properties=properties.to_dict(),
        commit=commit,
    )

on_contract_signed staticmethod

on_contract_signed(
    email,
    country_code,
    persona_type,
    properties,
    commit=True,
)

Dispatch contract signed event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_contract_signed(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType,
    properties: ProspectEventContractSignedProperties,
    commit: bool = True,
) -> None:
    """Dispatch contract signed event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.FLOW_CONTRACT_SIGNED,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=None,
        event_properties=properties.to_dict(),
        commit=commit,
    )

on_flow_capture_screen staticmethod

on_flow_capture_screen(
    email,
    country_code,
    persona_type,
    marketing_parameters,
    commit=True,
)

Dispatch flow capture contact event (email capture).

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_flow_capture_screen(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType,
    marketing_parameters: MarketingEventParameters | None,
    commit: bool = True,
) -> None:
    """Dispatch flow capture contact event (email capture)."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.FLOW_CAPTURE_SCREEN,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        event_properties=None,
        commit=commit,
    )

on_intercom_admin_replied staticmethod

on_intercom_admin_replied(
    email, team_id, tags=None, commit=True
)

Dispatch sales inbound mail/chat replied event based on tags.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_intercom_admin_replied(
    email: str,
    team_id: str | None,
    tags: list[str] | None = None,
    commit: bool = True,
) -> None:
    """Dispatch sales inbound mail/chat replied event based on tags."""
    from components.growth.internal.business_logic.crm.intercom import (
        is_assigned_to_inbound_sales,
    )

    country_code = get_current_app_country_code()

    if not email or email == "unknown":
        return

    if not team_id or not is_assigned_to_inbound_sales(
        country_code=country_code, inbox_id=team_id
    ):
        return

    prospect = get_prospect_by_email(email)
    if prospect is None:
        return

    # Determine event type based on tags
    is_chat = tags and SELF_SERVE_TAG_CHAT in tags
    event_name = (
        ProspectEventType.SALES_INBOUND_CHAT_SENT
        if is_chat
        else ProspectEventType.SALES_INBOUND_MAIL_SENT
    )

    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=event_name,
        country_code=country_code,
        persona_type=None,
        marketing_parameters=None,
        event_properties=None,
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_intercom_user_replied staticmethod

on_intercom_user_replied(
    email, team_id, tags=None, commit=True
)

Dispatch sales inbound mail/chat sent event based on tags.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_intercom_user_replied(
    email: str,
    team_id: str | None,
    tags: list[str] | None = None,
    commit: bool = True,
) -> None:
    """Dispatch sales inbound mail/chat sent event based on tags."""

    from components.growth.internal.business_logic.crm.intercom import (
        is_assigned_to_inbound_sales,
    )

    country_code = get_current_app_country_code()

    if not email or email == "unknown":
        return

    if not team_id or not is_assigned_to_inbound_sales(
        country_code=country_code, inbox_id=team_id
    ):
        return

    prospect = get_prospect_by_email(email)
    if prospect is None:
        return

    is_chat = tags and SELF_SERVE_TAG_CHAT in tags
    event_name = (
        ProspectEventType.SALES_INBOUND_CHAT_REPLIED
        if is_chat
        else ProspectEventType.SALES_INBOUND_MAIL_REPLIED
    )

    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=event_name,
        country_code=country_code,
        persona_type=None,
        marketing_parameters=None,
        event_properties=None,
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_meeting_request_attribution_survey staticmethod

on_meeting_request_attribution_survey(
    email,
    country_code,
    persona_type,
    properties,
    marketing_parameters=None,
    commit=True,
)

Dispatch attribution survey filled event for meeting requests.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_meeting_request_attribution_survey(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType | None,
    properties: ProspectEventMeetingRequestAttributionSurveyProperties,
    marketing_parameters: MarketingEventParameters | None = None,
    commit: bool = True,
) -> None:
    """Dispatch attribution survey filled event for meeting requests."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.MEETING_REQUEST_ATTRIBUTION_SURVEY,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        event_properties=properties.to_dict(),
        commit=commit,
    )

on_nurturing_mail_clicked staticmethod

on_nurturing_mail_clicked(email, properties, commit=True)

Dispatch nurturing mail clicked event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_nurturing_mail_clicked(
    email: str,
    properties: ProspectEventNurturingProperties,
    commit: bool = True,
) -> None:
    """Dispatch nurturing mail clicked event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.NURTURING_MAIL_CLICKED,
        country_code=None,
        persona_type=None,
        marketing_parameters=None,
        event_properties=properties.to_dict(),
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_nurturing_mail_read staticmethod

on_nurturing_mail_read(email, properties, commit=True)

Dispatch nurturing mail read event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_nurturing_mail_read(
    email: str,
    properties: ProspectEventNurturingProperties,
    commit: bool = True,
) -> None:
    """Dispatch nurturing mail read event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.NURTURING_MAIL_READ,
        persona_type=None,
        country_code=None,
        marketing_parameters=None,
        event_properties=properties.to_dict(),
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_nurturing_mail_sent staticmethod

on_nurturing_mail_sent(email, properties, commit=True)

Dispatch nurturing mail sent event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_nurturing_mail_sent(
    email: str,
    properties: ProspectEventNurturingProperties,
    commit: bool = True,
) -> None:
    """Dispatch nurturing mail sent event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.NURTURING_MAIL_SENT,
        country_code=None,
        persona_type=None,
        marketing_parameters=None,
        event_properties=properties.to_dict(),
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_page_view staticmethod

on_page_view(
    email,
    properties,
    marketing_parameters=None,
    commit=True,
)

Dispatch page view event for known prospects only.

Marketing parameters are only stored if different from the last recorded parameters for this prospect (across any event type).

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_page_view(
    email: str,
    properties: ProspectEventPageViewProperties,
    marketing_parameters: MarketingEventParameters | None = None,
    commit: bool = True,
) -> None:
    """Dispatch page view event for known prospects only.

    Marketing parameters are only stored if different from the last recorded
    parameters for this prospect (across any event type).
    """
    # Check deduplication before creating event
    prospect = get_prospect_by_email(email)
    if prospect is None:
        return None

    # Only pass marketing_parameters if different from last recorded
    should_store = should_store_marketing_parameters(
        prospect_id=prospect.id,
        marketing_parameters=marketing_parameters,
    )
    params_to_store = marketing_parameters if should_store else None

    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.PAGE_VIEW,
        country_code=None,
        persona_type=None,
        marketing_parameters=params_to_store,
        event_properties=properties.to_dict(),
        create_prospect_if_not_exists=False,
        commit=commit,
    )

on_quote_request staticmethod

on_quote_request(
    email,
    country_code,
    persona_type,
    marketing_parameters,
    commit=True,
)

Dispatch quote request event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_quote_request(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType,
    marketing_parameters: MarketingEventParameters | None,
    commit: bool = True,
) -> None:
    """Dispatch quote request event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.FLOW_QUOTE_REQUEST,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        event_properties=None,
        commit=commit,
    )

on_website_capture_form staticmethod

on_website_capture_form(
    email,
    country_code,
    persona_type,
    marketing_parameters,
    properties=None,
    commit=True,
)

Dispatch website capture form event.

Source code in components/growth/internal/business_logic/prospect/prospect_event_manager.py

@staticmethod
def on_website_capture_form(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType,
    marketing_parameters: MarketingEventParameters | None,
    properties: WebsiteCaptureFormProperties | None = None,
    commit: bool = True,
) -> None:
    """Dispatch website capture form event."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.WEBSITE_CAPTURE_FORM,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        event_properties=properties.to_dict() if properties else None,
        commit=commit,
    )

ProspectFactory

Bases: AlanBaseFactory['Prospect']

Meta

model class-attribute instance-attribute

model = Prospect

email class-attribute instance-attribute

email = Sequence(lambda n: f'prospect-email{n}@alan.eu')

ProspectRequestType module-attribute

ProspectRequestType = Literal[
    "meeting_request",
    "callback",
    "email",
    "chat",
    "account_creation_stuck",
    "quote_request",
    "email_capture",
    "contract_comparison",
]

QuoteRequestData dataclass

QuoteRequestData(
    *,
    customer_io_event_name,
    customer_io_event_attributes,
    should_be_sent_to_crm,
    quote_details,
    self_serve_flow_id
)

Bases: DataClassJsonMixin

customer_io_event_attributes instance-attribute

customer_io_event_attributes

customer_io_event_name instance-attribute

customer_io_event_name

quote_details instance-attribute

quote_details

self_serve_flow_id instance-attribute

self_serve_flow_id

should_be_sent_to_crm instance-attribute

should_be_sent_to_crm

ReferralActions

Actions for the referral lifecycle.

add_contract_to_referral staticmethod

add_contract_to_referral(referral_id, contract_ref)

Add a contract reference to be rewarded for the referral.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
def add_contract_to_referral(referral_id: UUID, contract_ref: str) -> None:
    """Add a contract reference to be rewarded for the referral."""
    # REQUIRES_NEW: this action must commit independently — callers may run
    # in contexts where the global session won't auto-commit
    with transaction(propagation=Propagation.REQUIRES_NEW) as session:
        referral = session.get(Referral, referral_id)

        current_logger.info(
            f"Setting Referral contract ref: Referral {referral_id} to {contract_ref}",
            referral_id=referral_id,
            contract_ref=contract_ref,
        )

        if not referral:
            raise ReferralContractRefException(
                f"Setting Referral contract ref {contract_ref}: Referral {referral_id} not found"
            )

        if referral.referred_contract_rewarded_on:
            raise ReferralContractRefException(
                f"Setting Referral contract ref {contract_ref}: Referral {referral_id} has been rewarded"
            )

        if (
            referral.referred_contract_ref
            and referral.referred_contract_ref != contract_ref
        ):
            raise ReferralContractRefException(
                f"Setting Referral contract ref {contract_ref}: Referral {referral_id} is set with a different contract ref {referral.referred_contract_ref}"
            )

        referral.referred_contract_ref = contract_ref
        session.add(referral)

        register_event(
            ReferralSignedContractEvent(
                referred_user_id=referral.referred_profile.user_id,
                referring_user_id=referral.referring_profile.user_id
                if referral.referring_profile
                else None,
            )
        )

consume_referral staticmethod

consume_referral(referral_id, month=None)

Consume a referral for the rewarded contract.

Creates a discount from the referral for a given month if the conditions are met.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
@enqueueable
def consume_referral(referral_id: UUID, month: Month | None = None) -> None:
    """
    Consume a referral for the rewarded contract.

    Creates a discount from the referral for a given month if the conditions are met.
    """
    with transaction(propagation=Propagation.REQUIRES_NEW) as session:
        referral = session.get(Referral, referral_id)

        if not referral:
            raise ReferralConsumptionException(
                f"Consuming Referral {referral_id}: Referral not found"
            )

        if not referral.referred_contract_ref:
            current_logger.info(
                f"Consuming Referral {referral_id}: Referral has no contract ref",
                referral_id=referral_id,
            )
            return None

        if referral.referred_contract_rewarded_on:
            current_logger.info(
                f"Consuming Referral {referral_id}: Referral has already been rewarded",
                referral_id=referral_id,
                rewarded_on=referral.referred_contract_rewarded_on,
            )
            return None

        # Check if a discount already exists (extra-safeguard)
        country_code = referral.referred_profile.country_code
        if ReferralCountryRouter.has_discount_for_referral(
            country_code=country_code,
            referral_id=referral.id,
        ):
            current_logger.info(
                f"Consuming Referral {referral_id}: Discount already exists",
                referral_id=referral_id,
            )
            if ReferralCountryRouter.should_mark_referral_as_rewarded_when_creating_discount(
                country_code
            ):
                referral.referred_contract_rewarded_on = Month.current()
                session.add(referral)
            return None

        # Get contract info once - eligibility can be checked for different months
        contract_info = ReferralCountryRouter.get_referred_contract(
            country_code=referral.referred_profile.country_code,
            contract_ref=referral.referred_contract_ref,
        )
        if not contract_info:
            current_logger.info(
                f"Consuming Referral {referral_id}: Contract not found",
                referral_id=referral_id,
                contract_ref=referral.referred_contract_ref,
            )
            return None

        contract_signed_at = (
            contract_info.signed_at.date() if contract_info.signed_at else None
        )

        # Determine reward month
        if month is not None:
            reward_month = month
            if contract_info.eligible_for_free_month(reward_month):
                _create_discount_for_referral(
                    referral,
                    reward_month,
                    session,
                    contract_signed_at=contract_signed_at,
                )
            else:
                current_logger.info(
                    f"Consuming Referral {referral_id}: contract not eligible for free month",
                    referral_id=referral_id,
                    month=str(month),
                )
        else:
            # Try current month or next month
            start_month = max(Month.current(), contract_info.start_date)
            if contract_info.eligible_for_free_month(start_month):
                _create_discount_for_referral(
                    referral,
                    start_month,
                    session,
                    contract_signed_at=contract_signed_at,
                )
            else:
                # Try next month
                next_month = start_month.one_month_later
                if contract_info.eligible_for_free_month(next_month):
                    _create_discount_for_referral(
                        referral,
                        next_month,
                        session,
                        contract_signed_at=contract_signed_at,
                    )
                else:
                    current_logger.info(
                        f"Consuming Referral {referral_id}: contract not eligible for free month on current or next month",
                        referral_id=referral_id,
                        start_month=str(start_month),
                        next_month=str(next_month),
                    )

create_referral staticmethod

create_referral(
    profile_service,
    referred_user_id,
    referral_token=None,
    referring_partner=None,
)

Create a referral.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
@inject_profile_service
def create_referral(
    profile_service: ProfileService,
    referred_user_id: str,
    referral_token: str | None = None,
    referring_partner: str | None = None,
) -> None | UUID:
    """Create a referral."""
    with transaction(propagation=Propagation.REQUIRES_NEW) as session:
        referred_user_profile = profile_service.get_user_profile(
            user_id=referred_user_id
        )
        if not referred_user_profile:
            current_logger.info(
                "Unable to create Referral: profile not found",
                user_id=referred_user_id,
            )
            return None

        referred_profile = get_or_create_referral_profile(referred_user_id)

        if referred_profile.received_referral:
            current_logger.info(
                "Unable to create Referral: referred profile has already been referred",
                referral_id=referred_profile.received_referral.id,
            )
            return (
                None
                if referred_profile.received_referral.referred_contract_ref
                else referred_profile.received_referral.id
            )

        if referral_token:
            referring_profile = (
                session.execute(
                    select(ReferralProfile).filter_by(referral_token=referral_token)
                )
                .scalars()
                .unique()
                .one_or_none()
            )

            if not referring_profile:
                current_logger.info(
                    f"Unable to create Referral: referral profile not found (token: {referral_token})"
                )
                return None

            if referring_profile.id == referred_profile.id:
                current_logger.info(
                    f"Unable to create Referral: referring profile is the same as the referred profile (token: {referral_token})"
                )
                return None

            referral = Referral(
                referred_profile=referred_profile,
                referring_profile=referring_profile,
                referred_country_code=referred_profile.country_code,
            )
            session.add(referral)
            session.flush()

            current_logger.info(
                f"Creating Referral {referral} from profile '{referring_profile.user_id}",
                referral_id=referral.id,
            )

            register_event(
                ReferralCreatedEvent(
                    referred_user_id=referred_profile.user_id,
                    referring_user_id=referring_profile.user_id,
                )
            )

            return referral.id

        elif referring_partner:
            referral = Referral(
                referring_partner=referring_partner,
                referred_profile=referred_profile,
                referred_country_code=referred_profile.country_code,
            )
            session.add(referral)
            session.flush()

            current_logger.info(
                f"Creating Referral {referral} from partner '{referring_partner}",
                referral_id=referral.id,
            )

            register_event(
                ReferralCreatedEvent(referred_user_id=referred_profile.user_id)
            )

            return referral.id

    return None

delete_referral staticmethod

delete_referral(referral_id)

Delete a GlobalReferral and its associated country-specific discount.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
def delete_referral(referral_id: UUID) -> None:
    """Delete a GlobalReferral and its associated country-specific discount."""
    referral = current_session.get(Referral, referral_id)
    if not referral:
        return
    ReferralCountryRouter.delete_discount_for_referral(
        country_code=referral.referred_country_code,
        referral_ref=str(referral.id),
    )
    current_session.delete(referral)

delete_referral_profile staticmethod

delete_referral_profile(user_id)

Delete a referral profile and all associated GlobalReferrals for a user.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
def delete_referral_profile(user_id: str) -> None:
    """Delete a referral profile and all associated GlobalReferrals for a user."""
    profile = (
        current_session.execute(
            select(ReferralProfile).filter(ReferralProfile.user_id == user_id)
        )
        .scalars()
        .one_or_none()
    )
    if not profile:
        return
    for sent_ref in profile.sent_referrals:
        ReferralCountryRouter.delete_discount_for_referral(
            country_code=sent_ref.referred_country_code,
            referral_ref=str(sent_ref.id),
        )
        current_session.delete(sent_ref)
    if profile.received_referral:
        received = profile.received_referral
        ReferralCountryRouter.delete_discount_for_referral(
            country_code=received.referred_country_code,
            referral_ref=str(received.id),
        )
        current_session.delete(received)
    current_session.delete(profile)

merge_referral_profiles staticmethod

merge_referral_profiles(
    duplicated_user_id, existing_user_id
)

Merge global referral profile of a duplicated user into an existing user's profile.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
def merge_referral_profiles(
    duplicated_user_id: str, existing_user_id: str
) -> list[str]:
    """Merge global referral profile of a duplicated user into an existing user's profile."""
    logs: list[str] = []
    dup_profile = (
        current_session.execute(
            select(ReferralProfile).filter(
                ReferralProfile.user_id == duplicated_user_id
            )
        )
        .scalars()
        .one_or_none()
    )
    if not dup_profile:
        return logs
    dup_global_referral = dup_profile.received_referral
    if dup_global_referral:
        existing_profile = get_or_create_referral_profile(existing_user_id)
        if existing_profile.received_referral is None:
            dup_global_referral.referred_profile_id = existing_profile.id
            current_session.add(dup_global_referral)
            logs.append(
                f"Update global referral {dup_global_referral.id} to user {existing_user_id}."
            )
        else:
            current_session.delete(dup_global_referral)
            logs.append(
                f"Delete global referral {dup_global_referral.id}, user was already referred."
            )
    current_session.delete(dup_profile)
    return logs

reassign_referral_profile staticmethod

reassign_referral_profile(
    from_user_id, to_user_id, *, dry_run=False
)

Reassign all global referrals from one user's profile to another, then delete the source profile.

Source code in components/growth/internal/business_logic/referral/actions/referral_actions.py

@staticmethod
def reassign_referral_profile(
    from_user_id: str, to_user_id: str, *, dry_run: bool = False
) -> list[str]:
    """Reassign all global referrals from one user's profile to another, then delete the source profile."""
    logs: list[str] = []
    from_profile = (
        current_session.execute(
            select(ReferralProfile).filter(ReferralProfile.user_id == from_user_id)
        )
        .scalars()
        .one_or_none()
    )
    if not from_profile:
        return logs
    to_profile = get_or_create_referral_profile(to_user_id)
    for sent_ref in from_profile.sent_referrals:
        current_logger.info(
            f"Assign global sent referral {sent_ref.id} to user {to_user_id}"
        )
        if not dry_run:
            sent_ref.referring_profile_id = to_profile.id
        logs.append(f"Reassigned sent referral {sent_ref.id} to user {to_user_id}.")
    if from_profile.received_referral:
        ref = from_profile.received_referral
        current_logger.info(
            f"Assign global received referral {ref.id} to user {to_user_id}"
        )
        if not dry_run:
            ref.referred_profile_id = to_profile.id
        logs.append(f"Reassigned received referral {ref.id} to user {to_user_id}.")
    if not dry_run:
        current_session.delete(from_profile)
    return logs

ReferralEntity dataclass

ReferralEntity(referral)

Bases: DataClassJsonMixin

Entity for a referral.

Source code in components/growth/internal/entities/referral.py

def __init__(self, referral: "GlobalReferral"):
    self.id = referral.id
    self.referred_country_code = referral.referred_profile.country_code
    self.referring_partner = referral.referring_partner
    self.referring_user_id = (
        referral.referring_profile.user_id if referral.referring_profile else None
    )
    self.referring_user_has_received_amount = (
        referral.referring_user_has_received_amount
    )
    self.referring_user_rewarded_at = referral.referring_user_rewarded_at
    self.referred_contract_ref = referral.referred_contract_ref
    self.referred_contract_rewarded_on = referral.referred_contract_rewarded_on
    self.created_at = (
        referral.created_at.date()
        if isinstance(referral.created_at, datetime)
        else referral.created_at
    )

created_at instance-attribute

created_at

id instance-attribute

id

referred_contract_ref instance-attribute

referred_contract_ref

referred_contract_rewarded_on instance-attribute

referred_contract_rewarded_on

referred_country_code instance-attribute

referred_country_code

referrer_name property

referrer_name

Get the referrer display name: user full name or partner name.

referring_partner instance-attribute

referring_partner

referring_user_has_received_amount instance-attribute

referring_user_has_received_amount

referring_user_id instance-attribute

referring_user_id

referring_user_rewarded_at instance-attribute

referring_user_rewarded_at

ReferralProfileFactory

Bases: AlanBaseFactory['ReferralProfile']

Meta

model class-attribute instance-attribute

model = ReferralProfile

country_code class-attribute instance-attribute

country_code = 'BE'

user_id class-attribute instance-attribute

user_id = Sequence(lambda n: f'test-user-{n}')

ReferralQueries

Queries for the referral.

get_by_id staticmethod

get_by_id(referral_id)

Get a referral by its ID.

Source code in components/growth/internal/business_logic/referral/queries/referral_queries.py

@staticmethod
def get_by_id(referral_id: UUID) -> ReferralEntity | None:
    """Get a referral by its ID."""
    with transaction(propagation=Propagation.READ_ONLY) as session:
        referral = session.get(
            GlobalReferral,
            referral_id,
            options=[
                selectinload(GlobalReferral.referring_profile),
                selectinload(GlobalReferral.referred_profile),
            ],
        )
        if referral:
            return ReferralEntity(referral=referral)
        return None

get_received_referral_for_profile staticmethod

get_received_referral_for_profile(user_id)

Get the received referral for a profile.

Source code in components/growth/internal/business_logic/referral/queries/referral_queries.py

@staticmethod
def get_received_referral_for_profile(user_id: str) -> ReferralEntity | None:
    """Get the received referral for a profile."""
    referral_profile = get_or_create_referral_profile(user_id)

    if referral_profile.received_referral:
        return ReferralEntity(referral=referral_profile.received_referral)

    return None

get_referral_by_contract_ref staticmethod

get_referral_by_contract_ref(contract_ref)

Find a referral by its referred contract reference.

Source code in components/growth/internal/business_logic/referral/queries/referral_queries.py

@staticmethod
def get_referral_by_contract_ref(contract_ref: str) -> ReferralEntity | None:
    """Find a referral by its referred contract reference."""
    with transaction(propagation=Propagation.READ_ONLY) as session:
        referral = (
            session.execute(
                select(GlobalReferral)
                .options(
                    selectinload(GlobalReferral.referring_profile),
                    selectinload(GlobalReferral.referred_profile),
                )
                .filter(GlobalReferral.referred_contract_ref == contract_ref)
            )
            .scalars()
            .unique()
            .one_or_none()
        )
        if referral:
            return ReferralEntity(referral=referral)
        return None

get_referral_link staticmethod

get_referral_link(user_id)

Get the referral link for a user, creating a profile if needed.

Source code in components/growth/internal/business_logic/referral/queries/referral_queries.py

@staticmethod
def get_referral_link(user_id: str) -> str:
    """Get the referral link for a user, creating a profile if needed."""
    from components.growth.internal.business_logic.referral.queries.referral_progress import (
        build_referral_link,
    )

    referral_profile = get_or_create_referral_profile(user_id)
    return build_referral_link(referral_profile.referral_token)

get_unrewarded_referrals_with_contract staticmethod

get_unrewarded_referrals_with_contract(
    country_code, contract_refs=None
)

Get referrals that have a contract but haven't been rewarded yet.

Also filters out referrals that already have a country-specific discount.

Parameters:

Name	Type	Description	Default
country_code	CountryCode	Country to filter referrals for.	required
contract_refs	list[str] | None	If provided, only return referrals matching these contract refs.	None

Source code in components/growth/internal/business_logic/referral/queries/referral_queries.py

@staticmethod
def get_unrewarded_referrals_with_contract(
    country_code: CountryCode,
    contract_refs: list[str] | None = None,
) -> list[ReferralEntity]:
    """Get referrals that have a contract but haven't been rewarded yet.

    Also filters out referrals that already have a country-specific discount.

    Args:
        country_code: Country to filter referrals for.
        contract_refs: If provided, only return referrals matching these contract refs.
    """
    from components.growth.internal.business_logic.referral.country.referral_country_router import (
        ReferralCountryRouter,
    )

    query = (
        select(GlobalReferral)
        .options(
            selectinload(GlobalReferral.referring_profile),
            selectinload(GlobalReferral.referred_profile),
        )
        .filter(
            GlobalReferral.referred_country_code == country_code,
            GlobalReferral.referred_contract_ref.isnot(None),
            GlobalReferral.referred_contract_rewarded_on.is_(None),
        )
    )
    if contract_refs is not None:
        query = query.filter(
            GlobalReferral.referred_contract_ref.in_(contract_refs)
        )

    with transaction(propagation=Propagation.READ_ONLY) as session:
        referrals = session.execute(query).scalars().unique().all()

    return [
        ReferralEntity(referral=r)
        for r in referrals
        if not ReferralCountryRouter.has_discount_for_referral(
            country_code=country_code,
            referral_id=r.id,
        )
    ]

ReferralRewardTargetType

Bases: AlanBaseEnum

Target type for referral reward amount lookup.

be_smc class-attribute instance-attribute

be_smc = 'be_smc'

company class-attribute instance-attribute

company = 'company'

fr_individual class-attribute instance-attribute

fr_individual = 'fr_individual'

ReferralType module-attribute

ReferralType = Literal[
    "Partner Referral",
    "Customer Referral",
    "Former Member Referral",
]

SELF_SERVE_PROPOSAL_EXPIRES_AFTER_IN_DAYS module-attribute

SELF_SERVE_PROPOSAL_EXPIRES_AFTER_IN_DAYS = 45

clean_phone_number

clean_phone_number(phone)

Clean a phone number by removing all non-digit characters and keeping the country code if it exists. Returns null if cleaned number is too short.

Source code in components/growth/internal/business_logic/crm/utils/phone_number.py

def clean_phone_number(phone: str) -> Optional[str]:
    """
    Clean a phone number by removing all non-digit characters and keeping the country code if it exists.
    Returns null if cleaned number is too short.
    """
    escaped = escape(phone)
    stripped = escaped.lstrip()

    # Remove everything except digits
    cleaned = str(re.sub(r"\D", "", stripped))

    # keep the code at the beginning if it exists
    if stripped.startswith("+"):
        cleaned = "+" + cleaned

    # We are limited by 16 characters
    if len(cleaned) > 16:
        return cleaned[:16]

    # Below 8 characters, we can consider the number invalid
    elif len(cleaned) < 8:
        return None

    return cleaned

create_attribution_signature_success

create_attribution_signature_success(
    alan_partner_code, contract_id
)

Source code in components/growth/internal/business_logic/crm/fr/attribution.py

def create_attribution_signature_success(
    alan_partner_code: str, contract_id: int
) -> None:
    if bool(allowed_format_regex.match(alan_partner_code)):
        today = datetime.today()

        # We override created_at and updated_at in order to not be able to link
        # the event to a particular user.
        try:
            current_session.add(
                AttributionSignatureSuccess(
                    alan_partner_code=alan_partner_code,
                    contract_id=contract_id,
                    created_at=datetime(year=today.year, month=today.month, day=1),
                    updated_at=datetime(year=today.year, month=today.month, day=1),
                )
            )
            current_session.commit()
        except IntegrityError:
            current_session.rollback()

create_or_retrieve_self_serve_health_plan

create_or_retrieve_self_serve_health_plan(
    coverage_level, siren, ccn_code, target_population="all"
)

Create a Plan from the top builder product for a given company and coverage level

Retrieve the existing Plan if it has already been created Raise an error if the builder product were not created

Returns the Plan id

Source code in components/growth/internal/business_logic/self_serve/fr/actions/company_health.py

def create_or_retrieve_self_serve_health_plan(
    coverage_level: CoverageLevelEnum,
    siren: str,
    ccn_code: str,
    target_population: TargetPopulation = "all",  # type: ignore[assignment] # noqa: ALN085
) -> int:
    """
    Create a Plan from the top builder product for a given company and coverage level

    Retrieve the existing Plan if it has already been created
    Raise an error if the builder product were not created

    Returns the Plan id
    """
    from components.growth.internal.business_logic.self_serve.fr.queries.company_health import (
        retrieve_self_serve_company_health_offers,
    )
    from components.offer_builder.public.api import (
        handle_offer_builder_command,
    )
    from components.offer_builder.public.commands.create_offer_from_builder_product_command import (
        CreateOfferFromBuilderProductCommand,
    )
    from components.offer_catalog.public.api import (
        get_offer_for_builder_product_version,
    )
    from components.offer_catalog.public.entities.offer import Offer

    is_for_cadres_non_cadres = (
        target_population == ProfessionalCategory.cadres.value
        or target_population == ProfessionalCategory.non_cadres.value
    )

    all_offers = (
        retrieve_self_serve_company_health_offers(
            siren=siren,
            ccn_code=ccn_code,
            is_for_cadres_non_cadres=is_for_cadres_non_cadres,
        )
        or []
    )
    offer = next(
        (
            offer
            for offer in all_offers
            if offer.coverage_level == coverage_level
            and offer.target_population == target_population
        ),
        None,
    )

    if not offer:
        raise ValueError(
            f"No offer for coverage level={coverage_level}, siren={siren} and ccn_code={ccn_code}"
        )

    health_plan: Offer[TargetPopulation] | None = get_offer_for_builder_product_version(
        builder_product_version_id=offer.builder_product_version_ref,
        target_population=target_population,
    )
    if not health_plan:
        handle_offer_builder_command(
            CreateOfferFromBuilderProductCommand(
                builder_product_id=offer.builder_product_ref,
                builder_product_version_id=offer.builder_product_version_ref,
            ),
            commit=False,
        )
        health_plan = mandatory(
            get_offer_for_builder_product_version(
                builder_product_version_id=offer.builder_product_version_ref,
                target_population=target_population,
            )
        )
    return cast("int", health_plan.id)

extract_marketing_parameters_from_params

extract_marketing_parameters_from_params(params)

Extract marketing parameters from params dict if present.

Source code in components/growth/internal/entities/prospect.py

def extract_marketing_parameters_from_params(
    params: dict[str, Any],
) -> MarketingEventParameters | None:
    """Extract marketing parameters from params dict if present."""
    if "marketing_parameters" not in params:
        return None
    return MarketingEventParameters.from_dict(
        params["marketing_parameters"], infer_missing=True
    )

get_accountant_by_email

get_accountant_by_email(email)

Source code in components/growth/internal/business_logic/accountant/fr/get_accountant.py

@cached_for(days=1)
def get_accountant_by_email(email: str) -> Accountant | None:
    from components.fr.internal.salesforce.api.main import (  # noqa: ALN043
        get_salesforce_accountant_by_email,
    )

    salesforce_accountant = get_salesforce_accountant_by_email(email)

    if not salesforce_accountant:
        return None

    return salesforce_accountant_presenter(salesforce_accountant)

get_accounting_firm_landing_page_link

get_accounting_firm_landing_page_link(partner_code)

Source code in components/growth/internal/business_logic/accountant/fr/get_accounting_firm_landing_page_link.py

def get_accounting_firm_landing_page_link(partner_code: str):  # type: ignore[no-untyped-def]
    return (f"{current_config['FRONT_END_BASE_URL']}/fr-fr/comptables/{partner_code}",)

get_livestorm_replay_url

get_livestorm_replay_url(session_id)

Get the replay URL for a given Livestorm session ID from Public API.

Source code in components/growth/internal/business_logic/crm/lead_generation_resources.py

def get_livestorm_replay_url(session_id: str) -> str:
    """
    Get the replay URL for a given Livestorm session ID from Public API.
    """
    if not session_id:
        raise ValueError("session_id must not be empty")

    url = f"https://api.livestorm.co/v1/sessions/{session_id}/recordings"

    # we use a public API key that does not require authentication
    headers = {
        "accept": "application/vnd.api+json",
        "Authorization": get_livestorm_config(),
    }

    response = requests.get(url, headers=headers, timeout=60)  # noqa: ALN096

    # Raise an error for bad responses
    response.raise_for_status()

    data = response.json()
    if "data" not in data or not data["data"] or len(data["data"]) == 0:
        raise ValueError(f"No recording found for session_id {session_id}")

    return str(data["data"][0]["attributes"]["url"])

get_referral_reward_amount

get_referral_reward_amount(country_code, target_type, date)

Returns the reward amount for the most recent reward active on or before the given date.

• If multiple entries share the same start_date, returns the largest amount.
• Returns 0 if no matching reward entry exists.

Source code in components/growth/internal/business_logic/referral/queries/reward_queries.py

def get_referral_reward_amount(
    country_code: CountryCode,
    target_type: ReferralRewardTargetType,
    date: date,
) -> int:
    """
    Returns the reward amount for the most recent reward active on or before the given date.

    - If multiple entries share the same start_date, returns the largest amount.
    - Returns 0 if no matching reward entry exists.
    """
    reward = (
        current_session.execute(
            select(GlobalReferralReward)
            .filter(
                GlobalReferralReward.country_code == country_code,
                GlobalReferralReward.type == target_type.value,
                GlobalReferralReward.start_date <= date,
            )
            .order_by(
                GlobalReferralReward.start_date.desc(),
                GlobalReferralReward.amount.desc(),
            )
        )
        .scalars()
        .unique()
        .first()
    )
    return reward.amount if reward else 0

individual_paid_acquisition_contract_conversion_value

individual_paid_acquisition_contract_conversion_value(
    product_name,
)

Source code in components/growth/internal/business_logic/crm/fr/attribution.py

def individual_paid_acquisition_contract_conversion_value(
    product_name: str,
) -> int | None:
    if TNS_CONVERSION_VALUE.get(product_name):
        return TNS_CONVERSION_VALUE[product_name]

    current_logger.error(
        "Unable to find contract conversion value for an individual contract",
        product_name=product_name,
    )

    return None

is_alaner_inbound_sales

is_alaner_inbound_sales(email)

Source code in components/growth/internal/business_logic/crm/user_and_roles.py

def is_alaner_inbound_sales(email: str) -> bool:
    user = get_app_growth_dependency().get_user_from_email(email)
    if not user:
        return False

    alaner_details = get_alaner_details_from_user(user)
    return alaner_details.is_inbound_sales if alaner_details else False

is_assigned_to_inbound_sales

is_assigned_to_inbound_sales(country_code, inbox_id)

Returns True if the inbox_id parameter is one of the Inbound Sales inbox for the selected country

Source code in components/growth/internal/business_logic/crm/intercom.py

def is_assigned_to_inbound_sales(country_code: CountryCode, inbox_id: str) -> bool:
    """
    Returns True if the inbox_id parameter is one of the Inbound Sales inbox for the selected country
    """
    return (
        inbox_id
        in get_self_serve_intercom_client(country_code).get_self_serve_team_ids()
    )

is_conversation_recently_closed

is_conversation_recently_closed(conversation)

Check if a conversation is closed and was updated within the last 30 days.

Source code in components/growth/internal/business_logic/crm/intercom.py

def is_conversation_recently_closed(conversation: dict[str, Any]) -> bool:
    """Check if a conversation is closed and was updated within the last 30 days."""
    if conversation.get("open", False):
        return False
    updated_at = conversation.get("updated_at")
    if not updated_at:
        return False
    return datetime.fromtimestamp(updated_at) > datetime.now() - timedelta(
        days=REOPEN_MAX_DELAY_IN_DAYS
    )

load_global_referral_rewards

load_global_referral_rewards(commit)

Seed GlobalReferralReward with initial FR/BE values.

Idempotent: skips if any rows already exist.

Source code in components/growth/internal/helpers/load_global_referral_rewards.py

def load_global_referral_rewards(commit: bool) -> None:
    """Seed GlobalReferralReward with initial FR/BE values.

    Idempotent: skips if any rows already exist.
    """
    count = current_session.execute(
        select(func.count()).select_from(GlobalReferralReward)
    ).scalar()
    if count:
        return

    rewards = [
        GlobalReferralReward(
            country_code="FR",
            type="fr_individual",
            amount=50,
            start_date=date(2016, 1, 1),
        ),
        GlobalReferralReward(
            country_code="FR",
            type="company",
            amount=50,
            start_date=date(2016, 1, 1),
        ),
        GlobalReferralReward(
            country_code="BE",
            type="company",
            amount=50,
            start_date=date(2016, 1, 1),
        ),
        GlobalReferralReward(
            country_code="BE",
            type="be_smc",
            amount=50,
            start_date=date(2016, 1, 1),
        ),
        GlobalReferralReward(
            country_code="ES",
            type="company",
            amount=50,
            start_date=date(2016, 1, 1),
        ),
    ]
    current_session.add_all(rewards)
    if commit:
        current_session.commit()

load_growth_settings

load_growth_settings(commit)

Load all growth settings to the database, used for flask data init

Source code in components/growth/internal/helpers/data_init.py

def load_growth_settings(commit: bool) -> None:
    """
    Load all growth settings to the database, used for flask data init
    """
    settings = [
        GrowthSetting(
            name=GrowthSettingEnum.fr_enable_chat,
            typed_value=True,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.fr_warning_in_contact_modal,
            typed_value=None,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.fr_enable_callback_button_on_first_step,
            typed_value=True,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.fr_backup_inbound_sales_emails,
            typed_value=None,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.be_backup_inbound_sales_emails,
            typed_value=None,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.inbound_sales_emails_to_exclude,
            typed_value=None,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.fr_enable_callback_button_on_email_capture_step,
            typed_value=False,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.fr_b2c_weekly_capacity_per_person,
            typed_value=0,
        ),
        GrowthSetting(
            name=GrowthSettingEnum.fr_b2b_weekly_capacity_per_person,
            typed_value=0,
        ),
    ]

    current_session.add_all(settings)

    if commit:
        current_session.commit()

notify_company_document_review

notify_company_document_review(
    email, document_review_params
)

Source code in components/growth/internal/business_logic/crm/fr/notify_document_review.py

def notify_company_document_review(
    email: str,
    document_review_params: DocumentReviewParams,
) -> None:
    prospect_document_review_company(email, document_review_params)

    proposal_url = f"{current_config['BASE_URL']}/contracting/proposal/{document_review_params.proposal_id}/proposal_items"
    care_conversation_queue = current_rq.get_queue(LOW_PRIORITY_QUEUE)
    care_conversation_queue.enqueue(
        prospect_crm_on_proposal_creation, email, proposal_url
    )

notify_individual_document_review

notify_individual_document_review(
    email, document_review_params
)

Source code in components/growth/internal/business_logic/crm/fr/notify_document_review.py

def notify_individual_document_review(
    email: str,
    document_review_params: DocumentReviewIndividualParams,
) -> None:
    prospect_document_review_individual(email, document_review_params)

    proposal_url = f"{current_config['BASE_URL']}/contracting/proposal/{document_review_params.proposal_id}/proposal_items"
    care_conversation_queue = current_rq.get_queue(EMAILING_QUEUE)
    care_conversation_queue.enqueue(
        prospect_crm_on_proposal_creation,
        email,
        proposal_url,
    )

notify_intercom_on_contract_signature

notify_intercom_on_contract_signature(country_code, email)

Source code in components/growth/internal/business_logic/crm/actions/notify_intercom_on_contract_signature.py

def notify_intercom_on_contract_signature(
    country_code: CountryCode,
    email: str,
) -> None:
    country_code = check_country_code(
        country_code_string=country_code, available_countries=["BE"]
    )

    intercom_client = get_self_serve_intercom_client(country_code=country_code)
    duplicated_conversations = retrieve_duplicate_conversations_for_prospect(
        country_code,
        email=email,
    )
    if duplicated_conversations is None:
        return

    for conversation in duplicated_conversations.open_conversations:
        current_logger.info(
            "Closing conversation as a prospect with the same email just signed a contract",
            conversation_id=conversation["id"],
        )
        intercom_client.leave_note(
            conversation_id=conversation["id"],
            text="A prospect with the same email just signed a contract",
            close=True,
        )

notify_nurturing_company_contract_signature

notify_nurturing_company_contract_signature(
    user_id,
    approver_email,
    company_id,
    products,
    contract_id=None,
)

Source code in components/growth/internal/business_logic/crm/fr/notify_contract_signature.py

def notify_nurturing_company_contract_signature(
    user_id: int | None,
    approver_email: str,
    company_id: int,
    products: list[str],
    contract_id: int | None = None,
) -> None:
    user: User | None
    user_email = approver_email
    if user_id:
        user = get_or_raise_missing_resource(User, user_id)
    else:
        user = get_user_from_email(approver_email)

    if user is not None and user.pro_email_with_perso_fallback:
        user_email = user.pro_email_with_perso_fallback

    company = get_or_raise_missing_resource(Company, company_id)

    send_event_to_customerio_async(
        email=user_email,
        event_name="self_serve_company_contract_signed",
        declared_employment_status=DeclaredEmploymentStatus.company,
    )

    ProspectEventManager.on_contract_signed(
        email=user_email,
        country_code="FR",
        persona_type=PersonaType.Company,
        properties=ProspectEventContractSignedProperties(
            contract_id=str(contract_id) if contract_id else None,
        ),
    )

    current_logger.info(
        "Enqueuing prospect_crm_on_contract_signature",
        user_id=user.id if user else None,
    )
    care_conversation_queue = current_rq.get_queue(EMAILING_QUEUE)
    care_conversation_queue.enqueue(
        prospect_crm_on_contract_signature,
        user_email,
        ContracteeType.company,
        company.name,
        products,
        None,
        approver_email,
        company.siren,
    )

notify_nurturing_individual_contract_signature

notify_nurturing_individual_contract_signature(
    user_id,
    approver_email,
    segment_for_slack_message,
    persona_type,
    siren_or_siret=None,
    contract_id=None,
)

Source code in components/growth/internal/business_logic/crm/fr/notify_contract_signature.py

def notify_nurturing_individual_contract_signature(
    user_id: int | None,
    approver_email: str,
    segment_for_slack_message: str | None,
    persona_type: PersonaType | None,
    siren_or_siret: str | None = None,
    contract_id: int | None = None,
) -> None:
    user: User | None
    user_email = approver_email
    if user_id:
        user = get_or_raise_missing_resource(User, user_id)
    else:
        user = get_user_from_email(approver_email)

    if user and user.pro_email_with_perso_fallback:
        user_email = user.pro_email_with_perso_fallback

    if user and user.first_name:
        contractee_name = user.first_name
    else:
        contractee_name = "John Doe"

    send_event_to_customerio_async(
        email=user_email,
        event_name="self_serve_individual_contract_signed",
    )

    if persona_type:
        ProspectEventManager.on_contract_signed(
            email=user_email,
            country_code="FR",
            persona_type=persona_type,
            properties=ProspectEventContractSignedProperties(
                contract_id=str(contract_id) if contract_id else None,
            ),
        )

    care_conversation_queue = current_rq.get_queue(EMAILING_QUEUE)
    care_conversation_queue.enqueue(
        prospect_crm_on_contract_signature,
        user_email,
        ContracteeType.individual,
        contractee_name,
        ["Health Insurance"],
        segment_for_slack_message,
        approver_email,
        siren_or_siret,
    )

paid_acquisition_contract_conversion_value

paid_acquisition_contract_conversion_value(
    coverage_level=None, ccn_code=None, has_prevoyance=False
)

Source code in components/growth/internal/business_logic/crm/fr/attribution.py

def paid_acquisition_contract_conversion_value(
    coverage_level: CoverageLevelEnum | None = None,
    ccn_code: str | None = None,
    has_prevoyance: bool = False,
) -> int:
    if coverage_level and ccn_code:
        prevoyance_key = "with_prevoyance" if has_prevoyance else "without_prevoyance"
        value = (
            COVERAGE_LEVEL_CONVERSION_VALUE.get(
                ccn_code, COVERAGE_LEVEL_CONVERSION_VALUE["other"]
            )
            .get(coverage_level, {})
            .get(prevoyance_key, None)
        )
        if value:
            return value

    return DEFAULT_CONVERSION_VALUE

prospect_conversation_detect_duplicates

prospect_conversation_detect_duplicates(
    current_conversation, prospect_id, user_id
)

This function will try to detect if Lead/Prospect have started several conversation, it will rely on both the fact that we have a Prospect object, but also if the conversation are owned by a Inbound Sales alaner or if the conversation was in the Lead Inbox.

We will also try to detect when a conversation is open via multiple prospect for the SIREN, it actually happens quite a lot, we can have the HR that come ask some questions, but then when the CEO is ready to sign they might ask new questions.

Source code in components/growth/internal/business_logic/crm/fr/conversation_detect_duplicates.py

def prospect_conversation_detect_duplicates(
    current_conversation: dict[str, Any],
    prospect_id: int | None,
    user_id: int | str | None,
) -> None:
    """This function will try to detect if Lead/Prospect have started several conversation,
    it will rely on both the fact that we have a Prospect object, but also if the conversation
    are owned by a Inbound Sales alaner or if the conversation was in the Lead Inbox.

    We will also try to detect when a conversation is open via multiple prospect for the SIREN,
    it actually happens quite a lot, we can have the HR that come ask some questions, but then
    when the CEO is ready to sign they might ask new questions.
    """
    try:
        current_logger.info(
            "Starting the duplicate detection for the conversation.",
            conversation_id=current_conversation["id"],
            prospect_id=prospect_id,
        )

        duplicates = retrieve_fr_duplicate_conversations_for_prospect(
            current_conversation=current_conversation,
            email=None,
            prospect_id=prospect_id,
            user_id=user_id,
        )
        if not duplicates:
            return

        current_logger.info(
            "Multiple conversations detected. Adding a note to the conversation."
        )
        intercom_client = get_self_serve_intercom_client("FR")

        message = message_warning_about_duplicate_conversation(
            open_conversations=duplicates.open_conversations,
            closed_conversations=duplicates.closed_conversations,
            open_conversations_via_siren=duplicates.open_conversations_via_siren,
            closed_conversations_via_siren=duplicates.closed_conversations_via_siren,
            open_conversations_via_secondary_email=duplicates.open_conversations_via_secondary_email,
            closed_conversations_via_secondary_email=duplicates.closed_conversations_via_secondary_email,
            open_conversations_via_phone=duplicates.open_conversations_via_phone,
            closed_conversations_via_phone=duplicates.closed_conversations_via_phone,
        )
        intercom_client.leave_note(
            conversation_id=current_conversation["id"], text=message
        )
    except Exception:
        # We do a big catch all to make sure to not impact UCE logic
        current_logger.exception(
            "Unable do detect duplicates for a prospect conversation"
        )

register_new_prospect

register_new_prospect(
    email,
    phone=None,
    attribution_survey_response=None,
    save=False,
)

Register a new prospect in the database. If the prospect already exists, it will update the phone number and attribution survey if provided.

Source code in components/growth/internal/business_logic/prospect/register_new_prospect.py

def register_new_prospect(
    email: str,
    phone: str | None = None,
    attribution_survey_response: AttributionSurveyResponse | None = None,
    save: bool | None = False,
) -> Prospect:
    """
    Register a new prospect in the database. If the prospect already exists, it will update the phone number and attribution survey if provided.
    """
    prospect = Prospect.from_email(email_address=email) or Prospect(email=email)
    if phone:
        prospect.phone = clean_phone_number(phone)
    if attribution_survey_response:
        prospect.attribution_survey_response = attribution_survey_response
    current_session.add(prospect)

    if save:
        current_session.commit()
    else:
        current_session.flush()

    return prospect

reopen_inbound_sales_conversation_if_possible

reopen_inbound_sales_conversation_if_possible(
    intercom_client, conversation, country_code
)

Reopen a closed conversation if assigned to inbound sales and the admin is available.

Source code in components/growth/internal/business_logic/crm/intercom.py

def reopen_inbound_sales_conversation_if_possible(
    intercom_client: Any,
    conversation: dict[str, Any],
    country_code: CountryCode,
) -> None:
    """Reopen a closed conversation if assigned to inbound sales and the admin is available."""
    team_assignee_id = str(conversation.get("team_assignee_id") or "")
    if not team_assignee_id or not is_assigned_to_inbound_sales(
        country_code, team_assignee_id
    ):
        return

    admin_assignee_id = conversation.get("admin_assignee_id")
    if admin_assignee_id and str(
        admin_assignee_id
    ) not in get_available_inbound_sales_intercom_ids(country_code):
        return

    intercom_client.reopen_conversation(conversation_id=str(conversation["id"]))
    current_logger.info(
        "Reopened inbound sales conversation",
        conversation_id=conversation["id"],
    )

retrieve_eligible_prevoyance_offers_from_account_id

retrieve_eligible_prevoyance_offers_from_account_id(
    account_id,
)

Source code in components/growth/internal/business_logic/self_serve/fr/queries/company_prevoyance.py

def retrieve_eligible_prevoyance_offers_from_account_id(
    account_id: UUID,
) -> dict[int, SelfServeCompanyPrevoyanceOffers | None]:
    from components.fr.internal.models.account import Account  # noqa: ALN069
    from components.fr.internal.models.contract import Contract  # noqa: ALN069
    from components.fr.internal.models.prevoyance_contract import (  # noqa: ALN069
        PrevoyanceContract,
    )

    account: Account | None = (
        current_session.query(Account)  # noqa: ALN085
        .options(
            joinedload(Account.companies).options(
                joinedload(Company.prevoyance_contracts).options(
                    joinedload(
                        PrevoyanceContract.contract_populations,
                    ),
                ),
                joinedload(Company.contracts).options(
                    joinedload(Contract.contract_versions),
                    joinedload(Contract.signed_documents),
                ),
                joinedload(Company.ccn),
            ),
        )
        .filter(Account.id == account_id)
        .first()
    )

    if not account:
        raise BaseErrorCode.missing_resource(message=f"account {account_id} not found")

    return {
        company.id: _retrieve_eligible_prevoyance_offers_from_company_model(company)
        for company in account.companies
    }

retrieve_eligible_prevoyance_offers_from_company_id

retrieve_eligible_prevoyance_offers_from_company_id(
    company_id,
)

Conditions to meet: - Active Health Contract - No Active Prevoyance Contract - No suspended Prevoyance contract - No Tailored offer in the account

Return only self-serve prevoyance offers (if there are some for the CCN)

Source code in components/growth/internal/business_logic/self_serve/fr/queries/company_prevoyance.py

def retrieve_eligible_prevoyance_offers_from_company_id(
    company_id: int,
) -> SelfServeCompanyPrevoyanceOffers | None:
    """
    Conditions to meet:
    - Active Health Contract
    - No Active Prevoyance Contract
    - No suspended Prevoyance contract
    - No Tailored offer in the account

    Return only self-serve prevoyance offers (if there are some for the CCN)
    """

    company = get_or_raise_missing_resource(Company, company_id)

    return _retrieve_eligible_prevoyance_offers_from_company_model(company)

submit_contact_form

submit_contact_form(
    params, ip_address, marketing_parameters=None
)

Source code in components/growth/internal/business_logic/crm/fr/push_sales_contact.py

def submit_contact_form(
    params: dict[str, Any],
    ip_address: str | None,
    marketing_parameters: MarketingEventParameters | None = None,
) -> dict[str, Any]:
    # Can be overridden depending on partner (via token param - e.g. shine)
    dispatch_form_to = "sales"

    if params.get("number_of_employees", 0) < get_minimum_number_of_employees_for_sales(
        params.get("country", "")
    ):
        dispatch_form_to = "sales_inbound"

    token = params.get("token")
    if token:
        if not validate_token(token):
            current_logger.error(
                "Invalid token provided",
                token=token,
            )
            return {"error": "Invalid token provided"}

        alan_partner_code = get_partner_code_from_token(token)

        if alan_partner_code == "partnership-shine":
            dispatch_form_to = "sales_inbound"
            # shine only in France for now
            params["country"] = "France"
            params["language"] = "French"
            params["alan_partner_code"] = alan_partner_code
            params["context_for_sales"] = (
                f"EXTERNAL PARTNERSHIP QUALIFIED LEAD\nSIRET: {params.get('company_identification', 'Not Available')}"
            )
            params["request_type"] = "callback" if params.get("phone") else "email"

    try:
        current_logger.info(
            "Submitting contact form",
            dispatch_form_to=dispatch_form_to,
            params=params,
        )

        country_code = hubspot_country_to_country_code(params.get("country"))

        # check params dataclass before using them and returning typed instance
        if dispatch_form_to == "sales_inbound":
            sales_inbound_params = get_contact_sales_inbound_params(params)
            response = submit_form_to_sales_inbound(sales_inbound_params)
            if response.get("ok", 0) > 0:
                persona_type = (
                    PersonaType(sales_inbound_params.persona_type)
                    if sales_inbound_params.persona_type
                    else None
                )
                ProspectEventManager.on_contact_sales(
                    email=sales_inbound_params.email,
                    country_code=country_code,
                    persona_type=persona_type,
                    properties=create_contact_sales_inbound_properties(
                        sales_inbound_params
                    ),
                    marketing_parameters=marketing_parameters,
                )
                if sales_inbound_params.attribution_survey:
                    ProspectEventManager.on_meeting_request_attribution_survey(
                        email=sales_inbound_params.email,
                        country_code=country_code,
                        persona_type=persona_type,
                        properties=ProspectEventMeetingRequestAttributionSurveyProperties(
                            source=sales_inbound_params.attribution_survey,
                            civil_servant=sales_inbound_params.civil_servant_type,
                        ),
                        marketing_parameters=marketing_parameters,
                    )
            return response
        else:
            sales_params = get_contact_sales_params(params)
            response = submit_form_to_sales(sales_params, ip_address)
            if response.get("ok", 0) > 0:
                persona_type = (
                    PersonaType(sales_params.persona_type)
                    if sales_params.persona_type
                    else None
                )
                ProspectEventManager.on_contact_sales(
                    email=sales_params.email,
                    country_code=country_code,
                    persona_type=persona_type,
                    properties=create_contact_sales_properties(sales_params),
                    marketing_parameters=marketing_parameters,
                )
                if sales_params.attribution_survey:
                    ProspectEventManager.on_meeting_request_attribution_survey(
                        email=sales_params.email,
                        country_code=country_code,
                        persona_type=persona_type,
                        properties=ProspectEventMeetingRequestAttributionSurveyProperties(
                            source=sales_params.attribution_survey,
                            role=sales_params.job_title,
                            industry=sales_params.industry,
                            civil_servant=sales_params.civil_servant_type,
                        ),
                        marketing_parameters=marketing_parameters,
                    )
            return response
    except (TypeError, ValueError) as e:
        current_logger.error(
            "Error submitting contact form due to invalid parameters",
            dispatch_form_to=dispatch_form_to,
            params=params,
            error=e,
        )
        return {"error": "Invalid parameters"}

submit_form_to_hubspot

submit_form_to_hubspot(
    portal_id,
    form_id,
    fields,
    ip_address,
    page=None,
    marketing_parameters=None,
)

Submit form to hubspot API

Source code in components/growth/internal/business_logic/crm/hubspot.py

def submit_form_to_hubspot(
    portal_id: str,
    form_id: str,
    fields: dict[str, Any],
    ip_address: str | None,
    page: str | None = None,
    marketing_parameters: MarketingEventParameters | None = None,
) -> requests.models.Response:
    """
    Submit form to hubspot API
    """
    current_logger.info(
        "Submitting form to Hubspot", form_id=form_id, portal_id=portal_id
    )

    form_url = f"https://forms.hubspot.com/uploads/form/v2/{portal_id}/{form_id}"

    hs_context_dict = json.loads(fields["hs_context"]) if "hs_context" in fields else {}

    if "ipAddress" not in hs_context_dict and ip_address is not None:
        hs_context_dict["ipAddress"] = ip_address

    hubspot_data = fields.copy()
    hubspot_data["hs_context"] = json.dumps(hs_context_dict)

    response = requests.post(
        form_url,
        data=hubspot_data,
        headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"},
        timeout=60,
    )
    if response.ok:
        current_logger.info(
            "Submitting form to Hubspot - Success",
            form_id=form_id,
        )
        if form_id == WEBSITE_CAPTURE_FORM_ID:
            ProspectEventManager.on_website_capture_form(
                email=fields["email"],
                country_code=hubspot_country_to_country_code(
                    fields["form_submission_country"]
                ),
                persona_type=fields["form_submission_persona_type"],
                properties=WebsiteCaptureFormProperties(
                    company=fields.get("company"),
                    employee_count=fields.get("nombre_de_salaries_couverts"),
                    civil_servant_type=fields.get("form_submission_civilservant_type"),
                    job_title_category=fields.get("form_submission_job_title_category"),
                    industry=fields.get("form_submission_industry"),
                    origin=fields.get("form_submission_last_origin"),
                    origin_details=fields.get("form_submission_last_origin_details"),
                    origin_details_label=fields.get(
                        "form_submission_last_origin_details_label"
                    ),
                    origin_link=fields.get("form_submission_last_origin_link"),
                    origin_category=fields.get("form_submission_last_origin_category"),
                    page=page,
                ),
                marketing_parameters=marketing_parameters,
            )

    else:
        current_logger.error(
            "Submitting form to Hubspot - Failed",
            form_id=form_id,
            error=response.content,
        )
    return response

unsubscribe_email

unsubscribe_email(
    email, unsubscribe_token, unsubscribe_target_type
)

Unsubscribe user or prospect from commercial emails.

Return success. Note that in the failure case, it's important that we return the same value (False) whether it's the email that does not exist or the token which is incorrect. This is because we don't want a snooper to be able to deduce the email addresses in our system by brute forcing this function.

Source code in components/growth/internal/business_logic/prospect/fr/commercial_emails_unsubscription.py

def unsubscribe_email(email, unsubscribe_token, unsubscribe_target_type):  # type: ignore[no-untyped-def]
    """Unsubscribe user or prospect from commercial emails.

    Return success.
    Note that in the failure case, it's important that we return the same value (False)
    whether it's the email that does not exist or the token which is incorrect.
    This is because we don't want a snooper to be able to deduce the email addresses in
    our system by brute forcing this function.
    """
    if email is None or email == "":
        return False

    if unsubscribe_target_type == UnsubscribeTargetType.user:
        entity_to_unsubscribe = User.from_email(email)
        if entity_to_unsubscribe is None:
            """
            - this check all employments rather than the last one on purpose.
            - this will never raise ValueError since email is not None.
            - this is safe to do because of the exclusion constraint on the Employment model
            that prevents two distinct users from sharing an `invite_email`.
            """
            employment = get_last_non_cancelled_employment(invite_email=email)
            entity_to_unsubscribe = employment.user if employment else None

    elif unsubscribe_target_type == UnsubscribeTargetType.prospect:
        entity_to_unsubscribe = Prospect.from_email(email)
    else:
        raise ErrorCode.unknown_unsubscribe_target_type()

    if not entity_to_unsubscribe:
        # Fail silently
        return False

    response = entity_to_unsubscribe.unsubscribe(unsubscribe_token)
    current_session.commit()

    if response and isinstance(entity_to_unsubscribe, User):
        publish_user_email_unsubscribed(entity_to_unsubscribe)

    return response

unsubscribe_email_without_token

unsubscribe_email_without_token(email)

Unsubscribe user or prospect from commercial emails.

WARNING: This doesn't require a token, and should only be called in a context where we're sure the request is legit. (eg webhook)

Source code in components/growth/internal/business_logic/prospect/fr/commercial_emails_unsubscription.py

def unsubscribe_email_without_token(email: str) -> None:
    """Unsubscribe user or prospect from commercial emails.

    WARNING: This doesn't require a token, and should only be called in a context where we're sure the request is legit.
    (eg webhook)
    """
    user_to_unsubscribe = User.from_email(email)
    if user_to_unsubscribe is None:
        employment = get_last_non_cancelled_employment(invite_email=email)
        user_to_unsubscribe = employment.user if employment else None

    prospect_to_unsubscribe = Prospect.from_email(email)

    user_unsubscribed = False
    prospect_unsubscribed = False
    if user_to_unsubscribe and user_to_unsubscribe.has_opted_in_to_commercial_emails:
        current_logger.info(
            "Unsubscribing user from commercial emails without token",
            user_id=user_to_unsubscribe.id,
        )
        # overriding the unsubscribe_token
        user_unsubscribed = user_to_unsubscribe.unsubscribe(
            user_to_unsubscribe.unsubscribe_token
        )

    if (
        prospect_to_unsubscribe
        and prospect_to_unsubscribe.has_opted_in_to_commercial_emails
    ):
        current_logger.info(
            "Unsubscribing prospect from commercial emails without token",
            prospect_id=prospect_to_unsubscribe.id,
        )
        # overriding the unsubscribe_token
        prospect_unsubscribed = prospect_to_unsubscribe.unsubscribe(
            prospect_to_unsubscribe.unsubscribe_token
        )

    current_session.commit()

    if user_unsubscribed and isinstance(user_to_unsubscribe, User):
        current_logger.info(
            "Unsubscribed user from commercial emails without token",
            user_id=user_to_unsubscribe.id,
        )
        publish_user_email_unsubscribed(user_to_unsubscribe)

    if prospect_unsubscribed:
        current_logger.info(
            "Unsubscribed prospect from commercial emails without token",
            prospect_id=prospect_to_unsubscribe.id,
        )

components.growth.public.dependencies

GrowthDependency

Bases: ABC

create_conversation_context abstractmethod

create_conversation_context(
    user_id, conversation_type, context_data
)

Implement create_conversation_context

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def create_conversation_context(
    self,
    user_id: str | None,
    conversation_type: IntercomConversationType,
    context_data: dict[str, Any],
) -> str | None:
    """Implement create_conversation_context"""

get_company_quote_request_data abstractmethod

get_company_quote_request_data(
    language,
    phone_number,
    firstname,
    lastname,
    company_name,
    number_of_employees,
    qualification,
)

Implement get_company_quote_request_event

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_company_quote_request_data(
    self,
    language: str,
    phone_number: str | None,
    firstname: str | None,
    lastname: str | None,
    company_name: str | None,
    number_of_employees: int | None,
    qualification: dict[str, Any],
) -> QuoteRequestData:
    """Implement get_company_quote_request_event"""

get_email_capture_event_data abstractmethod

get_email_capture_event_data(self_serve_flow_id, language)

Implement get_company_email_capture_event_data

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_email_capture_event_data(
    self,
    self_serve_flow_id: UUID,
    language: str,
) -> EmailCaptureData:
    """Implement get_company_email_capture_event_data"""

get_extra_tags_for_country abstractmethod

get_extra_tags_for_country(prospect_id, request_type)

Implement get_extra_tags_for_country

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_extra_tags_for_country(
    self,
    prospect_id: int | None,
    request_type: ProspectRequestType,
) -> list[str]:
    """Implement get_extra_tags_for_country"""

get_inbound_sales_team abstractmethod

get_inbound_sales_team()

Implement get_inbound_sales_team

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_inbound_sales_team(
    self,
) -> InboundSalesTeam:
    """Implement get_inbound_sales_team"""

get_intercom_admin_id_from_user_id abstractmethod

get_intercom_admin_id_from_user_id(user_id)

Implement get_intercom_admin_id_from_user_id

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_intercom_admin_id_from_user_id(
    self,
    user_id: str,
) -> str | None:
    """Implement get_intercom_admin_id_from_user_id"""

get_persona_type_via_flow_id abstractmethod

get_persona_type_via_flow_id(self_serve_flow_id)

Implement get_persona_type_via_flow_id

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_persona_type_via_flow_id(
    self,
    self_serve_flow_id: UUID,
) -> PersonaType:
    """Implement get_persona_type_via_flow_id"""

get_prospects_emails_from_phone_number abstractmethod

get_prospects_emails_from_phone_number(phone_number)

Implement get_prospects_emails_from_phone_number

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_prospects_emails_from_phone_number(
    self,
    phone_number: str,
) -> list[str]:
    """Implement get_prospects_emails_from_phone_number"""

get_self_serve_flow_id_from_qualification abstractmethod

get_self_serve_flow_id_from_qualification(qualification)

Implement get_self_serve_flow_id_from_qualification

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_self_serve_flow_id_from_qualification(
    self, qualification: dict[str, Any] | None
) -> str | None:
    """Implement get_self_serve_flow_id_from_qualification"""

get_user abstractmethod

get_user(user_id)

Implement get_user

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_user(
    self,
    user_id: str,
) -> GrowthUser | None:
    """Implement get_user"""

get_user_from_email abstractmethod

get_user_from_email(email)

Implement get_user_id_from_email

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_user_from_email(
    self,
    email: str,
) -> GrowthUser | None:
    """Implement get_user_id_from_email"""

get_user_from_intercom_admin_id abstractmethod

get_user_from_intercom_admin_id(intercom_admin_id)

Implement get_user_from_intercom_admin_id

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def get_user_from_intercom_admin_id(
    self,
    intercom_admin_id: str,
) -> GrowthUser | None:
    """Implement get_user_from_intercom_admin_id"""

is_admin_sales abstractmethod

is_admin_sales(admin_id)

Implement is_admin_sales

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def is_admin_sales(
    self,
    admin_id: str,
) -> bool:
    """Implement is_admin_sales"""

is_request_likely_for_care abstractmethod

is_request_likely_for_care(
    prospect_request_type, prospect_email, prospect_id
)

Implement is_request_likely_for_care it should return a tuple with: - a bool if the request is likely for care - a str with the reason if the request is likely for care

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def is_request_likely_for_care(
    self,
    prospect_request_type: ProspectRequestType,
    prospect_email: str,
    prospect_id: int | None,
) -> tuple[bool, str | None]:
    """Implement is_request_likely_for_care
    it should return a tuple with:
    - a bool if the request is likely for care
    - a str with the reason if the request is likely for care
    """

register_prospect abstractmethod

register_prospect(
    email,
    phone=None,
    attribution_survey_source=None,
    qualification_raw_company=None,
    qualification_raw_individual=None,
    qualification_raw_website=None,
)

Implement register_prospect

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def register_prospect(
    self,
    email: str,
    phone: str | None = None,
    attribution_survey_source: str | None = None,
    qualification_raw_company: dict[str, Any] | None = None,
    qualification_raw_individual: dict[str, Any] | None = None,
    qualification_raw_website: dict[str, Any] | None = None,
) -> int | None:
    """Implement register_prospect"""

send_unauthenticated_user_async_message_to_intercom abstractmethod

send_unauthenticated_user_async_message_to_intercom(
    email, subject, message
)

Implement send_unauthenticated_user_async_message_to_intercom

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

@abstractmethod
def send_unauthenticated_user_async_message_to_intercom(
    self,
    email: str,
    subject: str,
    message: str,
) -> None:
    """Implement send_unauthenticated_user_async_message_to_intercom"""

growth_dependency module-attribute

growth_dependency = FranceGrowthDependency()

set_app_growth_dependency

set_app_growth_dependency(dependency)

Set the growth dependency in the current app context

Source code in components/growth/internal/business_logic/dependencies/growth_dependency.py

def set_app_growth_dependency(dependency: GrowthDependency) -> None:
    """
    Set the growth dependency in the current app context
    """
    from flask import current_app

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