Api reference

components.growth.public.business_logic ¶

conversation_management ¶

assign_conversation `module-attribute` ¶

assign_conversation = assign_conversation

get_intercom_admin_id_from_email `module-attribute` ¶

get_intercom_admin_id_from_email = (
    get_intercom_admin_id_from_email
)

is_assigned_to_inbound_sales `module-attribute` ¶

is_assigned_to_inbound_sales = is_assigned_to_inbound_sales

notify_intercom_on_contract_signature `module-attribute` ¶

notify_intercom_on_contract_signature = (
    notify_intercom_on_contract_signature
)

hubspot ¶

get_hubspot_country `module-attribute` ¶

get_hubspot_country = get_hubspot_country

get_hubspot_language `module-attribute` ¶

get_hubspot_language = get_hubspot_language

submit_form_to_hubspot `module-attribute` ¶

submit_form_to_hubspot = submit_form_to_hubspot

lead_generation_resources ¶

get_livestorm_replay_url `module-attribute` ¶

get_livestorm_replay_url = get_livestorm_replay_url

prospect_request_contact ¶

prospect_request_contact `module-attribute` ¶

prospect_request_contact = (
    prospect_request_contact_protected
)

self_serve_feedback ¶

create_self_serve_feedback ¶

create_self_serve_feedback(
    email,
    country_code,
    flow_type,
    survey_response,
    commit=True,
)

Create a new self serve feedback form entry

Source code in components/growth/public/business_logic/self_serve_feedback.py

def create_self_serve_feedback(
    email: str | None,
    country_code: str,
    flow_type: SelfServeFeedbackTypeEnum,
    survey_response: dict[str, Any],
    commit: bool = True,
) -> None:
    """Create a new self serve feedback form entry"""
    from components.growth.internal.models.self_serve_feedback import (
        SelfServeFeedback,
    )
    from shared.helpers.db import current_session

    if email is None:
        raise BaseErrorCode.invalid_arguments(
            description="Email is required for feedback submission."
        )

    if survey_response == {}:
        raise BaseErrorCode.invalid_arguments(
            description="Survey response cannot be empty."
        )

    valid_country_code = check_country_code(
        country_code_string=country_code,
        available_countries=["BE", "FR", "ES", "CA"],
    )

    self_serve_feedback = SelfServeFeedback(
        email=email,
        survey_response=survey_response,
        country_code=valid_country_code,
        flow_type=flow_type,
    )
    current_session.add(self_serve_feedback)
    if commit:
        current_session.commit()

    return

self_serve_flow ¶

create_self_serve_flow_progress ¶

create_self_serve_flow_progress(
    country_code,
    flow_type,
    language,
    alan_partner_code,
    referring_user_token,
    referring_partner,
    user_ref,
    commit=True,
)

Create the self-serve flow progress

Source code in components/growth/public/business_logic/self_serve_flow.py

def create_self_serve_flow_progress(
    country_code: CountryCode,
    flow_type: SelfServeFlowTypeEnum,
    language: Lang,
    alan_partner_code: str | None,
    referring_user_token: str | None,
    referring_partner: str | None,
    user_ref: str | None,
    commit: bool = True,
) -> SelfServeFlowProgressEntity:
    """Create the self-serve flow progress"""
    from components.growth.internal.helpers.country_code import check_country_code

    check_country_code(country_code_string=country_code, available_countries=["FR"])

    if flow_type == SelfServeFlowTypeEnum.company:
        self_serve_flow_progress = SelfServeFlowProgress(
            country_code=country_code,
            flow_type=SelfServeFlowTypeEnum.company,
            state={"language": language},
            user_ref=user_ref,
        )

        if alan_partner_code:
            self_serve_flow_progress.state["alan_partner_code"] = alan_partner_code

        if referring_user_token:
            self_serve_flow_progress.state["referring_user_token"] = (
                referring_user_token
            )
        elif referring_partner:
            self_serve_flow_progress.state["referring_partner"] = referring_partner

        current_session.add(self_serve_flow_progress)

        if commit:
            current_session.commit()

        return SelfServeFlowProgressEntity(
            id=self_serve_flow_progress.id,
            flow_type=self_serve_flow_progress.flow_type,
            country_code=self_serve_flow_progress.country_code,
            state=self_serve_flow_progress.state,
            proposal_ref=self_serve_flow_progress.proposal_ref,
            user_ref=self_serve_flow_progress.user_ref,
        )

    raise ValueError(f"Invalid flow type {flow_type}. Only 'company' is supported.")

get_multiple_self_serve_flow_progress ¶

get_multiple_self_serve_flow_progress(
    self_serve_flow_progress_ids, user_ref
)

Get multiple self-serve flow progress records by ID.

Source code in components/growth/public/business_logic/self_serve_flow.py

def get_multiple_self_serve_flow_progress(
    self_serve_flow_progress_ids: list[UUID] | None,
    user_ref: str | None,
) -> list[SelfServeFlowProgressEntity]:
    """Get multiple self-serve flow progress records by ID."""
    query = current_session.query(SelfServeFlowProgress)  # noqa: ALN085

    if self_serve_flow_progress_ids:
        query = query.filter(SelfServeFlowProgress.id.in_(self_serve_flow_progress_ids))

    if user_ref:
        query = query.filter(SelfServeFlowProgress.user_ref == user_ref)

    flow_progress_objs = query.order_by(SelfServeFlowProgress.updated_at.desc()).all()

    return [
        SelfServeFlowProgressEntity(
            id=progress.id,
            flow_type=progress.flow_type,
            country_code=progress.country_code,
            state=progress.state,
            proposal_ref=progress.proposal_ref,
            user_ref=progress.user_ref,
        )
        for progress in flow_progress_objs
    ]

get_self_serve_flow_progress ¶

get_self_serve_flow_progress(self_serve_flow_progress_id)

Get the self-serve flow progress

Source code in components/growth/public/business_logic/self_serve_flow.py

def get_self_serve_flow_progress(
    self_serve_flow_progress_id: UUID,
) -> SelfServeFlowProgressEntity:
    """Get the self-serve flow progress"""
    self_serve_flow_progress = get_or_raise_missing_resource(
        SelfServeFlowProgress, self_serve_flow_progress_id
    )

    return SelfServeFlowProgressEntity(
        id=self_serve_flow_progress.id,
        flow_type=self_serve_flow_progress.flow_type,
        country_code=self_serve_flow_progress.country_code,
        state=self_serve_flow_progress.state,
        proposal_ref=self_serve_flow_progress.proposal_ref,
        user_ref=self_serve_flow_progress.user_ref,
    )

update_self_serve_flow_progress ¶

update_self_serve_flow_progress(
    self_serve_flow_progress_id,
    params=None,
    proposal_ref=None,
    user_ref=None,
    commit=True,
)

Update the self-serve flow progress

Source code in components/growth/public/business_logic/self_serve_flow.py

def update_self_serve_flow_progress(
    self_serve_flow_progress_id: UUID,
    params: dict[str, Any] | None = None,
    proposal_ref: UUID | None = None,
    user_ref: str | None = None,
    commit: bool = True,
) -> SelfServeFlowProgressEntity:
    """Update the self-serve flow progress"""
    self_serve_flow_progress = get_or_raise_missing_resource(
        SelfServeFlowProgress,
        self_serve_flow_progress_id,
        with_for_update=True,
    )

    # If a parameter is passed with value None, it will overwrite (and reset) the existing value.
    if params:
        for key, value in params.items():
            self_serve_flow_progress.state[key] = value

    if proposal_ref:
        self_serve_flow_progress.proposal_ref = proposal_ref

    if user_ref:
        self_serve_flow_progress.user_ref = user_ref

    if commit:
        current_session.commit()

    return SelfServeFlowProgressEntity(
        id=self_serve_flow_progress.id,
        flow_type=self_serve_flow_progress.flow_type,
        country_code=self_serve_flow_progress.country_code,
        state=self_serve_flow_progress.state,
        proposal_ref=self_serve_flow_progress.proposal_ref,
        user_ref=self_serve_flow_progress.user_ref,
    )

settings ¶

get_integer_setting_value `module-attribute` ¶

get_integer_setting_value = get_integer_setting_value

slack_reporting ¶

notify_sales_inbound_of_assigned_leads `module-attribute` ¶

notify_sales_inbound_of_assigned_leads = (
    notify_sales_inbound_of_assigned_leads
)

notify_sales_inbound_of_contract_signature `module-attribute` ¶

notify_sales_inbound_of_contract_signature = (
    notify_sales_inbound_of_contract_signature
)

user_and_roles ¶

get_available_inbound_sales_intercom_ids `module-attribute` ¶

get_available_inbound_sales_intercom_ids = (
    get_available_inbound_sales_intercom_ids
)

is_alaner_inbound_sales `module-attribute` ¶

is_alaner_inbound_sales = is_alaner_inbound_sales

retrieve_alaner_details_from_intercom_admin_id `module-attribute` ¶

retrieve_alaner_details_from_intercom_admin_id = (
    retrieve_alaner_details_from_intercom_admin_id
)

components.growth.public.commands ¶

growth ¶

growth_commands `module-attribute` ¶

growth_commands = AppGroup(
    name="growth",
    help="Main command group for the Growth component",
)

components.growth.public.dependencies ¶

COMPONENT_NAME `module-attribute` ¶

COMPONENT_NAME = 'growth'

GrowthDependency ¶

Bases: ABC