Skip to content

Api reference

components.contracting.public.admin_segment_fr

AdminSegment

Bases: Segment

Segment used for emailing to select Admin users based on their responsibilities within an account and optionally specific companies. TODO @jsagl - AdminSegment should probably be owned and exposed by a component responsible for managing admin/rights

list_recipients 

list_recipients(settings, limit=None)

List Admin recipients for account_id provided in settings. Admins are selected based on the responsibilities provided in settings. At least one of account responsibilities or company responsibilities must be provided. If no company_ids are provided but company responsibilities are provided, all companies are considered. If company_ids are provided, only admins with responsibilities for those companies are considered.

Source code in components/contracting/public/admin_segment_fr.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def list_recipients(
    self,
    settings: dict[str, Any],
    limit: int | None = None,  # noqa: ARG002
) -> list[EmailRecipient]:
    """
    List Admin recipients for account_id provided in settings.
    Admins are selected based on the responsibilities provided in settings.
    At least one of account responsibilities or company responsibilities must be provided.
    If no company_ids are provided but company responsibilities are provided, all companies are considered.
    If company_ids are provided, only admins with responsibilities for those companies are considered.
    """
    from components.fr.internal.business_logic.company.queries.admin import (  # noqa: ALN043, ALN039
        get_admins_with_responsibilities_for_account,
        get_admins_with_responsibilities_for_companies_in_account,
    )

    segment_settings = AdminSegmentSettings.from_dict(settings)

    account_id = segment_settings.account_id
    company_ids = segment_settings.company_ids
    account_responsibilities = (
        segment_settings.at_least_one_account_responsibilities
    )
    company_responsibilities = (
        segment_settings.at_least_one_company_responsibilities
    )

    if not account_responsibilities and not company_responsibilities:
        raise ValueError(
            "At least one of account_responsibilities or company_responsibilities must be set"
        )

    account_responsibilities_set: set[AdminResponsibility] | None = (
        set(account_responsibilities) if account_responsibilities else None
    )
    company_responsibilities_set: set[AdminResponsibility] | None = (
        set(company_responsibilities) if company_responsibilities else None
    )

    account_admins_recipients = []
    company_admins_recipients = []

    if account_responsibilities_set:
        account_admins = get_admins_with_responsibilities_for_account(
            account_id=UUID(account_id)
        )

        account_admins_recipients = [
            EmailRecipient(
                user_ref=str(admin.user_id),
                email_address=admin.email,
                user_context={"account_id": account_id},
                recipient_type=RecipientType.admin,
            )
            for admin in account_admins
            if account_responsibilities_set.intersection(
                _get_admin_responsibilities_set(admin=admin)
            )
        ]

    if company_responsibilities_set:
        company_admins = get_admins_with_responsibilities_for_companies_in_account(
            account_id=UUID(account_id)
        )

        company_admins_recipients = [
            EmailRecipient(
                user_ref=str(admin.user_id),
                email_address=admin.email,
                user_context={
                    "company_id": admin.company_id,
                    "account_id": account_id,
                },
                recipient_type=RecipientType.admin,
            )
            for admin in company_admins
            if company_responsibilities_set.intersection(
                _get_admin_responsibilities_set(admin=admin)
            )
            and admin.company_id is not None
            and (not company_ids or admin.company_id in company_ids)
        ]

    unique_recipients = {}
    for recipient in company_admins_recipients + account_admins_recipients:
        if recipient.user_ref not in unique_recipients:
            unique_recipients[recipient.user_ref] = recipient

    return list(unique_recipients.values())

name class-attribute instance-attribute 

name = 'admin_segment'

recipient_type class-attribute instance-attribute 

recipient_type = admin

AdminSegmentSettings dataclass 

AdminSegmentSettings(
    account_id,
    company_ids=None,
    at_least_one_account_responsibilities=None,
    at_least_one_company_responsibilities=None,
)

Bases: DataClassJsonMixin

Settings for the AdminSegment.

account_id instance-attribute 

account_id

at_least_one_account_responsibilities class-attribute instance-attribute 

at_least_one_account_responsibilities = None

at_least_one_company_responsibilities class-attribute instance-attribute 

at_least_one_company_responsibilities = None

company_ids class-attribute instance-attribute 

company_ids = None

components.contracting.public.contract

create_contract_termination module-attribute 

create_contract_termination = create_contract_termination

components.contracting.public.customer_insights

components.contracting.public.entities

DocumentProviderType

Bases: AlanBaseEnum

hellosign class-attribute instance-attribute 

hellosign = 'hellosign'

manual class-attribute instance-attribute 

manual = 'manual'

EmployeeNotificationType

Bases: AlanBaseEnum

Renewal class-attribute instance-attribute 

Renewal = 'renewal'

Standard class-attribute instance-attribute 

Standard = 'standard'

HealthPrices dataclass 

HealthPrices(base_price, option_prices)

Bases: BaseProductPrices

base_price instance-attribute 

base_price

has_changed 

has_changed(other)
Source code in components/contracting/external/price/fr/health_insurance.py 
443
444
445
446
447
448
449
450
451
452
def has_changed(self, other: "BaseProductPrices") -> bool:
    if not isinstance(other, HealthPrices):
        raise TypeError(f"Cannot compare HealthPrices with {type(other)}: {other}")

    base_changed = self.base_price.has_changed(other.base_price)
    options_changed = (
        price.has_changed(other_price)
        for price, other_price in zip(self.option_prices, other.option_prices)
    )
    return base_changed or any(options_changed)

has_price_decreased 

has_price_decreased(other)

Returns True if the price has decreased compared to the given 'other' price.

So, returns True if no dimensions of the price are greater than its counterpart in the 'other' price and if at least one dimension is lower.

Source code in components/contracting/external/price/fr/health_insurance.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
def has_price_decreased(self, other: "BaseProductPrices") -> bool:
    """
    Returns True if the price has decreased compared to the given 'other' price.

    So, returns True if no dimensions of the price are greater than its counterpart in the 'other' price
    and if at least one dimension is lower.
    """
    if not isinstance(other, HealthPrices):
        raise TypeError(f"Cannot compare HealthPrices with {type(other)}: {other}")

    if self.has_price_increased(other=other):
        return False

    return any(d < 0 for d in self._get_price_dimension_diffs(other=other))

has_price_increased 

has_price_increased(other)

Returns True if the price has increase compared to the given 'other' price.

So, returns True if at least one dimension of the price is greater than its counterpart in the 'other' price.

Source code in components/contracting/external/price/fr/health_insurance.py 
454
455
456
457
458
459
460
461
462
463
def has_price_increased(self, other: "BaseProductPrices") -> bool:
    """
    Returns True if the price has increase compared to the given 'other' price.

    So, returns True if at least one dimension of the price is greater than its counterpart in the 'other' price.
    """
    if not isinstance(other, HealthPrices):
        raise TypeError(f"Cannot compare HealthPrices with {type(other)}: {other}")

    return any(d > 0 for d in self._get_price_dimension_diffs(other=other))

is_same 

is_same(other)
Source code in components/contracting/external/price/fr/health_insurance.py 
480
481
def is_same(self, other: "HealthPrices") -> bool:
    return not self.has_changed(other)

number_of_options property 

number_of_options

option_prices class-attribute instance-attribute 

option_prices = field(hash=False)

InsightsReportType

Bases: AlanBaseEnum

loi_evin class-attribute instance-attribute 

loi_evin = 'loi_evin'

MetricsPerHealthSubscriptionEntity dataclass 

MetricsPerHealthSubscriptionEntity(
    account_ref,
    subscriptor_ref,
    subscription_ref,
    contract_size,
    n_employees=0,
    n_live_primaries=0,
    n_live_partners=0,
    n_live_children=0,
    n_live_families=0,
    n_primaries_with_children=0,
    n_1st_or_2nd_children=0,
    n_live_primaries_option=0,
    n_live_partners_option=0,
    n_live_families_option=0,
    n_1st_or_2nd_children_option=0,
    n_primaries_with_children_option=0,
    n_live_primaries_option2=0,
    n_1st_or_2nd_children_option2=0,
    n_primaries_with_children_option2=0,
)

This entity is used to expose the metrics per subscription. These insights are refreshed monthly.

account_ref instance-attribute 

account_ref

contract_size instance-attribute 

contract_size

n_1st_or_2nd_children class-attribute instance-attribute 

n_1st_or_2nd_children = 0

n_1st_or_2nd_children_option class-attribute instance-attribute 

n_1st_or_2nd_children_option = 0

n_1st_or_2nd_children_option2 class-attribute instance-attribute 

n_1st_or_2nd_children_option2 = 0

n_employees class-attribute instance-attribute 

n_employees = 0

n_live_children class-attribute instance-attribute 

n_live_children = 0

n_live_families class-attribute instance-attribute 

n_live_families = 0

n_live_families_option class-attribute instance-attribute 

n_live_families_option = 0

n_live_partners class-attribute instance-attribute 

n_live_partners = 0

n_live_partners_option class-attribute instance-attribute 

n_live_partners_option = 0

n_live_primaries class-attribute instance-attribute 

n_live_primaries = 0

n_live_primaries_option class-attribute instance-attribute 

n_live_primaries_option = 0

n_live_primaries_option2 class-attribute instance-attribute 

n_live_primaries_option2 = 0

n_primaries_with_children class-attribute instance-attribute 

n_primaries_with_children = 0

n_primaries_with_children_option class-attribute instance-attribute 

n_primaries_with_children_option = 0

n_primaries_with_children_option2 class-attribute instance-attribute 

n_primaries_with_children_option2 = 0

subscription_ref instance-attribute 

subscription_ref

subscriptor_ref instance-attribute 

subscriptor_ref

PrevoyancePrices dataclass 

PrevoyancePrices(ta, tb, tc)

Bases: BaseProductPrices

has_changed 

has_changed(other)
Source code in components/contracting/external/price/fr/prevoyance_insurance.py 
14
15
16
17
18
19
20
def has_changed(self, other: "BaseProductPrices") -> bool:
    if not isinstance(other, PrevoyancePrices):
        raise TypeError(
            f"Cannot compare PrevoyancePrices with {type(other)}: {other}"
        )

    return self.ta != other.ta or self.tb != other.tb or self.tc != other.tc

has_price_decreased 

has_price_decreased(other)

Returns True if the price has decreased compared to the given 'other' price. So, returns True if no dimensions of the price are greater than its counterpart in the 'other' price and if at least one dimension is lower.

Source code in components/contracting/external/price/fr/prevoyance_insurance.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def has_price_decreased(self, other: "BaseProductPrices") -> bool:
    """
    Returns True if the price has decreased compared to the given 'other' price.
    So, returns True if no dimensions of the price are greater than its counterpart in the 'other' price
    and if at least one dimension is lower.
    """
    if not isinstance(other, PrevoyancePrices):
        raise TypeError(
            f"Cannot compare PrevoyancePrices with {type(other)}: {other}"
        )

    if self.has_price_increased(other=other):
        return False

    return self.ta < other.ta or self.tb < other.tb or self.tc < other.tc

has_price_increased 

has_price_increased(other)

Returns True if the price has increase compared to the given 'other' price. So, returns True if at least one dimension of the price is greater than its counterpart in the 'other' price.

Source code in components/contracting/external/price/fr/prevoyance_insurance.py 
22
23
24
25
26
27
28
29
30
31
32
def has_price_increased(self, other: "BaseProductPrices") -> bool:
    """
    Returns True if the price has increase compared to the given 'other' price.
    So, returns True if at least one dimension of the price is greater than its counterpart in the 'other' price.
    """
    if not isinstance(other, PrevoyancePrices):
        raise TypeError(
            f"Cannot compare PrevoyancePrices with {type(other)}: {other}"
        )

    return self.ta > other.ta or self.tb > other.tb or self.tc > other.tc

ta instance-attribute 

ta

tb instance-attribute 

tb

tc instance-attribute 

tc

SafeCustomerInsightsDashboard dataclass 

SafeCustomerInsightsDashboard()

Bases: DataClassJsonMixin

__abstract__ class-attribute instance-attribute 

__abstract__ = True

SubscriptorLegalStatus

Bases: AlanBaseEnum

company class-attribute instance-attribute 

company = 'company'

individual class-attribute instance-attribute 

individual = 'individual'

WorkflowRunUpdate dataclass 

WorkflowRunUpdate(
    started_at=None,
    completed_at=None,
    status=None,
    document_check_result_status=None,
    document_check_result_sub_status=None,
)

completed_at class-attribute instance-attribute 

completed_at = None

document_check_result_status class-attribute instance-attribute 

document_check_result_status = None

document_check_result_sub_status class-attribute instance-attribute 

document_check_result_sub_status = None

started_at class-attribute instance-attribute 

started_at = None

status class-attribute instance-attribute 

status = None

components.contracting.public.errors

ProposalErrorCodes

Bases: AlanBaseEnum

all_codes classmethod 

all_codes()
Source code in components/contracting/utils/error_codes.py 
117
118
119
120
@classmethod
@memory_only_cache
def all_codes(cls) -> set["ProposalErrorCodes"]:
    return {code for code in cls}

company_has_no_iban class-attribute instance-attribute 

company_has_no_iban = 'company_has_no_iban'

compliance_with_1_5_rule class-attribute instance-attribute 

compliance_with_1_5_rule = 'compliance_with_1_5_rule'

conflicting_proposal_item_exists class-attribute instance-attribute 

conflicting_proposal_item_exists = (
    "conflicting_proposal_item_exists"
)

direct_billing_only_work_when_participation_for_children_and_partner_is_zero class-attribute instance-attribute 

direct_billing_only_work_when_participation_for_children_and_partner_is_zero = "direct_billing_only_work_when_participation_for_children_and_partner_is_zero"

expiration_date_in_past class-attribute instance-attribute 

expiration_date_in_past = 'expiration_date_in_past'

health_compliance class-attribute instance-attribute 

health_compliance = 'health_compliance'

health_contract_has_future_amendment class-attribute instance-attribute 

health_contract_has_future_amendment = (
    "health_contract_has_future_amendment"
)

health_contract_has_no_ongoing_subscription class-attribute instance-attribute 

health_contract_has_no_ongoing_subscription = (
    "health_contract_has_no_ongoing_subscription"
)

health_subscription_not_active_at_prevoyance_start class-attribute instance-attribute 

health_subscription_not_active_at_prevoyance_start = (
    "health_subscription_not_active_at_prevoyance_start"
)

include_due_while_setup_act_is_not_due class-attribute instance-attribute 

include_due_while_setup_act_is_not_due = (
    "include_due_while_setup_act_is_not_due"
)

internal_server_error class-attribute instance-attribute 

internal_server_error = 'internal_server_error'

invalid_product_for_individual_tns_freelancer class-attribute instance-attribute 

invalid_product_for_individual_tns_freelancer = (
    "invalid_product_for_individual_tns_freelancer"
)

legacy_contract_cancellation_date class-attribute instance-attribute 

legacy_contract_cancellation_date = (
    "legacy_contract_cancellation_date"
)

legacy_health_contract_first_day_of_month class-attribute instance-attribute 

legacy_health_contract_first_day_of_month = (
    "legacy_health_contract_first_day_of_month"
)

members_will_be_unsubscribed_from_options_by_default class-attribute instance-attribute 

members_will_be_unsubscribed_from_options_by_default = (
    "members_will_be_unsubscribed_from_options_by_default"
)

new_proposal_item_starts_before_latest_offer_version class-attribute instance-attribute 

new_proposal_item_starts_before_latest_offer_version = (
    "new_proposal_item_starts_before_latest_offer_version"
)

no_manual_renewal_proposal_on_automatic_renewal_account class-attribute instance-attribute 

no_manual_renewal_proposal_on_automatic_renewal_account = "no_manual_renewal_proposal_on_automatic_renewal_account"

no_subscription_matching_legacy_termination_target class-attribute instance-attribute 

no_subscription_matching_legacy_termination_target = (
    "no_subscription_matching_legacy_termination_target"
)

no_switch_to_affiliation_based_billing class-attribute instance-attribute 

no_switch_to_affiliation_based_billing = (
    "no_switch_to_affiliation_based_billing"
)

plan_is_not_compatible_with_one_of_the_target class-attribute instance-attribute 

plan_is_not_compatible_with_one_of_the_target = (
    "plan_is_not_compatible_with_one_of_the_target"
)

proposal_does_not_start_on_first_day_of_month class-attribute instance-attribute 

proposal_does_not_start_on_first_day_of_month = (
    "proposal_does_not_start_on_first_day_of_month"
)

proposal_item_has_no_start_date class-attribute instance-attribute 

proposal_item_has_no_start_date = (
    "proposal_item_has_no_start_date"
)

proposal_item_product_is_same_as_current class-attribute instance-attribute 

proposal_item_product_is_same_as_current = (
    "proposal_item_product_is_same_as_current"
)

proposal_start_date_first_day_of_month class-attribute instance-attribute 

proposal_start_date_first_day_of_month = (
    "proposal_start_date_first_day_of_month"
)

proposal_start_date_in_the_past class-attribute instance-attribute 

proposal_start_date_in_the_past = (
    "proposal_start_date_too_far_in_the_past"
)

renewal_month_should_be_january class-attribute instance-attribute 

renewal_month_should_be_january = (
    "renewal_month_should_be_january"
)

requested_amendment_offer_is_the_same_as_ongoing_subscription_period class-attribute instance-attribute 

requested_amendment_offer_is_the_same_as_ongoing_subscription_period = "requested_amendment_offer_is_the_same_as_ongoing_subscription_period"

self_served_amendment_not_implemented class-attribute instance-attribute 

self_served_amendment_not_implemented = (
    "self_served_amendment_not_implemented"
)

self_served_proposal_cannot_downgrade_under_commitment class-attribute instance-attribute 

self_served_proposal_cannot_downgrade_under_commitment = (
    "self_served_proposal_cannot_downgrade_under_commitment"
)

self_served_proposal_item_should_have_one_target class-attribute instance-attribute 

self_served_proposal_item_should_have_one_target = (
    "self_served_proposal_item_should_have_one_target"
)

self_served_proposal_policy_not_found_for_individual_user class-attribute instance-attribute 

self_served_proposal_policy_not_found_for_individual_user = "self_served_proposal_policy_not_found_for_individual_user"

self_served_proposal_user_is_not_individual class-attribute instance-attribute 

self_served_proposal_user_is_not_individual = (
    "self_served_proposal_user_is_not_individual"
)

start_date_too_far_in_the_future class-attribute instance-attribute 

start_date_too_far_in_the_future = (
    "start_date_too_far_in_the_future"
)

switch_to_dsn_billing class-attribute instance-attribute 

switch_to_dsn_billing = 'switch_to_dsn_billing'

uncategorized class-attribute instance-attribute 

uncategorized = 'uncategorized'

user_has_no_billing_iban class-attribute instance-attribute 

user_has_no_billing_iban = 'user_has_no_billing_iban'

user_has_no_birth_date class-attribute instance-attribute 

user_has_no_birth_date = 'user_has_no_birth_date'

user_has_no_iban class-attribute instance-attribute 

user_has_no_iban = 'user_has_no_iban'

user_has_no_ssn_or_ntt class-attribute instance-attribute 

user_has_no_ssn_or_ntt = 'user_has_no_ssn_or_ntt'

user_with_active_contract class-attribute instance-attribute 

user_with_active_contract = 'user_with_active_contract'

user_with_active_exemption class-attribute instance-attribute 

user_with_active_exemption = 'user_with_active_exemption'

user_with_debts class-attribute instance-attribute 

user_with_debts = 'user_with_debts'

user_with_policy_not_in_ani class-attribute instance-attribute 

user_with_policy_not_in_ani = 'user_with_policy_not_in_ani'

TerminationErrorCodes

Bases: AlanBaseEnum

cannot_terminate_contract_for_key_account class-attribute instance-attribute 

cannot_terminate_contract_for_key_account = (
    "cannot_terminate_contract_for_key_account"
)

cannot_terminate_contract_that_was_already_terminated class-attribute instance-attribute 

cannot_terminate_contract_that_was_already_terminated = (
    "cannot_terminate_contract_that_was_already_terminated"
)

cannot_terminate_health_contract_without_terminating_prevoyance_contract class-attribute instance-attribute 

cannot_terminate_health_contract_without_terminating_prevoyance_contract = "cannot_terminate_health_contract_without_terminating_prevoyance_contract"

cannot_terminate_unbalanced_contract class-attribute instance-attribute 

cannot_terminate_unbalanced_contract = (
    "cannot_terminate_unbalanced_contract"
)

self_termination_not_supported_for_individuals class-attribute instance-attribute 

self_termination_not_supported_for_individuals = (
    "self_termination_not_supported_for_individuals"
)

termination_date_cannot_be_in_the_past class-attribute instance-attribute 

termination_date_cannot_be_in_the_past = (
    "termination_date_cannot_be_in_the_past"
)

termination_date_cannot_be_more_than_3_months_from_now class-attribute instance-attribute 

termination_date_cannot_be_more_than_3_months_from_now = (
    "termination_date_cannot_be_more_than_3_months_from_now"
)

user_does_not_have_access_to_all_health_contracts class-attribute instance-attribute 

user_does_not_have_access_to_all_health_contracts = (
    "user_does_not_have_access_to_all_health_contracts"
)

user_does_not_have_access_to_all_prevoyance_contracts class-attribute instance-attribute 

user_does_not_have_access_to_all_prevoyance_contracts = (
    "user_does_not_have_access_to_all_prevoyance_contracts"
)

ValidationContext dataclass 

ValidationContext(
    should_bypass_all_warnings=False,
    warnings_to_bypass=list(),
    warnings_to_consider=list(),
    should_bypass_all_blockers=False,
    blockers_to_bypass=list(),
    noncompliance_acknowledgement_link=None,
    should_persist_errors=True,
    mark_blockers_as_seen=tuple(),
    mark_warnings_as_seen=tuple(),
)

Bases: DataClassJsonMixin

This class is used to configure how the validation warnings & blockers should be handled. You can define which warnings & blockers should be ignored using their code, or if all of them should be ignored. Blockers can't be ignored in production mode, this is only provided to made testing easier.

__post_init__ 

__post_init__()
Source code in components/contracting/utils/validation.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __post_init__(self) -> None:
    if self.should_bypass_all_blockers and self.blockers_to_bypass:
        raise ValueError(
            "You can't bypass all blockers and specific errors at the same time"
        )
    if self.should_bypass_all_warnings and self.warnings_to_bypass:
        raise ValueError(
            "You can't bypass all warnings and specific warnings at the same time"
        )
    if not self.should_bypass_all_warnings and self.warnings_to_consider:
        raise ValueError(
            "You can't consider all warnings and specific warnings at the same time"
        )

    if not self.should_persist_errors and (
        self.mark_blockers_as_seen or self.mark_warnings_as_seen
    ):
        raise ValueError("You can't mark errors as seen if you don't persist them")

blockers_to_bypass class-attribute instance-attribute 

blockers_to_bypass = field(default_factory=list)

mark_blockers_as_seen class-attribute instance-attribute 

mark_blockers_as_seen = tuple()

mark_warnings_as_seen class-attribute instance-attribute 

mark_warnings_as_seen = tuple()

noncompliance_acknowledgement_link = None

should_bypass_all_blockers class-attribute instance-attribute 

should_bypass_all_blockers = False

should_bypass_all_warnings class-attribute instance-attribute 

should_bypass_all_warnings = False

should_ignore_message 

should_ignore_message(message)
Source code in components/contracting/utils/validation.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def should_ignore_message(self, message: ContractingMessage) -> bool:
    if (
        message.is_warning
        and (
            self.should_bypass_all_warnings
            or message.code in self.warnings_to_bypass
        )
        and message.code not in self.warnings_to_consider
    ):
        if message.require_acknowledgement_link:
            return self.noncompliance_acknowledgement_link is not None
        else:
            return True
    if message.is_blocker and (
        self.should_bypass_all_blockers or message.code in self.blockers_to_bypass
    ):
        return True
    return False

should_mark_as_seen 

should_mark_as_seen(message)
Source code in components/contracting/utils/validation.py 
59
60
61
62
63
64
65
66
67
def should_mark_as_seen(self, message: ContractingMessage) -> bool:
    if message.severity_level == ContractingMessageSeverityLevel.warning:
        return message.code in self.mark_warnings_as_seen

    elif message.severity_level == ContractingMessageSeverityLevel.blocker:
        return message.code in self.mark_blockers_as_seen

    else:
        raise ValueError("Unknown severity level")

should_persist_errors class-attribute instance-attribute 

should_persist_errors = True

warnings_to_bypass class-attribute instance-attribute 

warnings_to_bypass = field(default_factory=list)

warnings_to_consider class-attribute instance-attribute 

warnings_to_consider = field(default_factory=list)

components.contracting.public.exceptions

ProposalInTerminalStateException

Bases: Exception

ProposalNotFoundException

Bases: Exception

RenewalNotFoundException

Bases: Exception

SubscriptionNotFoundException

Bases: Exception

components.contracting.public.helpers

app_group

contracting module-attribute 

contracting = AppGroup('contracting')

controllers

get_contracting_controllers 

get_contracting_controllers()
Source code in components/contracting/public/helpers/controllers.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def get_contracting_controllers():  # type: ignore[no-untyped-def]  # noqa: D103
    from components.contracting.subcomponents.account_hub.internal.controllers.base_controllers import (
        AccountHubChurnRiskAlertController,
        AccountHubLegalDocumentController,
        AccountHubSettingsController,
    )
    from components.contracting.subcomponents.account_hub.internal.controllers.churn_risk_alert import (  # noqa: F401
        create_churn_risk_alert,
        edit_churn_risk_alert,
        list_churn_risk_alerts,
    )
    from components.contracting.subcomponents.account_hub.internal.controllers.download_legal_document import (  # noqa: F401
        download_legal_document,
    )
    from components.contracting.subcomponents.account_hub.internal.controllers.get_account_settings import (  # noqa: F401
        get_account_settings_view,
    )
    from components.contracting.subcomponents.account_hub.internal.controllers.set_account_settings import (  # noqa: F401
        set_account_settings_view,
    )
    from components.contracting.subcomponents.account_hub.internal.controllers.upload_legal_document import (  # noqa: F401
        upload_legal_document,
    )
    from components.contracting.subcomponents.dashboard.public.controllers.builder_product_cards import (
        BuilderProductCardsController,
    )
    from components.contracting.subcomponents.dashboard.public.controllers.legal_documents import (
        LegalDocumentsController,
    )
    from components.contracting.subcomponents.dashboard.public.controllers.renewal import (
        RenewalController,
    )
    from components.contracting.subcomponents.dashboard.public.controllers.signature_request import (
        SignatureRequestController,
    )
    from components.contracting.subcomponents.dashboard.public.controllers.subscription import (
        SubscriptionController,
    )
    from components.contracting.subcomponents.legal_document.internal.controllers.legal_clause import (
        LegalClauseController,
    )
    from components.contracting.subcomponents.legal_document.internal.controllers.legal_content import (
        LegalContentController,
    )
    from components.contracting.subcomponents.legal_document.internal.controllers.legal_template import (
        LegalTemplateController,
    )
    from components.contracting.subcomponents.legal_document.internal.controllers.legal_viewer import (
        LegalViewerController,
    )

    return [
        AccountHubChurnRiskAlertController,
        AccountHubLegalDocumentController,
        AccountHubSettingsController,
        HealthAmendmentProposalController,
        PrevoyanceAmendmentProposalController,
        LegalDocumentsController,
        BuilderProductCardsController,
        SignatureRequestController,
        SubscriptionController,
        SubscriptionPeriodNotificationSettingsController,
        LegalClauseController,
        LegalContentController,
        LegalTemplateController,
        LegalViewerController,
        RenewalController,
    ]

components.contracting.public.insights

components.contracting.public.language

get_supported_languages 

get_supported_languages(app_name)

Gets the supported languages for the given app_name

Source code in components/contracting/public/language.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def get_supported_languages(app_name: AppName) -> set[Lang]:
    """Gets the supported languages for the given app_name"""
    from components.contracting.external.i18n.be.language import (
        get_supported_languages as get_supported_languages_be,
    )
    from components.contracting.external.i18n.es.language import (
        get_supported_languages as get_supported_languages_es,
    )
    from components.contracting.external.i18n.fr.language import (
        get_supported_languages as get_supported_languages_fr,
    )

    if app_name == AppName.ALAN_FR:
        return get_supported_languages_fr()
    elif app_name == AppName.ALAN_BE:
        return get_supported_languages_be()
    elif app_name == AppName.ALAN_ES:
        return get_supported_languages_es()
    else:
        raise ValueError(f"Unknown app_name {app_name}")

components.contracting.public.legal_document

EsAlanEssentialCPInputs

Bases: Inputs

Used to represent inputs needed to render a Spanish Alan Essential CP document

company_info instance-attribute 

company_info

dates instance-attribute 

dates

flags instance-attribute 

flags

options_data instance-attribute 

options_data

references instance-attribute 

references

signatory instance-attribute 

signatory

EsAlanEssentialNoticeInputs

Bases: Inputs

Used to represent inputs needed to render a Spanish Alan Essential Notice document

dates instance-attribute 

dates

options_data instance-attribute 

options_data

EsAlanHealthCPInputCoverageDetails

Used to represent coverage details needed to render a Spanish Alan Health CP document

covered_therapy_sessions instance-attribute 

covered_therapy_sessions

EsAlanHealthCPInputDates

Bases: EsCPInputDates

Used to represent dates needed to render a Spanish Alan Health CP document

waiting_periods_from_date instance-attribute 

waiting_periods_from_date

EsAlanHealthCPInputFlags

Bases: EsCPInputFlags

Used to represent flags needed to render a Spanish Alan Health CP document

has_copayments instance-attribute 

has_copayments

has_optical instance-attribute 

has_optical

has_pharmacy instance-attribute 

has_pharmacy

has_physio_nutrition instance-attribute 

has_physio_nutrition

has_pre_existing_conditions instance-attribute 

has_pre_existing_conditions

has_premium_reimbursements instance-attribute 

has_premium_reimbursements

has_waiting_periods instance-attribute 

has_waiting_periods

is_premium_reimbursements_optional instance-attribute 

is_premium_reimbursements_optional

EsAlanHealthCPInputs

Bases: Inputs

Used to represent inputs needed to render a Spanish Alan Health CP document

company_info instance-attribute 

company_info

coverage instance-attribute 

coverage

dates instance-attribute 

dates

flags instance-attribute 

flags

insuree_info instance-attribute 

insuree_info

options_data instance-attribute 

options_data

references instance-attribute 

references

signatory instance-attribute 

signatory

EsAlanHealthNoticeInputCoverageDetails

Used to represent coverage details needed to render a Spanish Alan Health Notice document

covered_therapy_sessions instance-attribute 

covered_therapy_sessions

EsAlanHealthNoticeInputDates

Bases: EsNoticeInputDates

Used to represent dates needed to render a Spanish Alan Health Notice document

waiting_periods_from_date instance-attribute 

waiting_periods_from_date

EsAlanHealthNoticeInputFlags

Used to represent flags needed to render a Spanish Alan Health Notice document

has_copayments instance-attribute 

has_copayments

has_optical instance-attribute 

has_optical

has_pharmacy instance-attribute 

has_pharmacy

has_physio_nutrition instance-attribute 

has_physio_nutrition

has_pre_existing_conditions instance-attribute 

has_pre_existing_conditions

has_premium_reimbursements instance-attribute 

has_premium_reimbursements

has_waiting_periods instance-attribute 

has_waiting_periods

is_premium_reimbursements_optional instance-attribute 

is_premium_reimbursements_optional

EsAlanHealthNoticeInputs

Bases: Inputs

Used to represent inputs needed to render a Spanish Alan Health Notice document

coverage instance-attribute 

coverage

dates instance-attribute 

dates

flags instance-attribute 

flags

options_data instance-attribute 

options_data

EsAlanOutpatientCPInputFlags

Bases: EsCPInputFlags

Used to represent flags needed to render a Spanish Alan Outpatient CP document

has_optical instance-attribute 

has_optical

has_pharmacy instance-attribute 

has_pharmacy

has_physio_nutrition instance-attribute 

has_physio_nutrition

has_pre_existing_conditions instance-attribute 

has_pre_existing_conditions

has_zero_percent_participation instance-attribute 

has_zero_percent_participation

EsAlanOutpatientCPInputs

Bases: Inputs

Used to represent inputs needed to render a Spanish Alan Outpatient CP document

company_info instance-attribute 

company_info

coverage instance-attribute 

coverage

dates instance-attribute 

dates

flags instance-attribute 

flags

options_data instance-attribute 

options_data

references instance-attribute 

references

signatory instance-attribute 

signatory

EsAlanOutpatientNoticeInputFlags

Used to represent flags needed to render a Spanish Alan Outpatient Notice document

has_optical instance-attribute 

has_optical

has_pharmacy instance-attribute 

has_pharmacy

has_physio_nutrition instance-attribute 

has_physio_nutrition

has_pre_existing_conditions instance-attribute 

has_pre_existing_conditions

has_zero_percent_participation instance-attribute 

has_zero_percent_participation

EsAlanOutpatientNoticeInputs

Bases: Inputs

Used to represent inputs needed to render a Spanish Alan Outpatient Notice document

coverage instance-attribute 

coverage

dates instance-attribute 

dates

flags instance-attribute 

flags

options_data instance-attribute 

options_data

EsCGInputFlags

Used to represent flags needed to render a Spanish general conditions document

has_premium_reimbursements instance-attribute 

has_premium_reimbursements

EsCGInputs

Bases: Inputs

A class used to represent the inputs needed to render CG document for ES Health.

Coverage

Used to represent coverage details needed to render a Spanish Alan Health CG document

covered_therapy_sessions instance-attribute 
covered_therapy_sessions

coverage instance-attribute 

coverage

flags instance-attribute 

flags

EsCPInputDates

Used to represent shared dates needed to render a Spanish CP document

amendment_start_date instance-attribute 

amendment_start_date

contract_end_date instance-attribute 

contract_end_date

contract_start_date instance-attribute 

contract_start_date

signature_date instance-attribute 

signature_date

EsCPInputFlags

Used to represent the shared flags needed to render a Spanish CP document

has_sepa_debit_payment_method instance-attribute 

has_sepa_debit_payment_method

is_amendment instance-attribute 

is_amendment

is_preview instance-attribute 

is_preview

EsCPInputReferences

Used to represent references needed to render a Spanish CP document

contract_number instance-attribute 

contract_number

EsCPOptions

Used to represent options needed to render a Spanish CP document

optical_data instance-attribute 

optical_data

pharmacy_data instance-attribute 

pharmacy_data

physio_nutrition_data instance-attribute 

physio_nutrition_data

pre_existing_conditions_data instance-attribute 

pre_existing_conditions_data

premium_reimbursements_data instance-attribute 

premium_reimbursements_data

product_data instance-attribute 

product_data

EsNoticeInputDates

Used to represent shared dates needed to render a Spanish Notice document

amendment_start_date instance-attribute 

amendment_start_date

contract_start_date instance-attribute 

contract_start_date

FrCGAANInputs

Bases: Inputs

A class used to represent the inputs needed to render a French CG document.

It differs from FrCGInputs in the sense that it's intended at the association and thus include information about the price and coverage as well as legal content.

bundle_version instance-attribute 

bundle_version

FrCGInputs

Bases: Inputs

A class used to represent the inputs needed to render a French CG document.

FrCertificateHealthInputs

Bases: Inputs

A class used to represent the inputs needed to render a Health Certificate document.

bundle_version instance-attribute 

bundle_version

company_info instance-attribute 

company_info

contract_number instance-attribute 

contract_number

insuree_info instance-attribute 

insuree_info

is_amendment instance-attribute 

is_amendment

is_offline instance-attribute 

is_offline

is_preview class-attribute instance-attribute 

is_preview = False

is_quote_request class-attribute instance-attribute 

is_quote_request = False

is_tacit instance-attribute 

is_tacit

signatory instance-attribute 

signatory

FrCompaniesCPHealthInputs

Bases: Inputs

A class used to represent the inputs needed to render Health CP and Notice for companies and collective_retirees segments in France. FR Health Companies CP: https://docs.google.com/document/d/1a976b0FOF1K69r24xn5-4PphcJJ1MSMTID5ytf83gd0/edit ⧉ FR Health Companies Notice: https://docs.google.com/document/d/1P8fCRYn-uqxOX08qddW7A2NZyxCOqHZI/edit ⧉ FR Health Collective Retirees CP: https://docs.google.com/document/d/1OmugyJpuzjGpf_y234hZi5PqNs5jaKM0DnZD9jP5154/edit ⧉ FR Health Collective Retirees Notice: https://docs.google.com/document/d/1RzhOM67-V-4bJA-3LTL-8XfkRWAqKVua/edit ⧉ TODO: Haven't renamed to avoid too many conflicts for now, but should be renamed to something less specific

bundle_version instance-attribute 

bundle_version

company_info instance-attribute 

company_info

company_participation instance-attribute 

company_participation

contract_number instance-attribute 

contract_number

custom_clauses instance-attribute 

custom_clauses

dates instance-attribute 

dates

flags instance-attribute 

flags

get_cache_key 

get_cache_key(legal_document_identifier)

Returns a cache key for the given legal document identifier, None if not applicable.

Source code in components/contracting/subcomponents/legal_document/public/inputs.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def get_cache_key(
    self, legal_document_identifier: LegalDocumentIdentifier
) -> str | None:
    """
    Returns a cache key for the given legal document identifier, None if not applicable.
    """
    if LegalDocumentIdentifierMatcher().notice(legal_document_identifier):
        # ⚠️ This is based on the knowledge of the notice_companies template, the fact
        # that it doesn't use certain information from the inputs (eg: company info,
        # contract numbers, dates, and so on).
        # ⚠️ We ignore document ID as we're only caching unsigned document and we
        # don't expect them to be needed for a "split by marker ID" coming back from
        # signature provider.
        relevant_info_for_cache_key = ":".join(
            [
                str(info)
                for info in (
                    self.bundle_version,
                    self.flags.is_mandatory_children,
                    self.flags.is_mandatory_partner,
                    self.flags.has_direct_billing,
                    self.flags.has_direct_billing_option,
                    hash(self.population),
                    hash(self.company_participation),
                    "/".join([str(hash(c)) for c in self.custom_clauses]),
                )
            ]
        )
        return f"fr_health:notice:{relevant_info_for_cache_key}"

    return None

option_contract_numbers instance-attribute 

option_contract_numbers

population instance-attribute 

population

signatory instance-attribute 

signatory

surco_contract_number instance-attribute 

surco_contract_number

FrCpHealthInputFlags

Used to represent flags needed to render a French health CP document

has_direct_billing instance-attribute 

has_direct_billing

has_direct_billing_option instance-attribute 

has_direct_billing_option

is_amendment instance-attribute 

is_amendment

is_mandatory_children instance-attribute 

is_mandatory_children

is_mandatory_partner instance-attribute 

is_mandatory_partner

is_preview class-attribute instance-attribute 

is_preview = False

is_quote_request class-attribute instance-attribute 

is_quote_request = False

FrIndivAANHealthInputs

Bases: Inputs

A class used to represent the inputs needed to render Health Notice for individual_tns and individual_evin segments in France. FR Health Individual Notice:

bundle_version instance-attribute 

bundle_version

contract_number instance-attribute 

contract_number

discounts instance-attribute 

discounts

insuree_info instance-attribute 

insuree_info

is_amendment instance-attribute 

is_amendment

is_preview class-attribute instance-attribute 

is_preview = False

is_quote_request class-attribute instance-attribute 

is_quote_request = False

is_tacit instance-attribute 

is_tacit

signatory instance-attribute 

signatory

FrIndivPrivateOrRetireeInputs

Bases: Inputs

Used to represent the inputs needed to render Health Notice for individual_private and individual_retirees segments in France.

bundle_version instance-attribute 

bundle_version

contract_number instance-attribute 

contract_number

insuree_info instance-attribute 

insuree_info

is_amendment instance-attribute 

is_amendment

is_preview class-attribute instance-attribute 

is_preview = False

is_quote_request class-attribute instance-attribute 

is_quote_request = False

is_tacit instance-attribute 

is_tacit

signatory instance-attribute 

signatory

FrPrevoyanceInputs

Bases: Inputs

A class used to represent the inputs needed to render Prevoyance for companies and companies_cnp segments in France. FR Prevoyance Companies Notice: https://docs.google.com/document/d/1ez66ziAuLshKMqo7pKLqC57Kj4k_HDhtDmwz3f0VBY0/edit ⧉

company_info instance-attribute 

company_info

contract_number instance-attribute 

contract_number

coverage_table_data instance-attribute 

coverage_table_data

custom_clauses instance-attribute 

custom_clauses

dates instance-attribute 

dates

flags instance-attribute 

flags

min_tenure_in_months instance-attribute 

min_tenure_in_months

participation instance-attribute 

participation

population instance-attribute 

population

pricing instance-attribute 

pricing

product_code instance-attribute 

product_code

product_id instance-attribute 

product_id

signatory instance-attribute 

signatory

Inputs

Base data type to represent the inputs we'll need to render a legal document. See render_legal_document for more details.

This class has the bare minimum that will be needed to resolve/lookup at the right template according to their release policy. Each context might require different inputs, in which case subclassing Inputs is the way.

Those inputs are sent to the relevant legal document context's init function.

contract_start_date instance-attribute 

contract_start_date

document_id class-attribute instance-attribute 

document_id = None

This ID will be injected in the document template so it's possible to get it back from a PDF, or a merged PDF's specific page. See SIGNED_DOCUMENT_ID_MARKER.

generate_dummy classmethod 

generate_dummy(legal_document_identifier, **overrides)

Generates dummy inputs for the given legal document identifier.

This is used to generate previews in the UI.

Source code in components/contracting/subcomponents/legal_document/public/inputs.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@classmethod
def generate_dummy(
    cls,
    legal_document_identifier: LegalDocumentIdentifier,
    **overrides: Any,
) -> Self:
    """
    Generates dummy inputs for the given legal document identifier.

    This is used to generate previews in the UI.
    """
    if not cls._does_support(legal_document_identifier):
        raise UnsupportedLegalDocumentIdentifierError(
            f"{cls.__name__} does not support {legal_document_identifier}"
        )

    return cls._generate_dummy(overrides=overrides)

new_start_date instance-attribute 

new_start_date

on_date property 

on_date

Calculate the reference date for legal document template selection

LegalDocumentIdentifier dataclass 

LegalDocumentIdentifier(
    subscription_scope, segment, document_type, language
)

Bases: DataClassJsonMixin

With a legal document identifier, a context will be able to generate the proper template variables, and to fetch the proper document template to render as well.

document_type instance-attribute 

document_type

language instance-attribute 

language

segment instance-attribute 

segment

subscription_scope instance-attribute 

subscription_scope

LegalDocumentIdentifierMatcher 

LegalDocumentIdentifierMatcher(conditions=tuple())
Source code in components/contracting/subcomponents/legal_document/external/contexts/utils.py 
22
23
24
25
26
def __init__(
    self,
    conditions: tuple[Callable[[LegalDocumentIdentifier], bool], ...] = tuple(),
) -> None:
    self._conditions = conditions

__call__ 

__call__(key)
Source code in components/contracting/subcomponents/legal_document/external/contexts/utils.py 
19
20
def __call__(self, key: LegalDocumentIdentifier) -> bool:
    return all(condition(key) for condition in self._conditions)

certificat property 

certificat

cg property 

cg

collective_retirees property 

collective_retirees

companies property 

companies

companies_essential property 

companies_essential

companies_outpatient property 

companies_outpatient

cp property 

cp

fr_health property 

fr_health

fr_prevoyance property 

fr_prevoyance

individual_collective_retirees property 

individual_collective_retirees

individual_evin property 

individual_evin

individual_ex_employees property 

individual_ex_employees

individual_private property 

individual_private

individual_retirees property 

individual_retirees

individual_tns property 

individual_tns

notice property 

notice

legal_document_cache 

legal_document_cache(initial_cache_content=NOT_SET)
Source code in components/contracting/subcomponents/legal_document/internal/helpers/caching.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@contextmanager
def legal_document_cache(
    initial_cache_content: NotSet[LegalDocumentCache] = NOT_SET,
) -> Iterator[None]:
    if is_set(initial_cache_content):
        cache = initial_cache_content
    else:
        if previous_cache_content := _legal_document_cache.get():
            cache = {**previous_cache_content}
        else:
            cache = {}

    reset_token = _legal_document_cache.set(cache)

    try:
        yield
    finally:
        _legal_document_cache.reset(reset_token)

legal_document_overrides 

legal_document_overrides(new_overrides)
Source code in components/contracting/subcomponents/legal_document/internal/helpers/overrides.py 
29
30
31
32
33
34
35
36
37
38
39
40
@contextmanager
def legal_document_overrides(new_overrides: Overrides) -> Iterator[None]:
    overrides: Overrides = {
        **get_overrides(),
        **new_overrides,
    }
    reset_token = _legal_document_overrides.set(overrides)

    try:
        yield
    finally:
        _legal_document_overrides.reset(reset_token)

render_legal_document(
    legal_document_identifier,
    inputs,
    file_format,
    fragment_overrides=NOT_SET,
    _include_drafts_only_for_preview=False,
)
Source code in components/contracting/subcomponents/legal_document/internal/actions/rendering.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
@contracting_tracer_wrap()
def render_legal_document(
    legal_document_identifier: LegalDocumentIdentifier,
    inputs: Inputs,
    file_format: Literal["pdf", "html"],
    fragment_overrides: NotSet[Mapping[FragmentName, MarkdownContent]] = NOT_SET,
    _include_drafts_only_for_preview: bool = False,
) -> RenderedLegalDocument:
    from components.contracting.subcomponents.legal_document.internal.helpers.overrides import (
        get_overrides,
    )

    trace = Trace.start_trace(
        namespace="contracting.legal_document",
        metric="render_legal_document",
        log_message=f"Rendering {legal_document_identifier}",
    )
    # Add those to the tags of each metric
    dd_tags = {
        "subscription_scope": legal_document_identifier.subscription_scope.value,
        "segment": legal_document_identifier.segment.value,
        "document_type": legal_document_identifier.document_type.value,
        "lang": legal_document_identifier.language.value,
        "file_format": file_format,
    }

    tracker = TemplateUsageTracker()

    context = get_context(
        subscription_scope=legal_document_identifier.subscription_scope,
        inputs=inputs,
    )

    cache_key = context.get_cache_key(
        legal_document_identifier=legal_document_identifier
    )

    if file_format == "pdf":
        # Check for custom PDF content first
        custom_pdf_content, custom_pdf_identifier = context.get_custom_pdf_content(
            legal_document_identifier=legal_document_identifier,
        )
        if custom_pdf_content and custom_pdf_identifier:
            tracker.record_template_usage(
                template_name="custom_pdf_content",
                template_id=custom_pdf_identifier,
            )
            template_manifest = TemplateManifest(
                entries=tracker.get_entries(),
                generated_at=utcnow(),
                commit_hash=get_application_version(),
            )

            trace.end_trace(
                dimensions={"custom_pdf_content": "yes", **dd_tags},
                log_message=f"Rendered {legal_document_identifier} successfully",
            )

            return RenderedLegalDocument(
                file=custom_pdf_content,
                template_args={},
                metadata={"from_custom_pdf_content": True},
                template_manifest=template_manifest,
            )

        # Then check the cached version
        if cache_key and (
            cached_value := get_cached_document(legal_document_cache_key=cache_key)
        ):
            trace.end_trace(
                dimensions={"custom_pdf_content": "no", "cache": "yes", **dd_tags},
                log_message=f"Rendered {legal_document_identifier} successfully",
            )

            # Copuy the cached file to a new temp file to be able to read them independently
            cached_value.file.seek(0)
            temp_file = NamedTemporaryFile(suffix=".pdf")
            temp_file.write(cached_value.file.read())
            temp_file.seek(0)

            return RenderedLegalDocument(
                file=temp_file,
                template_args={},
                metadata={"from_cache": True, "cache_key": cache_key},
                template_manifest=cached_value.template_manifest,
            )

    template = context.get_template(
        legal_document_identifier=legal_document_identifier,
    )

    # Resolver for the templates that can come from different sources...
    loaders: list[BaseLoader] = []

    if is_set(fragment_overrides):
        loaders.append(
            DictLoader(
                {
                    f"{fragment_name}.markdown": fragment_content
                    for fragment_name, fragment_content in fragment_overrides.items()
                }
            )
        )

    # Loaders are dependent to the release date of the documents we're targetting.
    on_date = get_overrides().get("on_date", inputs.on_date)

    loaders.append(
        # Data-stored templates (legal content) available for this identifier and date
        LegalDocumentTemplateLoader(
            legal_document_identifier=legal_document_identifier,
            on_date=on_date,
            _include_drafts_only_for_preview=_include_drafts_only_for_preview,
        ),
    )

    if not is_production_mode() and bool_feature_flag(
        feature_flag_key="kill-switch-legal-docs-fallback-to-turing",
        default_value=True,
    ):
        loaders.append(
            # Meant for acceptance/demo, looks up if the template is available in prod via Turing
            TuringLoader(
                legal_document_identifier=legal_document_identifier,
                on_date=on_date,
                _include_drafts_only_for_preview=_include_drafts_only_for_preview,
            ),
        )

    # Loader for the partials and shared stuff
    loaders.extend([context.loader, _shared_loader])

    # Generate the template args, including the tricky doc_info which depends on the
    # loader and root template document.
    template_args = context.get_template_args(
        legal_document_identifier=legal_document_identifier,
        doc_info=DocumentInformation(
            root_release_year=str(template.release_date.year),
            root_release_date=str(template.release_date),
            legal_content_version=_get_legal_content_version(
                loader=ChoiceLoader(loaders),
            ),
        ),
    )
    template_args_dict = attrs.asdict(template_args, recurse=False)

    # Tracking of the loader
    tracker.record_template_usage(
        template_name="root",
        template_id=str(template.id),
    )
    loader = ChoiceLoader([TrackingWrapper(l, tracker) for l in loaders])

    file_io = _render(
        template_name=mandatory(template.file_name),
        template_args=template_args_dict,
        loader=loader,
        file_format=file_format,
    )

    template_manifest = TemplateManifest(
        entries=tracker.get_entries(),
        generated_at=utcnow(),
        commit_hash=get_application_version(),
    )

    trace.end_trace(
        dimensions={"custom_pdf_content": "no", **dd_tags},
        log_message=f"Rendered {legal_document_identifier} successfully",
    )

    if cache_key:
        set_cached_document(
            legal_document_cache_key=cache_key,
            legal_document_cache_value=LegalDocumentCacheValue(
                file=file_io,
                template_manifest=template_manifest,
            ),
        )

    return RenderedLegalDocument(
        file=file_io,
        template_args=template_args_dict,
        metadata={},
        template_manifest=template_manifest,
    )

components.contracting.public.population

components.contracting.public.proposal

ActivityType

Bases: AlanBaseEnum

employed class-attribute instance-attribute 

employed = 'employed'

is_employed classmethod 

is_employed(activity_type)
Source code in components/contracting/utils/population.py 
46
47
48
@classmethod
def is_employed(cls, activity_type: "ActivityType") -> bool:
    return activity_type == cls.employed

is_retired classmethod 

is_retired(activity_type)
Source code in components/contracting/utils/population.py 
42
43
44
@classmethod
def is_retired(cls, activity_type: "ActivityType") -> bool:
    return activity_type == cls.retired

retired class-attribute instance-attribute 

retired = 'retired'

ApprovalNotificationType

Bases: AlanBaseEnum

admin_invitation_email_with_signature_redirect class-attribute instance-attribute 

admin_invitation_email_with_signature_redirect = (
    "admin_invitation_email_with_signature_redirect"
)

direct_link = 'direct_link'

signature_email class-attribute instance-attribute 

signature_email = 'signature_email'

ApprovalRequest dataclass 

ApprovalRequest(
    id,
    proposal_id,
    approver_first_name,
    approver_last_name,
    approver_email,
    created_at,
    signed_bundle_id,
    requester_ref,
    requester_full_name,
    approver_ref,
    notification_type,
    activity_logs,
    is_withdrawn,
    is_signed,
    approved_at,
    auto_approve_after,
    document_provider_type,
    cc_recipients,
    can_be_forwarded,
    reminder_scheduled_at,
    reminder_sent_at,
    onfido_workflow_run_id,
    client_signed_at,
)

activity_logs instance-attribute 

activity_logs

approved_at instance-attribute 

approved_at

approver_email instance-attribute 

approver_email

approver_first_name instance-attribute 

approver_first_name

approver_full_name property 

approver_full_name

approver_last_name instance-attribute 

approver_last_name

approver_ref instance-attribute 

approver_ref

auto_approve_after instance-attribute 

auto_approve_after

can_be_forwarded instance-attribute 

can_be_forwarded

can_be_manually_approved property 

can_be_manually_approved

cc_recipients instance-attribute 

cc_recipients

client_signed_at instance-attribute 

client_signed_at

created_at instance-attribute 

created_at

document_provider property 

document_provider

document_provider_type instance-attribute 

document_provider_type

forward_sign_url property 

forward_sign_url

from_model classmethod 

from_model(model)
Source code in components/contracting/subcomponents/proposal/api/entities/approval_request.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@classmethod
def from_model(cls, model: "ApprovalRequestSQLA") -> "ApprovalRequest":
    return ApprovalRequest(
        id=model.id,
        proposal_id=model.proposal_id,
        approver_first_name=model.approver_first_name,
        approver_last_name=model.approver_last_name,
        approver_email=model.approver_email,
        created_at=model.created_at,
        signed_bundle_id=model.signed_bundle_id,
        requester_ref=(
            model.signed_bundle.requester_ref if model.signed_bundle else None
        ),
        requester_full_name=(
            model.signed_bundle.requester_full_name
            if model.signed_bundle and model.signed_bundle.requester_full_name
            else "unknown"
        ),
        approver_ref=model.approver_ref,
        can_be_forwarded=model.can_be_forwarded,
        notification_type=ApprovalNotificationType(model.notification_type),
        activity_logs=(
            [ApprovalActivityLog.from_dict(x) for x in model.activity_logs]
            if model.activity_logs
            else []
        ),
        is_withdrawn=model.is_withdrawn,
        is_signed=(
            model.signed_bundle.is_signed
            if model.signed_bundle
            else bool(model.manual_documents)
        ),
        auto_approve_after=model.auto_approve_after,
        approved_at=model.approved_at,
        document_provider_type=model.document_provider_type,
        cc_recipients=model.cc_recipients,
        reminder_scheduled_at=model.reminder_scheduled_at,
        reminder_sent_at=model.reminder_sent_at,
        onfido_workflow_run_id=model.onfido_workflow_run_id,
        client_signed_at=model.client_signed_at,
    )

id instance-attribute 

id

is_approved property 

is_approved

is_approver_external property 

is_approver_external

is_approver_internal property 

is_approver_internal

is_live property 

is_live

is_login_required property 

is_login_required

is_signable property 

is_signable

is_signed instance-attribute 

is_signed

is_withdrawn instance-attribute 

is_withdrawn

notification_type instance-attribute 

notification_type

onfido_workflow_run_id instance-attribute 

onfido_workflow_run_id

proposal_id instance-attribute 

proposal_id

reminder_scheduled_at instance-attribute 

reminder_scheduled_at

reminder_sent_at instance-attribute 

reminder_sent_at

requester_full_name instance-attribute 

requester_full_name

requester_ref instance-attribute 

requester_ref

review_sign_url property 

review_sign_url

short_id property 

short_id

sign_url property 

sign_url

signed_bundle_id instance-attribute 

signed_bundle_id

EmployeeNotificationType

Bases: AlanBaseEnum

Renewal class-attribute instance-attribute 

Renewal = 'renewal'

Standard class-attribute instance-attribute 

Standard = 'standard'

FrMemberNotification dataclass 

FrMemberNotification(
    notify_members_on_approved,
    notification_type,
    employee_communication_scheduled_at,
)

employee_communication_scheduled_at instance-attribute 

employee_communication_scheduled_at

from_settings_dict classmethod 

from_settings_dict(settings)
Source code in components/contracting/subcomponents/proposal/plugins/shared/fr/notification.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@classmethod
def from_settings_dict(cls, settings: dict) -> Optional["MemberNotification"]:  # type: ignore[type-arg]
    communication_date = settings.get("employee_communication_scheduled_at")
    return (
        cls(
            notify_members_on_approved=settings["notify_members_on_approved"],
            notification_type=settings["notification_type"],
            employee_communication_scheduled_at=(
                datetime.strptime(communication_date, "%Y-%m-%d").date()
                if communication_date
                else None
            ),
        )
        if "notify_members_on_approved" in settings
        and "notification_type" in settings
        else None
    )

notification_type instance-attribute 

notification_type

notify_members_on_approved instance-attribute 

notify_members_on_approved

to_settings_dict 

to_settings_dict()
Source code in components/contracting/subcomponents/proposal/plugins/shared/fr/notification.py 
34
35
36
37
38
39
40
41
42
43
def to_settings_dict(self) -> dict:  # type: ignore[type-arg]
    return {
        "notify_members_on_approved": self.notify_members_on_approved,
        "notification_type": self.notification_type,
        "employee_communication_scheduled_at": (
            str(self.employee_communication_scheduled_at)
            if self.employee_communication_scheduled_at
            else None
        ),
    }

LightProposal dataclass 

LightProposal(
    id,
    created_at,
    creator_ref,
    creator_display_name,
    subscriptor_names,
    state,
    proposal_items,
    expire_after,
    finalizor_ref,
    finalizor_display_name,
    origin,
    tags,
    language,
    app_name,
    name=None,
    updated_at=None,
)

__hash__ 

__hash__()
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
98
99
def __hash__(self):  # type: ignore[no-untyped-def]
    return hash(self.id)

app_name instance-attribute 

app_name

created_at instance-attribute 

created_at

creator_display_name instance-attribute 

creator_display_name

creator_ref instance-attribute 

creator_ref

expire_after instance-attribute 

expire_after

finalizor_display_name instance-attribute 

finalizor_display_name

finalizor_ref instance-attribute 

finalizor_ref

from_model classmethod 

from_model(model)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def from_model(cls, model: "ProposalSQLA") -> "LightProposal":
    from components.contracting.external.subscriptor.api.subscriptor import (
        get_subscriptors,
    )

    app_name = AppName[AppName.get_key_from_value(model.app_name)]
    subscriptor_scope = [
        SubscriptorKey.from_dict(s) for s in model.subscriptor_scope
    ]
    subscriptor_legal_status = one_or_none(
        uniquify([s.legal_status for s in subscriptor_scope]),
        message=f"Subscriptors in the scope of proposal {model.id} should all have the same legal status",
    )
    subscriptors = (
        get_subscriptors(
            ids=[s.ref for s in subscriptor_scope],
            legal_status=subscriptor_legal_status,
            app_name=app_name,
        )
        if subscriptor_legal_status
        else []
    )
    # Could we store the whole subscriptors on LightProposal? We already query the DB to retrieve the names
    subscriptor_names = [s.name for s in subscriptors]

    return cls(
        id=model.id,
        name=model.name,
        state=ProposalState(model.state),
        creator_display_name=model.creator_display_name,
        creator_ref=model.creator_ref,
        created_at=model.created_at,
        subscriptor_names=subscriptor_names,
        updated_at=model.updated_at,
        finalizor_ref=model.finalizor_ref,
        finalizor_display_name=model.finalizor_display_name,
        expire_after=model.expire_after,
        proposal_items=list(map(ProposalItem.from_model, model.proposal_items)),
        origin=ProposalOrigin(model.origin),
        tags=model.tags or [],
        language=Lang(model.language),
        app_name=app_name,
    )

id instance-attribute 

id

language instance-attribute 

language

name class-attribute instance-attribute 

name = None

origin instance-attribute 

origin

plugins property 

plugins

Lists the various plugins involved in the proposal items.

proposal_items instance-attribute 

proposal_items

state instance-attribute 

state

subscriptor_names instance-attribute 

subscriptor_names

tags instance-attribute 

tags

updated_at class-attribute instance-attribute 

updated_at = None

PluginId

Bases: AlanBaseEnum

collective_retiree_individual_health_amendment_fr class-attribute instance-attribute 

collective_retiree_individual_health_amendment_fr = (
    "collective_retiree_individual_health_amendment_fr"
)

collective_retiree_individual_health_subscription_fr class-attribute instance-attribute 

collective_retiree_individual_health_subscription_fr = (
    "collective_retiree_individual_health_subscription_fr"
)

company_retiree_health_amendment_be class-attribute instance-attribute 

company_retiree_health_amendment_be = (
    "company_retiree_health_amendment_be"
)

company_retiree_health_subscription_be class-attribute instance-attribute 

company_retiree_health_subscription_be = (
    "company_retiree_health_subscription_be"
)

health_amendment_be class-attribute instance-attribute 

health_amendment_be = 'health_amendment_be'

health_amendment_es class-attribute instance-attribute 

health_amendment_es = 'health_amendment_es'

health_amendment_fr class-attribute instance-attribute 

health_amendment_fr = 'health_amendment_fr'

health_subscription_be class-attribute instance-attribute 

health_subscription_be = 'health_subscription_be'

health_subscription_es class-attribute instance-attribute 

health_subscription_es = 'health_subscription_es'

health_subscription_fr class-attribute instance-attribute 

health_subscription_fr = 'health_subscription_fr'

individual_health_amendment_be class-attribute instance-attribute 

individual_health_amendment_be = (
    "individual_health_amendment_be"
)

individual_health_amendment_fr class-attribute instance-attribute 

individual_health_amendment_fr = (
    "individual_health_amendment_fr"
)

individual_health_subscription_be class-attribute instance-attribute 

individual_health_subscription_be = (
    "individual_health_subscription_be"
)

individual_health_subscription_fr class-attribute instance-attribute 

individual_health_subscription_fr = (
    "individual_health_subscription_fr"
)

individual_retiree_health_amendment_be class-attribute instance-attribute 

individual_retiree_health_amendment_be = (
    "individual_retiree_health_amendment_be"
)

individual_retiree_health_subscription_be class-attribute instance-attribute 

individual_retiree_health_subscription_be = (
    "individual_retiree_health_subscription_be"
)

legacy_termination_fr class-attribute instance-attribute 

legacy_termination_fr = 'legacy_termination_fr'

prevoyance_amendment_fr class-attribute instance-attribute 

prevoyance_amendment_fr = 'prevoyance_amendment_fr'

prevoyance_subscription_fr class-attribute instance-attribute 

prevoyance_subscription_fr = 'prevoyance_subscription_fr'

Population dataclass 

Population(
    professional_category,
    ccn,
    activity_type=ActivityType.employed,
    is_apec_for_cadres=False,
    custom_beneficiary_clause_for_cadres=None,
    custom_beneficiary_clause_for_non_cadres=None,
)

Bases: DataClassJsonMixin

__hash__ 

__hash__()
Source code in components/contracting/utils/population.py 
105
106
def __hash__(self):  # type: ignore[no-untyped-def]
    return hash((self.professional_category, self.ccn_code, self.activity_type))

__repr__ 

__repr__()
Source code in components/contracting/utils/population.py 
108
109
def __repr__(self: "Population") -> str:
    return f"<Population professional_category={self.professional_category}, ccn_code={self.ccn_code}, activity_type={self.activity_type}>"

activity_type class-attribute instance-attribute 

activity_type = field(default=employed)

ccn class-attribute instance-attribute 

ccn = field(
    metadata=config(
        encoder=ccn_to_code, decoder=code_to_ccn
    )
)

ccn_code property 

ccn_code

custom_beneficiary_clause_for_cadres class-attribute instance-attribute 

custom_beneficiary_clause_for_cadres = field(default=None)

custom_beneficiary_clause_for_non_cadres class-attribute instance-attribute 

custom_beneficiary_clause_for_non_cadres = field(
    default=None
)

do_overlap 

do_overlap(other)
Source code in components/contracting/utils/population.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def do_overlap(self, other: "Population") -> bool:
    # Disjoint professional category (None is considered overlapping)
    if (
        self.professional_category
        in (ProfessionalCategory.cadres, ProfessionalCategory.non_cadres)
        and other.professional_category
        in (ProfessionalCategory.cadres, ProfessionalCategory.non_cadres)
        and self.professional_category != other.professional_category
    ):
        return False

    # Disjoint CCN (None is considered overlapping)
    if self.ccn and other.ccn and self.ccn.code != other.ccn.code:
        return False

    # Disjoint activity type
    if self.activity_type != other.activity_type:
        return False

    return True

friendly_name property 

friendly_name

from_ccn_code classmethod 

from_ccn_code(
    professional_category,
    ccn_code,
    activity_type=None,
    is_apec_for_cadres=False,
)
Source code in components/contracting/utils/population.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@classmethod
def from_ccn_code(
    cls,
    professional_category: ProfessionalCategory | None,
    ccn_code: str | None,
    activity_type: ActivityType | None = None,
    is_apec_for_cadres: bool = False,
) -> "Population":
    return cls(
        professional_category=professional_category,
        is_apec_for_cadres=is_apec_for_cadres,
        ccn=code_to_ccn(ccn_code),
        **{"activity_type": activity_type} if activity_type else {},
    )

from_contract_population classmethod 

from_contract_population(model)
Source code in components/contracting/utils/population.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
@classmethod
def from_contract_population(cls, model: "ContractPopulation") -> "Population":
    is_apec_for_cadres = False
    custom_beneficiary_clause_for_cadres = None
    custom_beneficiary_clause_for_non_cadres = None

    # ContractPopulation is a French model, so we know we're in the context of the French app
    # and we can ask for the APEC thing!
    if model.company_id is not None and model.professional_category is not None:
        from components.contracting.external.subscriptor.fr.company import (
            get_custom_beneficiary_clause,
            get_is_apec_for_cadres,
        )

        is_apec_for_cadres = get_is_apec_for_cadres(str(model.company_id))
        (
            custom_beneficiary_clause_for_cadres,
            custom_beneficiary_clause_for_non_cadres,
        ) = get_custom_beneficiary_clause(str(model.company_id))

    return cls(
        professional_category=model.professional_category,
        ccn=CCN.from_ccn(model.ccn),
        activity_type=(
            ActivityType.retired
            if model.is_collective_retirees_population
            else ActivityType.employed
        ),
        is_apec_for_cadres=is_apec_for_cadres,
        custom_beneficiary_clause_for_cadres=custom_beneficiary_clause_for_cadres,
        custom_beneficiary_clause_for_non_cadres=custom_beneficiary_clause_for_non_cadres,
    )

from_model classmethod 

from_model(model)
Source code in components/contracting/utils/population.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@classmethod
def from_model(cls, model: "ProposalItemTarget") -> "Population":
    is_apec_for_cadres = False
    custom_beneficiary_clause_for_cadres = None
    custom_beneficiary_clause_for_non_cadres = None

    if (
        model.app_name == AppName.ALAN_FR
        and model.subscriptor_legal_status == SubscriptorLegalStatus.company
        and model.subscriptor_ref is not None
    ):
        # Only for French companies we're gonna attempt to get the APEC setting
        from components.contracting.external.subscriptor.fr.company import (
            get_custom_beneficiary_clause,
            get_is_apec_for_cadres,
        )

        is_apec_for_cadres = get_is_apec_for_cadres(model.subscriptor_ref)
        (
            custom_beneficiary_clause_for_cadres,
            custom_beneficiary_clause_for_non_cadres,
        ) = get_custom_beneficiary_clause(model.subscriptor_ref)

    return cls(
        professional_category=model.professional_category,
        ccn=CCN.from_ccn(model.ccn),
        activity_type=model.activity_type,
        is_apec_for_cadres=is_apec_for_cadres,
        custom_beneficiary_clause_for_cadres=custom_beneficiary_clause_for_cadres,
        custom_beneficiary_clause_for_non_cadres=custom_beneficiary_clause_for_non_cadres,
    )

get_explicit_professional_category property 

get_explicit_professional_category

is_apec_for_cadres class-attribute instance-attribute 

is_apec_for_cadres = field(default=False)

is_collective_retiree property 

is_collective_retiree

is_same_professional_category 

is_same_professional_category(other)
Source code in components/contracting/utils/population.py 
301
302
303
304
305
306
307
308
309
310
def is_same_professional_category(self, other: str) -> bool:
    return (
        self.professional_category == other
        or (
            self.professional_category == ProfessionalCategory.all and other is None
        )
        or (
            self.professional_category is None and other == ProfessionalCategory.all
        )
    )

professional_category instance-attribute 

professional_category

professional_category_dutch_friendly_name property 

professional_category_dutch_friendly_name

professional_category_english_friendly_name property 

professional_category_english_friendly_name

professional_category_friendly_name property 

professional_category_friendly_name

professional_category_spanish_friendly_name property 

professional_category_spanish_friendly_name

professional_category_value property 

professional_category_value

Proposal dataclass 

Proposal(
    id,
    created_at,
    creator_ref,
    creator_display_name,
    state,
    proposal_items,
    approval_requests,
    expire_after,
    finalizor_ref,
    finalizor_display_name,
    lifecycle_notifications,
    origin,
    tags,
    errors,
    language,
    app_name,
    subscriptors_in_scope,
    name=None,
    updated_at=None,
    renewal_campaign_name=None,
)

__post_init__ 

__post_init__()
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
183
184
185
186
187
def __post_init__(self):  # type: ignore[no-untyped-def]
    if len(uniquify(s.legal_status for s in self.subscriptors_in_scope)) > 1:
        raise ValueError(
            f"Subscriptors in the scope of proposal {self.id} should all have the same legal status"
        )

all_targets property 

all_targets

allow_auto_approval property 

allow_auto_approval

app_name instance-attribute 

app_name

approval_requests instance-attribute 

approval_requests

can_be_approved property 

can_be_approved

can_be_cancelled property 

can_be_cancelled

can_be_edited property 

can_be_edited

can_be_finalized property 

can_be_finalized

can_be_tacitly_approved property 

can_be_tacitly_approved

Tacit approve is something defined for renewal

can_request_approval property 

can_request_approval

companies_in_scope property 

companies_in_scope

created_at instance-attribute 

created_at

creator_display_name instance-attribute 

creator_display_name

creator_ref instance-attribute 

creator_ref

errors instance-attribute 

errors

estimated_signed_pages_count property 

estimated_signed_pages_count

expire_after instance-attribute 

expire_after

finalizor_display_name instance-attribute 

finalizor_display_name

finalizor_ref instance-attribute 

finalizor_ref

from_model classmethod 

from_model(model)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
@classmethod
def from_model(cls, model: "ProposalSQLA") -> "Proposal":
    from components.contracting.external.subscriptor.api.subscriptor import (
        get_subscriptors,
    )

    app_name = AppName[AppName.get_key_from_value(model.app_name)]
    subscriptor_scope = [
        SubscriptorKey.from_dict(s) for s in model.subscriptor_scope
    ]
    subscriptor_legal_status = one_or_none(
        uniquify([s.legal_status for s in subscriptor_scope]),
        message=f"Subscriptors in the scope of proposal {model.id} should all have the same legal status",
    )
    subscriptors = (
        get_subscriptors(
            ids=[s.ref for s in subscriptor_scope],
            legal_status=subscriptor_legal_status,
            app_name=app_name,
        )
        if subscriptor_legal_status
        else []
    )

    return cls(
        id=model.id,
        name=model.name,
        state=ProposalState(model.state),
        creator_display_name=model.creator_display_name,
        creator_ref=model.creator_ref,
        created_at=model.created_at,
        updated_at=model.updated_at,
        finalizor_ref=model.finalizor_ref,
        finalizor_display_name=model.finalizor_display_name,
        expire_after=model.expire_after,
        proposal_items=list(map(ProposalItem.from_model, model.proposal_items)),
        approval_requests=list(
            map(
                ApprovalRequest.from_model,
                model.approval_requests,
            )
        ),
        lifecycle_notifications=(
            model.lifecycle_notifications if model.lifecycle_notifications else []
        ),
        origin=ProposalOrigin(model.origin),
        tags=model.tags or [],
        renewal_campaign_name=model.renewal_campaign_name,
        errors=(
            [ProposalError.from_model(x) for x in model.proposal_errors]
            if model.proposal_errors
            else []
        ),
        language=Lang(model.language),
        app_name=app_name,
        subscriptors_in_scope=list(subscriptors),
    )

from_offer_builder property 

from_offer_builder

get_approval_request 

get_approval_request(approval_request_id)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
439
440
441
442
443
def get_approval_request(self, approval_request_id: UUID) -> ApprovalRequest:
    for approval_request in self.approval_requests:
        if approval_request.id == approval_request_id:
            return approval_request
    raise Exception(f"Approval request with id {approval_request_id} not found")

get_approval_requests 

get_approval_requests(
    notification_types=None,
    is_live=None,
    approver_emails=None,
    reminder_scheduled_at=None,
    document_provider_type=None,
)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
def get_approval_requests(
    self,
    notification_types: Optional[set["ApprovalNotificationType"]] = None,
    is_live: Optional[bool] = None,
    approver_emails: Optional[set[str]] = None,
    reminder_scheduled_at: Optional[date] = None,
    document_provider_type: Optional[DocumentProviderType] = None,
) -> list["ApprovalRequest"]:
    return [
        approval_request
        for approval_request in self.approval_requests
        if (is_live is None or approval_request.is_live == is_live)
        and (
            notification_types is None
            or approval_request.notification_type in notification_types
        )
        and (
            approver_emails is None
            or approval_request.approver_email in approver_emails
        )
        and (
            reminder_scheduled_at is None
            or approval_request.reminder_scheduled_at == reminder_scheduled_at
        )
        and (
            document_provider_type is None
            or approval_request.document_provider_type == document_provider_type
        )
    ]

get_proposal_item 

get_proposal_item(proposal_item_id)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
427
428
429
430
431
432
433
def get_proposal_item(self, proposal_item_id: UUID) -> ProposalItem:
    for proposal_item in self.proposal_items:
        if proposal_item.id == proposal_item_id:
            return proposal_item
    raise Exception(
        f"Offer with id {proposal_item_id} not found in proposal {self.id}"
    )

get_proposal_items 

get_proposal_items(
    target_subscription_ref=None,
    target_subscription_type=None,
)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
def get_proposal_items(
    self,
    target_subscription_ref: Optional[str] = None,
    target_subscription_type: Optional[SubscriptionType] = None,
) -> list[ProposalItem]:
    return [
        proposal_item
        for proposal_item in self.proposal_items
        for target in proposal_item.targets
        if (
            target_subscription_ref is None
            or target.subscription_ref == target_subscription_ref
        )
        and (
            target_subscription_type is None
            or target.subscription_type == target_subscription_type
        )
    ]

get_proposal_items_for_plugin 

get_proposal_items_for_plugin(plugin_id)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
389
390
391
392
393
394
395
396
def get_proposal_items_for_plugin(
    self, plugin_id: "PluginId"
) -> list["ProposalItem"]:
    return [
        proposal_item
        for proposal_item in self.proposal_items
        if proposal_item.plugin_id == plugin_id
    ]

get_proposal_items_for_plugins 

get_proposal_items_for_plugins(plugin_ids)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
382
383
384
385
386
387
def get_proposal_items_for_plugins(
    self, plugin_ids: list["PluginId"]
) -> list[ProposalItem]:
    return flatten_list(
        self.get_proposal_items_for_plugin(pi) for pi in uniquify(plugin_ids)
    )

get_proposal_items_for_product_ref 

get_proposal_items_for_product_ref(product_ref)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
398
399
400
401
402
403
404
405
def get_proposal_items_for_product_ref(
    self, product_ref: str
) -> list[ProposalItem]:
    return [
        proposal_item
        for proposal_item in self.proposal_items
        if proposal_item.product.id == product_ref
    ]

get_targeted_subscription_refs 

get_targeted_subscription_refs(plugin_ids)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
247
248
249
250
251
def get_targeted_subscription_refs(self, plugin_ids: list["PluginId"]) -> list[str]:
    return uniquify(
        target.subscription_ref  # type: ignore[misc]
        for target in self.targets_for_plugins(plugin_ids=plugin_ids)
    )

get_targeted_subscriptor_refs 

get_targeted_subscriptor_refs(plugin_ids)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
241
242
243
244
245
def get_targeted_subscriptor_refs(self, plugin_ids: list["PluginId"]) -> list[str]:
    return uniquify(
        target.subscriptor_ref
        for target in self.targets_for_plugins(plugin_ids=plugin_ids)
    )

has_been_approved property 

has_been_approved

has_been_auto_approved property 

has_been_auto_approved

has_expired property 

has_expired

has_live_approval_request property 

has_live_approval_request

has_only_companies 

has_only_companies()
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
475
476
477
478
479
480
481
482
483
def has_only_companies(self) -> bool:
    has_only_companies = (
        len(self.companies_in_scope) > 0 and len(self.users_in_scope) == 0
    )
    if len(self.companies_in_scope) > 0 and len(self.users_in_scope) > 0:
        current_logger.error(
            f"Proposal {self.id} has both companies and individuals in scope"
        )
    return has_only_companies

has_only_individuals 

has_only_individuals()
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
485
486
487
488
489
490
491
492
493
def has_only_individuals(self) -> bool:
    has_only_individual = (
        len(self.users_in_scope) > 0 and len(self.companies_in_scope) == 0
    )
    if len(self.companies_in_scope) > 0 and len(self.users_in_scope) > 0:
        current_logger.error(
            f"Proposal {self.id} has both companies and individuals in scope"
        )
    return has_only_individual

has_plugin 

has_plugin(plugin_id)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
349
350
def has_plugin(self, plugin_id: "PluginId") -> bool:
    return plugin_id in self.plugin_ids

id instance-attribute 

id

is_approval_processing property 

is_approval_processing

A proposal is in approval processing if - it has an approval request that is flagged as signed from client (through the params client_signed_at which is set as soon as the client signs the approval request within our electronic signature provider) - and the proposal is not yet approved (document processed & subscription created / updated)

is_large property 

is_large

language instance-attribute 

language

lifecycle_notifications instance-attribute 

lifecycle_notifications

name class-attribute instance-attribute 

name = None

origin instance-attribute 

origin

plugin_ids property 

plugin_ids

Lists the ID of the various plugins involved in the proposal items

plugins property 

plugins

Lists the various plugins involved in the proposal items.

proposal_items instance-attribute 

proposal_items

proposal_items_by_plugin property 

proposal_items_by_plugin

renewal_campaign_name class-attribute instance-attribute 

renewal_campaign_name = None

short_id property 

short_id

state instance-attribute 

state

subscriptor_legal_status

subscriptors_in_scope instance-attribute 

subscriptors_in_scope

tags instance-attribute 

tags

targeted_account_refs property 

targeted_account_refs

targeted_company_refs property 

targeted_company_refs

targeted_subscriptor_refs property 

targeted_subscriptor_refs

targeted_user_refs property 

targeted_user_refs

targets_for_plugins 

targets_for_plugins(plugin_ids)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal.py 
224
225
226
227
228
229
230
231
def targets_for_plugins(self, plugin_ids: list["PluginId"]) -> list["Target"]:
    return [
        target
        for proposal_item in self.get_proposal_items_for_plugins(
            plugin_ids=plugin_ids
        )
        for target in proposal_item.targets
    ]

unseen_errors property 

unseen_errors

updated_at class-attribute instance-attribute 

updated_at = None

users_in_scope property 

users_in_scope

ProposalItem dataclass 

ProposalItem(
    id,
    plugin_id,
    name,
    start_date,
    targets,
    product,
    proposal_id,
    settings,
    noncompliance_acknowledgement_link,
    created_at,
    updated_at,
    internal_metadata,
    origin,
    templates,
)

__post_init__ 

__post_init__()
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def __post_init__(self) -> None:
    # Compute employee communication preview from proposal item
    if (
        self.plugin.subscription_type is not None
        and self.plugin.plugin_id in employee_communication_related_plugins
        and self.start_date is not None
        and len(self.targets) > 0
    ):
        if self.plugin_id == PluginId.health_amendment_fr:
            from components.emailing.public.shoot import (
                get_preview_shoot_url,
            )

            preview_url = get_preview_shoot_url(
                template_name="employee_health_amendment_fr",
                segment_name="health_employee_segment",
                audience_settings={
                    "subscription_id": self.targets[0].subscription_ref,
                    "on_date": self.start_date.isoformat(),
                },
                template_settings={
                    "proposal_item_id": str(self.id),
                    "subscription_id": self.targets[0].subscription_ref,
                    "notification_type": self.settings.get(
                        "notification_type", "standard"
                    ),
                },
            )
        else:
            preview_url = (
                None  # TODO: @jean.saglio implement the preview in the new system
            )
        link_settings = self.internal_metadata.get("link_settings", {})
        link_settings["notify_members_on_approved"] = preview_url
        self.internal_metadata["link_settings"] = link_settings

    try:
        # Handle health participation in legacy settings
        settings = {**self.settings}
        participation = settings.pop("participation", None)
        cover_partner = settings.pop("cover_partner", False)
        cover_children = settings.pop("cover_children", False)
        if (
            "participation_primary" in settings
            and "participation_partner" in settings
            and "participation_children" in settings
        ):
            # Coaerce to Decimal
            for key in (
                "participation_primary",
                "participation_partner",
                "participation_children",
            ):
                settings[key] = Decimal(settings[key])

        elif participation is not None:
            # Infer from deprecated settings
            legacy_participation = Decimal(participation)
            settings["participation_primary"] = legacy_participation
            settings["participation_partner"] = (
                legacy_participation if cover_partner else Decimal(0)
            )
            settings["participation_children"] = (
                legacy_participation if cover_children else Decimal(0)
            )
        self.settings = settings

        # Handle prevoyance participation in legacy settings
        if (
            "participation_tranche_a" in self.settings
            and "participation_tranche_b_c" in self.settings
        ):
            # Coaerce participation from string to Decimal
            for key in ("participation_tranche_a", "participation_tranche_b_c"):
                self.settings[key] = Decimal(self.settings[key])

    except Exception as e:
        current_logger.debug(
            "Unable to handle participation settings",
            exc_info=e,
            plugin_id=self.plugin_id,
            proposal_id=self.proposal_id,
        )

created_at instance-attribute 

created_at

differences 

differences(other)

Returns a dictionary of differences between this proposal item and another one for the following fields: (name, start_date, product_ref, settings)

  • keys are the names of the fields that differ.
  • values are tuples representing on previous value and the new one.
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def differences(self, other: "ProposalItem") -> dict[str, tuple[Any, Any]]:
    """
    Returns a dictionary of differences between this proposal item and another one for the following fields:
    (name, start_date, product_ref, settings)

    - keys are the names of the fields that differ.
    - values are tuples representing on previous value and the new one.
    """
    differences: dict[str, tuple[Any, Any]] = {}
    if self.name != other.name:
        differences["name"] = (self.name, other.name)
    if self.start_date != other.start_date:
        differences["start_date"] = (self.start_date, other.start_date)
    if self.product.id != other.product.id:
        differences["product_ref"] = (self.product.id, other.product.id)
    settings_union_keys = {*self.settings.keys(), *other.settings.keys()}
    for key in settings_union_keys:
        if self.settings.get(key, None) != other.settings.get(key, None):
            differences[key] = (
                self.settings.get(key, None),
                other.settings.get(key, None),
            )
    return differences

does_target_individual_only property 

does_target_individual_only

Returns True if all targets are about individual only.

estimated_signed_pages_count property 

estimated_signed_pages_count

Returns the estimated number of pages that will be generated for this proposal item to be signed.

excluded_templates property 

excluded_templates

from_model classmethod 

from_model(model)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
@classmethod
def from_model(cls, model: "ProposalItemSQLA") -> "ProposalItem":
    from components.contracting.subcomponents.proposal.plugins.shared.common.helpers.templating import (
        contracting_plugin_for,
    )

    plugin = contracting_plugin_for(plugin_id=model.plugin_id)
    product = one(
        plugin.get_products_by_ref(model.product_ref),
        f"Item can't be loaded because the selected product {model.product_ref} does not exist",
    )
    targets = [
        Target.from_model(model=t, subscription_type=plugin.subscription_type)
        for t in model.targets
    ]
    all_templates_by_type_and_language = {
        (t.type, t.language): t
        for t in plugin.get_templates_for_settings(
            product_ref=product.id,
            item_settings=model.settings,
            targets=targets,
            item_metadata=model.internal_metadata,
        )
    }

    templates = []
    for t in model.templates:
        # All information are not stored in the model, so we need to merge the model with the template
        std_template: Template | None = all_templates_by_type_and_language.get(
            (t.template_type, t.language), None
        )
        if not std_template:
            current_logger.error(
                f"No template found for template_type {t.template_type}",
                proposal_id=model.proposal_id,
                template_type=t.template_type,
                language=t.language,
                proposal_item_id=model.id,
            )
            continue

        templates.append(
            attrs.evolve(std_template, custom_file_uri=t.custom_template_uri)
        )

    return cls(
        id=model.id,
        plugin_id=PluginId(model.plugin_id),
        name=model.name,
        created_at=model.created_at,
        updated_at=model.updated_at,
        product=product,
        start_date=model.start_date,
        settings=model.settings,
        proposal_id=model.proposal_id,
        targets=targets,
        noncompliance_acknowledgement_link=model.noncompliance_acknowledgement_link,
        internal_metadata=model.internal_metadata,
        origin=ProposalOrigin(model.proposal.origin),
        templates=templates,
    )

get_target 

get_target(subscription_ref)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
257
258
259
260
261
def get_target(self, subscription_ref: str) -> Target:
    return one(
        [t for t in self.targets if t.subscription_ref == subscription_ref],
        f"Unable to find target for subscription {subscription_ref}",
    )

get_targets 

get_targets(subscriptor_ref)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
254
255
def get_targets(self, subscriptor_ref: str) -> list[Target]:
    return [t for t in self.targets if t.subscriptor_ref == subscriptor_ref]

get_template_by_type 

get_template_by_type(template_type)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
237
238
def get_template_by_type(self, template_type: str) -> Optional["Template"]:
    return one_or_none([t for t in self.templates if t.type == template_type])

get_templates 

get_templates(
    documents_types=(
        DocumentsType.unsigned_documents,
        DocumentsType.signed_documents,
    ),
    template_types=None,
    languages=None,
)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_item.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def get_templates(
    self,
    documents_types: Sequence["DocumentsType"] = (
        DocumentsType.unsigned_documents,
        DocumentsType.signed_documents,
    ),
    template_types: Optional[set[str]] = None,
    languages: Optional[set[Lang]] = None,
) -> list["Template"]:
    templates: list[Template] = []

    if DocumentsType.signed_documents in documents_types:
        templates += [t for t in self.templates if t.should_be_signed]

    if DocumentsType.unsigned_documents in documents_types:
        templates += [t for t in self.templates if not t.should_be_signed]

    return [
        t
        for t in templates
        if template_types is None or t.type in template_types
        if languages is None or t.language in languages
    ]

id instance-attribute 

id

internal_metadata instance-attribute 

internal_metadata

is_large property 

is_large

Defines when the proposal item is considered as large. This implies that some action might require to be done async.

is_new_subscription property 

is_new_subscription

name instance-attribute 

name

noncompliance_acknowledgement_link

origin instance-attribute 

origin

plugin property 

plugin

plugin_id instance-attribute 

plugin_id

product instance-attribute 

product

proposal_id instance-attribute 

proposal_id

settings instance-attribute 

settings

short_id property 

short_id

should_be_approved_before property 

should_be_approved_before

Returns the date before which the proposal item should be approved.

start_date instance-attribute 

start_date

subscription_refs property 

subscription_refs

targets instance-attribute 

targets

templates instance-attribute 

templates

updated_at instance-attribute 

updated_at

ProposalOrigin

Bases: AlanBaseEnum

alan_served class-attribute instance-attribute 

alan_served = 'alan_served'

friendly_name property 

friendly_name

is_renewal classmethod 

is_renewal(value)
Source code in components/contracting/subcomponents/proposal/api/entities/proposal_origin.py 
30
31
32
33
34
35
36
@classmethod
def is_renewal(cls, value: Union[str, "ProposalOrigin"]) -> bool:
    return (
        cls(value) == cls.renewal2022
        or cls(value) == cls.renewal2023
        or cls(value) == cls.renewal_bot
    )

renewal2022 class-attribute instance-attribute 

renewal2022 = 'renewal2022'

renewal2023 class-attribute instance-attribute 

renewal2023 = 'renewal2023'

renewal_bot class-attribute instance-attribute 

renewal_bot = 'renewal_bot'

self_served class-attribute instance-attribute 

self_served = 'self_served'

ProposalState

Bases: AlanBaseEnum

active_states staticmethod 

active_states()

Active states are composed of all states not trashed or archived, ie: not in the bin

Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
106
107
108
109
110
111
@staticmethod
def active_states() -> set["ProposalState"]:
    """
    Active states are composed of all states not trashed or archived, ie: not in the bin
    """
    return set(state for state in ProposalState) - ProposalState.archived_states()

after_waiting_for_approval_states staticmethod 

after_waiting_for_approval_states()
Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
91
92
93
94
95
96
97
@staticmethod
def after_waiting_for_approval_states() -> set["ProposalState"]:
    return {
        ProposalState.waiting_for_approval,
        ProposalState.approved,
        ProposalState.auto_approved,
    }

approved class-attribute instance-attribute 

approved = 'approved'

approved_states staticmethod 

approved_states()
Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
67
68
69
70
71
72
@staticmethod
def approved_states() -> set["ProposalState"]:
    return {
        ProposalState.approved,
        ProposalState.auto_approved,
    }

archived_states staticmethod 

archived_states()
Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
 99
100
101
102
103
104
@staticmethod
def archived_states() -> set["ProposalState"]:
    return {
        ProposalState.expired,
        ProposalState.cancelled,
    }

auto_approved class-attribute instance-attribute 

auto_approved = 'auto_approved'

blocking_states staticmethod 

blocking_states()

Blocking states are states that prevent other proposals with the same targets from being finalized.

Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
113
114
115
116
117
118
119
120
121
@staticmethod
def blocking_states() -> set["ProposalState"]:
    """
    Blocking states are states that prevent other proposals with the same targets from being finalized.
    """
    return {
        ProposalState.waiting_for_approval,
        ProposalState.finalized,
    }

cancelled class-attribute instance-attribute 

cancelled = 'cancelled'

expired class-attribute instance-attribute 

expired = 'expired'

finalized class-attribute instance-attribute 

finalized = 'finalized'

is_approved staticmethod 

is_approved(state)
Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
60
61
62
63
64
65
@staticmethod
def is_approved(state: "ProposalState") -> bool:
    return state in [
        ProposalState.approved,
        ProposalState.auto_approved,
    ]

non_terminal_states staticmethod 

non_terminal_states()
Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
83
84
85
86
87
88
89
@staticmethod
def non_terminal_states() -> set["ProposalState"]:
    return {
        ProposalState.started,
        ProposalState.finalized,
        ProposalState.waiting_for_approval,
    }

started class-attribute instance-attribute 

started = 'started'

terminal_states staticmethod 

terminal_states()
Source code in components/contracting/subcomponents/proposal/internals/models/proposal.py 
74
75
76
77
78
79
80
81
@staticmethod
def terminal_states() -> set["ProposalState"]:
    return {
        ProposalState.expired,
        ProposalState.approved,
        ProposalState.auto_approved,
        ProposalState.cancelled,
    }

waiting_for_approval class-attribute instance-attribute 

waiting_for_approval = 'waiting_for_approval'

SubscriptionType

Bases: AlanBaseEnum

DEPRECATED_mind class-attribute instance-attribute 

DEPRECATED_mind = 'mind'

dutch_friendly_name property 

dutch_friendly_name

english_friendly_name property 

english_friendly_name

friendly_name property 

friendly_name

health_insurance class-attribute instance-attribute 

health_insurance = 'health_insurance'

prevoyance class-attribute instance-attribute 

prevoyance = 'prevoyance'

spanish_friendly_name property 

spanish_friendly_name

to_contract_type 

to_contract_type()
Source code in components/contracting/external/subscription/api/entities/subscription_type.py 
56
57
58
59
60
61
62
def to_contract_type(self) -> ContractType:
    if self == SubscriptionType.health_insurance:
        return ContractType.health
    elif self == SubscriptionType.prevoyance:
        return ContractType.prevoyance
    else:
        raise ValueError(f"Unknown offer category {self}")

SubscriptorKey dataclass 

SubscriptorKey(ref, legal_status)

Bases: DataClassJsonMixin

is_company property 

is_company

is_individual property 

is_individual

legal_status instance-attribute 

legal_status

ref instance-attribute 

ref

SubscriptorLegalStatus

Bases: AlanBaseEnum

company class-attribute instance-attribute 

company = 'company'

individual class-attribute instance-attribute 

individual = 'individual'

Target dataclass 

Target(
    subscriptor_ref,
    subscriptor_legal_status,
    account_ref,
    subscription_ref,
    subscription_type,
    population,
    app_name,
    metadata=dict(),
)

Bases: DataClassJsonMixin

__repr__ 

__repr__()
Source code in components/contracting/utils/types.py 
51
52
def __repr__(self: "Target") -> str:
    return f"<Target subscriptor_ref={self.subscriptor_ref}, subscriptor_legal_status={self.subscriptor_legal_status}, subscription_ref={self.subscription_ref}, subscription_type={self.subscription_type}, population={self.population}, app_name={self.app_name}>"

account_ref instance-attribute 

account_ref

app_name instance-attribute 

app_name

company_subscriptor property 

company_subscriptor

from_model classmethod 

from_model(model, subscription_type)
Source code in components/contracting/utils/types.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def from_model(
    cls,
    model: "ProposalItemTargetSQLA",
    subscription_type: Optional[SubscriptionType],
) -> "Target":
    return cls(
        subscription_ref=model.subscription_ref,
        # ignoring model.subscription_type for retro-compatibility
        # subscription_type usually comes from the proposal_item.plugin
        subscription_type=subscription_type,
        population=Population.from_model(model),
        metadata=model.internal_metadata,
        account_ref=model.account_ref,
        subscriptor_ref=model.subscriptor_ref,
        subscriptor_legal_status=model.subscriptor_legal_status,
        app_name=model.app_name,
    )

from_subscription classmethod 

from_subscription(subscription)
Source code in components/contracting/utils/types.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@classmethod
def from_subscription(cls, subscription: "Subscription") -> "Target":
    return cls(
        subscription_ref=subscription.id,
        subscription_type=subscription.subscription_type,
        subscriptor_legal_status=subscription.contractee.legal_status,
        subscriptor_ref=subscription.contractee.id,
        account_ref=(
            str(mandatory(subscription.company.account_id))
            if subscription.is_for_company
            else None
        ),
        population=subscription.population,
        metadata={},
        app_name=subscription.contractee.app_name,
    )

from_subscription_period classmethod 

from_subscription_period(subscription_period)
Source code in components/contracting/utils/types.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@classmethod
def from_subscription_period(
    cls, subscription_period: "SubscriptionPeriod"
) -> "Target":
    return cls(
        subscription_ref=subscription_period.subscription_id,
        subscription_type=subscription_period.subscription_type,
        subscriptor_legal_status=subscription_period.contractee.legal_status,
        subscriptor_ref=subscription_period.contractee.id,
        account_ref=(
            str(mandatory(subscription_period.company.account_id))
            if subscription_period.is_for_company
            else None
        ),
        population=subscription_period.population,
        metadata={},
        app_name=subscription_period.contractee.app_name,
    )

is_amendment property 

is_amendment

is_company property 

is_company

is_individual property 

is_individual

is_new_subscription property 

is_new_subscription

metadata class-attribute instance-attribute 

metadata = field(
    default_factory=dict, hash=False, compare=False
)

population instance-attribute 

population

subscription_ref instance-attribute 

subscription_ref

subscription_type instance-attribute 

subscription_type

subscriptor property 

subscriptor

subscriptor_legal_status

subscriptor_ref instance-attribute 

subscriptor_ref

user_subscriptor property 

user_subscriptor

TargetOriginProposal dataclass 

TargetOriginProposal(
    proposal_id, target_ref, target_type, metadata
)

from_model classmethod 

from_model(model)
Source code in components/contracting/subcomponents/proposal/api/entities/target_origin_proposal.py 
35
36
37
38
39
40
41
42
@classmethod
def from_model(cls, model: "TargetOriginProposalSQLA") -> "TargetOriginProposal":
    return cls(
        proposal_id=model.proposal_id,
        target_ref=model.target_ref,
        target_type=model.target_type,
        metadata=model.internal_metadata,
    )

metadata instance-attribute 

metadata

proposal_id instance-attribute 

proposal_id

target_ref instance-attribute 

target_ref

target_type instance-attribute 

target_type

TargetOriginProposalMetadata dataclass 

TargetOriginProposalMetadata(origin, is_tacit)

Bases: DataClassJsonMixin

is_tacit instance-attribute 

is_tacit

origin instance-attribute 

origin

TargetType

Bases: AlanBaseEnum

health_contract class-attribute instance-attribute 

health_contract = 'health_contract'

health_contract_version class-attribute instance-attribute 

health_contract_version = 'health_contract_version'

prevoyance_contract class-attribute instance-attribute 

prevoyance_contract = 'prevoyance_contract'

prevoyance_contract_version class-attribute instance-attribute 

prevoyance_contract_version = 'prevoyance_contract_version'

ValidationContext dataclass 

ValidationContext(
    should_bypass_all_warnings=False,
    warnings_to_bypass=list(),
    warnings_to_consider=list(),
    should_bypass_all_blockers=False,
    blockers_to_bypass=list(),
    noncompliance_acknowledgement_link=None,
    should_persist_errors=True,
    mark_blockers_as_seen=tuple(),
    mark_warnings_as_seen=tuple(),
)

Bases: DataClassJsonMixin

This class is used to configure how the validation warnings & blockers should be handled. You can define which warnings & blockers should be ignored using their code, or if all of them should be ignored. Blockers can't be ignored in production mode, this is only provided to made testing easier.

__post_init__ 

__post_init__()
Source code in components/contracting/utils/validation.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __post_init__(self) -> None:
    if self.should_bypass_all_blockers and self.blockers_to_bypass:
        raise ValueError(
            "You can't bypass all blockers and specific errors at the same time"
        )
    if self.should_bypass_all_warnings and self.warnings_to_bypass:
        raise ValueError(
            "You can't bypass all warnings and specific warnings at the same time"
        )
    if not self.should_bypass_all_warnings and self.warnings_to_consider:
        raise ValueError(
            "You can't consider all warnings and specific warnings at the same time"
        )

    if not self.should_persist_errors and (
        self.mark_blockers_as_seen or self.mark_warnings_as_seen
    ):
        raise ValueError("You can't mark errors as seen if you don't persist them")

blockers_to_bypass class-attribute instance-attribute 

blockers_to_bypass = field(default_factory=list)

mark_blockers_as_seen class-attribute instance-attribute 

mark_blockers_as_seen = tuple()

mark_warnings_as_seen class-attribute instance-attribute 

mark_warnings_as_seen = tuple()

noncompliance_acknowledgement_link = None

should_bypass_all_blockers class-attribute instance-attribute 

should_bypass_all_blockers = False

should_bypass_all_warnings class-attribute instance-attribute 

should_bypass_all_warnings = False

should_ignore_message 

should_ignore_message(message)
Source code in components/contracting/utils/validation.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def should_ignore_message(self, message: ContractingMessage) -> bool:
    if (
        message.is_warning
        and (
            self.should_bypass_all_warnings
            or message.code in self.warnings_to_bypass
        )
        and message.code not in self.warnings_to_consider
    ):
        if message.require_acknowledgement_link:
            return self.noncompliance_acknowledgement_link is not None
        else:
            return True
    if message.is_blocker and (
        self.should_bypass_all_blockers or message.code in self.blockers_to_bypass
    ):
        return True
    return False

should_mark_as_seen 

should_mark_as_seen(message)
Source code in components/contracting/utils/validation.py 
59
60
61
62
63
64
65
66
67
def should_mark_as_seen(self, message: ContractingMessage) -> bool:
    if message.severity_level == ContractingMessageSeverityLevel.warning:
        return message.code in self.mark_warnings_as_seen

    elif message.severity_level == ContractingMessageSeverityLevel.blocker:
        return message.code in self.mark_blockers_as_seen

    else:
        raise ValueError("Unknown severity level")

should_persist_errors class-attribute instance-attribute 

should_persist_errors = True

warnings_to_bypass class-attribute instance-attribute 

warnings_to_bypass = field(default_factory=list)

warnings_to_consider class-attribute instance-attribute 

warnings_to_consider = field(default_factory=list)

add_proposal_item 

add_proposal_item(
    proposal_id,
    plugin_id,
    name,
    targets,
    product_ref,
    start_date,
    settings,
    templates=None,
    internal_metadata=None,
    validation_context=None,
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
@contracting_tracer_wrap()
def add_proposal_item(
    proposal_id: UUID,
    plugin_id: PluginId,
    name: Optional[str],
    targets: list[Target],
    product_ref: str,
    start_date: date,
    settings: dict,  # type: ignore[type-arg]
    templates: Optional[list["Template"]] = None,
    internal_metadata: Optional[dict] = None,  # type: ignore[type-arg]
    validation_context: Optional[ValidationContext] = None,
    commit: bool = True,
) -> ProposalItem:
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    proposal_item = add_item_to_proposal(
        proposal=proposal,
        plugin_id=plugin_id,
        name=name,
        targets=targets,
        product_ref=product_ref,
        start_date=start_date,
        settings=settings,
        templates=templates,
        validation_context=validation_context,
        internal_metadata=internal_metadata,
    )

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

    proposal_item_entity = ProposalItem.from_model(proposal_item)

    current_logger.info(
        "A new proposal_item has been created",
        proposal_item_id=proposal_item_entity.id,
        proposal_id=proposal.id,
        templates=proposal_item_entity.templates,
        commit=commit,
    )

    return proposal_item_entity

append_activity 

append_activity(
    approval_request_id,
    message,
    link=None,
    created_at=None,
    logged_username=None,
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
@contracting_tracer_wrap()
def append_activity(
    approval_request_id: UUID,
    message: str,
    link: Optional[str] = None,
    created_at: Optional[datetime] = None,
    logged_username: Optional[str] = None,
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.proposal.internals.approval import (
        append_activity as append_activity_bl,
    )

    new_activity = append_activity_bl(
        approval_request_id=approval_request_id,
        message=message,
        link=link,
        created_at=created_at,
        logged_username=logged_username,
    )

    if new_activity is None:
        return

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

    current_logger.info(
        "Approval activity added",
        approval_request_id=approval_request_id,
        activity=new_activity,
    )

approve_proposal 

approve_proposal(
    proposal_id, approval_request_id, commit=True
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
@contracting_tracer_wrap()
@enqueueable
def approve_proposal(
    proposal_id: UUID,
    approval_request_id: UUID,
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        approve,
    )

    with AdvisoryLock(_proposal_approval_advisory_lock_key(proposal_id)):
        proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)
        approval_request = get_or_raise_missing_resource(
            ApprovalRequestSQLA, approval_request_id
        )

        if approval_request.document_provider_type == DocumentProviderType.hellosign:
            signed_bundle = approval_request.signed_bundle
            if signed_bundle and signed_bundle.is_signed:
                append_activity(
                    approval_request.id,
                    "Documents signed and attached to proposal",
                    commit=False,
                )
                current_logger.info(
                    "Proposal's documents have been signed", proposal_id=proposal_id
                )
            else:
                append_activity(
                    approval_request.id,
                    "Documents have not been signed, unsigned documents will be attached to proposal",
                    commit=False,
                )

        with side_effects(trigger_on_exit=commit):
            approve(
                proposal=proposal,
                approval_request=approval_request,
            )
            if commit:
                current_session.commit()
            else:
                current_session.flush()
        current_logger.info(
            f"Proposal {proposal.id} approved",
            proposal_id=proposal.id,
            approval_request_id=approval_request.id,
            # Time between actual client signature (client_signed_at) and now
            signature_processing_time_in_sec=(
                (datetime.now() - approval_request.client_signed_at).total_seconds()
                if approval_request.client_signed_at
                else None
            ),
        )

assign_documents_to_request 

assign_documents_to_request(
    approval_request_id,
    proposal_item_id,
    documents,
    target,
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
def assign_documents_to_request(
    approval_request_id: UUID,
    proposal_item_id: UUID,
    documents: list[Document],
    target: Target,
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.proposal.internals.document_providers.manual_provider import (
        ManualDocumentProvider,
    )
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )

    approval_request: ApprovalRequestSQLA = get_or_raise_missing_resource(
        ApprovalRequestSQLA, approval_request_id
    )

    if approval_request.document_provider_type != DocumentProviderType.manual:
        raise ValueError(
            "Assigning documents allowed only for manual document provider type"
        )

    ManualDocumentProvider().assign_documents(
        proposal_item_id=proposal_item_id,
        approval_request=approval_request,
        documents=documents,
        target=target,
    )

    append_activity(
        approval_request.id,
        f"{len(documents)} documents have been manually assigned to this approval request",
        commit=False,
    )

    if commit:
        current_session.commit()
    else:
        current_session.flush()
        current_session.expire(approval_request, attribute_names=["manual_documents"])

auto_approve_proposal 

auto_approve_proposal(
    proposal_id, approval_request_id, commit=True
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
@contracting_tracer_wrap()
def auto_approve_proposal(
    proposal_id: UUID,
    approval_request_id: UUID,
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        auto_approve,
    )

    with AdvisoryLock(_proposal_approval_advisory_lock_key(proposal_id)):
        proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)
        approval_request = get_or_raise_missing_resource(
            ApprovalRequestSQLA, approval_request_id
        )

        if approval_request.approved_at:
            current_logger.info(
                f"Proposal {proposal.id} already approved",
                proposal_id=proposal.id,
                approval_request_id=approval_request.id,
                commit=commit,
            )
            return

        with side_effects(trigger_on_exit=commit):
            auto_approve(
                proposal=proposal,
                approval_request=approval_request,
            )
            if commit:
                current_session.commit()
            else:
                current_session.flush()

        current_logger.info(
            f"Proposal {proposal.id} auto-approved",
            proposal_id=proposal.id,
            approval_request_id=approval_request.id,
            commit=commit,
        )

build_subscriptions_from_proposal_items 

build_subscriptions_from_proposal_items(
    proposal_items, target=None, signature_status=None
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
@contracting_tracer_wrap()
def build_subscriptions_from_proposal_items(
    proposal_items: list[ProposalItem],
    target: Optional["Target"] = None,
    signature_status: SignatureStatus | None = None,
) -> list[Subscription]:
    from components.contracting.subcomponents.proposal.internals.subscription import (
        build_subscriptions_for_proposal_items as build_subscriptions_for_proposal_items_bl,
    )

    current_logger.info(
        f"Building subscriptions for {len(proposal_items)} items",
        proposal_items=proposal_items,
        target=target,
    )
    return build_subscriptions_for_proposal_items_bl(
        proposal_items=proposal_items, target=target, signature_status=signature_status
    )

cancel_proposal 

cancel_proposal(
    proposal_id,
    withdraw_approval_requests=False,
    commit=True,
)

Flag the proposal as cancelled. It's terminal state and no further action can be done on it.

:param proposal_id: proposal id to cancel :param withdraw_approval_requests: automatically withdraw all approval requests :param commit: commit the changes

Source code in components/contracting/subcomponents/proposal/api/main.py 
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
@contracting_tracer_wrap()
def cancel_proposal(
    proposal_id: UUID,
    withdraw_approval_requests: bool = False,
    commit: bool = True,
) -> None:
    """
    Flag the proposal as cancelled. It's terminal state and no further action can be done on it.

    :param proposal_id: proposal id to cancel
    :param withdraw_approval_requests: automatically withdraw all approval requests
    :param commit: commit the changes
    """
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        cancel_proposal as cancel_proposal_bl,
    )

    proposal: ProposalSQLA = get_or_raise_missing_resource(ProposalSQLA, proposal_id)
    cancel_proposal_bl(
        proposal=proposal, withdraw_approval_requests=withdraw_approval_requests
    )
    if commit:
        current_session.commit()
    else:
        current_session.flush()

    current_logger.info(
        "Proposal cancelled",
        proposal=proposal,
        proposal_id=proposal.id,
        withdraw_approval_requests=withdraw_approval_requests,
        commit=commit,
    )

create_proposal 

create_proposal(
    creator_ref,
    creator_display_name,
    subscriptor_scope_refs,
    subscriptor_scope_legal_status,
    name,
    lifecycle_notifications,
    origin=None,
    tags=None,
    commit=True,
    language=Lang.french,
    app_name=AppName.ALAN_FR,
    cancel_after=None,
    renewal_campaign_name=None,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
@contracting_tracer_wrap()
def create_proposal(
    creator_ref: str,
    creator_display_name: str,
    subscriptor_scope_refs: list[str],
    subscriptor_scope_legal_status: SubscriptorLegalStatus,
    name: Optional[str],
    lifecycle_notifications: list[LifecycleNotification],
    origin: Optional[ProposalOrigin] = None,
    tags: Optional[list[str]] = None,
    commit: bool = True,
    language: Lang = Lang.french,
    app_name: AppName = AppName.ALAN_FR,
    cancel_after: Optional[date] = None,
    renewal_campaign_name: Optional[str] = None,
) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.proposal import (
        create_proposal as create_proposal_bl,
    )

    proposal = create_proposal_bl(
        creator_ref=creator_ref,
        creator_display_name=creator_display_name,
        subscriptor_scope_refs=subscriptor_scope_refs,
        subscriptor_scope_legal_status=subscriptor_scope_legal_status,
        name=name,
        expire_after=None,
        cancel_after=cancel_after,
        lifecycle_notifications=lifecycle_notifications,
        origin=origin or ProposalOrigin.alan_served,
        tags=tags or [],
        language=language,
        app_name=app_name,
        renewal_campaign_name=renewal_campaign_name,
    )

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

    current_logger.info(
        "Proposal created",
        proposal=proposal,
        commit=commit,
        proposal_id=proposal.id,
        tags=tags,
        app_name=app_name,
    )

    return Proposal.from_model(proposal)

default_lifecycle_notifications 

default_lifecycle_notifications(
    creator_ref=None, app_name=None
)

Default lifecycle notifications for a proposal. Slack integration is enabled by default using the proposal creator user firstname.lastname (not robust)

Source code in components/contracting/subcomponents/proposal/views/create_proposal.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def default_lifecycle_notifications(
    creator_ref: str | None = None,
    app_name: AppName | None = None,
) -> list[LifecycleNotification]:
    """
    Default lifecycle notifications for a proposal.
    Slack integration is  enabled by default using the proposal creator user firstname.lastname (not robust)
    """
    alan_email = None
    slack_handle = None
    current_user = getattr(g, "current_user", None)

    # Figure out the alaner email from the creator_ref
    if current_user:
        if current_user.alan_employee:
            alan_email = current_user.alan_employee.alan_email
    elif creator_ref and app_name:
        if alaners := list_alaners(user_ids=[creator_ref], app_name=app_name):
            alan_email = alaners[0].alan_email

    if alan_email:
        slack_handle = alan_email.split("@")[0]
    elif current_user:
        slack_handle = (
            f"{current_user.first_name.lower()}.{current_user.last_name.lower()}"
        )

    if slack_handle:
        return [
            LifecycleNotification(
                integration_type=LifecycleNotificationIntegrationType.slack,
                value=slack_handle,
            )
        ]
    else:
        return []

delete_proposal_item 

delete_proposal_item(proposal_item_id, commit=True)
Source code in components/contracting/subcomponents/proposal/api/main.py 
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
@contracting_tracer_wrap()
def delete_proposal_item(proposal_item_id: UUID, commit: bool = True) -> ProposalItem:
    from components.contracting.subcomponents.proposal.internals.models.proposal_item import (
        ProposalItem as ProposalItemSQLA,
    )

    proposal_item: ProposalItemSQLA = get_or_raise_missing_resource(
        ProposalItemSQLA, proposal_item_id
    )
    delete_item_from_proposal(proposal_item.proposal, proposal_item)
    if commit:
        current_session.commit()
    current_logger.info(
        "A proposal_item has been removed",
        proposal_item=proposal_item,
        proposal_id=proposal_item.proposal_id,
    )
    return ProposalItem.from_model(proposal_item)

duplicate_proposal 

duplicate_proposal(
    proposal_id,
    creator_ref,
    creator_display_name,
    origin,
    commit=True,
)

Duplicate a proposal and all its items. Notice that target metadata are not duplicated.

Source code in components/contracting/subcomponents/proposal/api/main.py 
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
@contracting_tracer_wrap()
def duplicate_proposal(
    proposal_id: UUID,
    creator_ref: str,
    creator_display_name: str,
    origin: ProposalOrigin,
    commit: bool = True,
) -> Proposal:
    """
    Duplicate a proposal and all its items.
    Notice that target metadata are not duplicated.
    """
    NON_DUPLICABLE_TARGET_METADATA = ["prevoyance_contract_id", "health_contract_id"]
    proposal = get_proposal(proposal_id=proposal_id)
    new_proposal: Proposal = create_proposal(
        creator_ref=creator_ref,
        creator_display_name=creator_display_name,
        subscriptor_scope_refs=[s.id for s in proposal.subscriptors_in_scope],
        subscriptor_scope_legal_status=mandatory(proposal.subscriptor_legal_status),
        name=f"{proposal.name} (duplicate)",
        lifecycle_notifications=proposal.lifecycle_notifications,
        origin=origin,
        tags=proposal.tags,
        renewal_campaign_name=proposal.renewal_campaign_name,
        commit=False,
        app_name=proposal.app_name,
        language=proposal.language,
    )
    for item in proposal.proposal_items:
        add_proposal_item(
            proposal_id=new_proposal.id,
            targets=[
                dataclasses.replace(
                    target,
                    metadata={
                        k: v
                        for k, v in target.metadata.items()
                        if k not in NON_DUPLICABLE_TARGET_METADATA
                    },
                )
                for target in item.targets
            ],
            name=item.name,
            product_ref=item.product.id,
            start_date=mandatory(item.start_date),
            plugin_id=item.plugin_id,
            settings=item.settings,
            internal_metadata=item.internal_metadata,
            validation_context=ValidationContext(
                should_bypass_all_blockers=True,
                should_bypass_all_warnings=True,
                noncompliance_acknowledgement_link=(
                    item.noncompliance_acknowledgement_link or "duplicating"
                ),
            ),
            commit=False,
            templates=item.templates,
        )

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

    current_logger.info("Proposal duplicated", proposal_id=proposal_id)

    return new_proposal

edit_approval_request 

edit_approval_request(
    approval_request_id, reminder_sent_at=None
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
@contracting_tracer_wrap()
def edit_approval_request(
    approval_request_id: UUID,
    reminder_sent_at: Optional[datetime] = None,
) -> None:
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )

    approval_request: ApprovalRequestSQLA = get_or_raise_missing_resource(
        ApprovalRequestSQLA, approval_request_id
    )

    approval_request.reminder_sent_at = reminder_sent_at

    current_session.commit()

    current_logger.info(
        "Approval request edited",
        proposal_id=approval_request.proposal.id,
        approval_request_id=approval_request_id,
        reminder_sent_at=reminder_sent_at,
    )

edit_proposal 

edit_proposal(
    proposal_id,
    subscriptor_scope_refs,
    subscriptor_scope_legal_status,
    name,
    expire_after,
    lifecycle_notifications,
    language=Lang.french,
    tags=None,
    renewal_campaign_name=None,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
@contracting_tracer_wrap()
def edit_proposal(
    proposal_id: UUID,
    subscriptor_scope_refs: list[str],
    subscriptor_scope_legal_status: SubscriptorLegalStatus,
    name: Optional[str],
    expire_after: Optional[date],
    lifecycle_notifications: list[LifecycleNotification],
    language: Lang = Lang.french,
    tags: Optional[list[str]] = None,
    renewal_campaign_name: Optional[str] = None,
) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        update_proposal,
    )

    proposal: ProposalSQLA = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    update_proposal(
        proposal=proposal,
        subscriptor_scope_refs=subscriptor_scope_refs,
        subscriptor_scope_legal_status=subscriptor_scope_legal_status,
        name=name,
        expire_after=expire_after,
        lifecycle_notifications=lifecycle_notifications,
        tags=tags or [],
        language=language,
        renewal_campaign_name=renewal_campaign_name,
    )

    current_session.commit()
    current_logger.info(
        "Proposal item edited", proposal=proposal, proposal_id=proposal.id
    )
    return Proposal.from_model(proposal)

edit_proposal_item 

edit_proposal_item(
    proposal_item_id,
    name,
    targets,
    product_ref,
    start_date,
    settings,
    templates=None,
    internal_metadata=None,
    validation_context=None,
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
@contracting_tracer_wrap()
def edit_proposal_item(
    proposal_item_id: UUID,
    name: Optional[str],
    targets: list[Target],
    product_ref: str,
    start_date: date,
    settings: dict,  # type: ignore[type-arg]
    templates: Optional[list["Template"]] = None,
    internal_metadata: Optional[dict] = None,  # type: ignore[type-arg]
    validation_context: Optional[ValidationContext] = None,
    commit: bool = True,
) -> ProposalItem:
    from components.contracting.subcomponents.proposal.internals.models.proposal_item import (
        ProposalItem as ProposalItemSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal_item import (
        edit_item_in_proposal,
    )

    previous_proposal_item: ProposalItemSQLA = get_or_raise_missing_resource(
        ProposalItemSQLA, proposal_item_id
    )

    new_proposal_item = edit_item_in_proposal(
        proposal=previous_proposal_item.proposal,
        proposal_item=previous_proposal_item,
        name=name,
        targets=targets,
        product_ref=product_ref,
        start_date=start_date,
        settings=settings,
        validation_context=validation_context,
        internal_metadata=internal_metadata,
        templates=templates,
    )

    if commit:
        current_session.commit()

    new_proposal_item_entity = ProposalItem.from_model(new_proposal_item)

    current_logger.info(
        "A proposal_item has been edited",
        proposal_item_id=new_proposal_item_entity.id,
        proposal_id=previous_proposal_item.proposal.id,
        templates=new_proposal_item_entity.templates,
    )
    return new_proposal_item_entity

edit_proposal_tags 

edit_proposal_tags(proposal_id, tags, commit=True)
Source code in components/contracting/subcomponents/proposal/api/main.py 
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
@contracting_tracer_wrap()
def edit_proposal_tags(
    proposal_id: UUID,
    tags: list[str],
    commit: bool = True,
) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        update_proposal_tags,
    )

    proposal: ProposalSQLA = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    update_proposal_tags(
        proposal=proposal,
        tags=tags,
    )

    if commit:
        current_session.commit()

    current_logger.info(
        "Proposal tags edited", proposal=proposal, proposal_id=proposal.id
    )
    return Proposal.from_model(proposal)

finalize_proposal 

finalize_proposal(
    proposal_id,
    finalizor_ref,
    finalizor_display_name,
    validation_context=None,
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
@contracting_tracer_wrap()
def finalize_proposal(
    proposal_id: UUID,
    finalizor_ref: str,
    finalizor_display_name: str,
    validation_context: Optional[ValidationContext] = None,
    commit: bool = True,
) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        finalize,
    )

    proposal: ProposalSQLA = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    with side_effects(trigger_on_exit=commit):
        finalize(
            proposal,
            finalizor_ref,
            finalizor_display_name,
            validation_context,
        )
        if commit:
            current_session.commit()
        else:
            current_session.flush()

    current_logger.info(
        "Proposal finalized", proposal=proposal, proposal_id=proposal.id
    )
    return Proposal.from_model(proposal)

forward_approval_request 

forward_approval_request(
    approval_request_id,
    approver_first_name,
    approver_last_name,
    approver_email,
    requester_id,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
@contracting_tracer_wrap()
@enqueueable
def forward_approval_request(
    approval_request_id: UUID,
    approver_first_name: str,
    approver_last_name: str,
    approver_email: str,
    requester_id: str,
) -> ApprovalRequest:
    from components.contracting.external.subscriptor.api.user import (
        get_users,
    )

    proposal = get_proposal_by_approval_request_id(approval_request_id)
    requester = one(get_users(user_ids=[requester_id], app_name=proposal.app_name))
    new_approval_request = request_approval(
        proposal_id=proposal.id,
        approver_first_name=approver_first_name,
        approver_last_name=approver_last_name,
        approver_email=approver_email,
        approver_ref=None,
        requester_id=requester.id,
        notification_type=ApprovalNotificationType.signature_email,
    )
    append_activity(
        approval_request_id,
        f"Approval request forwarded to #{new_approval_request.short_id}",
        logged_username=requester.name,
    )
    current_logger.info(
        "Approval request forwarded",
        approval_request_id=new_approval_request.id,
        proposal_id=proposal.id,
    )
    return new_approval_request

get_all_documents 

get_all_documents(
    proposal_id,
    documents_types=(DocumentsType.signed_documents,),
)

Best effort to retrieve all documents for an approved proposal. As - we do not store final unsigned document else where than in final subscription (we should) - we do not have any link between proposal and subscription (e.g amendment / prevoyance contract) We can only retrieve document based on the start date of the proposal items.

Source code in components/contracting/subcomponents/proposal/api/main.py 
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
@contracting_tracer_wrap()
def get_all_documents(
    proposal_id: UUID,
    documents_types: Sequence[DocumentsType] = (DocumentsType.signed_documents,),
) -> list[Document]:
    """
    Best effort to retrieve all documents for an approved proposal.
    As
      - we do not store final unsigned document else where than in final subscription (we should)
      - we do not have any link between proposal and subscription (e.g amendment / prevoyance contract)
    We can only retrieve document based on the start date of the proposal items.
    """
    from components.contracting.external.document.be.utils import (
        get_documents_for_period as get_documents_for_period_be,
    )
    from components.contracting.external.document.es.utils import (
        get_documents_for_period as get_documents_for_period_es,
    )
    from components.contracting.external.document.fr.utils import (
        get_documents_for_period as get_documents_for_period_fr,
    )

    proposal = get_proposal(proposal_id=proposal_id)
    if proposal.state not in (ProposalState.approved, ProposalState.auto_approved):
        raise Exception("Proposal must be approved to download signed documents")

    documents = []
    for item in proposal.proposal_items:
        subscriptions = get_subscriptions_for_targets(item.targets)
        for subscription in subscriptions:
            period = subscription.get_ongoing_period(mandatory(item.start_date))
            if not period:
                current_logger.warning(
                    "No period found for subscription - the subscription targeted by the proposal was likely terminated",
                    subscription_id=subscription.id,
                    proposal_item_id=item.id,
                )
                continue

            if proposal.app_name == AppName.ALAN_FR:
                documents.extend(
                    get_documents_for_period_fr(
                        period,
                        documents_types=documents_types,
                        templates=item.templates,
                    )
                )
            elif proposal.app_name == AppName.ALAN_ES:
                documents.extend(
                    get_documents_for_period_es(
                        period=period,
                        documents_types=documents_types,
                    )
                )
            elif proposal.app_name == AppName.ALAN_BE:
                documents.extend(
                    get_documents_for_period_be(
                        period=period,
                        documents_types=documents_types,
                    )
                )
            else:
                raise NotImplementedError(
                    f"Document retrieval not implemented for app {proposal.app_name}"
                )

    return documents

get_proposal 

get_proposal(proposal_id)
Source code in components/contracting/subcomponents/proposal/api/main.py 
398
399
400
401
402
403
404
405
@contracting_tracer_wrap()
def get_proposal(proposal_id: UUID | str) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal_sqla = get_or_raise_missing_resource(ProposalSQLA, proposal_id)
    return Proposal.from_model(proposal_sqla)

get_proposal_alert 

get_proposal_alert(
    admin_id, account_id, company_ids=None, **_
)

Return an alert when a proposal is waiting for approval by the admin.

Can be called with either: - company_id: returns alert for proposals of that company - account_id: returns alert for proposals of all companies in the account

At least one of company_id or account_id must be provided.

Source code in components/contracting/subcomponents/proposal/api/main.py 
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
def get_proposal_alert(
    admin_id: int,
    account_id: UUID,
    company_ids: Optional[list[int]] = None,
    **_: Any,
) -> Optional["Alert"]:
    """
    Return an alert when a proposal is waiting for approval by the admin.

    Can be called with either:
    - company_id: returns alert for proposals of that company
    - account_id: returns alert for proposals of all companies in the account

    At least one of company_id or account_id must be provided.
    """
    from components.contracting.subcomponents.proposal.api.entities.proposal import (
        LightProposal,
    )
    from components.contracting.subcomponents.renewal.public.contractee_renewal_operation import (
        is_admin_dashboard_renewal_experience_enabled_for_account,
    )
    from components.contracting.subcomponents.renewal.public.proposal import (
        is_alternative_renewal,
        is_manual_renewal,
        is_renewal,
    )
    from components.fr.internal.models.account import Account  # noqa: ALN069
    from components.fr.internal.models.user import User  # noqa: ALN069
    from shared.models.helpers.enum import AlanBaseEnum

    class ProposalAlertWording(AlanBaseEnum):
        base = "base"
        alternative_renewal = "alternative_renewal"
        manual_renewal = "manual_renewal"

    admin = get_or_raise_missing_resource(User, admin_id)

    if not admin.email:
        return None

    if company_ids:
        company_refs = [str(company_id) for company_id in company_ids]
    else:
        account = get_or_raise_missing_resource(Account, account_id)
        company_refs = [str(company.id) for company in account.companies]

    proposals: list[LightProposal] = paginate_proposals(
        page=1,
        per_page=len(
            company_refs
        ),  # Assumption: 1 proposal waiting approval at maximum per company
        company_refs=company_refs,
        states=[ProposalState.waiting_for_approval],
    ).items

    if not proposals:
        return None

    alert_parameters = []
    proposals_with_alerts = []
    for light_proposal in proposals:
        proposal = get_proposal(light_proposal.id)

        if proposal.is_approval_processing:
            continue

        # Check if renewal and admin dashboard renewal experience is enabled
        if is_renewal(proposal):
            if not is_admin_dashboard_renewal_experience_enabled_for_account(
                account_id=account_id
            ):
                continue

        approval_requests = proposal.get_approval_requests(
            approver_emails={m for m in [admin.email, admin.pro_email] if m},
            is_live=True,
            document_provider_type=DocumentProviderType.hellosign,
        )

        if approval_requests:
            approval_request = approval_requests[-1]
            alert_parameters.append(
                {
                    "proposal_id": proposal.id,
                    "approval_request_id": approval_request.id,
                    "approval_request_link": get_signature_link(
                        approval_request_id=str(approval_request.id),
                        app_name=proposal.app_name,
                    ),
                }
            )
            proposals_with_alerts.append(proposal)

    if not alert_parameters:
        return None

    if any(is_alternative_renewal(proposal) for proposal in proposals_with_alerts):
        alert_wording = ProposalAlertWording.alternative_renewal
    elif any(is_manual_renewal(proposal) for proposal in proposals_with_alerts):
        alert_wording = ProposalAlertWording.manual_renewal
    else:
        alert_wording = ProposalAlertWording.base

    return Alert(
        type=AlertType.proposal_waiting_for_approval,
        category=AlertCategory.contract,
        priority=AlertPriority.medium,
        parameters=alert_parameters,
        wording=alert_wording,
    )

get_proposal_by_approval_request_id 

get_proposal_by_approval_request_id(approval_request_id)
Source code in components/contracting/subcomponents/proposal/api/main.py 
465
466
467
468
469
470
471
472
473
474
475
@contracting_tracer_wrap()
def get_proposal_by_approval_request_id(approval_request_id: UUID) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )

    request: ApprovalRequestSQLA = get_or_raise_missing_resource(
        ApprovalRequestSQLA, approval_request_id
    )

    return get_proposal(request.proposal_id)

get_proposal_by_proposal_item_id 

get_proposal_by_proposal_item_id(proposal_item_id)
Source code in components/contracting/subcomponents/proposal/api/main.py 
495
496
497
498
499
500
501
502
503
504
@contracting_tracer_wrap()
def get_proposal_by_proposal_item_id(proposal_item_id: UUID) -> Proposal:
    from components.contracting.subcomponents.proposal.internals.models.proposal_item import (
        ProposalItem as ProposalItemSQLA,
    )

    proposal_item: ProposalItemSQLA = get_or_raise_missing_resource(
        ProposalItemSQLA, proposal_item_id
    )
    return get_proposal(proposal_item.proposal_id)

get_proposal_from_target 

get_proposal_from_target(target_ref, target_type)

Return the proposal from a target

Source code in components/contracting/subcomponents/proposal/api/main.py 
452
453
454
455
456
457
458
459
460
461
462
@contracting_tracer_wrap()
def get_proposal_from_target(
    target_ref: str, target_type: TargetType
) -> Proposal | None:
    """
    Return the proposal from a target
    """
    target = get_target_origin_proposal(target_ref, target_type)
    if target:
        return get_proposal(target.proposal_id)
    return None

get_requester_id_by_approval_request_id 

get_requester_id_by_approval_request_id(
    approval_request_id,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
@contracting_tracer_wrap()
def get_requester_id_by_approval_request_id(approval_request_id: UUID) -> str:
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )

    request: ApprovalRequestSQLA = get_or_raise_missing_resource(
        ApprovalRequestSQLA, approval_request_id
    )

    if not request.signed_bundle or not request.signed_bundle.requester_ref:
        # if the approval is a manual approval for example, this function should not be called
        raise ValueError(f"Approval request {approval_request_id} has no requester")

    return str(request.signed_bundle.requester_ref)

get_signature_link(
    approval_request_id, app_name, language=None
)

Get the link to the signature page for the given approval request.

Source code in components/contracting/subcomponents/proposal/api/main.py 
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
def get_signature_link(
    approval_request_id: str,
    app_name: AppName,  # TODO: remove this parameter when the feature is enabled for all apps
    language: Lang | None = None,
) -> str:
    """
    Get the link to the signature page for the given approval request.
    """
    from flask import current_app

    return current_app.front_end_url.build_url(  # type: ignore[attr-defined,no-any-return]
        additional_path=get_signature_path(
            approval_request_id=approval_request_id,
            app_name=app_name,
            language=language,
        )
    )

get_target_origin_proposal 

get_target_origin_proposal(target_ref, target_type)
Source code in components/contracting/subcomponents/proposal/api/main.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
@contracting_tracer_wrap()
def get_target_origin_proposal(
    target_ref: str, target_type: TargetType
) -> TargetOriginProposal | None:
    from components.contracting.subcomponents.proposal.internals.models.target_origin_proposal import (
        TargetOriginProposal as TargetOriginProposalSQLA,
    )

    target_origin_proposal_sqla = (
        current_session.query(TargetOriginProposalSQLA)  # noqa: ALN085
        .filter(
            TargetOriginProposalSQLA.target_ref == target_ref,
            TargetOriginProposalSQLA.target_type == target_type,
        )
        .one_or_none()
    )

    return (
        TargetOriginProposal.from_model(target_origin_proposal_sqla)
        if target_origin_proposal_sqla
        else None
    )

get_waiting_signature_links(
    company_refs, user_ref, plugin_ids, approver_emails=None
)

Return the signature links for the given admin, company or user.

:param company_refs: if provided return the signature links for the companies :param user_ref: if provided return the signature links for the user :param approver_emails: if provided return only the signature links where the approver email match :param plugin_ids: if provided return the signature links for the given plugins

Source code in components/contracting/subcomponents/proposal/api/main.py 
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
def get_waiting_signature_links(
    company_refs: list[str] | None,
    user_ref: str | None,
    plugin_ids: list[PluginId],
    approver_emails: set[str] | None = None,
) -> list[str] | None:
    """
    Return the signature links for the given admin, company or user.

    :param company_refs: if provided return the signature links for the companies
    :param user_ref: if provided return the signature links for the user
    :param approver_emails: if provided return only the signature links where the approver email match
    :param plugin_ids: if provided return the signature links for the given plugins
    """
    proposals: list[LightProposal] = paginate_proposals(
        page=1,
        per_page=100,
        user_refs=[user_ref] if user_ref else None,
        company_refs=company_refs if company_refs else None,
        states=[ProposalState.waiting_for_approval],
        plugin_ids=plugin_ids,
    ).items

    if not proposals:
        return None

    signature_links = []
    for light_proposal in proposals:
        proposal = get_proposal(light_proposal.id)
        if proposal.is_approval_processing:
            continue
        approval_requests = proposal.get_approval_requests(
            approver_emails=approver_emails if approver_emails else None,
            is_live=True,
            document_provider_type=DocumentProviderType.hellosign,
        )
        if approval_requests:
            signature_links.append(
                get_signature_link(
                    approval_request_id=str(approval_requests[-1].id),
                    app_name=proposal.app_name,
                )
            )

    return signature_links

insert_proposal_item_in_subscription 

insert_proposal_item_in_subscription(
    subscription,
    proposal_item,
    source=None,
    signature_status=None,
)

Returns a new (non persisted) subscription that includes this proposal item. It inserts the proposal item in the subscription periods list.

Source code in components/contracting/subcomponents/proposal/api/main.py 
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
@contracting_tracer_wrap()
def insert_proposal_item_in_subscription(
    subscription: Subscription,
    proposal_item: ProposalItem,
    source: Optional[SubscriptionSource] = None,
    signature_status: Optional[SignatureStatus] = None,
) -> Subscription:
    """
    Returns a new (non persisted) subscription that includes this proposal item.
    It inserts the proposal item in the subscription periods list.
    """

    new_period = subscription_period_from_proposal_item(
        proposal_item=proposal_item,
        target=subscription.target,
        source=source
        or subscription_source_from_origin(
            origin=proposal_item.origin, app_name=subscription.contractee.app_name
        ),
        signature_status=signature_status,
    )
    new_subscription_periods = insert_period(new_period, subscription)

    return replace(subscription, periods=new_subscription_periods)

manual_approve_proposal 

manual_approve_proposal(
    approval_request_id,
    all_documents,
    document_types=(DocumentsType.signed_documents,),
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
@enqueueable
@contracting_tracer_wrap()
def manual_approve_proposal(
    approval_request_id: UUID,
    all_documents: dict[
        tuple[UUID, Target, str],  # proposal_item.id, target, template.type
        bytes,
    ],
    document_types: Sequence[DocumentsType] = (DocumentsType.signed_documents,),
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.proposal.internals.models.approval_request import (
        ApprovalRequest as ApprovalRequestSQLA,
    )

    approval_request = get_or_raise_missing_resource(
        ApprovalRequestSQLA, approval_request_id
    )
    proposal_id = approval_request.proposal_id

    with AdvisoryLock(_proposal_approval_advisory_lock_key(proposal_id)):
        proposal = get_proposal(proposal_id=proposal_id)
        for proposal_item in proposal.proposal_items:
            for target in proposal_item.targets:
                documents = []
                for template in proposal_item.get_templates(document_types):
                    file_bytes = all_documents[
                        (proposal_item.id, target, template.type)
                    ]
                    documents.append(
                        Document(
                            id=0,
                            name=template.file_name,
                            file=BytesIO(file_bytes),
                            template=template,
                            template_args={},
                            metadata={"language": template.language},
                            is_preview=False,
                        )
                    )
                assign_documents_to_request(
                    approval_request_id, proposal_item.id, documents, target, False
                )

        approve_proposal(proposal.id, approval_request_id, commit)

paginate_proposals 

paginate_proposals(
    page=1,
    per_page=100,
    proposal_ids=None,
    created_by=None,
    user_refs=None,
    account_refs=None,
    company_refs=None,
    contract_refs=None,
    product_refs=None,
    targets=None,
    states=None,
    origins=None,
    proposal_name=None,
    plugin_ids=None,
    tags=None,
    tags_to_exclude=None,
    expire_before=None,
    cancel_before=None,
    max_per_page=100,
    app_name=None,
    renewal_campaign_name=None,
    subscription_type=None,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@contracting_tracer_wrap()
def paginate_proposals(
    page: int = 1,  # start at 1
    per_page: int = 100,
    proposal_ids: Optional[list[UUID]] = None,
    created_by: Optional[str] = None,
    user_refs: Optional[list[str]] = None,
    account_refs: Optional[list[str] | list[UUID]] = None,
    company_refs: Optional[list[str]] = None,
    contract_refs: Optional[list[str]] = None,
    product_refs: Optional[list[str]] = None,
    targets: Optional[list[Target]] = None,
    states: Optional[list[ProposalState]] = None,
    origins: Optional[list[ProposalOrigin]] = None,
    proposal_name: Optional[str] = None,
    plugin_ids: Optional[list[PluginId]] = None,
    tags: Optional[list[str]] = None,
    tags_to_exclude: Optional[list[str]] = None,
    expire_before: Optional[date] = None,
    cancel_before: Optional[date] = None,
    max_per_page: int = 100,
    app_name: Optional[AppName] = None,
    renewal_campaign_name: Optional[str] = None,
    subscription_type: Optional[SubscriptionType] = None,
) -> Paginate[LightProposal]:
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal_item import (
        ProposalItem as ProposalItemSQLA,
    )
    from components.contracting.subcomponents.proposal.internals.proposal import (
        build_proposal_query,
    )
    from shared.errors.error_code import BaseErrorCode

    if not app_name:
        app_name = get_current_app_name()

    query = build_proposal_query(
        proposal_ids=proposal_ids if proposal_ids is not None else None,
        states=states if states else None,
        creator_refs=[created_by] if created_by else None,
        user_refs=user_refs,
        account_refs=account_refs,
        company_refs=company_refs,
        contract_refs=contract_refs,
        product_refs=product_refs,
        targets=targets,
        origins=origins,
        proposal_name=proposal_name,
        plugin_ids=plugin_ids if plugin_ids else None,
        tags=tags if tags else None,
        tags_to_exclude=tags_to_exclude,
        expire_before=expire_before,
        cancel_before=cancel_before,
        renewal_campaign_name=renewal_campaign_name,
        app_name=app_name,
        subscription_type=subscription_type,
    )
    query = query.options(
        selectinload(ProposalSQLA.proposal_items).options(
            selectinload(ProposalItemSQLA.targets),  # type: ignore[arg-type]
            selectinload(ProposalItemSQLA.templates),  # type: ignore[arg-type]
        )
    ).order_by(ProposalSQLA.created_at.desc())

    pagination = paginate(
        query=query,
        page=page,
        per_page=per_page,
        max_per_page=max_per_page,
    )

    def preload_subscriptors() -> None:
        # Preload subscriptors so from_model doesn't issue too many queries for proposals **with a single subscriptor**
        subscription_ref_lists_by_legal_status = group_by(
            cast(
                "list[ProposalSQLA]",
                [
                    proposal
                    for proposal in pagination.items
                    if proposal.subscriptor_scope
                ],
            ),
            # Assume all subscriptors of a proposal have the same legal status
            key_fn=lambda p: p.subscriptor_scope[0]["legal_status"],
            value_fn=lambda p: [s["ref"] for s in p.subscriptor_scope],
        )
        for legal_status, ref_lists in subscription_ref_lists_by_legal_status.items():
            get_subscriptors(
                ids=[ref for refs in ref_lists for ref in refs],
                legal_status=legal_status,
                app_name=app_name,
            )

    catch_and_log(
        preload_subscriptors,
        exceptions=[ValueError, BaseErrorCode],
        message="Some subscriptors are unable to be (pre)loaded",
        log_level=logging.INFO,
    )

    return Paginate(
        items=collections.compact(
            [
                catch_and_log(
                    LightProposal.from_model,
                    message=f"Proposal {proposal.id} unable to be loaded",
                    exceptions=[ValueError, BaseErrorCode],
                    log_level=logging.WARNING,
                    model=proposal,
                )
                for proposal in pagination.items
            ]
        ),
        page=mandatory(pagination.page),
        next_page=pagination.next_num,
        prev_page=pagination.prev_num,
        items_count=mandatory(pagination.total),
        pages_count=pagination.pages,
    )

process_hellosign_callback 

process_hellosign_callback(message)
Source code in components/contracting/subcomponents/proposal/api/main.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@contracting_tracer_wrap()
def process_hellosign_callback(message: str) -> None:
    from components.fr.internal.queuing.config import (  # noqa: ALN043
        CONTRACTING_QUEUE,
    )
    from shared.queuing.flask_rq import current_rq

    current_logger.info("Validate hellosign callback - start", message=message)
    signature_logic.validate_hellosign_callback(event_as_string=message)
    current_logger.info("Validate hellosign callback - valid", message=message)

    # Perform checks on the message now (and now in the async code because it won't have access to the request context)
    signature_logic.parse_callback_event(message, True)

    queue = current_rq.get_queue(CONTRACTING_QUEUE)
    queue.enqueue(process_hellosign_callback_async, message, job_timeout=60 * 15)

proposal_state_machine module-attribute 

proposal_state_machine = StateMachine('state')

register_event 

register_event(event_name, handler)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1292
1293
1294
1295
1296
1297
1298
@contracting_tracer_wrap()
def register_event(event_name: str, handler: Callable) -> None:  # type: ignore[type-arg]
    from components.contracting.utils.events import (
        event_handler,
    )

    event_handler.register_events(**{event_name: handler})

request_approval 

request_approval(
    proposal_id,
    approver_first_name,
    approver_last_name,
    approver_email,
    requester_id,
    notification_type,
    approver_ref=None,
    auto_approve_after=None,
    is_signable=True,
    document_provider_type=DocumentProviderType.hellosign,
    cc_recipients=None,
    can_be_forwarded=None,
    reminder_scheduled_at=None,
    require_identity_verification=False,
    validation_context=None,
    commit=True,
)

Request approval for a proposal.

Parameters:

Name Type Description Default
proposal_id UUID

proposal to request approval for

 required
approver_*

approver's information

 required
approver_ref Optional[str]

optional reference for the approver when he's an admin (and not an external user)

 None
requester_id str

id of the user requesting the approval

 required
notification_type ApprovalNotificationType

defines how the approver will be notified (directly by email or just generate a unique link to share)

 required
auto_approve_after Optional[date]

defines when the proposal should be automatically approved if nothing happens

 None
is_signable bool

defines if the proposal should be signable or not. Documents won't be pushed to HelloSign if False

 True
document_provider_type DocumentProviderType

defines how the documents will be approved (automatically through hellosign or manually)

 hellosign
cc_recipients Optional[list[str]]

defines the list of email addresses to be cced on the approval request

 None
can_be_forwarded Optional[bool]

defines if the approval request can be forwarded to another approver. If None the default behaviour is to allow forwarding.

 None
reminder_scheduled_at Optional[date]

defines when the reminder should be sent to the approver, no reminder will be sent if None

 None
require_identity_verification bool

defines if approver should verify their identity

 False
Source code in components/contracting/subcomponents/proposal/api/main.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@contracting_tracer_wrap()
@enqueueable
def request_approval(
    proposal_id: UUID,
    approver_first_name: str,
    approver_last_name: str,
    approver_email: str,
    requester_id: str,
    notification_type: ApprovalNotificationType,  # noqa: ALN072
    approver_ref: Optional[str] = None,
    auto_approve_after: Optional[date] = None,
    is_signable: bool = True,
    document_provider_type: DocumentProviderType = DocumentProviderType.hellosign,  # noqa: ALN072
    cc_recipients: Optional[list[str]] = None,
    can_be_forwarded: Optional[bool] = None,
    reminder_scheduled_at: Optional[date] = None,
    require_identity_verification: bool = False,  # IO-2966 diverge from 🇧🇪 as discussed in https://alanhealth.slack.com/archives/C19FZEB41/p1684313913641489
    validation_context: Optional[ValidationContext] = None,
    commit: bool = True,
) -> ApprovalRequest:
    """
    Request approval for a proposal.

    Args:
        proposal_id: proposal to request approval for
        approver_*: approver's information
        approver_ref: optional reference for the approver when he's an admin (and not an external user)
        requester_id: id of the user requesting the approval
        notification_type: defines how the approver will be notified (directly by email or just generate a unique link to share)
        auto_approve_after: defines when the proposal should be automatically approved if nothing happens
        is_signable: defines if the proposal should be signable or not. Documents won't be pushed to HelloSign if False
        document_provider_type: defines how the documents will be approved (automatically through hellosign or manually)
        cc_recipients: defines the list of email addresses to be cced on the approval request
        can_be_forwarded: defines if the approval request can be forwarded to another approver.
            If None the default behaviour is to allow forwarding.
        reminder_scheduled_at: defines when the reminder should be sent to the approver, no reminder will be sent if None
        require_identity_verification: defines if approver should verify their identity
    """
    from components.contracting.external.subscriptor.api.user import (
        get_users,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)
    requester = one(get_users(user_ids=[requester_id], app_name=proposal.app_name))

    with side_effects(trigger_on_exit=commit):
        # We skip ContractingValidationError because they have their own error handling and persistence
        with persist_error(
            proposal_id=proposal_id,
            skip=ContractingValidationError,
            username=requester.name,
            scope="request_approval",
        ):
            approval_request = add_request_approval(
                proposal,
                approver_first_name=approver_first_name,
                approver_last_name=approver_last_name,
                approver_email=approver_email.strip(),
                requester=requester,
                approver_ref=approver_ref,
                notification_type=notification_type,
                auto_approve_after=auto_approve_after,
                is_signable=is_signable,
                document_provider_type=document_provider_type,
                cc_recipients=cc_recipients,
                can_be_forwarded=bool(can_be_forwarded),
                reminder_scheduled_at=reminder_scheduled_at,
                require_identity_verification=require_identity_verification,
                validation_context=validation_context,
            )
            if commit:
                current_session.commit()
            else:
                current_session.flush()

    return ApprovalRequest.from_model(approval_request)

set_target_metadata 

set_target_metadata(
    proposal_item_id,
    subscriptor_ref,
    subscriptor_legal_status,
    subscription_ref,
    subscription_type,
    metadata,
    commit=True,
)
Source code in components/contracting/subcomponents/proposal/api/main.py 
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
@contracting_tracer_wrap()
def set_target_metadata(
    proposal_item_id: UUID,
    subscriptor_ref: str,
    subscriptor_legal_status: SubscriptorLegalStatus,
    subscription_ref: Optional[str],
    subscription_type: Optional[SubscriptionType],
    metadata: dict,  # type: ignore[type-arg]
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.proposal.internals import (
        proposal_item as bl,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal_item import (
        ProposalItem as ProposalItemSQLA,
    )

    proposal_item: ProposalItemSQLA = get_or_raise_missing_resource(
        ProposalItemSQLA, proposal_item_id
    )

    proposal_item_entity = ProposalItem.from_model(proposal_item)
    assert proposal_item_entity.plugin.subscription_type == subscription_type
    assert (
        proposal_item_entity.plugin.subscriptor_legal_status == subscriptor_legal_status
    )

    proposal_item_target = one(
        [
            t
            for t in proposal_item.targets
            if t.subscription_ref == subscription_ref
            and t.subscriptor_ref == subscriptor_ref
        ],
    )

    bl.set_proposal_item_target_metadata(
        proposal_item_target=proposal_item_target, metadata=metadata
    )

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

    current_logger.info(
        "Target metadata has been updated",
        proposal_id=proposal_item.proposal_id,
        proposal_item_id=proposal_item_id,
        proposal_item_target_id=proposal_item_target.id,
        subscriptor_ref=subscriptor_ref,
        subscription_ref=subscription_ref,
        subscription_type=subscription_type,
        commit=commit,
    )

unregister_event 

unregister_event(event_name, handler)
Source code in components/contracting/subcomponents/proposal/api/main.py 
1301
1302
1303
1304
1305
1306
1307
@contracting_tracer_wrap()
def unregister_event(event_name: str, handler: Callable) -> None:  # type: ignore[type-arg]
    from components.contracting.utils.events import (
        event_handler,
    )

    event_handler.unregister_events(**{event_name: handler})

update_proposal_creator 

update_proposal_creator(
    proposal_id,
    creator_ref,
    creator_display_name,
    commit=True,
)

This is only used is rare cases, when 2 users are merged together, and we need to update the creator_ref of the proposals created by the merged user.

Source code in components/contracting/subcomponents/proposal/api/main.py 
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
@contracting_tracer_wrap()
def update_proposal_creator(
    proposal_id: UUID, creator_ref: str, creator_display_name: str, commit: bool = True
) -> None:
    """
    This is only used is rare cases, when 2 users are merged together,
    and we need to update the creator_ref of the proposals created by the merged user.
    """
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    proposal.creator_ref = creator_ref
    proposal.creator_display_name = creator_display_name

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

update_proposal_expire_date 

update_proposal_expire_date(
    proposal_id, expire_after, commit=True
)

It's only used by self-serve, that is making sure proposals expire 45 days (= PRODUCT_BUILDER_VALIDITY_FOR_SELF_SERVE_IN_DAYS) after its creation date

Source code in components/contracting/subcomponents/proposal/api/main.py 
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
@contracting_tracer_wrap()
def update_proposal_expire_date(
    proposal_id: UUID, expire_after: date, commit: bool = True
) -> None:
    """
    It's only used by self-serve, that is making sure proposals expire 45 days
    (= PRODUCT_BUILDER_VALIDITY_FOR_SELF_SERVE_IN_DAYS) after its creation date
    """
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    # This condition has been put self-serve doesn't need to make the expiration date later. Maybe it can be removed for other use-cases
    if proposal.expire_after and proposal.expire_after < expire_after:
        raise ValueError(
            f"New expire date {expire_after} is after the current expire date {proposal.expire_after}"
        )

    proposal.expire_after = expire_after

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

update_proposal_target 

update_proposal_target(
    proposal_id, old_user_ref, new_user_ref, commit=False
)

This is only used is rare cases, when 2 users are merged together, and we need to replace the user_ref targeted by the proposals by the new_user_ref.

Source code in components/contracting/subcomponents/proposal/api/main.py 
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
@contracting_tracer_wrap()
def update_proposal_target(
    proposal_id: UUID, old_user_ref: str, new_user_ref: str, commit: bool = False
) -> None:
    """
    This is only used is rare cases, when 2 users are merged together,
    and we need to replace the user_ref targeted by the proposals by the new_user_ref.
    """
    from components.contracting.subcomponents.proposal.api.entities.subscriptor import (
        SubscriptorKey,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)

    # 1. Update ProposalItemTarget
    for proposal_item in proposal.proposal_items:
        for target in proposal_item.targets:
            if (
                target.subscriptor_ref == old_user_ref
                and target.subscriptor_legal_status == SubscriptorLegalStatus.individual
            ):
                target.subscriptor_ref = new_user_ref

    # 2. Update the subscriptor_scope
    subscriptors = [SubscriptorKey.from_dict(s) for s in proposal.subscriptor_scope]
    new_subscriptors = [
        s
        for s in subscriptors
        if s.legal_status != SubscriptorLegalStatus.individual or s.ref != old_user_ref
    ]
    new_subscriptors.append(
        SubscriptorKey(ref=new_user_ref, legal_status=SubscriptorLegalStatus.individual)
    )
    proposal.subscriptor_scope = [s.to_dict() for s in new_subscriptors]

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

view_documents 

view_documents(
    proposal_id,
    proposal_item_ids=None,
    targets=None,
    documents_types=_ONLY_SIGNED_DOCUMENTS,
    template_types=None,
    is_preview=True,
    approver_first_name=None,
    approver_last_name=None,
    approver_email=None,
    languages=None,
)

Returns a list of documents. :param proposal_id: the proposal id to generate documents for :param proposal_item_ids: returns only documents for the given proposal items :param targets: returns only documents for the given targets :param documents_types: returns only documents of the given types :param template_types: returns only documents of the given template types :param is_preview: defines if the documents should include the signature fields or the preview block :param approver_first_name: the first name of the approver - required if is_preview is False :param approver_last_name: the last name of the approver - required if is_preview is False :param approver_email: the email of the approver - required if is_preview is False :param languages: the languages of the documents

Source code in components/contracting/subcomponents/proposal/api/main.py 
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
@contracting_tracer_wrap()
def view_documents(
    proposal_id: UUID,
    proposal_item_ids: Optional[list[UUID]] = None,
    targets: Optional[list[Target]] = None,
    documents_types: Sequence[DocumentsType] = _ONLY_SIGNED_DOCUMENTS,
    template_types: Optional[set[str]] = None,
    is_preview: bool = True,
    approver_first_name: Optional[str] = None,
    approver_last_name: Optional[str] = None,
    approver_email: Optional[str] = None,
    languages: Optional[set[Lang]] = None,
) -> list[Document]:
    """Returns a list of documents.
    :param proposal_id: the proposal id to generate documents for
    :param proposal_item_ids: returns only documents for the given proposal items
    :param targets: returns only documents for the given targets
    :param documents_types: returns only documents of the given types
    :param template_types: returns only documents of the given template types
    :param is_preview: defines if the documents should include the signature fields or the preview block
    :param approver_first_name: the first name of the approver - required if is_preview is False
    :param approver_last_name: the last name of the approver - required if is_preview is False
    :param approver_email: the email of the approver - required if is_preview is False
    :param languages: the languages of the documents
    """
    from components.contracting.subcomponents.proposal.internals.document import (
        fetch_or_generate_preview_documents,
    )
    from components.contracting.subcomponents.proposal.internals.models.proposal import (
        Proposal as ProposalSQLA,
    )

    proposal = get_or_raise_missing_resource(ProposalSQLA, proposal_id)
    proposal_items = [
        pi
        for pi in proposal.proposal_items
        if proposal_item_ids is None or pi.id in proposal_item_ids
    ]

    if is_preview:
        return fetch_or_generate_preview_documents(
            proposal_items=proposal_items,
            documents_types=documents_types,
            targets=targets,
            template_types=template_types,
            languages=languages,
        )
    elif (
        not is_preview and approver_first_name and approver_last_name and approver_email
    ):
        return generate_proposal_items_documents(
            proposal_items=proposal_items,
            approver_first_name=approver_first_name,
            approver_last_name=approver_last_name,
            approver_email=approver_email,
            documents_types=documents_types,
            targets=targets,
            template_types=template_types,
            preview=False,
            languages=languages,
        )
    else:
        raise ValueError(
            "Approval details is required to generate not previewed signed documents"
        )

withdraw_approval_request 

withdraw_approval_request(approval_request_id, commit=True)
Source code in components/contracting/subcomponents/proposal/api/main.py 
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
@contracting_tracer_wrap()
def withdraw_approval_request(
    approval_request_id: UUID, commit: bool = True
) -> ApprovalRequest:
    from components.contracting.subcomponents.proposal.internals import (
        approval as bl,
    )

    approval_request = bl.withdraw_approval_request(approval_request_id)
    if commit:
        current_session.commit()
    current_logger.info(
        f"Approval request {approval_request_id} has been withdrawn",
        proposal_id=approval_request.proposal.id,
    )
    return ApprovalRequest.from_model(approval_request)

components.contracting.public.renewal

get_contractee_renewal_operation 

get_contractee_renewal_operation(
    account_id=None,
    user_id=None,
    contractee_renewal_operation_id=None,
    campaign_name=None,
    app_name=None,
    active_campaign=True,
)

Fetches the contractee renewal operation for a given account and renewal year. If the campaign is not found, it raised a RenewalNotFoundException.

Parameters:

Name Type Description Default
account_id UUID

The unique identifier of the account.

 None
user_id str

The unique identifier of the user.

 None
contractee_renewal_operation_id UUID

The unique identifier of the account renewal campaign.

 None
campaign_name str

The name of the campaign.

 None
app_name AppName

The name of the app.

 None
active_campaign bool

If True, only active campaign is considered.

 True
Source code in components/contracting/subcomponents/renewal/public/contractee_renewal_operation.py 
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
def get_contractee_renewal_operation(
    account_id: UUID | None = None,
    user_id: str | None = None,
    contractee_renewal_operation_id: UUID | None = None,
    campaign_name: str | None = None,
    app_name: AppName | None = None,
    active_campaign: bool | None = True,
) -> ContracteeRenewalOperation:
    """
    Fetches the contractee renewal operation for a given account and renewal year.
    If the campaign is not found, it raised a RenewalNotFoundException.

    Args:
        account_id (UUID): The unique identifier of the account.
        user_id (str): The unique identifier of the user.
        contractee_renewal_operation_id (UUID): The unique identifier of the account renewal campaign.
        campaign_name (str): The name of the campaign.
        app_name (AppName): The name of the app.
        active_campaign (bool): If True, only active campaign is considered.
    """
    if user_id and account_id:
        raise ValueError("user_id and account_id cannot be used together")

    if not contractee_renewal_operation_id and not (account_id or user_id):
        raise ValueError(
            "account_id or user_id is required when contractee_renewal_operation_id is not provided"
        )

    if (account_id or user_id) and not active_campaign and not campaign_name:
        raise ValueError("campaign_name is required when active_campaign is False")

    return one(
        _list_contractee_renewal_operation(
            contractee_renewal_operation_ids=(
                [contractee_renewal_operation_id]
                if contractee_renewal_operation_id
                else None
            ),
            account_ids=[account_id] if account_id else None,
            user_ids=[user_id] if user_id else None,
            campaign_name=campaign_name if campaign_name else None,
            app_name=app_name,
            active_campaign=active_campaign,
        ),
        CustomErrorClass=RenewalNotFoundException,
    )

get_contractee_renewal_operation_or_none 

get_contractee_renewal_operation_or_none(
    account_id=None,
    user_id=None,
    campaign_name=None,
    app_name=None,
    active_campaign=True,
)

Fetches the contractee renewal operation for a given account and renewal year. If the campaign is not found, it returns None.

Source code in components/contracting/subcomponents/renewal/public/contractee_renewal_operation.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
def get_contractee_renewal_operation_or_none(
    account_id: UUID | None = None,
    user_id: str | None = None,
    campaign_name: str | None = None,
    app_name: AppName | None = None,
    active_campaign: bool | None = True,
) -> ContracteeRenewalOperation | None:
    """
    Fetches the contractee renewal operation for a given account and renewal year.
    If the campaign is not found, it returns None.
    """
    try:
        renewal_operation = get_contractee_renewal_operation(
            account_id=account_id,
            user_id=user_id,
            campaign_name=campaign_name,
            app_name=app_name,
            active_campaign=active_campaign,
        )
        return renewal_operation
    except RenewalNotFoundException:
        return None

is_alternative_renewal 

is_alternative_renewal(proposal)

Check if the proposal is an alternative renewal proposal

Source code in components/contracting/subcomponents/renewal/public/proposal.py 
788
789
790
791
792
793
@contracting_tracer_wrap()
def is_alternative_renewal(proposal: Proposal | LightProposal) -> bool:
    """
    Check if the proposal is an alternative renewal proposal
    """
    return is_renewal(proposal) and ALTERNATIVE_RENEWAL_TAG in proposal.tags

is_automatic_renewal 

is_automatic_renewal(proposal)

Check if the proposal is an automatic renewal proposal

Source code in components/contracting/subcomponents/renewal/public/proposal.py 
780
781
782
783
784
785
@contracting_tracer_wrap()
def is_automatic_renewal(proposal: Proposal | LightProposal) -> bool:
    """
    Check if the proposal is an automatic renewal proposal
    """
    return is_renewal(proposal) and proposal.origin == ProposalOrigin.renewal_bot

is_manual_renewal 

is_manual_renewal(proposal)

Check if the proposal is a manual renewal proposal

Source code in components/contracting/subcomponents/renewal/public/proposal.py 
772
773
774
775
776
777
@contracting_tracer_wrap()
def is_manual_renewal(proposal: Proposal | LightProposal) -> bool:
    """
    Check if the proposal is a manual renewal proposal
    """
    return is_renewal(proposal) and proposal.origin == ProposalOrigin.alan_served

is_renewal_employee_communication 

is_renewal_employee_communication(
    subscription_period_ref,
    subscription_type,
    company_id,
    professional_category,
)

Check if the communication is a renewal communication :param subscription_period_ref: Either health_contract_version_id or prevoyance_contract_version_ref :param subscription_type: Either health_insurance or prevoyance

Source code in components/contracting/subcomponents/renewal/public/proposal.py 
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
@contracting_tracer_wrap()
def is_renewal_employee_communication(
    subscription_period_ref: str,
    subscription_type: SubscriptionType,
    company_id: int,
    professional_category: str,
) -> bool:
    """
    Check if the communication is a renewal communication
    :param subscription_period_ref: Either health_contract_version_id or prevoyance_contract_version_ref
    :param subscription_type: Either health_insurance or prevoyance
    """
    is_renewal_communication = False
    try:
        if subscription_type == subscription_type.health_insurance:
            target_type = TargetType.health_contract_version
        elif subscription_type == subscription_type.prevoyance:
            target_type = TargetType.prevoyance_contract_version
        else:
            raise ValueError(f"Subscription type {subscription_type} is not supported")
        target_origin = get_target_origin_proposal(
            target_ref=str(subscription_period_ref), target_type=target_type
        )
        if target_origin is not None:
            proposal = get_proposal(target_origin.proposal_id)
            items = [
                proposal_item
                for proposal_item in proposal.proposal_items
                if contracting_plugin_for(
                    plugin_id=proposal_item.plugin_id
                ).subscription_type
                == subscription_type
            ]
            targeted_item = one(
                [
                    item
                    for item in items
                    if any(
                        target.subscriptor_ref == str(company_id)
                        and target.population.is_same_professional_category(
                            professional_category
                        )
                        for target in item.targets
                    )
                ]
            )
            if (
                targeted_item.settings.get("notification_type")
                == EmployeeNotificationType.Renewal
            ):
                is_renewal_communication = True
        else:
            current_logger.error(
                "Employee communication email - No target origin found for subscription period",
                subscription_period_ref=subscription_period_ref,
                subscription_type=subscription_type,
            )
    except Exception as e:
        current_logger.error(
            "Employee communication email - Unexpected error while getting proposal from target origin",
            exception=e,
            subscription_period_ref=subscription_period_ref,
            subscription_type=subscription_type,
        )
        raise e
    return is_renewal_communication

components.contracting.public.self_serve_subscription

self_serve_active_company_proposal_from_user 

self_serve_active_company_proposal_from_user(
    user_id, include_approved=False
)

Self Serve Active Company Proposal From User

Source code in components/contracting/public/self_serve_subscription.py 
 9
10
11
12
13
14
15
16
17
def self_serve_active_company_proposal_from_user(
    user_id: int, include_approved: bool = False
) -> Proposal | None:
    """Self Serve Active Company Proposal From User"""
    from components.contracting.subcomponents.self_serve_subscription.internals.proposal import (
        self_serve_active_company_proposal_from_user,
    )

    return self_serve_active_company_proposal_from_user(user_id, include_approved)

self_serve_create_or_update_company_proposal_for_user 

self_serve_create_or_update_company_proposal_for_user(
    user_id, params
)

Create or update a user's company proposal for self-serve.

Source code in components/contracting/public/self_serve_subscription.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def self_serve_create_or_update_company_proposal_for_user(
    user_id: int,
    params: SignupParams,
) -> Proposal:
    """Create or update a user's company proposal for self-serve."""
    from components.contracting.subcomponents.self_serve_subscription.internals.step_company_signup import (
        self_serve_update_or_create_company_proposal,
    )

    _, proposal = self_serve_update_or_create_company_proposal(
        signup_params=params, user_id=user_id
    )

    return proposal

self_serve_individual_proposal_from_user 

self_serve_individual_proposal_from_user(
    user_id, include_approved=False
)

Self Serve Active Individual Proposal From User

Source code in components/contracting/public/self_serve_subscription.py 
20
21
22
23
24
25
26
27
28
def self_serve_individual_proposal_from_user(
    user_id: int, include_approved: bool = False
) -> Proposal | None:
    """Self Serve Active Individual Proposal From User"""
    from components.contracting.subcomponents.self_serve_subscription.internals.proposal import (
        self_serve_individual_proposal_from_user,
    )

    return self_serve_individual_proposal_from_user(user_id, include_approved)

components.contracting.public.subscription

api

initialize_subscription 

initialize_subscription(
    subscription_scope,
    owner_type,
    owner_ref,
    payload_ref=None,
    *,
    commit=True
)
Source code in components/contracting/public/subscription/api.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def initialize_subscription(  # noqa: D103
    subscription_scope: SubscriptionScope,
    owner_type: str,
    owner_ref: str,
    payload_ref: Optional[UUID] = None,
    *,
    commit: bool = True,
) -> BaseSubscription:
    from shared.helpers.db import current_session

    subscription = internal_initialize_subscription(
        subscription_scope=subscription_scope,
        owner_type=owner_type,
        owner_ref=owner_ref,
        payload_ref=payload_ref,
    )
    if commit:
        current_session.commit()

    return subscription

fr

health_insurance

Subscription dataclass 
Subscription(
    *,
    id,
    contractee,
    population,
    subscription_type,
    subscription_scope
)

Bases: Timeline[SubscriptionPeriod]

company property 
company
contractee instance-attribute 
contractee
dutch_friendly_name property 
dutch_friendly_name
earliest_period property 
earliest_period
english_friendly_name property 
english_friendly_name
friendly_name property 
friendly_name
get_period_by_id 
get_period_by_id(period_id)
Source code in components/contracting/external/subscription/api/entities/subscription.py 
305
306
307
308
309
def get_period_by_id(self, period_id: str) -> SubscriptionPeriod | None:
    return next(
        (period for period in self.periods if period.id == period_id),
        None,
    )
get_status 
get_status(on_date)
Source code in components/contracting/external/subscription/api/entities/subscription.py 
329
330
331
332
333
334
335
336
337
338
339
def get_status(self, on_date: date) -> SubscriptionStatusEnum:
    ongoing_period = self.get_ongoing_period(on_date)
    if self.end_date and self.end_date > on_date:
        return SubscriptionStatusEnum.ending
    elif ongoing_period:
        return ongoing_period.get_status(on_date)
    elif self.is_ended_on(on_date):
        return SubscriptionStatusEnum.ended

    else:
        return SubscriptionStatusEnum.upcoming
id instance-attribute 
id
is_for_company property 
is_for_company
ongoing_pricing property 
ongoing_pricing
population instance-attribute 
population
spanish_friendly_name property 
spanish_friendly_name
subscription_periods property 
subscription_periods
subscription_scope instance-attribute 
subscription_scope
subscription_type instance-attribute 
subscription_type
target property 
target
user property 
user
get_health_subscriptions_for_companies 
get_health_subscriptions_for_companies(
    *company_ids,
    include_ended=False,
    include_will_end=True,
    include_collective_retiree=False
)
Source code in components/contracting/external/subscription/fr/health_insurance.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
@contracting_tracer_wrap()
def get_health_subscriptions_for_companies(
    *company_ids: str,
    include_ended: bool = False,
    include_will_end: bool = True,
    include_collective_retiree: bool = False,
) -> list[Subscription]:
    return _get_health_subscriptions_for_subscriptor_refs(
        model_klass=Company,
        subscriptor_refs=company_ids,
        include_ended=include_ended,
        include_will_end=include_will_end,
        include_collective_retiree=include_collective_retiree,
    )
get_ongoing_or_upcoming_health_subscription_for_company 
get_ongoing_or_upcoming_health_subscription_for_company(
    company_id, include_ended=False, include_will_end=True
)
Source code in components/contracting/public/subscription/fr/health_insurance.py 
 9
10
11
12
13
14
15
16
def get_ongoing_or_upcoming_health_subscription_for_company(  # noqa: D103
    company_id: int,
    include_ended: bool = False,
    include_will_end: bool = True,
) -> list[Subscription]:
    return get_health_subscriptions_for_companies(
        str(company_id), include_ended=include_ended, include_will_end=include_will_end
    )

prevoyance_insurance

get_prevoyance_subscriptions_for_companies 
get_prevoyance_subscriptions_for_companies(
    *company_ids, include_ended=True, include_will_end=True
)
Source code in components/contracting/external/subscription/fr/prevoyance_insurance.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
@contracting_tracer_wrap()
def get_prevoyance_subscriptions_for_companies(
    *company_ids: str,
    include_ended: bool = True,
    include_will_end: bool = True,
) -> list[Subscription]:
    from components.contracting.subcomponents.legal_document.public.entities import (
        LegalClause,
    )
    from components.contracting.subcomponents.legal_document.public.legal_clause import (
        paginate_legal_clauses,
    )
    from components.fr.internal.contract.queries.prevoyance_contract_version import (  # noqa: ALN043
        get_prevoyance_contract_versions,
    )
    from components.fr.internal.models.company import (  # noqa: ALN069
        Company,
    )

    subscriptions: list[Subscription] = []
    companies = (
        current_session.query(Company)  # noqa: ALN085
        .options(
            selectinload(Company.prevoyance_contracts).options(
                selectinload(PrevoyanceContract.contract_populations),
                selectinload(PrevoyanceContract.amended_by_prevoyance_contracts),
                selectinload(PrevoyanceContract.signed_documents),
                selectinload(PrevoyanceContract.company).options(
                    joinedload(Company.address),
                    joinedload(Company.account),
                    selectinload(Company.ibans),
                    joinedload(Company.ccn),
                ),
                PrevoyanceContract.preload_subscription_timeline_option(
                    payload_options=[
                        joinedload(PrevoyanceSubscriptionPayload.prevoyance_plan),
                    ],
                ),
            )
        )
        .filter(Company.id.in_(company_ids))
    )

    contracts: list[PrevoyanceContract] = []

    for company in companies:
        for contract in company.prevoyance_contracts:
            if contract.is_cancelled:
                continue
            if contract.amended_prevoyance_contract_id is not None:
                continue
            if not contract.is_signed:
                continue

            contracts.append(contract)

    all_current_and_succeeding_contracts: list[PrevoyanceContract] = []

    for contract in contracts:
        current_prevoyance_contract: PrevoyanceContract | None = contract
        while current_prevoyance_contract is not None:
            all_current_and_succeeding_contracts.append(current_prevoyance_contract)
            current_prevoyance_contract = _get_succeeding_prevoyance_contract(
                current_prevoyance_contract
            )

    subscription_period_notifications_by_subscription_period_id = (
        find_all_subscription_period_notifications(
            notification_type=NotificationType.employee_notify_prevoyance_subscription,
            subscription_period_ids=[
                SubscriptionPeriodId(
                    subscription_type=SubscriptionType.prevoyance,
                    subscription_period_ref=version.period_ref,
                )
                for contract in all_current_and_succeeding_contracts
                for version in get_prevoyance_contract_versions(contract.id)
            ],
        )
    )

    clauses_by_subscription_period_ref: dict[str, set[LegalClause]] = defaultdict(set)
    for legal_clause in paginate_legal_clauses(
        page=1,
        per_page=10_000,
        subscription_scope=SubscriptionScope.fr_prevoyance,
        subscription_period_refs=[
            str(pcv.period_ref)
            for prevoyance_contract in contracts
            for pcv in get_prevoyance_contract_versions(prevoyance_contract.id)
        ],
    ).items:
        for period_ref in legal_clause.subscription_period_refs:
            clauses_by_subscription_period_ref[period_ref].add(legal_clause)

    for contract in contracts:
        subscription = subscription_from_prevoyance_contract(
            contract,
            subscription_period_notification_for_employee_notify_prevoyance_subscription=subscription_period_notifications_by_subscription_period_id,
            clauses_by_subscription_period_ref=clauses_by_subscription_period_ref,
        )
        subscriptions.append(subscription)
        request_caching.record_subscription(subscription)

    return [
        s
        for s in subscriptions
        if include_ended or not s.is_ended_on(on_date=utctoday())
        if include_will_end or (s.end_date is None or s.end_date < utctoday())
    ]

subscription

BaseSubscription dataclass 

BaseSubscription(
    id,
    subscription_scope,
    owner_type,
    owner_ref,
    payload_ref,
)
id instance-attribute 
id
owner_ref instance-attribute 
owner_ref
owner_type instance-attribute 
owner_type
payload_ref instance-attribute 
payload_ref
subscription_scope instance-attribute 
subscription_scope

BaseSubscriptionVersion

Bases: WithValidityPeriod, Protocol

operation_ref instance-attribute 
operation_ref

Subscription dataclass 

Subscription(
    id,
    subscription_scope,
    owner_type,
    owner_ref,
    payload_ref,
    validity_period,
    versions,
)

Bases: Generic[_T], BaseSubscription

validity_period instance-attribute 
validity_period
versions instance-attribute 
versions

SubscriptionScope

Bases: AlanBaseEnum

Subscription's API can be configured to operate within a specific scope.

DEPRECATED_mind class-attribute instance-attribute 
DEPRECATED_mind = 'mind'
be_health class-attribute instance-attribute 
be_health = 'be_health'
ca_health class-attribute instance-attribute 
ca_health = 'ca_health'
es_health class-attribute instance-attribute 
es_health = 'es_health'
fr_health class-attribute instance-attribute 
fr_health = 'fr_health'
fr_legacy_termination class-attribute instance-attribute 
fr_legacy_termination = 'fr_legacy_termination'
fr_prevoyance class-attribute instance-attribute 
fr_prevoyance = 'fr_prevoyance'
get_app_name 
get_app_name()

Returns the app name associated with the subscription scope.

Source code in components/contracting/subcomponents/subscription/public/entities.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_app_name(self) -> AppName:
    """Returns the app name associated with the subscription scope."""
    if self in {
        SubscriptionScope.fr_health,
        SubscriptionScope.fr_prevoyance,
        SubscriptionScope.fr_legacy_termination,
    }:
        return AppName.ALAN_FR

    if self == SubscriptionScope.be_health:
        return AppName.ALAN_BE

    if self == SubscriptionScope.ca_health:
        return AppName.ALAN_CA

    if self == SubscriptionScope.es_health:
        return AppName.ALAN_ES

    if self == SubscriptionScope.healthy_benefits:
        return AppName.ALAN_ES

    if self == SubscriptionScope.occupational_health:
        # It looks like this is only used by legal clauses at this point
        raise NotImplementedError(
            "Occupational Health has no concept of active app at this point"
        )

    raise ValueError(f"Unsupported subscription scope: {self}")
healthy_benefits class-attribute instance-attribute 
healthy_benefits = 'healthy_benefits'
occupational_health class-attribute instance-attribute 
occupational_health = 'occupational_health'
test class-attribute instance-attribute 
test = 'test'

SubscriptionType module-attribute 

SubscriptionType = SubscriptionType

SubscriptionUpdateModel

Bases: BaseModel

Represents a change in a Subscription.

A subscription is identified by it's scope and reference and isn't materialized in the database through a table.

It records a period of time where subscription properties must apply.

It gives the ability to clear/delete the previously stored information by setting is_deletion to true.

If clients need additional context, we could either add it in the payload OR in a new context JSONB column.

__repr__ 
__repr__()
Source code in components/contracting/subcomponents/subscription/internal/models/subscription_update.py 
83
84
def __repr__(self) -> str:
    return f"<{self.__class__.__name__} [{self.id}] for {self.subscription_scope} - {self.subscription_ref} {self.validity_period}>"
__table_args__ class-attribute instance-attribute 
__table_args__ = (
    Index(
        "ix_subscription_update_unique_revision_by_reference_by_scope",
        subscription_scope,
        subscription_ref,
        revision,
        unique=True,
    ),
    CheckConstraint(
        "(is_deletion is true) = (payload_ref is null)",
        name="subscription_is_deletion_or_have_payload",
    ),
    {"schema": CONTRACTING_SCHEMA},
)
__tablename__ class-attribute instance-attribute 
__tablename__ = 'subscription_update'
end_date class-attribute instance-attribute 
end_date = mapped_column(Date, nullable=True)
is_deletion class-attribute instance-attribute 
is_deletion = mapped_column(
    Boolean, nullable=False, default=False
)
operation_ref class-attribute instance-attribute 
operation_ref = mapped_column(String(255), nullable=True)
payload_ref class-attribute instance-attribute 
payload_ref = mapped_column(
    UUID(as_uuid=True), nullable=True
)
revision class-attribute instance-attribute 
revision = mapped_column(Integer, nullable=False)
start_date class-attribute instance-attribute 
start_date = mapped_column(Date, nullable=False)
subscription_ref class-attribute instance-attribute 
subscription_ref = mapped_column(String, nullable=False)
subscription_scope class-attribute instance-attribute 
subscription_scope = mapped_column(
    AlanBaseEnumTypeDecorator(SubscriptionScope),
    nullable=False,
)
validity_period property writable 
validity_period

SubscriptionVersionModel

Bases: BaseModel

Represents the evolution of the state of a Subscription.

A subscription is identified by it's scope and reference and isn't materialized in the database through a table.

It record the latest knowledge about which properties of the subscription should be active, and when.

__repr__ 
__repr__()
Source code in components/contracting/subcomponents/subscription/internal/models/subscription_version.py 
90
91
def __repr__(self) -> str:
    return f"<{self.__class__.__name__} [{self.id}] for {self.subscription_scope} - {self.subscription_ref}>"
__table_args__ class-attribute instance-attribute 
__table_args__ = (
    Index(
        "ix_subscription_version_scope_subscription_ref",
        subscription_scope,
        subscription_ref,
    ),
    Index(
        "ix_subscription_version_scope_and_payload_ref",
        subscription_scope,
        payload_ref,
    ),
    ExcludeConstraint(
        ("subscription_scope", "="),
        ("subscription_ref", "="),
        (
            text("daterange(start_date, end_date, '[]')"),
            "&&",
        ),
        name="no_overlapping_subscription_version",
        using="gist",
    ),
    {"schema": CONTRACTING_SCHEMA},
)
__tablename__ class-attribute instance-attribute 
__tablename__ = 'subscription_version'
end_date class-attribute instance-attribute 
end_date = mapped_column(Date, nullable=True)
operation_ref class-attribute instance-attribute 
operation_ref = mapped_column(String(255), nullable=True)
payload_ref class-attribute instance-attribute 
payload_ref = mapped_column(
    UUID(as_uuid=True), nullable=False
)
start_date class-attribute instance-attribute 
start_date = mapped_column(Date, nullable=False)
subscription class-attribute instance-attribute 
subscription = relationship(
    SubscriptionModel,
    primaryjoin=lambda: and_(
        subscription_scope == subscription_scope,
        foreign(subscription_ref) == cast(id, String),
    ),
    backref=backref(
        "subscription_versions",
        uselist=True,
        order_by=lambda: asc(),
    ),
    uselist=False,
    viewonly=True,
)
subscription_ref class-attribute instance-attribute 
subscription_ref = mapped_column(String, nullable=False)
subscription_scope class-attribute instance-attribute 
subscription_scope = mapped_column(
    AlanBaseEnumTypeDecorator(SubscriptionScope),
    nullable=False,
)
validity_period property writable 
validity_period

admin_tools_contracting_subscriptions_blueprint module-attribute 

admin_tools_contracting_subscriptions_blueprint = (
    AdminToolsContractingSubscriptionsBlueprint(
        name="admin_tools_contracting_subscriptions",
        import_name=__name__,
        template_folder="../internal/templates",
        static_folder="../internal/static",
    )
)

download_prevoyance_legal_document(
    subscription_period_id, document_type, lang=None
)
Source code in components/contracting/external/legal_documents.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def download_prevoyance_legal_document(
    subscription_period_id: str,
    document_type: "SupportedDocumentName",
    lang: Optional[str] = None,  # noqa: ARG001
) -> IO:  # type: ignore[type-arg]
    from components.fr.internal.contract.queries.prevoyance_contract_version import (  # noqa: ALN043
        get_prevoyance_contract_info_from_period_ref,
    )
    from components.fr.internal.models.enums.signable_document_type import (  # noqa: ALN069
        SignableDocumentType,
    )
    from components.fr.internal.models.signed_document import (  # noqa: ALN069
        SignedDocument as SignedDocumentSQLA,
    )
    from components.fr.internal.services.pdf_document import (  # noqa: ALN043
        get_pdf_document,
    )

    prevoyance_contract_version = get_prevoyance_contract_info_from_period_ref(
        period_ref=subscription_period_id
    ).prevoyance_contract_version
    document_type_to_use = "cg" if document_type == "cg-prevoyance" else document_type

    try:
        signed_document = prevoyance_contract_version.signed_document_for_document_type(
            document_type=SignableDocumentType(document_type_to_use)
        )
        current_logger.info(f"Found the signed document {signed_document.id}")
    except ValueError as e:
        if document_type == "cg-prevoyance":
            # Fallback on default doc if no CG in the signed documents
            current_logger.info("Fallback on default CG document")
            pdf_name = f"cg-prevoyance_v{prevoyance_contract_version.bundle_version}-{prevoyance_contract_version.prevoyance_plan_id}.pdf"
            result_file, _ = get_pdf_document(
                pdf_name,
                on_date=max(
                    mandatory(
                        prevoyance_contract_version.prevoyance_contract.start_date
                    ),
                    mandatory(prevoyance_contract_version.start_date).replace(
                        month=1, day=1
                    ),
                ),
            )
            return mandatory(result_file, "Only cg-assistance can be None")
        else:
            raise e

    return signed_document.get_or_download_file(
        uri_field=SignedDocumentSQLA.document_type_to_uri_column(document_type_to_use),
    )

get_subscription 

get_subscription(subscription_id, subscription_type)
Source code in components/contracting/external/subscription/fr/queries.py 
24
25
26
27
28
def get_subscription(
    subscription_id: str,
    subscription_type: SubscriptionType,
) -> Subscription:
    return one(get_subscriptions([subscription_id], subscription_type))

get_subscription_by_id 

get_subscription_by_id(
    subscription_id,
    subscription_scope,
    subscription_version_serializer,
)
Source code in components/contracting/subcomponents/subscription/public/queries.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_subscription_by_id(  # noqa: D103
    subscription_id: UUID,
    subscription_scope: SubscriptionScope,
    subscription_version_serializer: SubscriptionVersionSerializer[_T],
) -> Subscription | BaseSubscription:  # type: ignore[type-arg]
    subscription = (
        current_session.query(SubscriptionModel)  # noqa: ALN085
        .filter(
            SubscriptionModel.id == subscription_id,
            SubscriptionModel.subscription_scope == subscription_scope,
        )
        .one()
    )

    subscription_versions = get_subscription_version_timelines(
        subscription_scope=subscription_scope,
        subscription_refs=[subscription.id],
        subscription_version_serializer=subscription_version_serializer,
    )

    def _subscription_is_empty(subscription: SubscriptionModel) -> bool:
        return (
            subscription.id not in subscription_versions
            or subscription_versions[subscription.id].is_empty
        )

    if _subscription_is_empty(subscription):
        return BaseSubscription(
            id=subscription.id,
            subscription_scope=subscription.subscription_scope,
            owner_ref=subscription.owner_ref,
            owner_type=subscription.owner_type,
            payload_ref=subscription.payload_ref,
        )

    return Subscription(
        id=subscription.id,
        subscription_scope=subscription.subscription_scope,
        owner_ref=subscription.owner_ref,
        owner_type=subscription.owner_type,
        payload_ref=subscription.payload_ref,
        validity_period=ValidityPeriod(
            start_date=subscription_versions[subscription.id].start_date,
            end_date=subscription_versions[subscription.id].end_date,
        ),
        versions=subscription_versions[subscription.id],
    )

get_subscription_version_timeline 

get_subscription_version_timeline(
    subscription_scope,
    subscription_ref,
    subscription_version_serializer,
)
Source code in components/contracting/subcomponents/subscription/public/queries.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
def get_subscription_version_timeline(  # noqa: D103
    subscription_scope: SubscriptionScope,
    subscription_ref: _SubscriptionRefType,
    subscription_version_serializer: SubscriptionVersionSerializer[_T],
) -> Timeline[_T]:
    return next(
        timeline
        for timeline in get_subscription_version_timelines(
            subscription_scope=subscription_scope,
            subscription_refs=[subscription_ref],
            subscription_version_serializer=subscription_version_serializer,
        ).values()
    )

get_subscription_version_timelines 

get_subscription_version_timelines(
    subscription_scope,
    subscription_refs,
    subscription_version_serializer,
    subscription_preload_options=tuple(),
    payload_preload_options=tuple(),
)
Source code in components/contracting/subcomponents/subscription/public/queries.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def get_subscription_version_timelines(  # noqa: D103
    subscription_scope: SubscriptionScope,
    subscription_refs: Iterable[_SubscriptionRefType],
    subscription_version_serializer: SubscriptionVersionSerializer[_T],
    subscription_preload_options: Sequence[Load] = tuple(),
    payload_preload_options: Sequence[Load] = tuple(),
) -> dict[_SubscriptionRefType, Timeline[_T]]:
    from components.contracting.subcomponents.subscription.internal.models.subscription_version import (
        SubscriptionVersionModel,
    )
    from components.contracting.subcomponents.subscription.internal.serialization import (
        build_timeline,
    )

    def by_subscription_ref(subscription_version: SubscriptionVersionModel) -> str:
        return subscription_version.subscription_ref

    preload_options = []

    subscription_relation_name = f"{subscription_scope}_subscription"
    if hasattr(SubscriptionVersionModel, subscription_relation_name):
        preload_options.append(
            joinedload(
                getattr(SubscriptionVersionModel, subscription_relation_name)
            ).options(
                *subscription_preload_options,
            )
        )

    payload_relation_name = f"{subscription_scope}_payload"
    if hasattr(SubscriptionVersionModel, subscription_relation_name):
        preload_options.append(
            joinedload(
                getattr(SubscriptionVersionModel, payload_relation_name)
            ).options(
                *payload_preload_options,
            )
        )

    raw_versions = group_by(
        current_session.query(SubscriptionVersionModel)  # noqa: ALN085
        .filter(
            SubscriptionVersionModel.subscription_scope == subscription_scope,
            SubscriptionVersionModel.subscription_ref.in_(
                str(subscription_ref) for subscription_ref in subscription_refs
            ),
        )
        .options(*preload_options)
        .order_by(SubscriptionVersionModel.start_date.asc()),
        by_subscription_ref,
    )

    return {
        subscription_ref: build_timeline(
            raw_versions[str(subscription_ref)], subscription_version_serializer
        )
        for subscription_ref in subscription_refs
    }

get_subscription_version_updates 

get_subscription_version_updates(
    subscription_scope, subscription_refs
)
Source code in components/contracting/subcomponents/subscription/public/queries.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def get_subscription_version_updates(  # noqa: D103
    subscription_scope: SubscriptionScope,
    subscription_refs: Iterable[_SubscriptionRefType],
) -> dict[_SubscriptionRefType, list[SubscriptionUpdate]]:
    return group_by(
        [
            SubscriptionUpdate(
                subscription_scope=subscription_scope,
                subscription_ref=model.subscription_ref,
                validity_period=model.validity_period,
                payload_ref=model.payload_ref,
                operation_ref=model.operation_ref,
                is_deletion=model.is_deletion,
                revision=model.revision,
            )
            for model in current_session.query(SubscriptionUpdateModel)  # noqa: ALN085
            .filter(
                SubscriptionUpdateModel.subscription_scope == subscription_scope,
                SubscriptionUpdateModel.subscription_ref.in_(subscription_refs),
            )
            .order_by(SubscriptionUpdateModel.revision.asc())
        ],
        key_fn=lambda u: u.subscription_ref,
    )

get_subscriptions 

get_subscriptions(
    subscription_scope,
    subscription_version_serializer,
    period=None,
)
Source code in components/contracting/subcomponents/subscription/public/queries.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_subscriptions(  # noqa: D103
    subscription_scope: SubscriptionScope,
    subscription_version_serializer: SubscriptionVersionSerializer[_T],
    period: Optional[ValidityPeriod] = None,
) -> dict[str, GetSubscriptionsResult]:
    subscriptions_in_scope = (
        current_session.query(SubscriptionModel)  # noqa: ALN085
        .filter(SubscriptionModel.subscription_scope == subscription_scope)
        .all()
    )

    owner_refs = {subscription.owner_ref for subscription in subscriptions_in_scope}

    return {
        owner_ref: get_subscriptions_for(
            owner_ref=owner_ref,
            subscription_scope=subscription_scope,
            subscription_version_serializer=subscription_version_serializer,
            period=period,
        )
        for owner_ref in owner_refs
    }

get_subscriptions_for 

get_subscriptions_for(
    owner_ref,
    subscription_scope,
    subscription_version_serializer,
    period=None,
)

Return list of subscriptions for a given owner. :param owner_ref: Owner to return subscription to :param subscription_scope: filter subscription on this scope :param subscription_version_serializer: Your serialize for Subscription model :param period: if set, only subscription overlapping this period are returned

:return: GetSubscriptionsResult :raises EmptySubscriptionError: if only empty subscriptions are found

Source code in components/contracting/subcomponents/subscription/public/queries.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def get_subscriptions_for(
    owner_ref: str,
    subscription_scope: SubscriptionScope,
    subscription_version_serializer: SubscriptionVersionSerializer[_T],
    period: Optional[ValidityPeriod] = None,
) -> GetSubscriptionsResult:
    """

    Return list of subscriptions for a given owner.
    :param owner_ref: Owner to return subscription to
    :param subscription_scope: filter subscription on this scope
    :param subscription_version_serializer: Your serialize for Subscription model
    :param period: if set, only subscription overlapping this period are returned

    :return: GetSubscriptionsResult
    :raises EmptySubscriptionError: if only empty subscriptions are found
    """
    subscriptions: list[SubscriptionModel] = (
        current_session.query(SubscriptionModel)  # noqa: ALN085
        .filter(SubscriptionModel.owner_ref == str(owner_ref))
        .all()
    )

    subscription_versions = get_subscription_version_timelines(
        subscription_scope=subscription_scope,
        subscription_refs=[subscription.id for subscription in subscriptions],
        subscription_version_serializer=subscription_version_serializer,
    )

    def _subscription_is_empty(subscription: SubscriptionModel) -> bool:
        return (
            subscription.id not in subscription_versions
            or subscription_versions[subscription.id].is_empty
        )

    def _subscription_overlap_period(subscription: SubscriptionModel) -> bool:
        if period is None:
            return False
        subscription_versions_timeline = subscription_versions[subscription.id]
        return any(
            period.do_overlap(overlapping_period.validity_period)
            for overlapping_period in subscription_versions_timeline.periods
        )

    empty_subscriptions = [
        subscription
        for subscription in subscriptions
        if _subscription_is_empty(subscription)
    ]

    matched_subscriptions: list[SubscriptionModel] = [
        subscription
        for subscription in subscriptions
        if not _subscription_is_empty(subscription)
        and (period is None or _subscription_overlap_period(subscription))
    ]

    return GetSubscriptionsResult(
        subscriptions=[
            Subscription(
                id=subscription.id,
                subscription_scope=subscription.subscription_scope,
                owner_ref=subscription.owner_ref,
                owner_type=subscription.owner_type,
                payload_ref=subscription.payload_ref,
                validity_period=ValidityPeriod(
                    start_date=subscription_versions[subscription.id].start_date,
                    end_date=subscription_versions[subscription.id].end_date,
                ),
                versions=subscription_versions[subscription.id],
            )
            for subscription in matched_subscriptions
        ],
        empty_subscriptions=[
            BaseSubscription(
                id=subscription.id,
                subscription_scope=subscription.subscription_scope,
                owner_ref=subscription.owner_ref,
                owner_type=subscription.owner_type,
                payload_ref=subscription.payload_ref,
            )
            for subscription in empty_subscriptions
        ],
    )

get_subscriptions_for_companies 

get_subscriptions_for_companies(
    *company_ids, subscription_type
)
Source code in components/contracting/external/subscription/fr/queries.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_subscriptions_for_companies(
    *company_ids: str, subscription_type: SubscriptionType
) -> list[Subscription]:
    if subscription_type == SubscriptionType.health_insurance:
        from components.contracting.external.subscription.fr.health_insurance import (
            get_health_subscriptions_for_companies,
        )

        return get_health_subscriptions_for_companies(*company_ids)
    elif subscription_type == SubscriptionType.prevoyance:
        from components.contracting.external.subscription.fr.prevoyance_insurance import (
            get_prevoyance_subscriptions_for_companies,
        )

        return get_prevoyance_subscriptions_for_companies(*company_ids)
    else:
        raise ValueError(f"Unknown subscription type {subscription_type}")

initialize_subscription 

initialize_subscription(
    subscription_scope,
    owner_type,
    owner_ref,
    payload_ref=None,
)

Create a subscription, and a first version matching inputs Subscription reference is returned You should create at least 1 version to the subscription by calling 'record_subscription_updates'

Source code in components/contracting/subcomponents/subscription/internal/actions/initialize_subscribtion.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def initialize_subscription(
    subscription_scope: SubscriptionScope,
    owner_type: str,
    owner_ref: str,
    payload_ref: Optional[UUID] = None,
) -> BaseSubscription:
    """
    Create a subscription, and a first version matching inputs
    Subscription reference is returned
    You should create at least 1 version to the subscription by calling 'record_subscription_updates'
    """

    subscription = SubscriptionModel(
        owner_ref=owner_ref,
        owner_type=owner_type,
        subscription_scope=subscription_scope,
        payload_ref=payload_ref,
    )

    current_session.add(subscription)
    current_session.flush()

    return BaseSubscription(
        id=subscription.id,
        subscription_scope=subscription.subscription_scope,
        owner_ref=subscription.owner_ref,
        owner_type=subscription.owner_type,
        payload_ref=subscription.payload_ref,
    )

list_periods 

list_periods(
    app_name,
    account_id=None,
    contract_ids=None,
    subscription_type=None,
    has_end_date=None,
    latest_periods_only=None,
    period_date=None,
    product_id=None,
)

List subscription periods for a given account.

Parameters:

Name Type Description Default
account_id UUID

The unique identifier of the account.

 None
app_name AppName

The name of the application.

 required
subscription_type SubscriptionType

The type of subscription to filter by. Sending None will return all subscription types.

 None
has_end_date bool

Filter periods by whether they have an end date. Defaults to None.

 None
latest_periods_only bool

If True, only the latest subscription periods are returned. Defaults to None.

 None
period_date date

The date to filter ongoing periods. This is exclusive with latest_periods_only. Defaults to None.

 None
product_id str

The product ID to filter periods by. Defaults to None.

 None

Raises:

Type Description
ValueError

If both period_date and latest_periods_only are provided or both are None.

Returns:

Type Description
list[SubscriptionPeriod]

list[SubscriptionPeriod]: A list of subscription periods matching the criteria.
Source code in components/contracting/external/subscription/api/queries.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def list_periods(
    app_name: AppName,
    account_id: UUID | None = None,
    contract_ids: list[str] | None = None,
    subscription_type: SubscriptionType | None = None,
    has_end_date: bool | None = None,
    latest_periods_only: bool | None = None,
    period_date: date | None = None,
    product_id: str | None = None,
) -> list[SubscriptionPeriod]:
    """
    List subscription periods for a given account.

    Args:
        account_id (UUID): The unique identifier of the account.
        app_name (AppName): The name of the application.
        subscription_type (SubscriptionType, optional): The type of subscription to filter by. Sending None will return all subscription types.
        has_end_date (bool, optional): Filter periods by whether they have an end date. Defaults to None.
        latest_periods_only (bool, optional): If True, only the latest subscription periods are returned. Defaults to None.
        period_date (date, optional): The date to filter ongoing periods. This is exclusive with latest_periods_only. Defaults to None.
        product_id (str, optional): The product ID to filter periods by. Defaults to None.

    Raises:
        ValueError: If both period_date and latest_periods_only are provided or both are None.

    Returns:
        list[SubscriptionPeriod]: A list of subscription periods matching the criteria.
    """
    if (period_date is None) == (latest_periods_only is None):
        raise ValueError("Either period_date or latest_periods_only must be provided")

    if account_id is not None:
        subscriptions = get_subscriptions_for_account(
            account_id=account_id,
            app_name=app_name,
            subscription_type=subscription_type,
            include_collective_retiree=True,
        )
    elif contract_ids is not None:
        subscriptions = get_subscriptions_for_contracts(
            *contract_ids,
            subscription_type=mandatory(
                subscription_type,
                "Subscription type must be provided when filtering on contract id",
            ),
            app_name=app_name,
        )
    else:
        raise ValueError("Either account_id or contract_ids must be provided")

    periods = []
    for subscription in subscriptions:
        if (
            subscription_type is not None
            and subscription.subscription_type != subscription_type
        ):
            continue
        if not subscription.periods:
            continue

        period = None
        if latest_periods_only:
            period = subscription.last_period
        elif period_date:
            period = subscription.get_ongoing_period(period_date)

        if product_id is not None:
            period = period if period and period.product.id == product_id else None

        if has_end_date is not None:
            period = (
                period
                if period and (period.end_date is not None) == has_end_date
                else None
            )

        if period is not None:
            periods.append(period)

    return periods

list_subscribed_products 

list_subscribed_products(
    account_id,
    app_name,
    subscription_type=None,
    has_end_date=None,
    latest_periods_only=None,
    period_date=None,
    product_id=None,
)

Lists subscription periods grouped by product for a given account.

Returns:

Type Description
list[SubscribedProduct]

list[SubscribedProduct]: A list of subscribed products with their associated periods.
Source code in components/contracting/external/subscription/api/queries.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def list_subscribed_products(
    account_id: UUID,
    app_name: AppName,
    subscription_type: SubscriptionType | None = None,
    has_end_date: bool | None = None,
    latest_periods_only: bool | None = None,
    period_date: date | None = None,
    product_id: str | None = None,
) -> list[SubscribedProduct]:
    """
    Lists subscription periods grouped by product for a given account.

    Returns:
        list[SubscribedProduct]: A list of subscribed products with their associated periods.
    """
    periods = list_periods(
        account_id=account_id,
        subscription_type=subscription_type,
        has_end_date=has_end_date,
        latest_periods_only=latest_periods_only,
        period_date=period_date,
        product_id=product_id,
        app_name=app_name,
    )

    product_periods = group_by(
        periods,
        # We have some contract split cases where an 'all' product was signed on contracts not sharing its professional_category
        # We're thus grouping periods by product x period's professional_category
        lambda period: (period.product, period.population.professional_category),
    )

    return [
        SubscribedProduct(product=product, periods=periods)
        for (product, _), periods in product_periods.items()
    ]

record_subscription_updates 

record_subscription_updates(
    subscription_scope,
    subscription_ref,
    updates,
    commit=True,
)

Append the ordered updates to the existing subscription, fully rewriting its versions.

Source code in components/contracting/subcomponents/subscription/internal/actions/subscription_updates.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def record_subscription_updates(
    subscription_scope: SubscriptionScope,
    subscription_ref: _SubscriptionRefType,
    updates: list[SubscriptionUpdateRequest],
    commit: bool = True,
) -> None:
    "Append the ordered updates to the existing subscription, fully rewriting its versions."

    from components.contracting.subcomponents.subscription.internal.builders import (
        append_updates,
        build_subscription_versions,
    )

    assert subscription_ref is not None

    all_updates = append_updates(
        subscription_scope=subscription_scope,
        subscription_ref=subscription_ref,
        updates=updates,
    )

    current_session.add_all(all_updates)

    current_session.query(SubscriptionVersionModel).filter(  # noqa: ALN085
        SubscriptionVersionModel.subscription_scope == subscription_scope,
        SubscriptionVersionModel.subscription_ref == str(subscription_ref),
    ).delete(synchronize_session=False)

    new_versions = build_subscription_versions(
        subscription_ref=str(subscription_ref),
        subscription_scope=subscription_scope,
        from_ordered_updates=all_updates,
    )

    current_session.add_all(new_versions)

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

types

SubscriptionMixin

get_simple_subscription_version_timeline 
get_simple_subscription_version_timeline()
Source code in components/contracting/subcomponents/subscription/public/mixins/subscription.py 
106
107
108
109
110
111
def get_simple_subscription_version_timeline(  # noqa: D102
    self,
) -> Timeline[SimpleSubscriptionVersion]:
    return self.get_subscription_version_timeline(
        _simple_subscription_version_serializer
    )
get_subscription_version_timeline 
get_subscription_version_timeline(
    subscription_version_serializer,
)
Source code in components/contracting/subcomponents/subscription/public/mixins/subscription.py 
113
114
115
116
117
118
119
120
121
122
123
124
def get_subscription_version_timeline(  # noqa: D102
    self,
    subscription_version_serializer: SubscriptionVersionSerializer[_T],
) -> Timeline[_T]:
    from components.contracting.subcomponents.subscription.internal.serialization import (
        build_timeline,
    )

    return build_timeline(
        cast("list[SubscriptionVersionModel]", self._subscription_versions),
        subscription_version_serializer,
    )
preload_subscription_timeline_option classmethod 
preload_subscription_timeline_option(
    subscription_options=tuple(), payload_options=tuple()
)

Options to preload the graph of needed relationships to later serialize the timeline of subscription version efficiently.

Depending on the serialization function, some extra preload-options might be provided through subscription_options and payload_options.

Example to preload the timeline and get ready to use payload.prevoyance_plan:

PrevoyanceContract.query.options(
    PrevoyanceContract.preload_subscription_timeline_options(
        payload_options=[joinedload(PrevoyanceSubscriptionPayload.prevoyance_plan)],
    )
)

See https://docs.sqlalchemy.org/en/latest/orm/queryguide/columns.html ⧉

Source code in components/contracting/subcomponents/subscription/public/mixins/subscription.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@classmethod
def preload_subscription_timeline_option(
    cls,
    subscription_options: Sequence[AbstractLoad] = tuple(),
    payload_options: Sequence[AbstractLoad] = tuple(),
) -> Load:
    """
    Options to preload the graph of needed relationships to later serialize
    the timeline of subscription version efficiently.

    Depending on the serialization function, some extra preload-options might
    be provided through subscription_options and payload_options.

    Example to preload the timeline and get ready to use `payload.prevoyance_plan`:

        PrevoyanceContract.query.options(
            PrevoyanceContract.preload_subscription_timeline_options(
                payload_options=[joinedload(PrevoyanceSubscriptionPayload.prevoyance_plan)],
            )
        )

    See https://docs.sqlalchemy.org/en/latest/orm/queryguide/columns.html
    """
    scope = SubscriptionScope(cls.__subscription_scope__)  # type: ignore[attr-defined]

    return selectinload(cls._subscription_versions).options(  # type: ignore[return-value]
        joinedload(
            getattr(SubscriptionVersionModel, f"{scope}_subscription"),
        ).options(
            *subscription_options,
        ),
        joinedload(
            getattr(SubscriptionVersionModel, f"{scope}_payload"),
        ).options(
            *payload_options,
        ),
    )
record_subscription_updates 
record_subscription_updates(updates, commit=True)
Source code in components/contracting/subcomponents/subscription/public/mixins/subscription.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def record_subscription_updates(  # noqa: D102
    self,
    updates: list[SubscriptionUpdateRequest],
    commit: bool = True,
) -> None:
    from components.contracting.subcomponents.subscription.public.actions import (
        record_subscription_updates as record_subscription_updates_action,
    )

    # To help mypy...
    assert isinstance(self, BaseModel)

    record_subscription_updates_action(
        subscription_scope=SubscriptionScope(self.__subscription_scope__),  # type: ignore[attr-defined]
        subscription_ref=str(self.id),
        updates=updates,
        commit=commit,
    )

    current_session.expire(self, ["_subscription_versions"])

SubscriptionPayloadMixin

SubscriptionScope

Bases: AlanBaseEnum

Subscription's API can be configured to operate within a specific scope.

DEPRECATED_mind class-attribute instance-attribute 
DEPRECATED_mind = 'mind'
be_health class-attribute instance-attribute 
be_health = 'be_health'
ca_health class-attribute instance-attribute 
ca_health = 'ca_health'
es_health class-attribute instance-attribute 
es_health = 'es_health'
fr_health class-attribute instance-attribute 
fr_health = 'fr_health'
fr_legacy_termination class-attribute instance-attribute 
fr_legacy_termination = 'fr_legacy_termination'
fr_prevoyance class-attribute instance-attribute 
fr_prevoyance = 'fr_prevoyance'
get_app_name 
get_app_name()

Returns the app name associated with the subscription scope.

Source code in components/contracting/subcomponents/subscription/public/entities.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_app_name(self) -> AppName:
    """Returns the app name associated with the subscription scope."""
    if self in {
        SubscriptionScope.fr_health,
        SubscriptionScope.fr_prevoyance,
        SubscriptionScope.fr_legacy_termination,
    }:
        return AppName.ALAN_FR

    if self == SubscriptionScope.be_health:
        return AppName.ALAN_BE

    if self == SubscriptionScope.ca_health:
        return AppName.ALAN_CA

    if self == SubscriptionScope.es_health:
        return AppName.ALAN_ES

    if self == SubscriptionScope.healthy_benefits:
        return AppName.ALAN_ES

    if self == SubscriptionScope.occupational_health:
        # It looks like this is only used by legal clauses at this point
        raise NotImplementedError(
            "Occupational Health has no concept of active app at this point"
        )

    raise ValueError(f"Unsupported subscription scope: {self}")
healthy_benefits class-attribute instance-attribute 
healthy_benefits = 'healthy_benefits'
occupational_health class-attribute instance-attribute 
occupational_health = 'occupational_health'
test class-attribute instance-attribute 
test = 'test'

SubscriptionStatusEnum

Bases: AlanBaseEnum

active class-attribute instance-attribute 
active = 'active'
ended class-attribute instance-attribute 
ended = 'ended'
ending class-attribute instance-attribute 
ending = 'ending'
pending class-attribute instance-attribute 
pending = 'pending'
upcoming class-attribute instance-attribute 
upcoming = 'upcoming'

SubscriptionUpdateRequest

Bases: NamedTuple

is_deletion instance-attribute 
is_deletion
operation_ref instance-attribute 
operation_ref
payload_ref instance-attribute 
payload_ref
validity_period instance-attribute 
validity_period

SubscriptionVersionSerializer

Bases: Generic[_T], ABC

Describe the kind of parameters our subscription version serializers can receive. The subscription and payload arguments are optional.

__call__ abstractmethod 
__call__(
    *,
    subscription_scope,
    subscription_ref,
    validity_period,
    payload_ref,
    operation_ref,
    subscription,
    payload,
    timeline_proxy,
    **kwargs
)
Source code in components/contracting/subcomponents/subscription/public/entities.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@abstractmethod
def __call__(  # type: ignore[no-untyped-def]  # noqa: D102
    self,
    *,
    subscription_scope: SubscriptionScope,
    subscription_ref: str,
    validity_period: ValidityPeriod,
    payload_ref: UUID,
    operation_ref: str | None,
    subscription: "SubscriptionMixin",
    payload: "SubscriptionPayloadMixin",
    timeline_proxy: TimelineProxy[_T],
    **kwargs,
) -> _T: ...