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'

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

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,
    phone=None,
    create_prospect_if_not_exists=True,
    commit=False,
)

Create a prospect event in the database.

When phone is provided, it is stored on the prospect: set on the new prospect when one is created, or filled in on an existing prospect that has none yet (existing phone numbers are never overwritten).

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,
    phone: str | None = None,
    create_prospect_if_not_exists: bool = True,
    commit: bool = False,
) -> None:
    """Create a prospect event in the database.

    When ``phone`` is provided, it is stored on the prospect: set on the new prospect
    when one is created, or filled in on an existing prospect that has none yet
    (existing phone numbers are never overwritten).
    """
    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, phone=phone)
        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
    elif phone and not prospect.phone:
        prospect.phone = clean_phone_number(phone)

    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,
    phone,
    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,
    phone: str | None,
    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(),
        phone=phone,
        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_attribution_survey staticmethod

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

Dispatch flow attribution survey event for self-serve feedback.

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

@staticmethod
def on_flow_attribution_survey(
    email: str,
    country_code: CountryCode,
    persona_type: PersonaType | None,
    properties: ProspectEventFlowAttributionSurveyProperties,
    marketing_parameters: MarketingEventParameters | None = None,
    commit: bool = True,
) -> None:
    """Dispatch flow attribution survey event for self-serve feedback."""
    ProspectEventManager.create_prospect_event(
        email=email,
        event_name=ProspectEventType.FLOW_ATTRIBUTION_SURVEY,
        country_code=country_code,
        persona_type=persona_type,
        marketing_parameters=marketing_parameters,
        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,
    phone,
    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,
    phone: str | None,
    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(),
        phone=phone,
        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,
    phone,
    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,
    phone: str | None,
    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,
        phone=phone,
        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

            if any(
                r.referred_profile_id == referring_profile.id
                for r in referred_profile.sent_referrals
            ):
                current_logger.info(
                    "Unable to create Referral: mutual referral detected",
                    referring_profile_id=referring_profile.id,
                    referred_profile_id=referred_profile.id,
                )
                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 Referral 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 Referral 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 Referrals 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 Referrals 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 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 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_referral = dup_profile.received_referral
    if dup_referral:
        existing_profile = get_or_create_referral_profile(existing_user_id)
        if existing_profile.received_referral is None:
            dup_referral.referred_profile_id = existing_profile.id
            current_session.add(dup_referral)
            logs.append(
                f"Update referral {dup_referral.id} to user {existing_user_id}."
            )
        else:
            current_session.delete(dup_referral)
            logs.append(
                f"Delete referral {dup_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 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 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 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 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: "Referral"):
    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(
            Referral,
            referral_id,
            options=[
                selectinload(Referral.referring_profile),
                selectinload(Referral.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(Referral)
                .options(
                    selectinload(Referral.referring_profile),
                    selectinload(Referral.referred_profile),
                )
                .filter(Referral.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(Referral)
        .options(
            selectinload(Referral.referring_profile),
            selectinload(Referral.referred_profile),
        )
        .filter(
            Referral.referred_country_code == country_code,
            Referral.referred_contract_ref.isnot(None),
            Referral.referred_contract_rewarded_on.is_(None),
        )
    )
    if contract_refs is not None:
        query = query.filter(Referral.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'

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

delete_all_prospect_data_for_email_for_gdpr

delete_all_prospect_data_for_email_for_gdpr(
    email, dry_run=True
)

Delete all prospect data for GDPR erasure request. Does not remove the user.

Source code in components/growth/internal/business_logic/data_retention/fr/actions/prospection_data.py

def delete_all_prospect_data_for_email_for_gdpr(
    email: str, dry_run: bool = True
) -> None:
    """Delete all prospect data for GDPR erasure request. Does not remove the user."""
    deleter = Deleter(with_archiving=False)
    delete_all_prospect_data_for_email(
        prospect_email=email, deleter=deleter, remove_user=False, dry_run=dry_run
    )

extract_marketing_parameters_from_params

extract_marketing_parameters_from_params(params)

Extract marketing parameters from params dict.

Prefers the full marketing_parameters dict when present. Falls back to individual UTM params (utm_source, utm_medium, etc.).

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.

    Prefers the full marketing_parameters dict when present.
    Falls back to individual UTM params (utm_source, utm_medium, etc.).
    """
    marketing_params = params.get("marketing_parameters")
    if marketing_params is not None:
        return MarketingEventParameters.from_dict(marketing_params, infer_missing=True)
    return _build_marketing_parameters_from_utm_params(params)

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_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(ReferralReward)
            .filter(
                ReferralReward.country_code == country_code,
                ReferralReward.type == target_type.value,
                ReferralReward.start_date <= date,
            )
            .order_by(
                ReferralReward.start_date.desc(),
                ReferralReward.amount.desc(),
            )
        )
        .scalars()
        .unique()
        .first()
    )
    return reward.amount if reward else 0

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_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()

load_referral_rewards

load_referral_rewards(commit)

Seed ReferralReward with initial FR/BE values.

Idempotent: skips if any rows already exist.

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

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

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

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

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 = CrmCountryRouter.get_self_serve_intercom_client(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,
        )

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/country/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 = CrmCountryRouter.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,
    self_serve_flow_progress_id=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.

If self_serve_flow_progress_id is provided, the flow progress row is linked to the resulting prospect (overwriting any existing prospect_id).

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,
    self_serve_flow_progress_id: UUID | 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.

    If `self_serve_flow_progress_id` is provided, the flow progress row is linked to the
    resulting prospect (overwriting any existing prospect_id).
    """
    from components.growth.internal.business_logic.self_serve.self_serve_flow import (
        link_progress_to_prospect,
    )

    # Match from_email's lookup with the column validator (which strips), else an
    # existing prospect slips through dedup and crashes on insert.
    email = normalize_email_address_format(email)
    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)

    current_session.flush()

    if self_serve_flow_progress_id is not None:
        link_progress_to_prospect(
            self_serve_flow_progress_id=self_serve_flow_progress_id,
            prospect_id=prospect.id,
        )

    if save:
        current_session.commit()

    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."""
    raw_team_id = conversation.get("team_assignee_id")
    team_assignee_id = (
        str(raw_team_id) if is_intercom_team_assignee_id_present(raw_team_id) else ""
    )
    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 is_intercom_admin_assignee_id_present(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"],
    )

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

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_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"""

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"""

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
    )