Api reference

components.recovery.public.actions

formal_notice

Formal notice executor: sends a registered postal letter and notification email.

FormalNoticeAudience

Bases: StrEnum

Audience for notice letter used to adapt legal communication.

b2b = company contract; b2c = individual contract;

b2b class-attribute instance-attribute

b2b = 'b2b'

b2c class-attribute instance-attribute

b2c = 'b2c'

FormalNoticeEmailSender

Bases: Protocol

Contract for country-specific formal notice email dispatch.

Receives pre-formatted display strings and dispatches the notification email through the country's mail infrastructure.

__call__

__call__(
    *,
    email,
    language,
    recipient_name,
    contract_name,
    invoice_number,
    total_debt_amount,
    next_action_date,
    recovery_timeline_entries,
    virtual_iban_data,
    uses_professional_dashboard
)

Dispatch one formal notice notification email to a single recipient.

Source code in components/recovery/public/actions/formal_notice.py

def __call__(
    self,
    *,
    email: str,
    language: str,
    recipient_name: str,
    contract_name: str,
    invoice_number: str,
    total_debt_amount: str,
    next_action_date: str,
    recovery_timeline_entries: list[dict[str, str]],
    virtual_iban_data: dict[str, str] | None,
    uses_professional_dashboard: bool,
) -> Any:
    """Dispatch one formal notice notification email to a single recipient."""
    ...

FormalNoticeExecutor

FormalNoticeExecutor(
    adapter_class,
    *,
    country_code,
    invoicer,
    postal_use_case_key,
    payment_delay_days,
    letter_configs,
    send_email
)

Send a formal notice registered letter and notification email.

Reusable executor for any recovery plan that needs to send a formal notice letter and email notification. Requires: - BalanceRecoveryAdapter subclass for country-specific data retrieval - country_code for postal letter routing - invoicer AlanLegalEntity used to resolve the legal mention via BillingSubscription.get_legal_mention and the footer signature logo - postal_use_case_key matching a registered postal use case - payment_delay_days legal formal deadline for payment - letter_configs dict mapping language → FormalNoticeLetterConfig - FormalNoticeEmailSender for dispatching notification emails

Source code in components/recovery/public/actions/formal_notice.py

def __init__(
    self,
    adapter_class: type[BalanceRecoveryAdapter],
    *,
    country_code: CountryPrefix,
    invoicer: AlanLegalEntity,
    postal_use_case_key: str,
    payment_delay_days: int,
    letter_configs: dict[Language, FormalNoticeLetterConfig],
    send_email: FormalNoticeEmailSender,
) -> None:
    self._adapter_class = adapter_class
    self._country_code = country_code
    self._invoicer = invoicer
    self._postal_use_case_key = postal_use_case_key
    self._payment_delay_days = payment_delay_days
    self._letter_configs = letter_configs
    self._send_email = send_email

__call__

__call__(case)

Generate formal notice PDF, send via postal letter, send notification emails.

Source code in components/recovery/public/actions/formal_notice.py

def __call__(self, case: RecoveryCase) -> None:
    """Generate formal notice PDF, send via postal letter, send notification emails."""
    contract_identifier = case.contract_identifier
    adapter = self._adapter_class()

    total_debt_amount, currency = adapter.get_contract_balance(contract_identifier)
    letter_recipient = adapter.resolve_letter_recipients(contract_identifier)
    email_recipients = adapter.resolve_notification_recipients(contract_identifier)
    if not email_recipients:
        current_logger.warning(
            "No recipients for formal notice letter sent email",
            case_id=case.id,
            contract_ref=case.contract_ref,
            contract_type=case.contract_type,
        )

    contract_display_info = adapter.get_contract_display_info(
        contract_identifier, utctoday()
    )

    virtual_iban = mandatory(adapter.get_virtual_iban(contract_identifier))

    newest_unpaid_invoice = adapter.get_newest_unpaid_invoice(contract_identifier)
    invoice_number = (
        newest_unpaid_invoice.invoice_number if newest_unpaid_invoice else ""
    )

    self.send_letter(
        case=case,
        recipient=letter_recipient,
        contract_display_info=contract_display_info,
        total_debt_amount=total_debt_amount,
        currency=currency,
        virtual_iban=virtual_iban,
        payment_reference=invoice_number,
    )

    self.send_emails(
        case=case,
        adapter=adapter,
        recipients=email_recipients,
        contract_display_name=contract_display_info.display_name,
        invoice_number=invoice_number,
        total_debt_amount=total_debt_amount,
        currency=currency,
        virtual_iban=virtual_iban,
    )

send_emails

send_emails(
    *,
    case,
    adapter,
    recipients,
    contract_display_name,
    invoice_number,
    total_debt_amount,
    currency,
    virtual_iban
)

Send notification emails to all recipients.

Source code in components/recovery/public/actions/formal_notice.py

def send_emails(
    self,
    *,
    case: RecoveryCase,
    adapter: BalanceRecoveryAdapter,
    recipients: list[RecoveryNotificationRecipient],
    contract_display_name: str,
    invoice_number: str,
    total_debt_amount: int,
    currency: Currency,
    virtual_iban: VirtualIBAN,
) -> None:
    """Send notification emails to all recipients."""
    timeline = case.timeline

    virtual_iban_data: dict[str, str] = {
        "iban": virtual_iban.iban,
        "bic": virtual_iban.bic,
        "account_holder_name": virtual_iban.account_holder_name,
    }

    for recipient in recipients:
        language = recipient.language.value
        next_action_date = (
            long_date(timeline.next_upcoming.date, language)
            if timeline.next_upcoming
            else ""
        )
        if is_dry_run():
            current_logger.info(
                "Would have sent formal notice email",
                email=recipient.email,
                contract_ref=case.contract_ref,
                contract_type=case.contract_type,
            )
        else:
            self._send_email(
                email=recipient.email,
                language=language,
                recipient_name=recipient.full_name,
                contract_name=contract_display_name,
                invoice_number=invoice_number,
                total_debt_amount=currency.format_amount_with_symbol(
                    total_debt_amount, language
                ),
                next_action_date=next_action_date,
                recovery_timeline_entries=[
                    {
                        "date": long_date(e.date, language),
                        "description": adapter.translate_timeline_entry(
                            recipient.language,
                            str(case.contract_type),
                            e.action_name,
                        ),
                        "status": e.status.value,
                    }
                    for e in timeline.entry_values
                ],
                virtual_iban_data=virtual_iban_data,
                uses_professional_dashboard=recipient.uses_professional_dashboard,
            )

send_letter

send_letter(
    *,
    case,
    recipient,
    contract_display_info,
    total_debt_amount,
    currency,
    virtual_iban,
    payment_reference
)

Generate letter PDF and send as registered postal letter.

Source code in components/recovery/public/actions/formal_notice.py

def send_letter(
    self,
    *,
    case: RecoveryCase,
    recipient: RecoveryLetterRecipient,
    contract_display_info: ContractDisplayInfo,
    total_debt_amount: int,
    currency: Currency,
    virtual_iban: VirtualIBAN,
    payment_reference: str,
) -> None:
    """Generate letter PDF and send as registered postal letter."""
    language = recipient.language
    config = self._letter_configs.get(language)
    if not config:
        config = self._letter_configs[Language.ENGLISH]
    if (
        not recipient.address.street
        and recipient.address.postal_code
        and recipient.address.locality
        and recipient.address.country
    ):
        current_logger.error(
            "A complete address is required to send a formal notice postal letter",
            case_id=case.id,
            contract_ref=case.contract_ref,
            contract_type=case.contract_type,
            recipient=recipient,
        )

    if case.reference_date is None:
        current_logger.error(
            "Cannot send formal notice: case has no reference_date",
            case_id=case.id,
            contract_ref=case.contract_ref,
            contract_type=case.contract_type,
        )
        return

    is_company = recipient.is_company
    is_individual_contract = (
        not is_company and case.contract_type == ContractType.health
    )

    audience = FormalNoticeAudience.b2b if is_company else FormalNoticeAudience.b2c

    billing_subscription = BillingSubscription(
        contract_identifier=case.contract_identifier,
        customer_type=(
            BillingCustomerType.company
            if is_company
            else BillingCustomerType.individual
        ),
        invoicer=self._invoicer,
        country_of_business=self._country_code,
        currency=currency,
    )
    footer_legal_mention = billing_subscription.get_legal_mention(language)

    pdf_file = generate_formal_notice_pdf(
        language=language,
        total_debt_amount=currency.format_amount_with_symbol(
            total_debt_amount, language.value
        ),
        current_date=long_date(utctoday(), language.value),
        sending_city=config.sending_city,
        contract_display_name=contract_display_info.display_name,
        contract_display_ref=contract_display_info.display_ref,
        contract_start_date=long_date(
            contract_display_info.start_date, language.value
        ),
        first_unpaid_due_date=long_date(case.reference_date, language.value),
        full_name=recipient.full_name,
        is_company=is_company,
        is_individual_contract=is_individual_contract,
        vat_number_label=config.vat_number_label,
        vat_number=recipient.vat_number,
        virtual_iban=virtual_iban.iban,
        virtual_bic=virtual_iban.bic,
        account_holder_name=virtual_iban.account_holder_name,
        payment_reference=payment_reference,
        payment_delay_days=self._payment_delay_days,
        legal_paragraph=config.legal_paragraph[(case.contract_type, audience)],
        legal_reference_line=config.legal_reference_line[case.contract_type],
        footer_tagline=config.footer_tagline,
        footer_legal_text=footer_legal_mention.footer_text,
        footer_logo_filename=self._invoicer.miniature_logo_filename,
    )

    if is_dry_run():
        current_logger.info(
            "Would have sent formal notice letter",
            case_id=case.id,
            contract_ref=case.contract_ref,
            contract_type=case.contract_type,
        )
        return

    result = send_letter(
        SendPostalLetterRequest(
            context=PostalContext(
                country_code=self._country_code,
                use_case_key=self._postal_use_case_key,
            ),
            description=f"formal_notice_{case.contract_type}_{case.contract_ref}",
            postal_address=PostalAddress(
                recipient_name=recipient.full_name,
                street=recipient.address.street,
                postal_code=mandatory(recipient.address.postal_code),
                city=mandatory(recipient.address.locality),
                country=recipient.address.country,
            ),
            source_file=SourceFilePayload(
                source_file_type=SourceFileType.file,
                content=pdf_file,
            ),
            in_color=False,
            postage_type=PostageTypePriority.registered_letter,
            metadata={
                "case_id": str(case.id),
                "contract_ref": case.contract_ref,
                "contract_type": case.contract_type,
                "action": "formal_notice",
                "user_id": recipient.recipient_id,
            },
        )
    )
    current_logger.info(
        "Formal notice letter sent",
        letter_id=result.letter_id,
        case_id=case.id,
        contract_ref=case.contract_ref,
        contract_type=case.contract_type,
    )

FormalNoticeLetterConfig dataclass

FormalNoticeLetterConfig(
    *,
    sending_city,
    vat_number_label,
    legal_paragraph,
    legal_reference_line,
    footer_tagline=None
)

Per-language template config for formal notice postal letter documents.

footer_tagline class-attribute instance-attribute

footer_tagline = None

legal_paragraph instance-attribute

legal_paragraph

legal_reference_line instance-attribute

legal_reference_line

sending_city instance-attribute

sending_city

vat_number_label instance-attribute

vat_number_label

formal_notice_pdf

PDF generator for the formal notice letter.

generate_formal_notice_pdf

generate_formal_notice_pdf(
    *,
    language,
    total_debt_amount,
    current_date,
    sending_city,
    contract_display_name,
    contract_display_ref,
    contract_start_date,
    first_unpaid_due_date,
    full_name,
    is_company,
    is_individual_contract,
    vat_number_label,
    vat_number,
    virtual_iban,
    virtual_bic,
    account_holder_name,
    payment_reference,
    payment_delay_days,
    legal_paragraph,
    legal_reference_line,
    footer_tagline=None,
    footer_legal_text=None,
    footer_logo_filename=None
)

Generate formal notice PDF.

Source code in components/recovery/public/actions/formal_notice_pdf.py

def generate_formal_notice_pdf(
    *,
    language: Language,
    total_debt_amount: str,
    current_date: str,
    sending_city: str,
    contract_display_name: str,
    contract_display_ref: str,
    contract_start_date: str,
    first_unpaid_due_date: str,
    full_name: str,
    is_company: bool,
    is_individual_contract: bool,
    vat_number_label: str | None,
    vat_number: str | None,
    virtual_iban: str,
    virtual_bic: str,
    account_holder_name: str,
    payment_reference: str,
    payment_delay_days: int,
    legal_paragraph: str,
    legal_reference_line: str,
    footer_tagline: str | None = None,
    footer_legal_text: str | None = None,
    footer_logo_filename: str | None = None,
) -> IO[bytes]:
    """Generate formal notice PDF."""
    template = _TEMPLATE_BY_LANGUAGE.get(
        language, _TEMPLATE_BY_LANGUAGE[Language.ENGLISH]
    )
    rendered_file = tempfile.NamedTemporaryFile(suffix=".pdf")
    PDFWriter.template2pdf(
        template,
        rendered_file.name,
        TOTAL_DEBT_AMOUNT=total_debt_amount,
        CURRENT_DATE=current_date,
        SENDING_CITY=sending_city,
        CONTRACT_DISPLAY_NAME=contract_display_name,
        CONTRACT_DISPLAY_REF=contract_display_ref,
        CONTRACT_START_DATE=contract_start_date,
        FIRST_UNPAID_DUE_DATE=first_unpaid_due_date,
        FULL_NAME=full_name,
        IS_COMPANY=is_company,
        IS_INDIVIDUAL_CONTRACT=is_individual_contract,
        VAT_NUMBER_LABEL=vat_number_label,
        VAT_NUMBER=vat_number,
        VIRTUAL_IBAN=virtual_iban,
        VIRTUAL_BIC=virtual_bic,
        ACCOUNT_HOLDER_NAME=account_holder_name,
        PAYMENT_REFERENCE=payment_reference,
        PAYMENT_DELAY_DAYS=payment_delay_days,
        LEGAL_PARAGRAPH=legal_paragraph,
        LEGAL_REFERENCE_LINE=legal_reference_line,
        FOOTER_TAGLINE=footer_tagline,
        FOOTER_LEGAL_TEXT=footer_legal_text,
        INVOICER_MINIATURE_LOGO_FILENAME=footer_logo_filename,
    )
    return cast("IO[bytes]", rendered_file)

payment_failure

Payment failure notification executor and variant enum.

PaymentFailureEmailSender

Bases: Protocol

Contract for country-specific payment failure email dispatch.

Implementations receive pre-formatted display strings from the executor and are responsible for building the email and dispatching it through the country's mail infrastructure.

Example BE implementation: a function decorated with @be_async_mailer that returns MailerParams, the decorator handles async queueing, rendering, delivery, and audit logging.

__call__

__call__(
    *,
    email,
    language,
    recipient_name,
    contract_name,
    invoice_month,
    invoice_number,
    invoice_amount,
    total_debt_amount,
    variant,
    next_action_date,
    recovery_timeline_entries,
    virtual_iban_data,
    uses_professional_dashboard,
    show_prior_invoices_unpaid_amount_line
)

Dispatch one payment failure notification email to a single recipient.

Source code in components/recovery/public/actions/payment_failure.py

def __call__(
    self,
    *,
    email: str,
    language: str,
    recipient_name: str,
    contract_name: str,
    invoice_month: str,
    invoice_number: str,
    invoice_amount: str,
    total_debt_amount: str,
    variant: str,
    next_action_date: str,
    recovery_timeline_entries: list[dict[str, str]],
    virtual_iban_data: dict[str, str] | None,
    uses_professional_dashboard: bool,
    show_prior_invoices_unpaid_amount_line: bool,
) -> Any:
    """Dispatch one payment failure notification email to a single recipient."""
    ...

PaymentFailureNotificationExecutor

PaymentFailureNotificationExecutor(
    adapter_class, send_email
)

Send a payment failure email to the contractee.

Reusable executor for any recovery plan action that needs to notify about a failed payment. Requires a BalanceRecoveryAdapter subclass and a country-specific PaymentFailureEmailSender implementation::

executor = PaymentFailureNotificationExecutor(
    MyBalanceRecoveryAdapter, send_email=my_send_email_function
)

The email variant is determined from the failure or dispute reason of the latest payment on the newest unpaid invoice.

Pair with HasFailedPaymentCondition to guard plan actions so this executor only runs when a failed payment actually exists.

Source code in components/recovery/public/actions/payment_failure.py

def __init__(
    self,
    adapter_class: type[BalanceRecoveryAdapter],
    send_email: PaymentFailureEmailSender,
) -> None:
    self._adapter_class = adapter_class
    self._send_email = send_email

__call__

__call__(case)

Retrieve the last failed payment and total debt, determine the failure variant, and dispatch emails.

Source code in components/recovery/public/actions/payment_failure.py

def __call__(self, case: RecoveryCase) -> None:
    """Retrieve the last failed payment and total debt, determine the failure variant, and dispatch emails."""
    contract_identifier = case.contract_identifier
    adapter = self._adapter_class()

    failed_payment = adapter.get_last_payment_if_failed(contract_identifier)
    if failed_payment is None:
        current_logger.warning(
            "Payment failure action triggered but no failed payment found",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    invoice = adapter.get_newest_unpaid_invoice(contract_identifier)
    if invoice is None:
        current_logger.warning(
            "Payment failure action triggered but no unpaid invoice found",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    total_debt_amount, currency = adapter.get_contract_balance(contract_identifier)
    show_prior_invoices_unpaid_amount_line = invoice.amount != total_debt_amount
    variant = PaymentFailureVariant.from_reason(failed_payment.failure_reason)
    contract_name = adapter.get_contract_display_info(
        contract_identifier, utctoday()
    ).display_name
    recipients = adapter.resolve_notification_recipients(contract_identifier)
    virtual_iban = adapter.get_virtual_iban(contract_identifier)
    virtual_iban_data: dict[str, str] | None = (
        {
            "iban": virtual_iban.iban,
            "bic": virtual_iban.bic,
            "account_holder_name": virtual_iban.account_holder_name,
        }
        if virtual_iban
        else None
    )

    if not recipients:
        current_logger.warning(
            "No recipients resolved for payment failure notification",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    timeline = case.timeline

    for recipient in recipients:
        language = recipient.language.value
        next_action_date = (
            long_date(timeline.next_upcoming.date, language)
            if timeline.next_upcoming
            else ""
        )
        if is_dry_run():
            current_logger.info(
                "Would have sent a payment failure email",
                email=recipient.email,
                variant=variant.value,
                contract_ref=case.contract_ref,
                invoice_number=invoice.invoice_number,
            )
        else:
            self._send_email(
                email=recipient.email,
                language=language,
                recipient_name=recipient.full_name,
                contract_name=contract_name,
                invoice_month=month_year(invoice.period, language),
                invoice_number=invoice.invoice_number,
                invoice_amount=currency.format_amount_with_symbol(
                    invoice.amount, language
                ),
                total_debt_amount=currency.format_amount_with_symbol(
                    total_debt_amount, language
                ),
                variant=variant.value,
                next_action_date=next_action_date,
                recovery_timeline_entries=[
                    {
                        "date": long_date(e.date, language),
                        "description": adapter.translate_timeline_entry(
                            recipient.language,
                            str(case.contract_type),
                            e.action_name,
                        ),
                        "status": e.status.value,
                    }
                    for e in timeline.entry_values
                ],
                virtual_iban_data=virtual_iban_data,
                uses_professional_dashboard=recipient.uses_professional_dashboard,
                show_prior_invoices_unpaid_amount_line=show_prior_invoices_unpaid_amount_line,
            )

PaymentFailureVariant

Bases: StrEnum

Domain variants for the payment failure email.

cannot_debit class-attribute instance-attribute

cannot_debit = 'cannot_debit'

The debit was rejected due to a problem with the payment method itself.

from_reason classmethod

from_reason(reason)

Map a failure/dispute reason to the appropriate variant.

Source code in components/recovery/public/actions/payment_failure.py

@classmethod
def from_reason(
    cls,
    reason: PaymentFailureReason | PaymentDisputeReason | str | None,
) -> PaymentFailureVariant:
    """Map a failure/dispute reason to the appropriate variant."""
    if reason is None:
        return cls.payment_declined

    variant_map: dict[str, PaymentFailureVariant] = {
        PaymentFailureReason.insufficient_funds.value: cls.insufficient_funds,
        PaymentFailureReason.missing_mandate.value: cls.missing_mandate,
        PaymentFailureReason.incorrect_account_holder_name.value: cls.cannot_debit,
        PaymentFailureReason.invalid_account.value: cls.cannot_debit,
        PaymentDisputeReason.insufficient_funds.value: cls.insufficient_funds,
        PaymentDisputeReason.incorrect_account_details.value: cls.cannot_debit,
    }
    return variant_map.get(str(reason), cls.payment_declined)

insufficient_funds class-attribute instance-attribute

insufficient_funds = 'insufficient_funds'

The SEPA debit was attempted but the account balance was low.

missing_mandate class-attribute instance-attribute

missing_mandate = 'missing_mandate'

No valid SEPA direct debit mandate at the time of the debit.

payment_declined class-attribute instance-attribute

payment_declined = 'payment_declined'

Payment was declined, used for failed or disputed payments or as a generic fallback.

suspension

Benefits suspension executors.

SuspensionEmailToBeneficiarySender

Bases: Protocol

Protocol for country-specific suspension email dispatch to the beneficiary audience.

Implementations receive pre-formatted display values from the executor and are responsible for building the email and dispatching it through the country's mail infrastructure.

Example BE implementation: a function decorated with @be_async_mailer that returns MailerParams, the decorator handles async queueing, rendering, delivery, and audit logging.

__call__

__call__(
    *,
    email,
    language,
    contract_name,
    recipient_name,
    contract_termination_date
)

Send one suspension email to a single beneficiary.

Source code in components/recovery/public/actions/suspension.py

def __call__(
    self,
    *,
    email: str,
    language: str,
    contract_name: str,
    recipient_name: str,
    contract_termination_date: str,
) -> Any:
    """Send one suspension email to a single beneficiary."""
    ...

SuspensionEmailToContracteeSender

Bases: Protocol

Protocol for country-specific suspension email dispatch to the contractee audience.

Implementations receive pre-formatted display values from the executor and are responsible for building the email and dispatching it through the country's mail infrastructure.

Example BE implementation: a function decorated with @be_async_mailer that returns MailerParams, the decorator handles async queueing, rendering, delivery and audit logging.

__call__

__call__(
    *,
    email,
    language,
    contract_name,
    recipient_name,
    invoice_number,
    total_debt_amount,
    contract_termination_date,
    uses_professional_dashboard,
    virtual_iban_data,
    recovery_timeline_entries
)

Send one suspension email to a single contractee.

Source code in components/recovery/public/actions/suspension.py

def __call__(
    self,
    *,
    email: str,
    language: str,
    contract_name: str,
    recipient_name: str,
    invoice_number: str,
    total_debt_amount: str,
    contract_termination_date: str,
    uses_professional_dashboard: bool,
    virtual_iban_data: dict[str, str] | None,
    recovery_timeline_entries: list[dict[str, str]],
) -> Any:
    """Send one suspension email to a single contractee."""
    ...

SuspensionInstallExecutor

SuspensionInstallExecutor(
    adapter_class,
    *,
    send_email_to_beneficiary,
    send_email_to_contractee,
    termination_action_name
)

Suspend benefits of an entity in recovery and notify its recipients.

Reusable executor for any recovery plan action that needs to block reimbursements of an entity and inform the corresponsing contractees and beneficiaries that their benefits have been suspended.

Requires: - BalanceRecoveryAdapter subclass for country-specific data retrieval; - send_email_to_beneficiary country-specific email sender targeting the entity which is not responsible for settling the debt but is impacted by the suspension; - send_email_to_contractee country-specific email sender targeting the entity which contracted and is responsible for handling the debt; - termination_action_name country-specific name of the recovery plan action which terminates the contract;

Source code in components/recovery/public/actions/suspension.py

def __init__(
    self,
    adapter_class: type[BalanceRecoveryAdapter],
    *,
    send_email_to_beneficiary: SuspensionEmailToBeneficiarySender,
    send_email_to_contractee: SuspensionEmailToContracteeSender,
    termination_action_name: str,
) -> None:
    self._adapter_class = adapter_class
    self._send_email_to_beneficiary = send_email_to_beneficiary
    self._send_email_to_contractee = send_email_to_contractee
    self._termination_action_name = termination_action_name

__call__

__call__(case)

Install a reimbursement blocker on the recovery case's entity and notify the contractees and beneficiaries.

Source code in components/recovery/public/actions/suspension.py

def __call__(self, case: RecoveryCase) -> None:
    """
    Install a reimbursement blocker on the recovery case's entity
    and notify the contractees and beneficiaries.
    """
    contract_identifier = case.contract_identifier
    adapter = self._adapter_class()
    suspension_target = adapter.resolve_suspension_target(contract_identifier)

    install_reimbursement_blocker(
        entity_identifier=suspension_target.entity_identifier,
        entity_identifier_type=suspension_target.entity_identifier_type,
        country=suspension_target.country,
        reason=BlockReimbursementsReason.unpaid_invoices,
        block_from=utctoday(),
        save=False,
    )

    _dispatch_contractee_emails(
        case=case,
        adapter=adapter,
        send_email=self._send_email_to_contractee,
        termination_action_name=self._termination_action_name,
    )

    _dispatch_beneficiary_emails(
        case=case,
        adapter=adapter,
        send_email=self._send_email_to_beneficiary,
        termination_action_name=self._termination_action_name,
    )

SuspensionLiftExecutor

SuspensionLiftExecutor(adapter_class)

Lift the benefits suspension from an entity in recovery.

Reusable executor for any recovery plan action that needs to deactivate a reimbursement blocker.

Requires: - BalanceRecoveryAdapter subclass for country-specific data retrieval;

Source code in components/recovery/public/actions/suspension.py

def __init__(self, adapter_class: type[BalanceRecoveryAdapter]) -> None:
    self._adapter_class = adapter_class

__call__

__call__(case)

Lift a reimbursement blocker from the recovery case's entity.

Source code in components/recovery/public/actions/suspension.py

def __call__(self, case: RecoveryCase) -> None:
    """Lift a reimbursement blocker from the recovery case's entity."""
    adapter = self._adapter_class()
    suspension_target = adapter.resolve_suspension_target(case.contract_identifier)

    deactivate_reimbursement_blocker(
        entity_identifier=suspension_target.entity_identifier,
        entity_identifier_type=suspension_target.entity_identifier_type,
        country=suspension_target.country,
        reason=BlockReimbursementsReason.unpaid_invoices,
        commit=False,
    )

suspension_warning

Suspension warning notification executor.

SuspensionWarningEmailSender

Bases: Protocol

Contract for country-specific suspension warning email dispatch.

Implementations receive pre-formatted display strings from the executor and are responsible for building the email and dispatching it through the country's mail infrastructure.

Example BE implementation: a function decorated with @be_async_mailer that returns MailerParams, the decorator handles async queueing, rendering, delivery, and audit logging.

__call__

__call__(
    *,
    email,
    language,
    contract_name,
    recipient_name,
    invoice_number,
    total_debt_amount,
    contract_termination_date,
    recovery_timeline_entries,
    virtual_iban_data,
    uses_professional_dashboard
)

Dispatch one suspension warning email to a single recipient.

Source code in components/recovery/public/actions/suspension_warning.py

def __call__(
    self,
    *,
    email: str,
    language: str,
    contract_name: str,
    recipient_name: str,
    invoice_number: str,
    total_debt_amount: str,
    contract_termination_date: str,
    recovery_timeline_entries: list[dict[str, str]],
    virtual_iban_data: dict[str, str] | None,
    uses_professional_dashboard: bool,
) -> Any:
    """Dispatch one suspension warning email to a single recipient."""
    ...

SuspensionWarningExecutor

SuspensionWarningExecutor(
    adapter_class, *, send_email, termination_action_name
)

Send a suspension warning email to the contractee.

Reusable executor for any recovery plan action that needs to warn about an upcoming benefits suspension.

Requires: - BalanceRecoveryAdapter subclass for country-specific data retrieval - send_email country/product-specific email sender that builds and dispatches the email through the country's mail infrastructure. - termination_action_name contry/product-specific name of a recovery plan action that terminates the contract.

Source code in components/recovery/public/actions/suspension_warning.py

def __init__(
    self,
    adapter_class: type[BalanceRecoveryAdapter],
    *,
    send_email: SuspensionWarningEmailSender,
    termination_action_name: str,
) -> None:
    self._adapter_class = adapter_class
    self._send_email = send_email
    self._termination_action_name = termination_action_name

__call__

__call__(case)

Retrieve the newest unpaid invoice and total debt, then dispatch emails.

Source code in components/recovery/public/actions/suspension_warning.py

def __call__(self, case: RecoveryCase) -> None:
    """Retrieve the newest unpaid invoice and total debt, then dispatch emails."""
    contract_identifier = case.contract_identifier
    adapter = self._adapter_class()

    invoice = adapter.get_newest_unpaid_invoice(contract_identifier)
    if invoice is None:
        current_logger.warning(
            "Suspension warning triggered but no unpaid invoice found",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    total_debt_amount, currency = adapter.get_contract_balance(contract_identifier)
    contract_display_info = adapter.get_contract_display_info(
        contract_identifier, utctoday()
    )
    recipients = adapter.resolve_notification_recipients(contract_identifier)
    virtual_iban = adapter.get_virtual_iban(contract_identifier)
    virtual_iban_data: dict[str, str] | None = (
        {
            "iban": virtual_iban.iban,
            "bic": virtual_iban.bic,
            "account_holder_name": virtual_iban.account_holder_name,
        }
        if virtual_iban
        else None
    )

    if not recipients:
        current_logger.warning(
            "No recipients resolved for suspension warning",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    termination_entry = case.timeline.find_by_action_name(
        self._termination_action_name
    )

    for recipient in recipients:
        language = recipient.language

        contract_termination_date = (
            long_date(termination_entry.date, language.value)
            if termination_entry
            else ""
        )

        if is_dry_run():
            current_logger.info(
                "Would have sent a suspension warning email",
                email=recipient.email,
                contract_ref=case.contract_ref,
                invoice_number=invoice.invoice_number,
            )
            continue

        self._send_email(
            email=recipient.email,
            language=language.value,
            recipient_name=recipient.full_name,
            contract_name=contract_display_info.display_name,
            invoice_number=invoice.invoice_number,
            total_debt_amount=currency.format_amount_with_symbol(
                total_debt_amount, language.value
            ),
            contract_termination_date=contract_termination_date,
            recovery_timeline_entries=[
                {
                    "date": long_date(e.date, language.value),
                    "description": adapter.translate_timeline_entry(
                        language,
                        str(case.contract_type),
                        e.action_name,
                    ),
                    "status": e.status.value,
                }
                for e in case.timeline.entry_values
            ],
            virtual_iban_data=virtual_iban_data,
            uses_professional_dashboard=recipient.uses_professional_dashboard,
        )

unpaid_invoice

Unpaid invoice notification executor and variant enum.

UnpaidInvoiceEmailSender

Bases: Protocol

Contract for country-specific unpaid invoice email dispatch.

Implementations receive pre-formatted display strings from the executor and are responsible for building the email and dispatching it through the country's mail infrastructure.

Example BE implementation: a function decorated with @be_async_mailer that returns MailerParams, the decorator handles async queueing, rendering, delivery, and audit logging.

__call__

__call__(
    *,
    email,
    language,
    recipient_name,
    contract_name,
    invoice_month,
    invoice_number,
    invoice_amount,
    total_debt_amount,
    remaining_amount,
    variant,
    next_action_date,
    recovery_timeline_entries,
    virtual_iban_data,
    uses_professional_dashboard,
    show_prior_invoices_unpaid_amount_line
)

Dispatch one unpaid invoice notification email to a single recipient.

Source code in components/recovery/public/actions/unpaid_invoice.py

def __call__(
    self,
    *,
    email: str,
    language: str,
    recipient_name: str,
    contract_name: str,
    invoice_month: str,
    invoice_number: str,
    invoice_amount: str,
    total_debt_amount: str,
    remaining_amount: str,
    variant: str,
    next_action_date: str,
    recovery_timeline_entries: list[dict[str, str]],
    virtual_iban_data: dict[str, str] | None,
    uses_professional_dashboard: bool,
    show_prior_invoices_unpaid_amount_line: bool,
) -> Any:
    """Dispatch one unpaid invoice notification email to a single recipient."""
    ...

UnpaidInvoiceNotificationExecutor

UnpaidInvoiceNotificationExecutor(
    adapter_class, send_email
)

Send an unpaid invoice notification email to the contractee.

Reusable executor for any recovery plan action that needs to notify about an unpaid invoice when there is no failed, blocked, or disputed payment. Requires a BalanceRecoveryAdapter subclass and a country-specific UnpaidInvoiceEmailSender implementation::

executor = UnpaidInvoiceNotificationExecutor(
    MyBalanceRecoveryAdapter, send_email=my_send_email_function
)

The email variant (partial_payment, late_payment, missing_debit, existing_debt) is determined from the invoice settlement status, payment history, and contract payment method.

Pair with NotCondition(HasFailedPaymentCondition(...)) to guard plan actions so this executor only runs when no failed payment exists.

Source code in components/recovery/public/actions/unpaid_invoice.py

def __init__(
    self,
    adapter_class: type[BalanceRecoveryAdapter],
    send_email: UnpaidInvoiceEmailSender,
) -> None:
    self._adapter_class = adapter_class
    self._send_email = send_email

__call__

__call__(case)

Retrieve the newest unpaid invoice, determine the variant, and dispatch emails.

Source code in components/recovery/public/actions/unpaid_invoice.py

def __call__(self, case: RecoveryCase) -> None:
    """Retrieve the newest unpaid invoice, determine the variant, and dispatch emails."""
    contract_identifier = case.contract_identifier
    adapter = self._adapter_class()

    invoice = adapter.get_newest_unpaid_invoice(contract_identifier)
    if invoice is None:
        current_logger.warning(
            "Unpaid invoice notification triggered but no unpaid invoice found",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    total_debt_amount, currency = adapter.get_contract_balance(contract_identifier)
    show_prior_invoices_unpaid_amount_line = invoice.amount != total_debt_amount
    variant = self._determine_variant(invoice, adapter, contract_identifier)
    contract_name = adapter.get_contract_display_info(
        contract_identifier, utctoday()
    ).display_name
    recipients = adapter.resolve_notification_recipients(contract_identifier)
    virtual_iban = adapter.get_virtual_iban(contract_identifier)
    virtual_iban_data: dict[str, str] | None = (
        {
            "iban": virtual_iban.iban,
            "bic": virtual_iban.bic,
            "account_holder_name": virtual_iban.account_holder_name,
        }
        if virtual_iban
        else None
    )

    if not recipients:
        current_logger.warning(
            "No recipients resolved for unpaid invoice notification",
            case_id=str(case.id),
            contract_ref=case.contract_ref,
        )
        return

    timeline = case.timeline

    for recipient in recipients:
        language = recipient.language.value
        next_action_date = (
            long_date(timeline.next_upcoming.date, language)
            if timeline.next_upcoming
            else ""
        )
        if is_dry_run():
            current_logger.info(
                "Would have sent an unpaid invoice email",
                email=recipient.email,
                variant=variant.value,
                contract_ref=case.contract_ref,
                invoice_number=invoice.invoice_number,
            )
        else:
            self._send_email(
                email=recipient.email,
                language=language,
                recipient_name=recipient.full_name,
                contract_name=contract_name,
                invoice_month=month_year(invoice.period, language),
                invoice_number=invoice.invoice_number,
                invoice_amount=currency.format_amount_with_symbol(
                    invoice.amount, language
                ),
                total_debt_amount=currency.format_amount_with_symbol(
                    total_debt_amount, language
                ),
                remaining_amount=currency.format_amount_with_symbol(
                    invoice.remaining_balance, language
                ),
                variant=variant.value,
                next_action_date=next_action_date,
                recovery_timeline_entries=[
                    {
                        "date": long_date(e.date, language),
                        "description": adapter.translate_timeline_entry(
                            recipient.language,
                            str(case.contract_type),
                            e.action_name,
                        ),
                        "status": e.status.value,
                    }
                    for e in timeline.entry_values
                ],
                virtual_iban_data=virtual_iban_data,
                uses_professional_dashboard=recipient.uses_professional_dashboard,
                show_prior_invoices_unpaid_amount_line=show_prior_invoices_unpaid_amount_line,
            )

UnpaidInvoiceNotificationVariant

Bases: StrEnum

Domain variants for the unpaid invoice notification email.

existing_debt class-attribute instance-attribute

existing_debt = 'existing_debt'

Invoice is considered unsettled due to pre-existing old debt.

late_payment class-attribute instance-attribute

late_payment = 'late_payment'

Payment not yet received for contracts paid by credit transfer.

missing_debit class-attribute instance-attribute

missing_debit = 'missing_debit'

Invoice has no payments and was not charged, typically indicates a missing payment method.

partial_payment class-attribute instance-attribute

partial_payment = 'partial_payment'

Invoice was paid but only partially, some amount remains unpaid.

components.recovery.public.adapters

BalanceRecoveryAdapter

Bases: RecoveryAdapter

Adapter for balance-based recovery (unpaid invoices).

get_contract_balance abstractmethod

get_contract_balance(contract_identifier)

Recovery-relevant balance in minor currency units. Positive = debt.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_contract_balance(
    self, contract_identifier: ContractIdentifier
) -> tuple[int, Currency]:
    """Recovery-relevant balance in minor currency units. Positive = debt."""
    ...

get_contract_display_info abstractmethod

get_contract_display_info(contract_identifier, on_date)

Return display info for this contract as of on_date.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_contract_display_info(
    self, contract_identifier: ContractIdentifier, on_date: date
) -> ContractDisplayInfo:
    """Return display info for this contract as of ``on_date``."""
    ...

get_last_payment_if_failed abstractmethod

get_last_payment_if_failed(contract_identifier)

Return the last failed/blocked/disputed payment on the newest unpaid invoice.

Returns None if the latest payment is not failed or no unpaid invoice exists.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_last_payment_if_failed(
    self, contract_identifier: ContractIdentifier
) -> PaymentInfo | None:
    """Return the last failed/blocked/disputed payment on the newest unpaid invoice.

    Returns None if the latest payment is not failed or no unpaid invoice exists.
    """
    ...

get_newest_unpaid_invoice abstractmethod

get_newest_unpaid_invoice(contract_identifier)

Return the newest unpaid invoice for this contract.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_newest_unpaid_invoice(
    self, contract_identifier: ContractIdentifier
) -> UnpaidInvoice | None:
    """Return the newest unpaid invoice for this contract."""
    ...

get_oldest_unpaid_invoice abstractmethod

get_oldest_unpaid_invoice(contract_identifier)

Return the oldest unpaid invoice for this contract.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_oldest_unpaid_invoice(
    self, contract_identifier: ContractIdentifier
) -> UnpaidInvoice | None:
    """Return the oldest unpaid invoice for this contract."""
    ...

get_recovery_metadata

get_recovery_metadata(contract_identifier)

Return current balance and unpaid invoice ids as BalanceMetadata, attached to recovery events during detection and evaluation.

Source code in components/recovery/public/adapters.py

def get_recovery_metadata(
    self, contract_identifier: ContractIdentifier
) -> RecoveryEventMetadata:
    """Return current balance and unpaid invoice ids as ``BalanceMetadata``,
    attached to recovery events during detection and evaluation.
    """
    balance_amount, currency = self.get_contract_balance(contract_identifier)
    invoices = self.get_unpaid_invoices(contract_identifier)
    return BalanceMetadata(
        balance_amount=balance_amount,
        currency=currency.alphabetic_code,
        unpaid_invoice_ids=[str(inv.invoice_id) for inv in invoices],
    )

get_unpaid_invoices abstractmethod

get_unpaid_invoices(contract_identifier)

Return unpaid invoices for this contract. Used by metadata and useful for recovery actions and conditions.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_unpaid_invoices(
    self, contract_identifier: ContractIdentifier
) -> list[UnpaidInvoice]:
    """Return unpaid invoices for this contract. Used by metadata
    and useful for recovery actions and conditions.
    """
    ...

get_virtual_iban abstractmethod

get_virtual_iban(contract_identifier)

Return virtual IBAN details to settle debt by bank transfer payments.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_virtual_iban(
    self, contract_identifier: ContractIdentifier
) -> VirtualIBAN | None:
    """Return virtual IBAN details to settle debt by bank transfer payments."""
    ...

invoice_has_payments abstractmethod

invoice_has_payments(invoice_id)

Whether the invoice has any associated payments.

Source code in components/recovery/public/adapters.py

@abstractmethod
def invoice_has_payments(self, invoice_id: uuid.UUID) -> bool:
    """Whether the invoice has any associated payments."""
    ...

is_credit_transfer_contract abstractmethod

is_credit_transfer_contract(contract_identifier)

Whether this contract uses credit transfer (wire) as payment method.

Source code in components/recovery/public/adapters.py

@abstractmethod
def is_credit_transfer_contract(
    self, contract_identifier: ContractIdentifier
) -> bool:
    """Whether this contract uses credit transfer (wire) as payment method."""
    ...

is_resolved

is_resolved(contract_identifier)

Resolved when balance is zero or negative.

Source code in components/recovery/public/adapters.py

def is_resolved(self, contract_identifier: ContractIdentifier) -> bool:
    """Resolved when balance is zero or negative."""
    amount, _ = self.get_contract_balance(contract_identifier)
    return amount <= 0

resolve_letter_recipients abstractmethod

resolve_letter_recipients(contract_identifier)

Return the data needed to send a postal letter for this contract.

e.g: single recipient: the company itself for company contracts, the primary member for personal contracts.

Source code in components/recovery/public/adapters.py

@abstractmethod
def resolve_letter_recipients(
    self, contract_identifier: ContractIdentifier
) -> RecoveryLetterRecipient:
    """Return the data needed to send a postal letter for this contract.

    e.g: single recipient: the company itself for company contracts, the
    primary member for personal contracts.
    """
    ...

resolve_notification_recipients abstractmethod

resolve_notification_recipients(contract_identifier)

Return email recipients for recovery notifications for this contract.

Source code in components/recovery/public/adapters.py

@abstractmethod
def resolve_notification_recipients(
    self, contract_identifier: ContractIdentifier
) -> list[RecoveryNotificationRecipient]:
    """Return email recipients for recovery notifications for this contract."""
    ...

resolve_suspension_target abstractmethod

resolve_suspension_target(contract_identifier)

Return a suspension target for this contract.

Source code in components/recovery/public/adapters.py

@abstractmethod
def resolve_suspension_target(
    self, contract_identifier: ContractIdentifier
) -> SuspensionTarget:
    """Return a suspension target for this contract."""
    ...

translate_timeline_entry abstractmethod

translate_timeline_entry(
    language, contract_type, action_name
)

Return localized description for a recovery timeline action.

Source code in components/recovery/public/adapters.py

@abstractmethod
def translate_timeline_entry(
    self, language: Language, contract_type: str, action_name: str
) -> str:
    """Return localized description for a recovery timeline action."""
    ...

ContractDisplayInfo dataclass

ContractDisplayInfo(
    *, display_name, display_ref, start_date
)

Display info for a contract used in recovery notifications.

display_name instance-attribute

display_name

display_ref instance-attribute

display_ref

start_date instance-attribute

start_date

PaymentInfo dataclass

PaymentInfo(*, payment_id, status, failure_reason)

Info about a payment on an invoice.

failure_reason instance-attribute

failure_reason

Failure or dispute reason value (e.g. PaymentFailureReason.insufficient_funds).

payment_id instance-attribute

payment_id

status instance-attribute

status

RecoveryAdapter

Bases: ABC

Base adapter for any recovery type.

get_contracts_requiring_recovery_case abstractmethod

get_contracts_requiring_recovery_case(already_in_recovery)

Identify contracts that require opening a new recovery case.

Implementors must:

1. Query contracts matching recovery criteria.
2. Exclude contracts already_in_recovery.
3. Exclude contracts where is_excluded() returns True.
4. Apply local timing/threshold filters if any.

Example for balance recovery: query active contracts with positive balance, exclude those with balance <= 15 EUR, exclude those whose newest invoice is too recent for recovery.

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_contracts_requiring_recovery_case(
    self, already_in_recovery: set[ContractIdentifier]
) -> list[ContractIdentifier]:
    """Identify contracts that require opening a new recovery case.

    Implementors must:

    1. Query contracts matching recovery criteria.
    2. Exclude contracts ``already_in_recovery``.
    3. Exclude contracts where ``is_excluded()`` returns True.
    4. Apply local timing/threshold filters if any.

    Example for balance recovery: query active contracts with positive balance,
    exclude those with balance <= 15 EUR, exclude those whose newest invoice
    is too recent for recovery.
    """
    ...

get_recovery_metadata abstractmethod

get_recovery_metadata(contract_identifier)

Contextual data attached to recovery events created during engine detection and evaluation, for trail and debugging.

Example for balance recovery:: BalanceMetadata(balance_amount=15000, currency="EUR", unpaid_invoice_ids=[12345])

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_recovery_metadata(
    self, contract_identifier: ContractIdentifier
) -> RecoveryEventMetadata:
    """Contextual data attached to recovery events created during engine
    detection and evaluation, for trail and debugging.

    Example for balance recovery::
        BalanceMetadata(balance_amount=15000, currency="EUR", unpaid_invoice_ids=[12345])
    """
    ...

get_recovery_reference_date abstractmethod

get_recovery_reference_date(contract_identifier)

Date reference for plan action scheduling. Set once at case creation, never updated.

Example: if reference_date is Jan 31 (invoice due date), an action with delay_from_reference_date=timedelta(days=15) fires on or after Feb 15.

Returns None if no meaningful date exists (actions rely solely on delay_from_previous).

Source code in components/recovery/public/adapters.py

@abstractmethod
def get_recovery_reference_date(
    self, contract_identifier: ContractIdentifier
) -> date | None:
    """Date reference for plan action scheduling. Set once at case
    creation, never updated.

    Example: if reference_date is Jan 31 (invoice due date), an action
    with ``delay_from_reference_date=timedelta(days=15)`` fires on or
    after Feb 15.

    Returns None if no meaningful date exists (actions rely solely on
    ``delay_from_previous``).
    """
    ...

is_excluded abstractmethod

is_excluded(contract_identifier)

Whether this contract is exempt from recovery.

Examples: special agreement, contract under legal dispute, test/internal contract.

Existing recovery cases are closed immediately during evaluation. The same conditions should also be checked during detection by get_contracts_requiring_recovery_case to prevent opening a new case in the next engine run.

Source code in components/recovery/public/adapters.py

@abstractmethod
def is_excluded(self, contract_identifier: ContractIdentifier) -> bool:
    """Whether this contract is exempt from recovery.

    Examples: special agreement, contract under legal dispute,
    test/internal contract.

    Existing recovery cases are closed immediately during evaluation.
    The same conditions should also be checked during detection by
    ``get_contracts_requiring_recovery_case`` to prevent opening a new
    case in the next engine run.
    """
    ...

is_resolved abstractmethod

is_resolved(contract_identifier)

Whether the recovery condition is satisfied.

Example for balance recovery: resolved when outstanding balance <= 0 (debt fully paid or credited). Once resolved, the case enters a safeguard period before closing.

Source code in components/recovery/public/adapters.py

@abstractmethod
def is_resolved(self, contract_identifier: ContractIdentifier) -> bool:
    """Whether the recovery condition is satisfied.

    Example for balance recovery: resolved when outstanding balance <= 0
    (debt fully paid or credited). Once resolved, the case enters a safeguard
    period before closing.
    """
    ...

RecoveryLetterRecipient dataclass

RecoveryLetterRecipient(
    *,
    recipient_id,
    is_company,
    full_name,
    address,
    language,
    vat_number=None
)

The addressee of a recovery postal letter.

address instance-attribute

address

full_name instance-attribute

full_name

is_company instance-attribute

is_company

language instance-attribute

language

recipient_id instance-attribute

recipient_id

Company ID when is_company is True; User ID for personal contracts.

vat_number class-attribute instance-attribute

vat_number = None

Company VAT number when is_company is True; None for personal contracts.

RecoveryNotificationRecipient dataclass

RecoveryNotificationRecipient(
    *,
    email,
    language,
    full_name,
    uses_professional_dashboard=False
)

A recipient for recovery notification emails.

email instance-attribute

email

full_name instance-attribute

full_name

language instance-attribute

language

uses_professional_dashboard class-attribute instance-attribute

uses_professional_dashboard = False

True when the recipient uses the company (pro/admin) dashboard.

SuspensionTarget dataclass

SuspensionTarget(
    *, entity_identifier, entity_identifier_type, country
)

Entity whose benefits should be suspended.

An "entity" is any uniquely identifiable object that can be used to resolve a reimbursement beneficiary in the claims engine (e.g. account, company, profile, policy, contract).

This class contains properties required to identify an entity to install a reimbursement blocker on.

country instance-attribute

country

entity_identifier instance-attribute

entity_identifier

entity_identifier_type instance-attribute

entity_identifier_type

UnpaidInvoice dataclass

UnpaidInvoice(
    *,
    invoice_id,
    invoice_number,
    period,
    amount,
    currency,
    due_date,
    remaining_balance,
    settlement_status
)

Unpaid invoice details used for metadata, conditions, and notifications.

amount instance-attribute

amount

Total invoice amount in minor currency units (e.g. cents for EUR).

currency instance-attribute

currency

due_date instance-attribute

due_date

invoice_id instance-attribute

invoice_id

invoice_number instance-attribute

invoice_number

period instance-attribute

period

remaining_balance instance-attribute

remaining_balance

settlement_status instance-attribute

settlement_status

components.recovery.public.blueprints

recovery_blueprint

recovery_blueprint module-attribute

recovery_blueprint = CustomBlueprint(
    "recovery",
    __name__,
    cli_group=None,
    template_folder="templates",
)

components.recovery.public.commands

evaluate_recovery_case

evaluate_recovery_case(case_id, country, contract_type)

Evaluate a single recovery case.

concurrency_group prevents two jobs for the same case running in parallel. deduplicate=True prevents duplicate jobs: if a job of the same case already enqueued, it won't be enqueued again.

Source code in components/recovery/public/commands.py

@enqueueable(
    concurrency_group=lambda case_id, **_: str(case_id),
    deduplicate=True,
)
def evaluate_recovery_case(
    case_id: UUID,
    country: CountryPrefix,
    contract_type: ContractType,
) -> None:
    """Evaluate a single recovery case.

    ``concurrency_group`` prevents two jobs for the same case running in parallel.
    ``deduplicate=True`` prevents duplicate jobs: if a job of the same case already enqueued, it won't be enqueued again.
    """
    service = RecoveryService(country=country, contract_type=contract_type)

    try:
        result = service.evaluate_case(case_id)
        current_logger.info(
            "case_evaluated",
            case_id=case_id,
            lifecycle_event=result.lifecycle_event.event_name
            if result.lifecycle_event
            else None,
            action_event=result.action_event.event_name
            if result.action_event
            else None,
            pending_action=result.pending_action,
            noop_reason=result.noop_reason,
            actions_bypassed=result.actions_bypassed,
        )
    except Exception:
        current_logger.exception("case_evaluation_failed", case_id=case_id)

recovery_commands module-attribute

recovery_commands = AppGroup('global_recovery')

run_recovery

run_recovery(country, contract_type, run_async)

Detect new recovery cases and evaluate existing ones.

Source code in components/recovery/public/commands.py

@recovery_commands.command("run")
@click.option(
    "--country", required=True, type=CountryPrefix, help="2-letter country code"
)
@click.option("--contract-type", required=True, type=ContractType, help="Contract type")
@click.option(
    "--async",
    "run_async",
    is_flag=True,
    default=False,
    help="Case evaluations as async jobs",
)
@command_with_dry_run
def run_recovery(
    country: CountryPrefix, contract_type: ContractType, run_async: bool
) -> None:
    """Detect new recovery cases and evaluate existing ones."""
    service = RecoveryService(country=country, contract_type=contract_type)

    detection = service.detect_and_open_new_cases()
    current_logger.info(
        "detection_complete",
        created=detection.created,
        skipped=detection.skipped,
        errors=detection.errors,
    )

    cases = service.get_evaluable_cases()

    if run_async:
        if not is_dry_run():
            queue = current_rq.get_queue(BILLING_LOW_QUEUE)
            for case in cases:
                queue.enqueue(evaluate_recovery_case, case.id, country, contract_type)
            current_logger.info("Recovery case evaluations enqueued", count=len(cases))
        else:
            current_logger.info(
                "Would have enqueued recovery case evaluations", count=len(cases)
            )
        return

    for case in cases:
        try:
            result = service.evaluate_case(case.id)
            current_logger.info(
                "case_evaluated",
                case_id=case.id,
                lifecycle_event=result.lifecycle_event.event_name
                if result.lifecycle_event
                else None,
                action_event=result.action_event.event_name
                if result.action_event
                else None,
                pending_action=result.pending_action,
                noop_reason=result.noop_reason,
                actions_bypassed=result.actions_bypassed,
            )
        except Exception:
            current_logger.exception("case_evaluation_failed", case_id=case.id)

components.recovery.public.conditions

Shared recovery action conditions.

AndCondition

AndCondition(*conditions)

Conjunction of recovery action conditions: True if all wrapped conditions return True.

Source code in components/recovery/public/conditions.py

def __init__(self, *conditions: RecoveryActionCondition) -> None:
    self._conditions = conditions

__call__

__call__(case)

Return True if every wrapped condition returns True.

Source code in components/recovery/public/conditions.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return True if every wrapped condition returns True."""
    return all(condition(case) for condition in self._conditions)

HasFailedPaymentCondition

HasFailedPaymentCondition(adapter_class)

Recovery action condition: true when the newest unpaid invoice's last payment is failed, blocked, or disputed.

Source code in components/recovery/public/conditions.py

def __init__(self, adapter_class: type[BalanceRecoveryAdapter]) -> None:
    self._adapter_class = adapter_class

__call__

__call__(case)

Return True if the contract has a failed payment on its newest unpaid invoice.

Source code in components/recovery/public/conditions.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return True if the contract has a failed payment on its newest unpaid invoice."""
    adapter = self._adapter_class()
    failed_payment = adapter.get_last_payment_if_failed(case.contract_identifier)

    return failed_payment is not None

HasNewerUnpaidInvoiceCondition

HasNewerUnpaidInvoiceCondition(adapter_class)

Recovery action condition: true when the newest unpaid invoice's due date is later than the case's reference date.

Used to detect that a new billing cycle's invoice is unpaid (not just the original one the case opened on). Returns False if there's no newest unpaid.

Source code in components/recovery/public/conditions.py

def __init__(self, adapter_class: type[BalanceRecoveryAdapter]) -> None:
    self._adapter_class = adapter_class

__call__

__call__(case)

Return True if there is a newest unpaid invoice issued after the case's reference date.

Source code in components/recovery/public/conditions.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return True if there is a newest unpaid invoice issued after the case's reference date."""
    if case.reference_date is None:
        return False

    adapter = self._adapter_class()
    newest_unpaid = adapter.get_newest_unpaid_invoice(case.contract_identifier)
    if newest_unpaid is None:
        return False

    return newest_unpaid.due_date > case.reference_date

HasOutstandingBalanceCondition

HasOutstandingBalanceCondition(adapter_class)

Recovery action condition: true if contract balance is positive.

Source code in components/recovery/public/conditions.py

def __init__(self, adapter_class: type[BalanceRecoveryAdapter]) -> None:
    self._adapter_class = adapter_class

__call__

__call__(case)

Return True while the balance is still positive.

Source code in components/recovery/public/conditions.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return True while the balance is still positive."""
    adapter = self._adapter_class()
    amount, _ = adapter.get_contract_balance(case.contract_identifier)
    return amount > 0

NotCondition

NotCondition(condition)

Negates another recovery action condition.

Source code in components/recovery/public/conditions.py

def __init__(self, condition: RecoveryActionCondition) -> None:
    self._condition = condition

__call__

__call__(case)

Return the negation of the wrapped condition.

Source code in components/recovery/public/conditions.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return the negation of the wrapped condition."""
    return not self._condition(case)

ReferenceDateNotPassedCondition

Recovery action condition: true when the case's reference_date is today or in the future.

__call__

__call__(case)

Return True if reference_date is set and not yet passed.

Source code in components/recovery/public/conditions.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return True if reference_date is set and not yet passed."""
    return case.reference_date is not None and case.reference_date >= utctoday()

components.recovery.public.entities

RecoveryCase dataclass

RecoveryCase(
    *,
    id,
    contract_ref,
    contract_type,
    country,
    status,
    reference_date,
    created_at,
    updated_at,
    events
)

Read-only view of a recovery case.

contract_identifier property

contract_identifier

ContractIdentifier derived from contract_ref + contract_type.

contract_ref instance-attribute

contract_ref

contract_type instance-attribute

contract_type

country instance-attribute

country

created_at instance-attribute

created_at

events instance-attribute

events

id instance-attribute

id

reference_date instance-attribute

reference_date

status instance-attribute

status

timeline property

timeline

Build projected timeline by resolving plan from registry.

updated_at instance-attribute

updated_at

RecoveryCaseStatus

Bases: AlanBaseEnum

Status of a recovery case.

A status change occurs following: - Automatic transitions driven by RecoveryEngine lifecycle check. - Manual transitions (e.g. put_on_hold) via RecoveryService.

active class-attribute instance-attribute

active = 'active'

Case is actively being processed through recovery actions.

closed class-attribute instance-attribute

closed = 'closed'

Case is fully closed (after safeguard period for resolution, termination, or exclusion).

on_hold class-attribute instance-attribute

on_hold = 'on_hold'

Case is paused (e.g. deadline extension, manual hold).

resolved class-attribute instance-attribute

resolved = 'resolved'

Balance cleared, waiting resolution_safeguard_period before closing.

Prevents premature closure since payments can be disputed or reversed. If debt returns during safeguard, case is reactivated instead of opening a new case and restarting the recovery plan.

RecoveryEvent dataclass

RecoveryEvent(
    *,
    id,
    recovery_case_id,
    category,
    event_name,
    metadata,
    actor_global_user_id,
    created_at
)

Read-only view of a recovery event.

actor_global_user_id instance-attribute

actor_global_user_id

category instance-attribute

category

created_at instance-attribute

created_at

event_name instance-attribute

event_name

id instance-attribute

id

metadata instance-attribute

metadata

recovery_case_id instance-attribute

recovery_case_id

RecoveryEventCategory

Bases: AlanBaseEnum

Category of a recovery event.

lifecycle_event class-attribute instance-attribute

lifecycle_event = 'lifecycle_event'

A case status change, automatic (via engine evaluation) or manual (via service actions). See RecoveryLifecycleEventName.

recovery_action class-attribute instance-attribute

recovery_action = 'recovery_action'

A plan action was executed. Plans are country and contract specific sequences defined via RecoveryPlan.

RecoveryLifecycleEventName

Bases: AlanBaseEnum

Name of a lifecycle event on a recovery case.

case_opened class-attribute instance-attribute

case_opened = 'case_opened'

New case created during detection, identified by RecoveryAdapter.get_contracts_requiring_recovery_case (e.g. unpaid balance above threshold for balance recovery).

closed_on_resolve class-attribute instance-attribute

closed_on_resolve = 'closed_on_resolve'

Debt resolved on a plan with no safeguard period, the case is closed immediately instead of waiting in a resolved state.

excluded class-attribute instance-attribute

excluded = 'excluded'

Contract became exempt from recovery (e.g. under legal dispute, special agreement, or contracts not eligible for regular recovery), specific rules are defined by RecoveryAdapter.is_excluded.

hold_expired class-attribute instance-attribute

hold_expired = 'hold_expired'

Hold's expires_at reached, case automatically resumes.

hold_resumed class-attribute instance-attribute

hold_resumed = 'hold_resumed'

Case manually resumed from hold before expiry, or when hold doesn't have an expiry.

plan_completed class-attribute instance-attribute

plan_completed = 'plan_completed'

All plan actions executed (or bypassed) with no resolution. Case is closed.

put_on_hold class-attribute instance-attribute

put_on_hold = 'put_on_hold'

Case manually paused (e.g. payment deadline extension granted to the member). No actions progress while on hold. Optional HoldMetadata.expires_at for auto-expiry.

reactivated class-attribute instance-attribute

reactivated = 'reactivated'

Debt returned during safeguard period (e.g. payment reversed). Actions resume from where they left off instead of restarting the plan.

resolved class-attribute instance-attribute

resolved = 'resolved'

Recovery condition satisfied (e.g. debt fully paid). Case enters safeguard period (resolution_safeguard_period) before closing.

safeguard_completed class-attribute instance-attribute

safeguard_completed = 'safeguard_completed'

Safeguard period elapsed and resolution still holds. Case is closed.

components.recovery.public.plan

Public API for defining and registering recovery plans.

Contains plan types and action protocols.

RecoveryAction dataclass

RecoveryAction(
    *,
    name,
    execute,
    delay_from_previous=None,
    delay_from_reference_date=None,
    condition=None,
    is_projected_in_timeline=True
)

One step in a recovery plan's action sequence.

condition class-attribute instance-attribute

condition = None

Optional precondition. The action is bypassed once next action's timing arrives and this condition is still False.

delay_from_previous class-attribute instance-attribute

delay_from_previous = None

Min delay since last executed action (or case creation for first action).

delay_from_reference_date class-attribute instance-attribute

delay_from_reference_date = None

Delay relative to the case's reference date (e.g. invoice due date).

execute instance-attribute

execute

Executes the action (send email, terminate contract, etc.).

is_projected_in_timeline class-attribute instance-attribute

is_projected_in_timeline = True

Whether this action appears in projected recovery timeline entries (upcoming). Actions already executed always appear in the timeline regardless of this flag.

You can set to False for conditional actions whose future execution is uncertain, when showing them as upcoming could mislead the consumer into assuming they will happen.

name instance-attribute

name

Unique name within the plan (e.g. 'formal_notice', 'terminate_contract').

RecoveryActionCondition

Bases: Protocol

Checks whether an action's precondition is met before execution.

__call__

__call__(case)

Return True if the action should proceed.

Source code in components/recovery/public/plan.py

def __call__(self, case: RecoveryCase) -> bool:
    """Return True if the action should proceed."""
    ...

RecoveryActionExecutor

Bases: Protocol

Executes a recovery action (send email, terminate contract, etc.).

__call__

__call__(case)

Execute the action.

Source code in components/recovery/public/plan.py

def __call__(self, case: RecoveryCase) -> None:
    """Execute the action."""
    ...

RecoveryPlan dataclass

RecoveryPlan(
    *,
    country,
    contract_type,
    actions,
    resolution_safeguard_period,
    on_resolved=None,
    on_closed=None
)

Recovery plan blueprint.

actions instance-attribute

actions

Ordered sequence of recovery actions executed from first to last.

contract_type instance-attribute

contract_type

country instance-attribute

country

on_closed class-attribute instance-attribute

on_closed = None

Executor to run when case is closed after safeguard period.

on_resolved class-attribute instance-attribute

on_resolved = None

Executor to run when case is resolved (e.g. lift service suspension).

resolution_safeguard_period instance-attribute

resolution_safeguard_period

How long a resolved case stays open before auto-closing.

None means no safeguard: the case is closed immediately on resolve.

components.recovery.public.plan_registry

RecoveryPlanRegistry

Registry of recovery plans keyed by (country, contract_type).

Plans are registered at app startup by each country's bootstrap. Each entry co-locates a plan with its adapter class.

Example::

_entries = {
    (CountryPrefix.france, ContractType.health): (fr_health_recovery_plan, FrHealthRecoveryAdapter),
    (CountryPrefix.belgium, ContractType.health): (be_health_recovery_plan, BeHealthRecoveryAdapter),
}

get classmethod

get(country, contract_type)

Return (plan, adapter) for a (country, contract_type) pair.

The adapter_class is instantiated via the adapter_class constructor on each call to provide a fresh adapter instance per use.

Source code in components/recovery/public/plan_registry.py

@classmethod
def get(
    cls, country: CountryPrefix, contract_type: ContractType
) -> tuple[RecoveryPlan, RecoveryAdapter]:
    """Return (plan, adapter) for a (country, contract_type) pair.

    The adapter_class is instantiated via the adapter_class constructor on each call
    to provide a fresh adapter instance per use.
    """
    key = (country, contract_type)

    if key not in cls._entries:
        raise ValueError(f"No recovery plan registered for {key}")

    plan, adapter_cls = cls._entries[key]
    return plan, adapter_cls()

register classmethod

register(plan, adapter_class)

Register a recovery plan with its adapter class. Raises if key already taken.

Stores (plan, adapter_class) under the key (plan.country, plan.contract_type).

Example::

RecoveryPlanRegistry.register(BE_HEALTH_PLAN, BeHealthRecoveryAdapter)

Source code in components/recovery/public/plan_registry.py

@classmethod
def register(
    cls,
    plan: RecoveryPlan,
    adapter_class: type[RecoveryAdapter],
) -> None:
    """Register a recovery plan with its adapter class. Raises if key already taken.

    Stores ``(plan, adapter_class)`` under the key ``(plan.country, plan.contract_type)``.

    Example::

        RecoveryPlanRegistry.register(BE_HEALTH_PLAN, BeHealthRecoveryAdapter)
    """
    key = (plan.country, plan.contract_type)

    if key in cls._entries:
        raise ValueError(f"Plan already registered for {key}")

    cls._entries[key] = (plan, adapter_class)

components.recovery.public.service

CaseDetectionResult dataclass

CaseDetectionResult(*, created=0, skipped=0, errors=0)

Summary of a recovery cases detection run.

created class-attribute instance-attribute

created = 0

errors class-attribute instance-attribute

errors = 0

skipped class-attribute instance-attribute

skipped = 0

CaseEvaluationResult dataclass

CaseEvaluationResult(
    *,
    lifecycle_event=None,
    action_event=None,
    pending_action=None,
    noop_reason=None,
    actions_bypassed=()
)

Summary of evaluating a single recovery case.

action_event class-attribute instance-attribute

action_event = None

Recovery action event executed from the plan, if any.

actions_bypassed class-attribute instance-attribute

actions_bypassed = ()

Action names that were bypassed during this evaluation (condition never met, next action timing arrived).

lifecycle_event class-attribute instance-attribute

lifecycle_event = None

Lifecycle event applied (e.g. resolved, excluded, reactivated).

noop_reason class-attribute instance-attribute

noop_reason = None

Why the evaluation produced no effect (no lifecycle transition, no action executed).

pending_action class-attribute instance-attribute

pending_action = None

Next Action that couldn't execute yet (timing or condition not met).

EvaluationNoopReason

Bases: AlanBaseEnum

Why case evaluation produced no effect.

case_on_hold class-attribute instance-attribute

case_on_hold = 'case_on_hold'

Case is on hold — no lifecycle check or action progression.

case_resolved class-attribute instance-attribute

case_resolved = 'case_resolved'

Case is resolved and waiting for safeguard period to elapse.

condition_not_met class-attribute instance-attribute

condition_not_met = 'condition_not_met'

Next action's condition is False and following action's timing hasn't arrived yet.

from_status classmethod

from_status(status)

Map a non-active case status to its noop reason.

Source code in components/recovery/internal/engine/results.py

@classmethod
def from_status(cls, status: RecoveryCaseStatus) -> EvaluationNoopReason:
    """Map a non-active case status to its noop reason."""
    from components.recovery.internal.domain.enums import RecoveryCaseStatus

    match status:
        case RecoveryCaseStatus.on_hold:
            return cls.case_on_hold
        case RecoveryCaseStatus.resolved:
            return cls.case_resolved
        case _:
            raise ValueError(f"No noop reason for recovery case status {status}")

timing_not_met class-attribute instance-attribute

timing_not_met = 'timing_not_met'

Next action's timing constraints not yet satisfied.

RecoveryService

RecoveryService(*, country, contract_type)

Public facade for recovery case detection and evaluation.

Wraps RecoveryEngine for a specific (country, contract_type) pair, resolved from the RecoveryPlanRegistry. Callers (typically commands) use three methods:

1. detect_and_open_new_cases: find eligible contracts and create cases.
2. get_evaluable_cases: list non-closed cases to feed into evaluation.
3. evaluate_case: run lifecycle check + action progression on one case.

Each method manages its own transaction; callers need no session handling.

Source code in components/recovery/public/service.py

def __init__(self, *, country: CountryPrefix, contract_type: ContractType) -> None:
    self._country = country
    self._contract_type = contract_type
    plan, adapter = RecoveryPlanRegistry.get(country, contract_type)
    self._engine = RecoveryEngine(plan=plan, adapter=adapter)

detect_and_open_new_cases

detect_and_open_new_cases()

Scan for contracts eligible for recovery and open new cases.

Per-contract errors are caught and counted (not raised), so detection is partially tolerant to individual failures.

Source code in components/recovery/public/service.py

def detect_and_open_new_cases(self) -> CaseDetectionResult:
    """Scan for contracts eligible for recovery and open new cases.

    Per-contract errors are caught and counted (not raised), so detection
    is partially tolerant to individual failures.
    """
    return self._engine.detect_and_open_new_cases()

evaluate_case

evaluate_case(case_id)

Evaluate a single recovery case: lifecycle check then action progression.

Runs in its own transaction. Plan callbacks (on_resolved, on_closed) and action side-effects execute in nested sub-transactions, if any fails, the entire evaluation rolls back and the case remains unchanged.

Source code in components/recovery/public/service.py

def evaluate_case(self, case_id: uuid.UUID) -> CaseEvaluationResult:
    """Evaluate a single recovery case: lifecycle check then action progression.

    Runs in its own transaction. Plan callbacks (``on_resolved``, ``on_closed``)
    and action side-effects execute in nested sub-transactions, if any fails,
    the entire evaluation rolls back and the case remains unchanged.
    """
    return self._engine.evaluate_case(case_id)

get_evaluable_cases

get_evaluable_cases()

Return all non-closed cases as frozen public entities.

Source code in components/recovery/public/service.py

def get_evaluable_cases(self) -> list[RecoveryCase]:
    """Return all non-closed cases as frozen public entities."""
    internal_cases = self._engine.get_evaluable_cases()
    return [c.to_public() for c in internal_cases]

get_open_case_by_contract_identifier

get_open_case_by_contract_identifier(contract_identifier)

Return the open (non-closed) case for a contract identifier, or None if none is open.

Source code in components/recovery/public/service.py

def get_open_case_by_contract_identifier(
    self, contract_identifier: ContractIdentifier
) -> RecoveryCase | None:
    """Return the open (non-closed) case for a contract identifier, or None if none is open."""
    with transaction(propagation=Propagation.READ_ONLY) as session:
        repo = RecoveryCaseRepository(session)
        recovery_case = repo.get_open_case_by_contract_ref(
            self._country, self._contract_type, contract_identifier.contract_ref
        )
    return recovery_case.to_public() if recovery_case is not None else None

components.recovery.public.timeline

RecoveryTimeline

Bases: Timeline['RecoveryTimelineEntry']

Projected timeline of recovery actions for a recovery case.

entry_values property

entry_values

RecoveryTimelineEntry for each effective value, without the interval.

find_by_action_name

find_by_action_name(action_name)

Return the entry with matching action_name, or None if not in the timeline.

Source code in components/recovery/public/timeline.py

def find_by_action_name(self, action_name: str) -> RecoveryTimelineEntry | None:
    """Return the entry with matching ``action_name``, or None if not in the timeline."""
    return next(
        (e for e in self.entry_values if e.action_name == action_name), None
    )

last_executed property

last_executed

Last executed action, or None if no actions executed yet.

next_upcoming property

next_upcoming

Next upcoming action, or None if all actions executed.

RecoveryTimelineEntry dataclass

RecoveryTimelineEntry(*, action_name, status, date)

One entry in a recovery timeline projection.

action_name instance-attribute

action_name

Action name from the recovery plan (e.g. 'formal_notice').

date instance-attribute

date

Execution date if executed, projected earliest trigger date if upcoming.

status instance-attribute

status

Current status of this action for the recovery case (executed or upcoming).

TimelineEntryStatus

Bases: AlanBaseEnum

Status of a recovery timeline entry.

executed class-attribute instance-attribute

executed = 'executed'

Action was executed (event exists in recovery case history).

upcoming class-attribute instance-attribute

upcoming = 'upcoming'

Action has not been executed yet (projected from recovery plan actions).