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

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_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,
    should_mark_errors_as_seen=False,
)

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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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.should_mark_errors_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)

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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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_errors_as_seen class-attribute instance-attribute 

should_mark_errors_as_seen = False

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

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

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

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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@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

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

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 
 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
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
@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:
    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,
    }

    on_date = inputs.on_date

    tracker = TemplateUsageTracker()

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

    if file_format == "pdf":
        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={},
                template_manifest=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.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",
    )

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

components.contracting.public.population

components.contracting.public.proposal

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 
1741
1742
1743
1744
1745
1746
1747
1748
1749
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
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
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
1845
1846
1847
1848
1849
1850
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,
    )

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 
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
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
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 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
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 
296
297
298
299
300
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 
320
321
322
323
324
325
326
327
328
329
330
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 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
@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 
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
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
@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.signed_documents),
                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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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

    raise ValueError(f"Unsupported subscription scope: {self}")
healthy_benefits class-attribute instance-attribute 
healthy_benefits = 'healthy_benefits'
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 
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
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 
28
29
30
31
32
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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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

    raise ValueError(f"Unsupported subscription scope: {self}")
healthy_benefits class-attribute instance-attribute 
healthy_benefits = 'healthy_benefits'
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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@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: ...