Api reference

components.premium.public.api

PremiumAggregationService

A service used to interact with subscription fees (AKA premiums in the context of insurance).

Fees are created to represent the consumption of Alan products. Many services throughout the stack need to consume fees to function: - invoicing aggregates fees into invoices - payroll reports the structure of the fees to give clarity to admins - data monitors general financial performance levels

This service is meant to become the one-stop-shop for all these consumers.

Additionally, the way we manage fees is evolving. Previously, each country & product had it's own engine for computing fees, and format to store them. We are moving toward a global solution: same engine and same format across all countries and products. However this is a still a work in progress, and this service also aims to protect consumers from having to deal with different formats. All functions are meant to function the same no matter which engine was used to compute the fees and how they were stored.

get_premium_entries_for_invoice staticmethod

get_premium_entries_for_invoice(*, invoice_id)

Retrieve premium entries for a given invoice. If invoice has no components, then it retrieves legacy premium entries, and transforms them to PremiumEntry objects with components.

Parameters:

Name	Type	Description	Default
invoice_id	UUID	The ID of the invoice to retrieve premium entries for.	required

Returns:

Type	Description
list[PremiumEntry]	List of PremiumEntry objects for the invoice.

Source code in components/premium/public/api.py

@staticmethod
def get_premium_entries_for_invoice(
    *,
    invoice_id: UUID,
) -> list[PremiumEntry]:
    """
    Retrieve premium entries for a given invoice.
    If invoice has no components, then it retrieves legacy premium entries, and transforms them to PremiumEntry objects with components.

    Args:
        invoice_id: The ID of the invoice to retrieve premium entries for.

    Returns:
        List of PremiumEntry objects for the invoice.
    """
    dependencies = get_app_dependency()
    premium_repository = dependencies.get_premium_repository(session=None)

    premium_entries = premium_repository.get_premium_entries_for_invoice(invoice_id)

    if len(premium_entries) == 0:
        legacy_premium_repository = (
            dependencies.get_legacy_premium_entry_repository(session=None)
        )
        premium_entries = legacy_premium_repository.get_premium_entries_for_invoice(
            invoice_id
        )

    if len(premium_entries) > 0:
        return premium_entries

    raise ValueError("No premium entries found for invoice", invoice_id)

get_premiums_by_payroll_csv_id staticmethod

get_premiums_by_payroll_csv_id(session, *, payroll_csv_id)

Retrieve all premium entries stamped with a given payroll CSV ID.

Tries the global repository first. If no entries are found there, falls back to the legacy repository.

Parameters:

Name	Type	Description	Default
session	Session	SQLAlchemy session injected by @transactional decorator.	required
payroll_csv_id	UUID	The payroll CSV ID to filter on.	required

Returns:

Type	Description
list[PremiumEntry]	List of PremiumEntry objects linked to the payroll CSV.

Source code in components/premium/public/api.py

@staticmethod
@transactional(
    propagation=Propagation.READ_ONLY, acceptable_replica_lag=timedelta(minutes=15)
)
def get_premiums_by_payroll_csv_id(
    session: Session,
    *,
    payroll_csv_id: UUID,
) -> list[PremiumEntry]:
    """
    Retrieve all premium entries stamped with a given payroll CSV ID.

    Tries the global repository first. If no entries are found there,
    falls back to the legacy repository.

    Args:
        session: SQLAlchemy session injected by @transactional decorator.
        payroll_csv_id: The payroll CSV ID to filter on.

    Returns:
        List of PremiumEntry objects linked to the payroll CSV.
    """
    dependencies = get_app_dependency()
    repository = dependencies.get_premium_repository(session)
    legacy_repository = dependencies.get_legacy_premium_entry_repository(session)

    entries = repository.get_entries_by_payroll_csv_id(payroll_csv_id)
    if entries:
        return entries
    return legacy_repository.get_entries_by_payroll_csv_id(payroll_csv_id)

get_premiums_for_local_payroll staticmethod

get_premiums_for_local_payroll(
    session,
    *,
    enrollment_ids,
    target_month,
    force_premium_source
)

Retrieve all premium entries to be included in a local payroll CSV where

• payroll_csv_id IS NULL (global repo) or pay_csv_id IS NULL (legacy repo)
• period_start <= target_month

In practice the aggregation service is supposed to shield consumers from the 'source' of premiums. However in practice this function does not require that complexity, considering this is a temporary bridge between local & global callers know what they need.

Parameters:

Name	Type	Description	Default
session	Session	SQLAlchemy session injected by @transactional decorator.	required
enrollment_ids	list[UUID]	UUIDs of enrollments to retrieve entries for.	required
target_month	date	Only entries with period_start <= last day of this month are returned.	required
force_premium_source	Literal['legacy', 'global']	Which SOT to use when fetching premiums	required

Source code in components/premium/public/api.py

@staticmethod
@transactional(
    propagation=Propagation.READ_ONLY, acceptable_replica_lag=timedelta(minutes=15)
)
def get_premiums_for_local_payroll(
    session: Session,
    *,
    enrollment_ids: list[UUID],
    target_month: date,
    force_premium_source: Literal["legacy", "global"],
) -> list[PremiumEntry]:
    """
    Retrieve all premium entries to be included in a local payroll CSV where:
        - payroll_csv_id IS NULL (global repo) or pay_csv_id IS NULL (legacy repo)
        - period_start <= target_month

    In practice the aggregation service is supposed to shield consumers from the
    'source' of premiums.
    However in practice this function does not require that complexity,
    considering this is a temporary bridge between local & global callers know what
    they need.

    Args:
        session: SQLAlchemy session injected by @transactional decorator.
        enrollment_ids: UUIDs of enrollments to retrieve entries for.
        target_month: Only entries with period_start <= last day of this month are returned.
        force_premium_source: Which SOT to use when fetching premiums
    """
    dependencies = get_app_dependency()

    match force_premium_source:
        case "legacy":
            legacy_repository = dependencies.get_legacy_premium_entry_repository(
                session
            )
            return legacy_repository.get_entries_for_local_payroll(
                enrollment_ids=enrollment_ids, target_month=target_month
            )
        case "global":
            repository = dependencies.get_premium_repository(session)

            return repository.get_entries_for_local_payroll(
                enrollment_ids=enrollment_ids, target_month=target_month
            )

get_premiums_for_payroll staticmethod

get_premiums_for_payroll(
    session,
    *,
    enrollment_ids,
    target_period,
    created_at_lower_bound,
    created_at_upper_bound
)

Retrieve all entries where

• at least some part of the total amount is meant to be billed to the company.
• entries for the current period (targeted by target_period)
• regularisation entries for previous periods created within the specified created_at_lower_bound / created_at_upper_bound window

To shield callers from the details of how premiums are stored, this function reads from the global repository first and falls back to the legacy repository only when the global repository returns nothing. As soon as the global repository has any entry, its result is returned as-is and the legacy repository is not queried.

Considering this function is meant for the usage of payroll batch jobs, it has been designed with a certain amount of care when it comes to performance: - the read happens on a read-replica - we guarantee a set number of queries no matter how many enrollment IDs are passed (no N+1) - the process of reading 1,000 entries consumes around 19 MB of memory at peak - once all transient objects have been released by the garbage collector and only the dataclass instances remain, 1,000 entries weigh around 7 MB - you can find methodology and instructions to reproduce the measurements in the component's README

Parameters:

Name	Type	Description	Default
session	Session	SQLAlchemy session injected by @transactional decorator.	required
enrollment_ids	list[int | UUID]	IDs of enrollments to retrieve entries for.	required
target_period	date	First day of the period being processed.	required
created_at_lower_bound	date	Lower bound (exclusive) on created_at for regularization entries.	required
created_at_upper_bound	date	Upper bound (inclusive) on created_at for all entries.	required

Returns:

Type	Description
list[PremiumEntry]	List of complete PremiumEntry objects.

Source code in components/premium/public/api.py

@staticmethod
@transactional(
    propagation=Propagation.READ_ONLY, acceptable_replica_lag=timedelta(minutes=15)
)
def get_premiums_for_payroll(
    session: Session,
    *,
    enrollment_ids: list[int | UUID],
    target_period: date,
    created_at_lower_bound: date,
    created_at_upper_bound: date,
) -> list[PremiumEntry]:
    """
    Retrieve all entries where:
        - at least some part of the total amount is meant to be billed to the company.
        - entries for the current period (targeted by ``target_period``)
        - regularisation entries for previous periods created within the
          specified ``created_at_lower_bound`` / ``created_at_upper_bound`` window

    To shield callers from the details of how premiums are stored, this function
    reads from the global repository first and falls back to the legacy repository
    only when the global repository returns nothing. As soon as the global
    repository has any entry, its result is returned as-is and the legacy
    repository is not queried.

    Considering this function is meant for the usage of payroll batch jobs, it has
    been designed with a certain amount of care when it comes to performance:
        - the read happens on a read-replica
        - we guarantee a set number of queries no matter how many enrollment IDs are
          passed (no N+1)
        - the process of reading 1,000 entries consumes around 19 MB of memory
          at peak
        - once all transient objects have been released by the garbage collector
          and only the dataclass instances remain, 1,000 entries weigh around 7 MB
        - you can find methodology and instructions to reproduce the measurements
          in the component's README

    Args:
        session: SQLAlchemy session injected by @transactional decorator.
        enrollment_ids: IDs of enrollments to retrieve entries for.
        target_period: First day of the period being processed.
        created_at_lower_bound: Lower bound (exclusive) on ``created_at`` for
            regularization entries.
        created_at_upper_bound: Upper bound (inclusive) on ``created_at`` for all
            entries.

    Returns:
        List of complete PremiumEntry objects.
    """
    dependencies = get_app_dependency()
    repository = dependencies.get_premium_repository(session)
    legacy_repository = dependencies.get_legacy_premium_entry_repository(session)

    new_entries = repository.get_entries_for_payroll(
        enrollment_ids=enrollment_ids,
        target_period=target_period,
        created_at_lower_bound=created_at_lower_bound,
        created_at_upper_bound=created_at_upper_bound,
    )
    if new_entries:
        return new_entries

    return legacy_repository.get_entries_for_payroll(
        enrollment_ids=enrollment_ids,
        target_period=target_period,
        created_at_lower_bound=created_at_lower_bound,
        created_at_upper_bound=created_at_upper_bound,
    )

get_uninvoiced_premiums staticmethod

get_uninvoiced_premiums(
    *,
    policy: Policy,
    up_to: date,
    force_premium_source: Literal["legacy", "global"],
    debtor: InsuranceDebtor | None = None
) -> list[PremiumEntry]

get_uninvoiced_premiums(
    *,
    contract: Contract,
    up_to: date,
    force_premium_source: Literal["legacy", "global"],
    debtor: InsuranceDebtor | None = None
) -> list[PremiumEntry]

get_uninvoiced_premiums(
    *,
    enrollment_ids: list[int | UUID],
    debtor: InsuranceDebtor,
    up_to: date,
    force_premium_source: Literal["legacy"]
) -> list[PremiumEntry]

get_uninvoiced_premiums(
    *,
    enrollment_group_ids: list[UUID],
    debtor: InsuranceDebtor,
    up_to: date,
    force_premium_source: Literal["global"]
) -> list[PremiumEntry]

get_uninvoiced_premiums(
    *,
    policy=None,
    contract=None,
    enrollment_ids=None,
    enrollment_group_ids=None,
    up_to,
    force_premium_source,
    debtor=None
)

Retrieve premium entries with uninvoiced components, checking specifically if the company or primary part is invoiced.

Returns complete premium entries where at least one component has not been invoiced yet. The debtor is inferred from the contract type: - Policy: primary policy holder - Contract → company - you can override this mechanism by passing a debtor type directly

When passing enrollment_ids or enrollment_group_ids, debtor is required.

Exposing force_premium_source partially defeats the purpose of a strangler pattern — the caller should not need to know which engine produced the premiums.

However, auto-selecting the source is not yet safe: some enrollments have uninvoiced premiums in engine A but no entries in engine B (due to unrelated data issues). Merging both sources transparently would risk double-invoicing or including stale entries.

Until the data is cleaned up, having the caller declare the source explicitly is the safest option. The parameter is intentionally easy to remove once the migration is complete.

Parameters:

Name	Type	Description	Default
policy	Policy | None	Policy to retrieve premiums for (mutually exclusive with contract/enrollment_ids/enrollment_group_ids).	None
contract	Contract | None	Contract to retrieve premiums for (mutually exclusive with policy/enrollment_ids/enrollment_group_ids).	None
enrollment_ids	list[int | UUID] | None	Legacy enrollment IDs. Requires force_premium_source="legacy" and debtor.	None
enrollment_group_ids	list[UUID] | None	Core Stack enrollment group IDs. Requires force_premium_source="global" and debtor.	None
up_to	date	Only returns entries where period_end <= up_to.	required
force_premium_source	Literal['legacy', 'global']	Which repository to read from ("legacy" or "global").	required
debtor	InsuranceDebtor | None	Override for debtor resolution. When None, inferred from contract type.	None

Returns:

Type	Description
list[PremiumEntry]	List of complete PremiumEntry objects with uninvoiced components.

Source code in components/premium/public/api.py

@staticmethod
def get_uninvoiced_premiums(
    *,
    policy: Policy | None = None,
    contract: Contract | None = None,
    enrollment_ids: list[int | UUID] | None = None,
    enrollment_group_ids: list[UUID] | None = None,
    up_to: date,
    force_premium_source: Literal["legacy", "global"],
    debtor: InsuranceDebtor | None = None,
) -> list[PremiumEntry]:
    """
    Retrieve premium entries with uninvoiced components, checking specifically if
    the company or primary part is invoiced.

    Returns complete premium entries where at least one component has not been
    invoiced yet. The debtor is inferred from the contract type:
      - Policy: primary policy holder
      - Contract → company
      - you can override this mechanism by passing a debtor type directly

    When passing enrollment_ids or enrollment_group_ids, debtor is required.


    Exposing ``force_premium_source`` partially defeats the purpose of a strangler pattern —
    the caller should not need to know which engine produced the premiums.

    However, auto-selecting the source is not yet safe: some enrollments have uninvoiced
    premiums in engine A but no entries in engine B (due to unrelated data issues). Merging
    both sources transparently would risk double-invoicing or including stale entries.

    Until the data is cleaned up, having the caller declare the source explicitly is the
    safest option. The parameter is intentionally easy to remove once the migration is
    complete.


    Args:
        policy: Policy to retrieve premiums for (mutually exclusive with contract/enrollment_ids/enrollment_group_ids).
        contract: Contract to retrieve premiums for (mutually exclusive with policy/enrollment_ids/enrollment_group_ids).
        enrollment_ids: Legacy enrollment IDs. Requires force_premium_source="legacy" and debtor.
        enrollment_group_ids: Core Stack enrollment group IDs. Requires force_premium_source="global" and debtor.
        up_to: Only returns entries where period_end <= up_to.
        force_premium_source: Which repository to read from ("legacy" or "global").
        debtor: Override for debtor resolution. When None, inferred from contract type.

    Returns:
        List of complete PremiumEntry objects with uninvoiced components.
    """
    dependencies = get_app_dependency()

    if enrollment_ids is not None:
        if debtor is None:
            raise ValueError("debtor is required when passing enrollment_ids")
        if force_premium_source != "legacy":
            raise ValueError(
                "enrollment_ids requires force_premium_source='legacy'; "
                "use enrollment_group_ids for global mode"
            )
        legacy_repository = dependencies.get_legacy_premium_entry_repository(
            session=None
        )
        return legacy_repository.get_all_uninvoiced_entries_up_to(
            enrollment_ids=enrollment_ids,
            debtor=debtor,
            up_to=up_to,
        )

    if enrollment_group_ids is not None:
        if debtor is None:
            raise ValueError("debtor is required when passing enrollment_group_ids")
        if force_premium_source != "global":
            raise ValueError(
                "enrollment_group_ids requires force_premium_source='global'; "
                "use enrollment_ids for legacy mode"
            )
        repository = dependencies.get_premium_repository(session=None)
        return repository.get_all_uninvoiced_entries_up_to(
            enrollment_group_ids=enrollment_group_ids,
            debtor=debtor,
            up_to=up_to,
        )

    if policy:
        resolved_debtor = debtor if debtor is not None else InsuranceDebtor.primary
        resolved_enrollment_ids = [
            enrollment.id for enrollment in policy.enrollments
        ]
    elif contract:
        resolved_debtor = debtor if debtor is not None else InsuranceDebtor.company
        resolved_enrollment_ids = [
            enrollment.id
            for policy in contract.policies
            for enrollment in policy.enrollments
        ]
    else:
        raise ValueError("Either policy or contract must be provided")

    debtor = resolved_debtor

    match force_premium_source:
        case "global":
            repository = dependencies.get_premium_repository(session=None)
            return repository.get_all_uninvoiced_entries_up_to(
                enrollment_ids=resolved_enrollment_ids,
                debtor=debtor,
                up_to=up_to,
            )
        case "legacy":
            legacy_repository = dependencies.get_legacy_premium_entry_repository(
                session=None
            )
            return legacy_repository.get_all_uninvoiced_entries_up_to(
                enrollment_ids=resolved_enrollment_ids,
                debtor=debtor,
                up_to=up_to,
            )

mark_entries_for_payroll staticmethod

mark_entries_for_payroll(
    premium_entries: list[PremiumEntry],
    *,
    payroll_csv_id: UUID
) -> None

mark_entries_for_payroll(
    premium_entries: list[PremiumEntry],
    *,
    finiquito_id: UUID
) -> None

mark_entries_for_payroll(
    session,
    premium_entries,
    *,
    payroll_csv_id=None,
    finiquito_id=None
)

Mark premium entries as included in a payroll CSV or finiquito.

Exactly one of payroll_csv_id or finiquito_id must be provided. A finiquito is a Spain-specific pay_csv generated to recap the premiums of an employee when they leave the company. The finiquito_id path is Spain-specific - Belgium repositories will raise NotImplementedError.

Tries the global repository first. If no entries are found there, falls back to the legacy repository. (Spain legacy written by caller so this is just mirror with no legacy fallback.)

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
premium_entries	list[PremiumEntry]	Premium entries to mark.	required
payroll_csv_id	UUID | None	The payroll CSV ID to set on matching entries.	None
finiquito_id	UUID | None	The finiquito ID to set on matching entries (Spain only).	None

Source code in components/premium/public/api.py

@staticmethod
@transactional(force_manage=True)
def mark_entries_for_payroll(
    session: Session,
    premium_entries: list[PremiumEntry],
    *,
    payroll_csv_id: UUID | None = None,
    finiquito_id: UUID | None = None,
) -> None:
    """
    Mark premium entries as included in a payroll CSV or finiquito.

    Exactly one of ``payroll_csv_id`` or ``finiquito_id`` must be provided.
    A finiquito is a Spain-specific pay_csv generated to recap the premiums
    of an employee when they leave the company. The ``finiquito_id`` path is
    Spain-specific - Belgium repositories will raise NotImplementedError.

    Tries the global repository first. If no entries are found there,
    falls back to the legacy repository.
    (Spain legacy written by caller so this is just mirror with no legacy fallback.)

    Args:
        session: Database session.
        premium_entries: Premium entries to mark.
        payroll_csv_id: The payroll CSV ID to set on matching entries.
        finiquito_id: The finiquito ID to set on matching entries (Spain only).
    """
    if (payroll_csv_id is None) == (finiquito_id is None):
        raise ValueError(
            "Exactly one of payroll_csv_id or finiquito_id must be provided."
        )

    dependencies = get_app_dependency()
    repository = dependencies.get_premium_repository(session)

    if payroll_csv_id is not None:
        entries = [e.with_payroll_csv_id(payroll_csv_id) for e in premium_entries]
    else:
        assert finiquito_id is not None
        entries = [e.with_finiquito_id(finiquito_id) for e in premium_entries]

    updated_count = repository.update(entries)

    if get_current_app_name() == AppName.ALAN_ES:
        return

    if updated_count == 0:
        legacy_repository = dependencies.get_legacy_premium_entry_repository(
            session
        )
        legacy_repository.update(entries)

mark_premiums_as_invoiced staticmethod

mark_premiums_as_invoiced(
    session, premium_entries, *, invoice_id, debtor
)

Mark premium entries as invoiced by setting invoice_id on matching components. (Right now, Spain only writes to global mirror with no legacy fallback.)

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
premium_entries	list[PremiumEntry]	List of premium entries to mark as invoiced.	required
invoice_id	UUID	The invoice ID to set on matching components.	required
debtor	InsuranceDebtor	The billed entity whose components should be marked as invoiced.	required

Source code in components/premium/public/api.py

@staticmethod
@transactional
def mark_premiums_as_invoiced(
    session: Session,
    premium_entries: list[PremiumEntry],
    *,
    invoice_id: UUID,
    debtor: InsuranceDebtor,
) -> None:
    """
    Mark premium entries as invoiced by setting invoice_id on matching components.
    (Right now, Spain only writes to global mirror with no legacy fallback.)

    Args:
        session: Database session.
        premium_entries: List of premium entries to mark as invoiced.
        invoice_id: The invoice ID to set on matching components.
        debtor: The billed entity whose components should be marked as invoiced.
    """
    dependencies = get_app_dependency()
    repository = dependencies.get_premium_repository(session)

    entries = [
        e.with_components_marked_as_invoiced(invoice_id, debtor=debtor)
        for e in premium_entries
    ]

    updated_count = repository.update(entries)

    if get_current_app_name() == AppName.ALAN_ES:
        return

    if updated_count == 0:
        legacy_repository = dependencies.get_legacy_premium_entry_repository(
            session
        )
        legacy_repository.update(entries)

PremiumComputationService

Compute premiums.

Not responsible for storing the computed premiums

compute_insurance_premiums staticmethod

compute_insurance_premiums(
    session,
    *,
    policy,
    contract,
    start_month,
    end_month,
    engine_parameters=None
)

Compute how much money is owed to Alan and persist the results.

Parameters:

Name	Type	Description	Default
session	Session	SQLAlchemy session injected by @transactional decorator	required
policy	Policy	The insurance policy containing enrollments	required
contract	Contract	The contract defining pricing terms	required
start_month	Month	First month to calculate premiums for	required
end_month	Month	Last month to include in premium calculations	required
engine_parameters	EngineParameters | None	Optional engine configuration. If not provided, uses defaults for current application	None

Returns:

Type	Description
list[PremiumEntry]	All fresh entries returned by the computation.

Source code in components/premium/public/api.py

@staticmethod
@transactional()
def compute_insurance_premiums(
    session: Session,
    *,
    policy: Policy,
    contract: Contract,
    start_month: Month,
    end_month: Month,
    engine_parameters: EngineParameters | None = None,
) -> list[PremiumEntry]:
    """
    Compute how much money is owed to Alan and persist the results.

    Args:
        session: SQLAlchemy session injected by @transactional decorator
        policy: The insurance policy containing enrollments
        contract: The contract defining pricing terms
        start_month: First month to calculate premiums for
        end_month: Last month to include in premium calculations
        engine_parameters: Optional engine configuration. If not provided, uses
            defaults for current application

    Returns:
        All fresh entries returned by the computation.
    """
    repository = get_app_dependency().get_premium_repository(session)
    engine_parameters = engine_parameters or EngineParameters.default(
        app_name=get_current_app_name()
    )

    affiliation_builder = AffiliationTimelineBuilder(
        policy=policy,
        contract=contract,
        engine_parameters=engine_parameters,
    )

    affiliation_timeline = affiliation_builder.compute_affiliation_timeline(
        start_date=start_month.first_day,
        end_date=end_month.last_day,
    )

    app_dependency = get_app_dependency()
    cost_timeline = map_timeline(
        affiliation_timeline,
        lambda period: CostPeriod(
            validity_period=period.validity_period,
            start_reason=period.start_reason,
            cost=CorePriceAdapter.from_core_price_cost(
                PricingService.get_breakdown(
                    pricing_context=PricingContext(
                        member_specs=[
                            app_dependency.to_member_spec(spec)
                            for spec in period.beneficiary_specs
                        ],
                        rounding_strategy=engine_parameters.rounding_strategy,
                    ),
                    on_date=period.validity_period.start_date,
                )
            ),
        ),
    )

    all_new_entries = compute_subscription_fees(
        cost_timeline=cost_timeline,
        periodicity=Periodicity.monthly,
        start=start_month,
        end=end_month,
        engine_parameters=engine_parameters,
    )

    enrollment_ids = [enrollment.id for enrollment in policy.enrollments]
    existing_premiums = repository.get_latest_entries(
        enrollment_ids=enrollment_ids,
        lower_bound=start_month,
        upper_bound=end_month,
    )

    _reconcile_and_persist_premium_entries(
        session,
        all_new_entries,
        existing_premiums,
    )

    return all_new_entries

compute_insurance_premiums_for_core_stack staticmethod

compute_insurance_premiums_for_core_stack(
    session,
    enrollment_group_id,
    start_month,
    end_month,
    engine_parameters=None,
)

Core Stack premium computations, ES only for now.

Source code in components/premium/public/api.py

@staticmethod
@transactional()
def compute_insurance_premiums_for_core_stack(
    session: Session,
    enrollment_group_id: UUID,
    start_month: Month,
    end_month: Month,
    engine_parameters: EngineParameters | None = None,
) -> list[PremiumEntry]:
    """Core Stack premium computations, ES only for now."""
    if get_current_app_name() != AppName.ALAN_ES or not bool_feature_flag(
        "es-core-stack-premium-shadow", default_value=False
    ):
        current_logger.warning(
            "Core Stack premium computations only for ES under LD feature flag.",
            app_name=str(get_current_app_name()),
            flag_enabled=bool_feature_flag(
                "es-core-stack-premium-shadow", default_value=False
            ),
        )
        return []

    engine_parameters = engine_parameters or EngineParameters.default(
        app_name=get_current_app_name()
    )

    breakdown_timelines = PricingService.get_breakdown_timeline(
        pricing_source=EnrollmentGroupPricingSource(
            enrollment_group_id=enrollment_group_id
        ),
        period=(start_month.first_day, end_month.last_day),
        compact=True,
    )

    new_premiums = compute_premium_entries_from_breakdowns(
        breakdown_timelines,
        start_month,
        end_month,
        engine_parameters=engine_parameters,
    )

    participants = list(
        {
            pe.participant_identity
            for pe in new_premiums
            if isinstance(pe.participant_identity, CoreStackIdentity)
        }
    )

    if participants:
        repository = get_app_dependency().get_premium_repository(session)
        existing_premiums = repository.get_latest_entries(
            participants=participants,
            lower_bound=start_month,
            upper_bound=end_month,
        )
    else:
        existing_premiums = []

    _reconcile_and_persist_premium_entries(
        session,
        new_premiums,
        existing_premiums,
    )

    return new_premiums

components.premium.public.commands

components.premium.public.cost_types

Insurance cost types for the premium component.

This module contains premium's own definitions of insurance domain types, decoupling it from price_provider component implementations. Changes to price_provider's internal pricing logic should not require changes to premium's domain model.

Currency and rounding strategy are intentionally NOT duplicated - import them from components.core_price.public.primitives.

InsuranceBeneficiary

Bases: AlanBaseEnum

Represents the different types of insurance beneficiaries.

children class-attribute instance-attribute

children = 'children'

from_enrollment_type classmethod

from_enrollment_type(enrollment_type)

Convert EnrollmentType to InsuranceBeneficiary.

Source code in components/premium/public/cost_types.py

@classmethod
def from_enrollment_type(
    cls, enrollment_type: EnrollmentType
) -> "InsuranceBeneficiary":
    """Convert EnrollmentType to InsuranceBeneficiary."""
    if enrollment_type == EnrollmentType.primary:
        return cls.primary
    elif enrollment_type == EnrollmentType.partner:
        return cls.partner
    elif enrollment_type == EnrollmentType.child:
        return cls.children
    else:
        raise ValueError(f"Unknown enrollment type: {enrollment_type}")

partner class-attribute instance-attribute

partner = 'partner'

primary class-attribute instance-attribute

primary = 'primary'

to_enrollment_type

to_enrollment_type()

Convert InsuranceBeneficiary to EnrollmentType.

Source code in components/premium/public/cost_types.py

def to_enrollment_type(self) -> EnrollmentType:
    """Convert InsuranceBeneficiary to EnrollmentType."""
    if self == InsuranceBeneficiary.primary:
        return EnrollmentType.primary
    elif self == InsuranceBeneficiary.partner:
        return EnrollmentType.partner
    elif self == InsuranceBeneficiary.children:
        return EnrollmentType.child
    else:
        raise ValueError(f"Unknown beneficiary type: {self}")

InsuranceCollectionMethod

Bases: AlanBaseEnum

Methods for collecting an employee's participation.

direct_billing class-attribute instance-attribute

direct_billing = 'direct_billing'

The amount is paid directly by the employee or their partner, using the payment method of their choice.

flexben_fund class-attribute instance-attribute

flexben_fund = 'flexben_fund'

The amount is paid from the employee's flexben fund.

payroll class-attribute instance-attribute

payroll = 'payroll'

The amount is paid by the company but automatically deducted from the employee's payslip.

InsuranceContribution

Bases: AlanBaseEnum

Represents how an amount is used, what the money is going towards.

cost class-attribute instance-attribute

cost = 'cost'

membership_fee class-attribute instance-attribute

membership_fee = 'membership_fee'

social_fund class-attribute instance-attribute

social_fund = 'social_fund'

taxes class-attribute instance-attribute

taxes = 'taxes'

InsuranceCost dataclass

InsuranceCost(*, cost_components)

Represents the entire cost of an insurance service.

No safeties on cost components

There are no checks verifying there are no duplicates in components. The caller is responsible for ensuring the components represent a valid cost breakdown.

Source code in components/premium/public/cost_types.py

def __init__(self, *, cost_components: list[InsuranceCostComponent]):
    if len(cost_components) == 0:
        raise ValueError("No components to build a cost out of.")

    self.cost_components = cost_components
    self.currency = cost_components[0].currency

    # Compute some intermediary values frequently used by consumers, in O(n) time
    untaxed_amount_billed_to_primary = 0
    untaxed_amount_billed_to_company = 0

    taxes_billed_to_company = 0
    taxes_billed_to_primary = 0

    for component in cost_components:
        if component.currency != self.currency:
            raise ValueError("Cannot mix different currencies.")

        match component.debtor:
            case InsuranceDebtor.company:
                if component.contribution_type == InsuranceContribution.taxes:
                    taxes_billed_to_company += component.amount
                else:
                    untaxed_amount_billed_to_company += component.amount
            case InsuranceDebtor.primary:
                if component.contribution_type == InsuranceContribution.taxes:
                    taxes_billed_to_primary += component.amount
                else:
                    untaxed_amount_billed_to_primary += component.amount
            case debtor:
                raise NotImplementedError(f"Unimplemented debtor: {debtor}")

    taxed_amount_billed_to_primary = (
        untaxed_amount_billed_to_primary + taxes_billed_to_primary
    )
    taxed_amount_billed_to_company = (
        untaxed_amount_billed_to_company + taxes_billed_to_company
    )

    self.taxed_amount_billed_to_primary = taxed_amount_billed_to_primary
    self.untaxed_amount_billed_to_primary = untaxed_amount_billed_to_primary
    self.untaxed_amount_billed_to_company = untaxed_amount_billed_to_company
    self.taxed_amount_billed_to_company = taxed_amount_billed_to_company

    # Once initialised, the dataclass is frozen.
    # Enforced via custom __setattr__ implementation.
    self._initialized = True

__setattr__

__setattr__(name, value)

Setting attributes once the dataclass is initialized is forbidden. Reproducing the behavior of frozen=True.

Source code in components/premium/public/cost_types.py

def __setattr__(self, name: str, value: object) -> None:
    """
    Setting attributes once the dataclass is initialized is forbidden.
    Reproducing the behavior of frozen=True.
    """
    if hasattr(self, "_initialized") and self._initialized:
        raise AttributeError(
            f"Cannot modify {name} after initialization. InsuranceCost is immutable."
        )
    super().__setattr__(name, value)

cost_components instance-attribute

cost_components = cost_components

currency instance-attribute

currency = currency

from_components staticmethod

from_components(cost_components)

Construct an InsuranceCost from several cost components.

Raises:

Type	Description
NotImplementedError	if the debtor is not implemented yet
ValueError	if the currency used by all the components is not the same

Source code in components/premium/public/cost_types.py

@staticmethod
def from_components(
    cost_components: list[InsuranceCostComponent],
) -> "InsuranceCost":
    """
    Construct an InsuranceCost from several cost components.

    Raises:
        NotImplementedError: if the debtor is not implemented yet
        ValueError: if the currency used by all the components is not the same
    """
    return InsuranceCost(cost_components=cost_components)

merge staticmethod

merge(insurance_costs)

Merge several costs into the same cost objects.

When we evaluate the costs of coverage modules individually, merging them is useful to get the total aggregated costs.

Source code in components/premium/public/cost_types.py

@staticmethod
def merge(
    insurance_costs: list["InsuranceCost"],
) -> "InsuranceCost":
    """
    Merge several costs into the same cost objects.

    When we evaluate the costs of coverage modules individually, merging them is
    useful to get the total aggregated costs.
    """
    if len(insurance_costs) == 1:
        return insurance_costs[0]

    all_components: list[InsuranceCostComponent] = []
    for cost in insurance_costs:
        all_components.extend(cost.cost_components)
    return InsuranceCost.from_components(all_components)

taxed_amount_billed_to_company instance-attribute

taxed_amount_billed_to_company = (
    taxed_amount_billed_to_company
)

taxed_amount_billed_to_primary instance-attribute

taxed_amount_billed_to_primary = (
    taxed_amount_billed_to_primary
)

untaxed_amount_billed_to_company instance-attribute

untaxed_amount_billed_to_company = (
    untaxed_amount_billed_to_company
)

untaxed_amount_billed_to_primary instance-attribute

untaxed_amount_billed_to_primary = (
    untaxed_amount_billed_to_primary
)

InsuranceCostComponent dataclass

InsuranceCostComponent(
    *,
    service_type,
    module_id=None,
    contribution_type,
    beneficiary_type,
    debtor,
    collection_method=None,
    enrollment_id=None,
    enrollment_group_id=None,
    member_id=None,
    currency,
    amount,
    periodicity="monthly"
)

Represents part of a cost of insurance service.

__post_init__

__post_init__()

Validate collection_method when debtor != company, and identity shape consistency (legacy XOR Core Stack, all-None also valid).

Source code in components/premium/public/cost_types.py

def __post_init__(self) -> None:
    """Validate collection_method when debtor != company, and identity
    shape consistency (legacy XOR Core Stack, all-None also valid).
    """
    if self.debtor != InsuranceDebtor.company and self.collection_method is None:
        raise ValueError(
            f"Collection method must be set when debtor is {self.debtor}"
        )
    # enrollment_group_id and member_id must be set together.
    has_group = self.enrollment_group_id is not None
    has_member = self.member_id is not None
    if has_group != has_member:
        raise ValueError(
            "enrollment_group_id and member_id must be set together "
            "(both or neither)"
        )
    # Cannot mix legacy enrollment_id with Core Stack identity.
    if has_group and self.enrollment_id is not None:
        raise ValueError(
            "Cannot set both legacy enrollment_id and Core Stack "
            "(enrollment_group_id, member_id)"
        )

amount instance-attribute

amount

The amount for an entire period of service.

Can be negative, for example if the amount corresponds to a discount. Must be expressed in the minor unit of the chosen currency.

beneficiary_type instance-attribute

beneficiary_type

Who's coverage does this cost correspond to ?

collection_method class-attribute instance-attribute

collection_method = None

If an employee is responsible for paying this cost, how is the payment collected ?

compare_component_lists staticmethod

compare_component_lists(left, right)

Compare two lists of insurance cost components for equality.

Parameters:

Name	Type	Description	Default
left	list[InsuranceCostComponent]	First list of components to compare	required
right	list[InsuranceCostComponent]	Second list of components to compare	required

Returns:

Type	Description
bool	True if both lists contain the same components in any order, False otherwise

Source code in components/premium/public/cost_types.py

@staticmethod
def compare_component_lists(
    left: list["InsuranceCostComponent"], right: list["InsuranceCostComponent"]
) -> bool:
    """
    Compare two lists of insurance cost components for equality.

    Args:
        left: First list of components to compare
        right: Second list of components to compare

    Returns:
        True if both lists contain the same components in any order, False otherwise
    """
    return set(left) == set(right)

contribution_type instance-attribute

contribution_type

What part of the service is paid for by this cost?

currency instance-attribute

currency

debtor instance-attribute

debtor

Who is responsible for paying this cost ?

enrollment_group_id class-attribute instance-attribute

enrollment_group_id = None

Core Stack enrollment group id.

enrollment_id class-attribute instance-attribute

enrollment_id = None

Legacy country-specific enrollment id (e.g., EsEnrollment.id).

group_by_beneficiary_id staticmethod

group_by_beneficiary_id(components)

Groups insurance cost components by enrollment ID.

Parameters:

Name	Type	Description	Default
components	list[InsuranceCostComponent]	List of InsuranceCostComponent to group	required

Returns:

Type	Description
dict[int | UUID, list[InsuranceCostComponent]]	Dictionary mapping enrollment ID (int or UUID) to list of components

Raises:

Type	Description
ValueError	If any component does not have an enrollment ID

Source code in components/premium/public/cost_types.py

@staticmethod
def group_by_beneficiary_id(
    components: list["InsuranceCostComponent"],
) -> dict[int | UUID, list["InsuranceCostComponent"]]:
    """
    Groups insurance cost components by enrollment ID.

    Args:
        components: List of InsuranceCostComponent to group

    Returns:
        Dictionary mapping enrollment ID (int or UUID) to list of components

    Raises:
        ValueError: If any component does not have an enrollment ID
    """
    grouped: dict[int | UUID, list[InsuranceCostComponent]] = {}

    for component in components:
        if component.enrollment_id is None:
            raise ValueError("All components must have an enrollment ID")

        enrollment_id = component.enrollment_id
        if enrollment_id not in grouped:
            grouped[enrollment_id] = []
        grouped[enrollment_id].append(component)

    return grouped

member_id class-attribute instance-attribute

member_id = None

Core Stack member id (stringified user_id, per Core Stack convention). Set together with enrollment_group_id.

module_id class-attribute instance-attribute

module_id = None

Contract module the cost belongs to.

periodicity class-attribute instance-attribute

periodicity = 'monthly'

service_type instance-attribute

service_type

Tag identifying what this cost is for. ES values describe bundle elements (pharmacy, optical, ...); FR/BE/CA values describe coverage roles (base, option, ...).

InsuranceDebtor

Bases: AlanBaseEnum

Represents an entity paying for an insurance service.

primary covers both the primary member and partner — pricing always bills the primary, never the partner directly.

company class-attribute instance-attribute

company = 'company'

primary class-attribute instance-attribute

primary = 'primary'

InsuranceService

Bases: AlanBaseEnum

Tag identifying what a cost is for, on a fee component.

base class-attribute instance-attribute

base = 'base'

ca_core_product class-attribute instance-attribute

ca_core_product = 'ca_core_product'

core_product class-attribute instance-attribute

core_product = 'es_core_product'

extended class-attribute instance-attribute

extended = 'extended'

Base + option rolled into the same coverage module.

hospitalization class-attribute instance-attribute

hospitalization = 'hospitalization'

Neutral label for single-coverage plans (no base/extended distinction).

optical class-attribute instance-attribute

optical = 'es_optical'

option class-attribute instance-attribute

option = 'option'

pharmacy class-attribute instance-attribute

pharmacy = 'es_pharmacy'

physio_nutrition class-attribute instance-attribute

physio_nutrition = 'es_physio_nutrition'

preexisting_conditions class-attribute instance-attribute

preexisting_conditions = 'es_preexisting_conditions'

premium_reimbursement class-attribute instance-attribute

premium_reimbursement = 'es_premium_reimbursement'

components.premium.public.dependencies

COMPONENT_NAME module-attribute

COMPONENT_NAME = 'premium'

Canonical name of the premium component.

PremiumDependency

Bases: ABC

Represents all the dependency required by components/premium to function as expected.

get_flexben_timeline_for_all_enrollments_of_the_policy abstractmethod

get_flexben_timeline_for_all_enrollments_of_the_policy(
    policy_id, start_date=None, end_date=None
)

For each enrollment in the policy, return a list of periods where a flexben choice is enabled.

Considering flexben is a feature local to Belgium, we expect components/be to be the only caller to actually inject this dependency. Other components are expected to inject a function returning an empty dict.

Example

{ enrollment_id_1: [ ValidityPeriod(date(2024, 1, 1), date(2024, 6, 30)), ], enrollment_id_2: [ ValidityPeriod(date(2024, 3, 1), date(2024, 12, 31)), ], }

Source code in components/premium/public/dependencies.py

@abstractmethod
def get_flexben_timeline_for_all_enrollments_of_the_policy(
    self,
    policy_id: int | UUID,
    start_date: date | None = None,
    end_date: date | None = None,
) -> "dict[int | UUID, list[ValidityPeriod]]":
    """
    For each enrollment in the policy, return a list of periods where a flexben
    choice is enabled.

    Considering flexben is a feature local to Belgium, we expect `components/be` to
    be the only caller to actually inject this dependency.
    Other components are expected to inject a function returning an empty dict.

    Example:
        {
            enrollment_id_1: [
                ValidityPeriod(date(2024, 1, 1), date(2024, 6, 30)),
            ],
            enrollment_id_2: [
                ValidityPeriod(date(2024, 3, 1), date(2024, 12, 31)),
            ],
        }
    """
    ...

get_legacy_premium_entry_repository

get_legacy_premium_entry_repository(session)

Instantiate layer to retrieve premium entries based on legacy ones

Source code in components/premium/public/dependencies.py

def get_legacy_premium_entry_repository(
    self, session: Session | None
) -> PremiumEntryRepository:
    """
    Instantiate layer to retrieve premium entries based on legacy ones
    """
    return NoOpPremiumEntryRepository(session)

get_premium_repository abstractmethod

get_premium_repository(session)

Instantiate a persistence layer for premium repository

Source code in components/premium/public/dependencies.py

@abstractmethod
def get_premium_repository(self, session: Session | None) -> PremiumEntryRepository:
    """
    Instantiate a persistence layer for premium repository
    """
    ...

to_member_spec

to_member_spec(spec)

Convert a BeneficiarySpec to a MemberSpec for core_price pricing.

Default implementation delegates to CorePriceAdapter. Override in country-specific subclasses when the default conversion is not suitable (e.g. CA, which uses CaMemberSpec instead of CorePriceBeneficiarySpec).

Source code in components/premium/public/dependencies.py

def to_member_spec(self, spec: BeneficiarySpec) -> MemberSpec:
    """Convert a BeneficiarySpec to a MemberSpec for core_price pricing.

    Default implementation delegates to CorePriceAdapter. Override in
    country-specific subclasses when the default conversion is not suitable
    (e.g. CA, which uses CaMemberSpec instead of CorePriceBeneficiarySpec).
    """
    from components.premium.internal.adapters.core_price_adapter import (
        CorePriceAdapter,
    )

    return CorePriceAdapter.to_core_price_beneficiary_spec(spec)

get_app_dependency

get_app_dependency()

Function used to fetch the dependencies from the flask app.

Source code in components/premium/public/dependencies.py

def get_app_dependency() -> PremiumDependency:
    """Function used to fetch the dependencies from the flask app."""
    from flask import current_app

    return cast("CustomFlask", current_app).get_component_dependency(COMPONENT_NAME)  # type: ignore[no-any-return]

set_app_dependency

set_app_dependency(dependency)

Function used to actually inject the dependency class in the component.

Source code in components/premium/public/dependencies.py

def set_app_dependency(dependency: PremiumDependency) -> None:
    """Function used to actually inject the dependency class in the component."""
    from flask import current_app

    cast("CustomFlask", current_app).add_component_dependency(
        COMPONENT_NAME, dependency
    )

components.premium.public.entities

AgeStrategy

Bases: AlanBaseEnum

Strategy for handling age calculation in insurance premium calculations.

The effective birthday or age we use for pricing a beneficiary may differ from their actual birthday or age.

KEEP_EXACT_BIRTHDAY class-attribute instance-attribute

KEEP_EXACT_BIRTHDAY = 'exact_date'

Use the exact birthday for age calculation.

Example: If a member turns 23 on March 15, 2023, their age changes exactly on March 15, 2023.

MOVE_BIRTHDAY_TO_FIRST_DAY_OF_MONTH class-attribute instance-attribute

MOVE_BIRTHDAY_TO_FIRST_DAY_OF_MONTH = (
    "move_to_first_day_of_month"
)

Move birthday to the first day of the birth month for age calculation.

Example: If a member turns 23 on March 15, 2023, they will be priced as 23 years old starting March 1, 2023 for the entire month.

MOVE_BIRTHDAY_TO_JAN_OF_NEXT_YEAR class-attribute instance-attribute

MOVE_BIRTHDAY_TO_JAN_OF_NEXT_YEAR = 'jan_of_next_year'

Move birthday to January 1st of the year following the birth year.

Example: If a member turns 23 on March 15, 2023, they will be priced as 22 years old for the entirety of 2023, and 23 years old starting January 1, 2024.

BeneficiarySpec dataclass

BeneficiarySpec(
    *,
    enrollment_id,
    beneficiary_type,
    beneficiary_age=None,
    beneficiary_date_of_birth=None,
    coverage_module_name=None,
    has_active_flexben_choice=False,
    contract_version_id=None,
    province_code=None
)

Specification for a beneficiary used in premium calculations.

Contains all the information needed to compute insurance costs for a specific beneficiary within a stable timeline period. Age information is provided either as a computed age or birth date.

Design Decision: Single Module for Billing Domain

This component represents the billing domain, not the pricing domain. For billing purposes, we only need to know which single module to bill for a beneficiary, not all modules they might be subscribed to.

• Currently, only Belgium uses this component, and BE supports exactly one module per beneficiary
• Other countries (FR, ES, CA) can iterate on this design when they adopt the premium component
• The adapter converts from price_provider's modules list by taking modules[0]

If multiple modules per beneficiary are needed in the future, this design can be extended. For now, YAGNI (You Aren't Gonna Need It) applies.

Attributes:

Name	Type	Description
enrollment_id	int | UUID | None	Unique identifier for the enrollment this beneficiary belongs to
beneficiary_type	InsuranceBeneficiary	Type of insurance beneficiary (employee, spouse, child, etc.)
beneficiary_age	int | None	Computed age at the time of the period (may be None if birth_date used)
beneficiary_date_of_birth	date | None	Actual birth date (may be None if age is pre-computed)
coverage_module_name	str | None	Specific coverage module for this beneficiary (optional)
contract_version_id	UUID | None	Contract version for legacy pricing function
province_code	str | None	ISO 3166-2 subdivision without country prefix (e.g. "ON", "QC"). Used by products where tax rates or pricing depend on province of residence.

__post_init__

__post_init__()

Source code in components/premium/internal/domain/insurance_entities.py

def __post_init__(self) -> None:
    age_specified = self.beneficiary_age is not None
    birthdate_specified = self.beneficiary_date_of_birth is not None

    if not (age_specified or birthdate_specified):
        raise ValueError(
            "Either beneficiary_age or beneficiary_date_of_birth must be specified"
        )

    if age_specified and birthdate_specified:
        raise ValueError(
            "Cannot specify both beneficiary_age and beneficiary_date_of_birth"
        )

beneficiary_age class-attribute instance-attribute

beneficiary_age = None

beneficiary_date_of_birth class-attribute instance-attribute

beneficiary_date_of_birth = None

beneficiary_type instance-attribute

beneficiary_type

contract_version_id class-attribute instance-attribute

contract_version_id = None

coverage_module_name class-attribute instance-attribute

coverage_module_name = None

enrollment_id instance-attribute

enrollment_id

has_active_flexben_choice class-attribute instance-attribute

has_active_flexben_choice = False

province_code class-attribute instance-attribute

province_code = None

Contract

Bases: Protocol

get_simple_subscription_version_timeline

get_simple_subscription_version_timeline()

Source code in components/premium/internal/domain/insurance_entities.py

def get_simple_subscription_version_timeline(
    self,
) -> Timeline[SimpleSubscriptionVersion]: ...

is_cancelled instance-attribute

is_cancelled

policies instance-attribute

policies

CoreStackIdentity

Bases: NamedTuple

Core Stack identity for a PremiumEntry. The triple is atomic by construction — there's no way to express a partial Core Stack identity.

enrollment_group_id instance-attribute

enrollment_group_id

member_id instance-attribute

member_id

module_id instance-attribute

module_id

CostPeriod dataclass

CostPeriod(*, start_reason, cost)

Bases: GenericPeriodWithStartReason

Represents a timeline period with computed premium cost.

Final output of the premium calculation pipeline, containing the actual cost for the stable period. Each period guarantees the cost remains constant throughout its validity period.

Attributes:

Name	Type	Description
cost	Cost	The computed insurance cost for this period, calculated based on the stable beneficiary composition and contract terms.

cost instance-attribute

cost

EngineParameters dataclass

EngineParameters(
    *,
    age_strategy,
    rounding_strategy,
    prorata_strategy,
    default_child_age,
    default_adult_age
)

Parameters dictating how the engine behaves and which rules it follows.

Example

• which effective birth date to use
• which prorata formula to use
• which rounding strategy to use

age_strategy instance-attribute

age_strategy

default staticmethod

default(app_name)

Get default engine parameters for a given application. Application is used as a proxy for country.

Parameters:

Name	Type	Description	Default
name		The application name to get default parameters for	required

Raises:

Type	Description
ValueError	If the application is not yet supported

Source code in components/premium/internal/domain/engine_parameters.py

@staticmethod
def default(app_name: AppName) -> "EngineParameters":
    """
    Get default engine parameters for a given application.
    Application is used as a proxy for country.

    Args:
        name: The application name to get default parameters for

    Raises:
        ValueError: If the application is not yet supported
    """
    match app_name:
        case AppName.ALAN_BE:
            return EngineParameters(
                age_strategy=AgeStrategy.MOVE_BIRTHDAY_TO_FIRST_DAY_OF_MONTH,
                rounding_strategy=RoundingStrategy.BANKERS,
                prorata_strategy=ProrataStrategy.THIRTY_DAY_PRORATA_WITH_LARGEST_REMAINDER_DISTRIBUTION_ACROSS_FEE_COMPONENTS,
                default_child_age=17,
                default_adult_age=25,
            )
        case AppName.ALAN_FR:
            return EngineParameters(
                age_strategy=AgeStrategy.MOVE_BIRTHDAY_TO_JAN_OF_NEXT_YEAR,
                rounding_strategy=RoundingStrategy.BANKERS,
                prorata_strategy=ProrataStrategy.THIRTY_DAY_PRORATA_WITH_LARGEST_REMAINDER_DISTRIBUTION_ACROSS_FEE_COMPONENTS,
                default_child_age=17,
                default_adult_age=25,
            )
        case AppName.ALAN_ES:
            # age_strategy + default_*_age unused on the Core Stack shadow
            # write path; placeholder values mirror BE.
            return EngineParameters(
                age_strategy=AgeStrategy.MOVE_BIRTHDAY_TO_FIRST_DAY_OF_MONTH,
                rounding_strategy=RoundingStrategy.BANKERS,
                prorata_strategy=ProrataStrategy.THIRTY_DAY_PRORATA_WITH_LARGEST_REMAINDER_DISTRIBUTION_ACROSS_FEE_COMPONENTS,
                default_child_age=17,
                default_adult_age=25,
            )
        case AppName.ALAN_CA:
            # age_strategy + default_*_age unused on the Core Stack shadow
            # write path; placeholder values mirror BE.
            return EngineParameters(
                age_strategy=AgeStrategy.MOVE_BIRTHDAY_TO_FIRST_DAY_OF_MONTH,
                rounding_strategy=RoundingStrategy.BANKERS,
                prorata_strategy=ProrataStrategy.THIRTY_DAY_PRORATA_WITH_LARGEST_REMAINDER_DISTRIBUTION_ACROSS_FEE_COMPONENTS,
                default_child_age=17,
                default_adult_age=25,
            )
        case _:
            raise ValueError(f"Unsupported app_name: {app_name}")

default_adult_age instance-attribute

default_adult_age

Optional strategy for normalizing ages into specific buckets for pricing. When None, ages are used directly from beneficiary birthdates or defaults. When specified, the strategy determines how to group beneficiaries by age.

default_child_age instance-attribute

default_child_age

prorata_strategy instance-attribute

prorata_strategy

rounding_strategy instance-attribute

rounding_strategy

FeeComponent dataclass

FeeComponent(
    *,
    id=uuid4(),
    num_days,
    invoice_id,
    amount_before_prorata
)

Bases: CostComponent

A fee component represents one part of a fee that we intend to bill to a customer.

__eq__

__eq__(other)

Source code in components/premium/internal/domain/fee.py

def __eq__(self, other: object) -> bool:
    if not isinstance(other, FeeComponent):
        return False

    return (
        self.num_days == other.num_days
        and self.amount == other.amount
        and self.amount_before_prorata == other.amount_before_prorata
        and self.currency == other.currency
        and self.beneficiary_type == other.beneficiary_type
        and self.service_type == other.service_type
        and self.contribution_type == other.contribution_type
        and self.debtor == other.debtor
        and self.collection_method == other.collection_method
        and self.enrollment_id == other.enrollment_id
    )

__hash__

__hash__()

Source code in components/premium/internal/domain/fee.py

def __hash__(self) -> int:
    return hash(
        (
            self.num_days,
            self.amount,
            self.amount_before_prorata,
            self.currency.alphabetic_code,
            self.beneficiary_type,
            self.service_type,
            self.contribution_type,
            self.debtor,
            self.collection_method,
            str(self.enrollment_id) if self.enrollment_id else None,
        )
    )

amount_before_prorata instance-attribute

amount_before_prorata

Amount for full period coverage without prorata applied. Expressed in the minor unit of the currency. This is the full monthly amount that would be charged for complete period coverage.

as_tuple

as_tuple()

Return (debtor, contribution_type, amount)

Source code in components/premium/internal/domain/fee.py

def as_tuple(
    self,
) -> tuple["InsuranceDebtor", "InsuranceContribution", int]:
    """Return (debtor, contribution_type, amount)"""

    return (self.debtor, self.contribution_type, self.amount)

id class-attribute instance-attribute

id = field(default_factory=uuid4)

Unique identifier of the entry

inverted

inverted()

Returns a new FeeComponent offsetting the current one.

Source code in components/premium/internal/domain/fee.py

def inverted(self) -> "FeeComponent":
    """
    Returns a new FeeComponent offsetting the current one.
    """
    return FeeComponent(
        num_days=-self.num_days,
        amount=-self.amount,
        amount_before_prorata=-self.amount_before_prorata,
        currency=self.currency,
        beneficiary_type=self.beneficiary_type,
        service_type=self.service_type,
        module_id=self.module_id,
        contribution_type=self.contribution_type,
        debtor=self.debtor,
        collection_method=self.collection_method,
        enrollment_id=self.enrollment_id,
        enrollment_group_id=self.enrollment_group_id,
        member_id=self.member_id,
        invoice_id=None,
    )

invoice_id instance-attribute

invoice_id

If the component is included in an invoice, it is represented here.

num_days instance-attribute

num_days

Number of days this fee covers within the billing period.

LegacyBelgianNormalization dataclass

LegacyBelgianNormalization()

Groups beneficiaries into specific age buckets: - Children are treated as adults if they are 25 or above - Young adults (18-25) are priced as 25 or above if they are the primary policy holder or their partner - No birthdate available: default to age 25 - Anyone not falling in one of the buckets above: use their actual age

CHILD_AGE_LIMIT_YEARS class-attribute instance-attribute

CHILD_AGE_LIMIT_YEARS = 25

normalize_age

normalize_age(
    effective_birthdate, enrollment_type, reference_date
)

Source code in components/premium/internal/domain/age_normalization.py

def normalize_age(
    self,
    effective_birthdate: date | None,
    enrollment_type: EnrollmentType,
    reference_date: date,
) -> int:
    if not effective_birthdate:
        return self.CHILD_AGE_LIMIT_YEARS

    age = age_on(effective_birthdate, reference_date)

    if enrollment_type != EnrollmentType.child and age < self.CHILD_AGE_LIMIT_YEARS:
        return self.CHILD_AGE_LIMIT_YEARS

    if enrollment_type == EnrollmentType.child and age < self.CHILD_AGE_LIMIT_YEARS:
        return 17

    return age

ParticipantIdentity module-attribute

ParticipantIdentity = CoreStackIdentity | UUID

A PremiumEntry's identity is one of two shapes: - CoreStackIdentity: the full (group, member, module) triple - UUID: a legacy country-specific enrollment id The discriminated union enforces "exactly one shape" statically. Readers isinstance/match-discriminate before accessing the parts.

Policy

Bases: Historizable

enrollments instance-attribute

enrollments

id instance-attribute

id

PremiumEntry dataclass

PremiumEntry(
    *,
    id=uuid4(),
    participant_identity,
    components,
    period_start,
    period_end,
    version=1,
    cancelled_by_entry_id=None,
    cancelled_entry_id=None,
    payroll_csv_id=None,
    finiquito_id=None,
    original_ids=()
)

A subscription fee specific to insurance subscriptions.

__eq__

__eq__(other)

Source code in components/premium/internal/domain/fee.py

def __eq__(self, other: object) -> bool:
    if not isinstance(other, PremiumEntry):
        return False

    return (
        self.participant_identity == other.participant_identity
        and self.period_start == other.period_start
        and self.period_end == other.period_end
        and set(self.components) == set(other.components)
    )

__hash__

__hash__()

Source code in components/premium/internal/domain/fee.py

def __hash__(self) -> int:
    return hash(
        (
            self.participant_identity,
            self.period_start,
            self.period_end,
            frozenset(self.components),
        )
    )

__post_init__

__post_init__()

Source code in components/premium/internal/domain/fee.py

def __post_init__(self) -> None:
    if not self.components:
        raise ValueError(
            f"PremiumEntry {self.id} has no components — "
            f"a PremiumEntry without components is invalid"
        )

beneficiary_type property

beneficiary_type

cancelled

cancelled()

Returns two new entries: one marking this entry as cancelled, and one offsetting it.

Returns:

Type	Description
tuple[PremiumEntry, PremiumEntry | None]	tuple[PremiumEntry, PremiumEntry]: - First entry: Same as self but marked as cancelled - Second entry: New entry to offset self.

Source code in components/premium/internal/domain/fee.py

def cancelled(self) -> tuple["PremiumEntry", "PremiumEntry | None"]:
    """
    Returns two new entries: one marking this entry as cancelled, and one offsetting it.

    Returns:
        tuple[PremiumEntry, PremiumEntry]:
            - First entry: Same as self but marked as cancelled
            - Second entry: New entry to offset self.
    """

    # If B cancels A and we cancel B, it's equivalent to undoing the cancelation of
    # A
    # Therefore, canceling entries should never get canceled.
    if self.cancelled_entry_id:
        return self, None

    if self.cancelled_by_entry_id:
        return self, None

    cancelling_entry_id = uuid4()

    cancelled_entry = PremiumEntry(
        id=self.id,
        participant_identity=self.participant_identity,
        components=self.components,
        period_start=self.period_start,
        period_end=self.period_end,
        version=self.version,
        cancelled_entry_id=self.cancelled_entry_id,
        cancelled_by_entry_id=cancelling_entry_id,
        payroll_csv_id=self.payroll_csv_id,
        finiquito_id=self.finiquito_id,
        original_ids=self.original_ids,
    )

    offsetting_entry = PremiumEntry(
        id=cancelling_entry_id,
        participant_identity=self.participant_identity,
        components=[component.inverted() for component in self.components],
        period_start=self.period_start,
        period_end=self.period_end,
        version=self.version + 1,
        cancelled_by_entry_id=None,
        cancelled_entry_id=self.id,
        payroll_csv_id=self.payroll_csv_id,
        finiquito_id=self.finiquito_id,
        original_ids=self.original_ids,
    )

    return cancelled_entry, offsetting_entry

cancelled_by_entry_id class-attribute instance-attribute

cancelled_by_entry_id = field(default=None)

cancelled_entry_id class-attribute instance-attribute

cancelled_entry_id = field(default=None)

components instance-attribute

components

components_billed_to_debtor

components_billed_to_debtor(debtor)

Returns components that are actually billed to the specified debtor.

This method accounts for collection methods: - If debtor is company: includes company components AND primary components with (payroll or flexben_fund) - If debtor is primary: only includes primary components with direct_billing

Source code in components/premium/internal/domain/fee.py

def components_billed_to_debtor(
    self, debtor: "InsuranceDebtor"
) -> list[FeeComponent]:
    """
    Returns components that are actually billed to the specified debtor.

    This method accounts for collection methods:
    - If debtor is company: includes company components AND primary components with (payroll or flexben_fund)
    - If debtor is primary: only includes primary components with direct_billing
    """
    return [
        c for c in self.components if self._is_component_billed_to_debtor(c, debtor)
    ]

components_for_debtor

components_for_debtor(debtor)

The debtor is not necessarily the entity being billed

If a component's debtor is the primary but the collection method is payroll, the company gets billed.

If your purpose is to list components that should be invoiced to a given debtor, use components_billed_to_debtor().

Source code in components/premium/internal/domain/fee.py

def components_for_debtor(self, debtor: "InsuranceDebtor") -> list[FeeComponent]:
    """
    !!! warning "The debtor is not necessarily the entity being billed"

        If a component's debtor is the primary but the collection method is payroll,
        the company gets billed.

        If your purpose is to list components that should be invoiced to a given debtor,
        use `components_billed_to_debtor()`.
    """
    return [c for c in self.components if c.debtor == debtor]

currency property

currency

enrollment_id property

enrollment_id

Legacy enrollment id when the PE has a legacy-shape identity, else None. Convenience for legacy-only consumers — Core-Stack-aware code should pattern-match on participant_identity directly.

finiquito_id class-attribute instance-attribute

finiquito_id = field(default=None)

Spain-specific. A finiquito is a pay_csv generated to recap the premiums of an employee when they leave the company. In order to maintain compatibility between Spain payroll and Global premiums we need to maintain this property. However it's not required by the Global payroll system, therefore this property is temporary and will be dropped as soon as global payroll is online.

Always null in Belgium.

id class-attribute instance-attribute

id = field(default_factory=uuid4)

is_uninvoiced

is_uninvoiced(debtor=None)

Returns True if none of the relevant components have been invoiced.

If debtor is None, checks all components. If debtor is provided, checks only components billed to that specific debtor.

Source code in components/premium/internal/domain/fee.py

def is_uninvoiced(self, debtor: "InsuranceDebtor | None" = None) -> bool:
    """
    Returns True if none of the relevant components have been invoiced.

    If debtor is None, checks all components.
    If debtor is provided, checks only components billed to that specific debtor.
    """
    components = (
        self.components_billed_to_debtor(debtor)
        if debtor is not None
        else self.components
    )
    return all(c.invoice_id is None for c in components)

num_days property

num_days

original_ids class-attribute instance-attribute

original_ids = field(default=())

Provenance of this PremiumEntry: the legacy country-specific premium-entry rows (e.g. EsPremiumEntry.ids) that produced this entry via the legacy-to-global mapper. Empty for PEs that did not come from a legacy mapper (e.g. fresh PEs produced by compute_subscription_fees from a pricing breakdown).

Used by callers that need to write something back to the legacy side after they finished working with the global-shape PremiumEntry (notably the M3 invoice generator, which builds an EsInvoice from PE and then needs the legacy EsPremiumEntry rows to set invoice.premium_entries via the SQLAlchemy backref).

Migration-transient: removed at M5 when legacy storage goes away.

participant_identity instance-attribute

participant_identity

The PE's billable subject. CoreStackIdentity (group, member, module) for Core-Stack-native PEs, or a legacy country-specific enrollment id.

payroll_csv_id class-attribute instance-attribute

payroll_csv_id = field(default=None)

Belgium & Spain's local payroll systems requires to keep track of which premiums are included in which file.

In order to maintain compatibility between Belgium payroll and Global premiums we need to maintain this property. However it's not required by the Global payroll system, therefore this property is temporary and will be dropped as soon as global payroll is online.

period_end instance-attribute

period_end

period_start instance-attribute

period_start

pretax_amount_billed_to_debtor

pretax_amount_billed_to_debtor(debtor)

Returns the total pretax amount (membership fees + costs) billed to the specified debtor.

This includes components with contribution_type of: - InsuranceContribution.membership_fee: Base membership fees - InsuranceContribution.cost: Insurance coverage costs

This method accounts for collection methods: - If debtor is company: includes company components AND primary components with (payroll or flexben_fund) - If debtor is primary: only includes primary components with direct_billing

Parameters:

Name	Type	Description	Default
debtor	InsuranceDebtor	The entity to which costs are billed	required

Returns:

Type	Description
int	Total pretax amount in minor currency units

Source code in components/premium/internal/domain/fee.py

def pretax_amount_billed_to_debtor(self, debtor: "InsuranceDebtor") -> int:
    """
    Returns the total pretax amount (membership fees + costs) billed to the specified debtor.

    This includes components with contribution_type of:
    - InsuranceContribution.membership_fee: Base membership fees
    - InsuranceContribution.cost: Insurance coverage costs

    This method accounts for collection methods:
    - If debtor is company: includes company components AND primary components with (payroll or flexben_fund)
    - If debtor is primary: only includes primary components with direct_billing

    Args:
        debtor: The entity to which costs are billed

    Returns:
        Total pretax amount in minor currency units
    """
    return sum(
        c.amount
        for c in self.components
        if self._is_component_billed_to_debtor(c, debtor)
        and c.contribution_type
        in [
            InsuranceContribution.membership_fee,
            InsuranceContribution.cost,
        ]
    )

prorata_ratio

prorata_ratio()

Returns the prorata ratio for this premium entry.

The ratio is calculated as: - 1.0 (or -1.0 for negative days) if the premium covers the full month - num_days / 30 otherwise (using 30-day convention, rounded to 2 decimals)

Uses arithmetic rounding (ROUND_HALF_UP) for consistency with invoicing display.

Returns:

Type	Description
float	Prorata ratio as a float. Positive for regular premiums, negative for credits/reversals.

Source code in components/premium/internal/domain/fee.py

def prorata_ratio(self) -> float:
    """
    Returns the prorata ratio for this premium entry.

    The ratio is calculated as:
    - 1.0 (or -1.0 for negative days) if the premium covers the full month
    - num_days / 30 otherwise (using 30-day convention, rounded to 2 decimals)

    Uses arithmetic rounding (ROUND_HALF_UP) for consistency with invoicing display.

    Returns:
        Prorata ratio as a float. Positive for regular premiums, negative for credits/reversals.
    """
    from shared.helpers.math import arithmetic_round

    # TODO: @florian.gagnadre 2026-01-21
    # Temporary solution for occupational health subscription fees.
    # We should probably find a better way to manga pro-rata for different periods.
    if self.num_days in [365, 366]:
        return 1.0

    if self.num_days in [-365, -366]:
        return -1.0

    if abs(self.num_days) == Month(self.period_start).n_days:
        return 1.0 if self.num_days > 0 else -1.0
    else:
        return arithmetic_round(self.num_days / 30, ndigits=2)

service_type property

service_type

taxes_billed_to_debtor

taxes_billed_to_debtor(debtor)

Returns the total tax amount billed to the specified debtor.

This includes components with contribution_type of: - InsuranceContribution.taxes: Tax components

This method accounts for collection methods: - If debtor is company: includes company components AND primary components with (payroll or flexben_fund) - If debtor is primary: only includes primary components with direct_billing

Parameters:

Name	Type	Description	Default
debtor	InsuranceDebtor	The entity to which taxes are billed	required

Returns:

Type	Description
int	Total tax amount in minor currency units

Source code in components/premium/internal/domain/fee.py

def taxes_billed_to_debtor(self, debtor: "InsuranceDebtor") -> int:
    """
    Returns the total tax amount billed to the specified debtor.

    This includes components with contribution_type of:
    - InsuranceContribution.taxes: Tax components

    This method accounts for collection methods:
    - If debtor is company: includes company components AND primary components with (payroll or flexben_fund)
    - If debtor is primary: only includes primary components with direct_billing

    Args:
        debtor: The entity to which taxes are billed

    Returns:
        Total tax amount in minor currency units
    """
    return sum(
        c.amount
        for c in self.components
        if self._is_component_billed_to_debtor(c, debtor)
        and c.contribution_type == InsuranceContribution.taxes
    )

total_amount

total_amount()

Source code in components/premium/internal/domain/fee.py

def total_amount(self) -> int:
    return sum(c.amount for c in self.components)

total_amount_billed_to_debtor

total_amount_billed_to_debtor(debtor)

Returns the total amount for components actually billed to the specified debtor.

This method accounts for collection methods: - If debtor is company: includes company components AND primary components with (payroll or flexben_fund) - If debtor is primary: only includes primary components with direct_billing

Source code in components/premium/internal/domain/fee.py

def total_amount_billed_to_debtor(self, debtor: "InsuranceDebtor") -> int:
    """
    Returns the total amount for components actually billed to the specified debtor.

    This method accounts for collection methods:
    - If debtor is company: includes company components AND primary components with (payroll or flexben_fund)
    - If debtor is primary: only includes primary components with direct_billing
    """
    return sum(
        c.amount
        for c in self.components
        if self._is_component_billed_to_debtor(c, debtor)
    )

total_amount_for_debtor

total_amount_for_debtor(debtor)

The debtor is not necessarily the entity being billed

If a component's debtor is the primary but the collection method is payroll, the company gets billed.

If your purpose is to compute the total amount that should be invoiced to a given debtor, use total_amount_billed_to_debtor().

Source code in components/premium/internal/domain/fee.py

def total_amount_for_debtor(self, debtor: "InsuranceDebtor") -> int:
    """
    !!! warning "The debtor is not necessarily the entity being billed"

        If a component's debtor is the primary but the collection method is payroll,
        the company gets billed.

        If your purpose is to compute the total amount that should be invoiced to a given debtor,
        use `total_amount_billed_to_debtor()`.
    """
    return sum(c.amount for c in self.components if c.debtor == debtor)

total_billed_to_debtor_without_prorata

total_billed_to_debtor_without_prorata(debtor)

Returns the total amount before prorata for components actually billed to the specified debtor.

This method accounts for collection methods: - If debtor is company: includes company components AND primary components with (payroll or flexben_fund) - If debtor is primary: only includes primary components with direct_billing

Source code in components/premium/internal/domain/fee.py

def total_billed_to_debtor_without_prorata(self, debtor: "InsuranceDebtor") -> int:
    """
    Returns the total amount before prorata for components actually billed to the specified debtor.

    This method accounts for collection methods:
    - If debtor is company: includes company components AND primary components with (payroll or flexben_fund)
    - If debtor is primary: only includes primary components with direct_billing
    """
    return sum(
        c.amount_before_prorata
        for c in self.components
        if self._is_component_billed_to_debtor(c, debtor)
    )

total_for_debtor_without_prorata

total_for_debtor_without_prorata(debtor)

The debtor is not necessarily the entity being billed

If a component's debtor is the primary but the collection method is payroll, the company gets billed.

If your purpose is to compute the total amount without prorata that should be invoiced to a given debtor, use total_billed_to_debtor_without_prorata().

Source code in components/premium/internal/domain/fee.py

def total_for_debtor_without_prorata(self, debtor: "InsuranceDebtor") -> int:
    """
    !!! warning "The debtor is not necessarily the entity being billed"

        If a component's debtor is the primary but the collection method is payroll,
        the company gets billed.

        If your purpose is to compute the total amount without prorata that should be invoiced to a given debtor,
        use `total_billed_to_debtor_without_prorata()`.
    """
    return sum(
        c.amount_before_prorata for c in self.components if c.debtor == debtor
    )

version class-attribute instance-attribute

version = field(default=1)

with_components_marked_as_invoiced

with_components_marked_as_invoiced(invoice_id, *, debtor)

Returns a new PremiumEntry with components for the specified debtor marked as invoiced.

Parameters:

Name	Type	Description	Default
invoice_id	UUID	The invoice ID to set on matching components.	required
debtor	InsuranceDebtor	The billed entity whose components should be marked as invoiced.	required

Returns:

Type	Description
PremiumEntry	New PremiumEntry with updated components.

Source code in components/premium/internal/domain/fee.py

def with_components_marked_as_invoiced(
    self, invoice_id: UUID, *, debtor: "InsuranceDebtor"
) -> "PremiumEntry":
    """
    Returns a new PremiumEntry with components for the specified debtor marked as invoiced.

    Args:
        invoice_id: The invoice ID to set on matching components.
        debtor: The billed entity whose components should be marked as invoiced.

    Returns:
        New PremiumEntry with updated components.
    """
    updated_components = [
        FeeComponent(
            id=component.id,
            num_days=component.num_days,
            amount=component.amount,
            amount_before_prorata=component.amount_before_prorata,
            currency=component.currency,
            beneficiary_type=component.beneficiary_type,
            service_type=component.service_type,
            module_id=component.module_id,
            contribution_type=component.contribution_type,
            debtor=component.debtor,
            collection_method=component.collection_method,
            enrollment_id=component.enrollment_id,
            enrollment_group_id=component.enrollment_group_id,
            member_id=component.member_id,
            invoice_id=invoice_id
            if self._is_component_billed_to_debtor(component, debtor)
            else component.invoice_id,
        )
        for component in self.components
    ]

    return PremiumEntry(
        id=self.id,
        participant_identity=self.participant_identity,
        components=updated_components,
        period_start=self.period_start,
        period_end=self.period_end,
        version=self.version,
        cancelled_by_entry_id=self.cancelled_by_entry_id,
        cancelled_entry_id=self.cancelled_entry_id,
        payroll_csv_id=self.payroll_csv_id,
        finiquito_id=self.finiquito_id,
        original_ids=self.original_ids,
    )

with_finiquito_id

with_finiquito_id(finiquito_id)

Returns a new PremiumEntry with finiquito_id set. Spain-specific.

Source code in components/premium/internal/domain/fee.py

def with_finiquito_id(self, finiquito_id: UUID) -> "PremiumEntry":
    """Returns a new PremiumEntry with finiquito_id set. Spain-specific."""
    return PremiumEntry(
        id=self.id,
        participant_identity=self.participant_identity,
        components=self.components,
        period_start=self.period_start,
        period_end=self.period_end,
        version=self.version,
        cancelled_by_entry_id=self.cancelled_by_entry_id,
        cancelled_entry_id=self.cancelled_entry_id,
        payroll_csv_id=self.payroll_csv_id,
        finiquito_id=finiquito_id,
        original_ids=self.original_ids,
    )

with_payroll_csv_id

with_payroll_csv_id(payroll_csv_id)

Returns a new PremiumEntry with payroll_csv_id set.

Source code in components/premium/internal/domain/fee.py

def with_payroll_csv_id(self, payroll_csv_id: UUID) -> "PremiumEntry":
    """Returns a new PremiumEntry with payroll_csv_id set."""
    return PremiumEntry(
        id=self.id,
        participant_identity=self.participant_identity,
        components=self.components,
        period_start=self.period_start,
        period_end=self.period_end,
        version=self.version,
        cancelled_by_entry_id=self.cancelled_by_entry_id,
        cancelled_entry_id=self.cancelled_entry_id,
        payroll_csv_id=payroll_csv_id,
        finiquito_id=self.finiquito_id,
        original_ids=self.original_ids,
    )

with_version

with_version(version)

Returns a new PremiumEntry with a different version.

Source code in components/premium/internal/domain/fee.py

def with_version(self, version: int) -> "PremiumEntry":
    """Returns a new PremiumEntry with a different version."""
    return PremiumEntry(
        id=self.id,
        participant_identity=self.participant_identity,
        components=self.components,
        period_start=self.period_start,
        period_end=self.period_end,
        version=version,
        cancelled_by_entry_id=self.cancelled_by_entry_id,
        cancelled_entry_id=self.cancelled_entry_id,
        payroll_csv_id=self.payroll_csv_id,
        finiquito_id=self.finiquito_id,
        original_ids=self.original_ids,
    )

PremiumEntryRepository

Bases: Protocol

Repository protocol for persisting and retrieving premium entries.

get_all_entries

get_all_entries(
    *,
    enrollment_ids: list[int | UUID],
    period_start: Month | date | None,
    period_end: Month | date | None
) -> list[PremiumEntry]

get_all_entries(
    *,
    participants: list[CoreStackIdentity],
    period_start: Month | date | None,
    period_end: Month | date | None
) -> list[PremiumEntry]

get_all_entries(
    *,
    enrollment_ids=None,
    participants=None,
    period_start,
    period_end
)

Retrieve all premium entries for specified enrollments with an optional period.

Returns all stored premium entries including both cancelled and active entries.

Parameters:

Name	Type	Description	Default
enrollment_ids	list[int | UUID] | None	IDs (int or UUID) of enrollments to retrieve entries for. Some implementations may only support UUID IDs and will raise an error if int IDs are provided. Mutually exclusive with participants.	None
participants	list[CoreStackIdentity] | None	Core Stack identities. Mutually exclusive with enrollment_ids. Some implementations may only support this form.	None
period_start	Month | date | None	Optional lower bound for the entry period. If provided, only returns entries where entry.period_start == period_start. If Month is passed, use the first day of the month. If None, there is no filter on the start date of the period.	required
period_end	Month | date | None	Optional upper bound for the entry period. If provided, only returns entries where entry.period_end == period_end. If Month is passed, use the last day of the month. If None, there is no filter on the end date of the period.	required

Returns:

Type	Description
list[PremiumEntry]	List of PremiumEntry objects matching the filtering criteria.
list[PremiumEntry]	If both date parameters are None, returns all entries for the enrollments.

Source code in components/premium/internal/domain/repository.py

def get_all_entries(
    self,
    *,
    enrollment_ids: list[int | UUID] | None = None,
    participants: list[CoreStackIdentity] | None = None,
    period_start: Month | date | None,
    period_end: Month | date | None,
) -> list[PremiumEntry]:
    """
    Retrieve all premium entries for specified enrollments with an optional period.

    Returns all stored premium entries including both cancelled and active entries.

    Args:
        enrollment_ids: IDs (int or UUID) of enrollments to retrieve entries for.
            Some implementations may only support UUID IDs and will raise an error
            if int IDs are provided. Mutually exclusive with participants.
        participants: Core Stack identities. Mutually exclusive with enrollment_ids.
            Some implementations may only support this form.
        period_start: Optional lower bound for the entry period.
            If provided, only returns entries where entry.period_start == period_start.
            If Month is passed, use the first day of the month.
            If None, there is no filter on the start date of the period.
        period_end: Optional upper bound for the entry period.
            If provided, only returns entries where entry.period_end == period_end.
            If Month is passed, use the last day of the month.
            If None, there is no filter on the end date of the period.

    Returns:
        List of PremiumEntry objects matching the filtering criteria.
        If both date parameters are None, returns all entries for the enrollments.
    """
    ...

get_all_uninvoiced_entries_up_to

get_all_uninvoiced_entries_up_to(
    *,
    enrollment_ids: list[int | UUID],
    debtor: InsuranceDebtor,
    up_to: date
) -> list[PremiumEntry]

get_all_uninvoiced_entries_up_to(
    *,
    participants: list[CoreStackIdentity],
    debtor: InsuranceDebtor,
    up_to: date
) -> list[PremiumEntry]

get_all_uninvoiced_entries_up_to(
    *, enrollment_ids=None, participants=None, debtor, up_to
)

Retrieve premium entries that have uninvoiced components for a specific debtor.

An entry is included if it has at least one component where the debtor has not been invoiced (invoice_id IS NULL). The returned entries are complete - they contain all components regardless of their invoice status or debtor.

Parameters:

Name	Type	Description	Default
enrollment_ids	list[int | UUID] | None	IDs (int or UUID) of enrollments to retrieve entries for. Some implementations may only support UUID IDs and will raise an error if int IDs are provided. Mutually exclusive with participants.	None
participants	list[CoreStackIdentity] | None	Core Stack identities. Mutually exclusive with enrollment_ids. Some implementations may only support this form.	None
debtor	InsuranceDebtor	The billed entity (primary, company, etc.) to check for uninvoiced components.	required
up_to	date	Only returns entries where period_end <= up_to.	required

Returns:

Type	Description
list[PremiumEntry]	List of complete PremiumEntry objects.

Source code in components/premium/internal/domain/repository.py

def get_all_uninvoiced_entries_up_to(
    self,
    *,
    enrollment_ids: list[int | UUID] | None = None,
    participants: list[CoreStackIdentity] | None = None,
    debtor: InsuranceDebtor,
    up_to: date,
) -> list[PremiumEntry]:
    """
    Retrieve premium entries that have uninvoiced components for a specific debtor.

    An entry is included if it has at least one component where the debtor has not
    been invoiced (invoice_id IS NULL). The returned entries are complete - they
    contain all components regardless of their invoice status or debtor.

    Args:
        enrollment_ids: IDs (int or UUID) of enrollments to retrieve entries for.
            Some implementations may only support UUID IDs and will raise an error
            if int IDs are provided. Mutually exclusive with participants.
        participants: Core Stack identities. Mutually exclusive with enrollment_ids.
            Some implementations may only support this form.
        debtor: The billed entity (primary, company, etc.) to check for uninvoiced components.
        up_to: Only returns entries where period_end <= up_to.

    Returns:
        List of complete PremiumEntry objects.
    """
    ...

get_entries_by_payroll_csv_id

get_entries_by_payroll_csv_id(payroll_csv_id)

Retrieve all premium entries stamped with a given payroll CSV ID.

Parameters:

Name	Type	Description	Default
payroll_csv_id	UUID	The payroll CSV ID to filter on.	required

Returns:

Type	Description
list[PremiumEntry]	List of PremiumEntry objects linked to the payroll CSV.

Source code in components/premium/internal/domain/repository.py

def get_entries_by_payroll_csv_id(
    self,
    payroll_csv_id: UUID,
) -> list[PremiumEntry]:
    """
    Retrieve all premium entries stamped with a given payroll CSV ID.

    Args:
        payroll_csv_id: The payroll CSV ID to filter on.

    Returns:
        List of PremiumEntry objects linked to the payroll CSV.
    """
    ...

get_entries_for_local_payroll

get_entries_for_local_payroll(
    *, enrollment_ids: list[UUID], target_month: date
) -> list[PremiumEntry]

get_entries_for_local_payroll(
    *,
    participants: list[CoreStackIdentity],
    target_month: date
) -> list[PremiumEntry]

get_entries_for_local_payroll(
    *, enrollment_ids=None, participants=None, target_month
)

Retrieve all entries where payroll_csv_id IS NULL and period_start <= target_month.

Parameters:

Name	Type	Description	Default
enrollment_ids	list[UUID] | None	IDs of enrollments to retrieve entries for. Mutually exclusive with participants.	None
participants	list[CoreStackIdentity] | None	Core Stack identities. Mutually exclusive with enrollment_ids. Some implementations may only support this form.	None
target_month	date	Only entries with period_start on or before this date are returned.	required

Source code in components/premium/internal/domain/repository.py

def get_entries_for_local_payroll(
    self,
    *,
    enrollment_ids: list[UUID] | None = None,
    participants: list[CoreStackIdentity] | None = None,
    target_month: date,
) -> list[PremiumEntry]:
    """
    Retrieve all entries where payroll_csv_id IS NULL and period_start <= target_month.

    Args:
        enrollment_ids: IDs of enrollments to retrieve entries for.
            Mutually exclusive with participants.
        participants: Core Stack identities. Mutually exclusive with enrollment_ids.
            Some implementations may only support this form.
        target_month: Only entries with period_start on or before this date are returned.
    """
    ...

get_entries_for_payroll

get_entries_for_payroll(
    *,
    enrollment_ids: list[int | UUID],
    target_period: date,
    created_at_lower_bound: date,
    created_at_upper_bound: date
) -> list[PremiumEntry]

get_entries_for_payroll(
    *,
    participants: list[CoreStackIdentity],
    target_period: date,
    created_at_lower_bound: date,
    created_at_upper_bound: date
) -> list[PremiumEntry]

get_entries_for_payroll(
    *,
    enrollment_ids=None,
    participants=None,
    target_period,
    created_at_lower_bound,
    created_at_upper_bound
)

Retrieve premium entries relevant for a payroll run.

• entries for the current period (identified by target_period) and create before created_at upper bound
• regularisation entries for previous periods created between created at lower and upper bound

Parameters:

Name	Type	Description	Default
enrollment_ids	list[int | UUID] | None	IDs (int or UUID) of enrollments to retrieve entries for. Mutually exclusive with participants.	None
participants	list[CoreStackIdentity] | None	Core Stack identities. Mutually exclusive with enrollment_ids. Some implementations may only support this form.	None
target_period	date	First day of the period being processed.	required
created_at_lower_bound	date	Lower bound (exclusive) on created_at for regularization entries.	required
created_at_upper_bound	date	Upper bound (inclusive) on created_at for all entries.	required

Source code in components/premium/internal/domain/repository.py

def get_entries_for_payroll(
    self,
    *,
    enrollment_ids: list[int | UUID] | None = None,
    participants: list[CoreStackIdentity] | None = None,
    target_period: date,
    created_at_lower_bound: date,
    created_at_upper_bound: date,
) -> list[PremiumEntry]:
    """
    Retrieve premium entries relevant for a payroll run.

    - entries for the current period (identified by target_period) and create before
      `created_at` upper bound
    - regularisation entries for previous periods created between created at lower and upper bound

    Args:
        enrollment_ids: IDs (int or UUID) of enrollments to retrieve entries for.
            Mutually exclusive with participants.
        participants: Core Stack identities. Mutually exclusive with enrollment_ids.
            Some implementations may only support this form.
        target_period: First day of the period being processed.
        created_at_lower_bound: Lower bound (exclusive) on ``created_at`` for
            regularization entries.
        created_at_upper_bound: Upper bound (inclusive) on ``created_at`` for all
            entries.
    """
    ...

get_latest_entries

get_latest_entries(
    *,
    enrollment_ids: list[int | UUID],
    lower_bound: Month | date,
    upper_bound: Month | date
) -> list[PremiumEntry]

get_latest_entries(
    *,
    participants: list[CoreStackIdentity],
    lower_bound: Month | date,
    upper_bound: Month | date
) -> list[PremiumEntry]

get_latest_entries(
    *,
    enrollment_ids=None,
    participants=None,
    lower_bound,
    upper_bound
)

Retrieve the latest premium entries for specified enrollments for all periods contained within lower_bound and upper_bound.

Lower bound and upper bound may cover more than one period

The caller is free to pass any range they would like. We return one premium entry, per enrollment, per period in the range.

Parameters:

Name	Type	Description	Default
enrollment_ids	list[int | UUID] | None	IDs (int or UUID) of enrollments to retrieve entries for. Some implementations may only support UUID IDs and will raise an error if int IDs are provided. Mutually exclusive with participants.	None
participants	list[CoreStackIdentity] | None	Core Stack identities. Mutually exclusive with enrollment_ids. Some implementations may only support this form.	None
lower_bound	Month | date	Range start. If Month is passed, use the first day of the month.	required
upper_bound	Month | date	Range end. If Month is passed, use the last day of the month.	required

Returns:

Type	Description
list[PremiumEntry]	One premium entry per enrollment per period in the range.
list[PremiumEntry]	For each period, the premium returned is the latest as defined by the
list[PremiumEntry]	version.

Source code in components/premium/internal/domain/repository.py

def get_latest_entries(
    self,
    *,
    enrollment_ids: list[int | UUID] | None = None,
    participants: list[CoreStackIdentity] | None = None,
    lower_bound: Month | date,
    upper_bound: Month | date,
) -> list[PremiumEntry]:
    """
    Retrieve the latest premium entries for specified enrollments for all periods
    contained within lower_bound and upper_bound.

    !!! warning "Lower bound and upper bound may cover more than one period"

        The caller is free to pass any range they would like.
        We return one premium entry, per enrollment, per period in the range.

    Args:
        enrollment_ids: IDs (int or UUID) of enrollments to retrieve entries for.
            Some implementations may only support UUID IDs and will raise an error
            if int IDs are provided. Mutually exclusive with participants.
        participants: Core Stack identities. Mutually exclusive with enrollment_ids.
            Some implementations may only support this form.
        lower_bound: Range start. If Month is passed, use the first day of the month.
        upper_bound: Range end. If Month is passed, use the last day of the month.

    Returns:
        One premium entry per enrollment per period in the range.
        For each period, the premium returned is the latest as defined by the
        version.
    """
    ...

get_premium_entries_for_invoice

get_premium_entries_for_invoice(invoice_id)

Retrieve premium entries for a specific invoice. If invoice has no components, then it returns an empty list.

Parameters:

Name	Type	Description	Default
invoice_id	HybridId	The ID of the invoice to retrieve premium entries for.	required

Returns:

Type	Description
list[PremiumEntry]	List of PremiumEntry objects for the invoice.

Source code in components/premium/internal/domain/repository.py

def get_premium_entries_for_invoice(
    self, invoice_id: HybridId
) -> list[PremiumEntry]:
    """
    Retrieve premium entries for a specific invoice.
    If invoice has no components, then it returns an empty list.

    Args:
        invoice_id: The ID of the invoice to retrieve premium entries for.

    Returns:
        List of PremiumEntry objects for the invoice.
    """
    ...

insert

insert(entries)

Insert new premium entries into the database.

Parameters:

Name	Type	Description	Default
entries	list[PremiumEntry]	List of PremiumEntry objects to insert. Each entry's components will be added to the database.	required

Source code in components/premium/internal/domain/repository.py

def insert(self, entries: list[PremiumEntry]) -> None:
    """
    Insert new premium entries into the database.

    Args:
        entries: List of PremiumEntry objects to insert.
                 Each entry's components will be added to the database.
    """
    ...

session instance-attribute

session

update

update(entries)

Persist mutable fields of existing premium entries.

Only the following fields are updated — all others are immutable after insert: - cancelled_entry_id - cancelled_by_entry_id - invoice_id - payroll_csv_id - finiquito_id (Spain only — always null in Belgium)

Parameters:

Name	Type	Description	Default
entries	list[PremiumEntry]	List of PremiumEntry objects to update.	required

Returns:

Type	Description
int	Number of rows updated. Returns > 0 if at least one component was updated.

Source code in components/premium/internal/domain/repository.py

def update(self, entries: list[PremiumEntry]) -> int:
    """
    Persist mutable fields of existing premium entries.

    Only the following fields are updated — all others are immutable after insert:
    - ``cancelled_entry_id``
    - ``cancelled_by_entry_id``
    - ``invoice_id``
    - ``payroll_csv_id``
    - ``finiquito_id`` (Spain only — always null in Belgium)

    Args:
        entries: List of PremiumEntry objects to update.

    Returns:
        Number of rows updated. Returns > 0 if at least one component was updated.
    """
    ...

ProrataStrategy

Bases: AlanBaseEnum

Strategy for applying prorata to cost components.

When multiple cost components need prorating (e.g., base cost, membership fee, taxes), different strategies can be used that may produce slightly different results due to rounding.

THIRTY_DAY_PRORATA class-attribute instance-attribute

THIRTY_DAY_PRORATA = '30_day_prorata'

If the service covers the whole month, bill the full amount for 1 month. A month is a full month if the number of days covered equals the number of days in the month. If the service does not cover the whole month, prorate on the base of 30 days.

Formula: monthly_price * days_covered / 30

This prorata is applied to all the components of the subscription fee and may carry rounding errors, meaning the total of components may exceed the initial total.

THIRTY_DAY_PRORATA_WITH_LARGEST_REMAINDER_DISTRIBUTION_ACROSS_FEE_COMPONENTS class-attribute instance-attribute

THIRTY_DAY_PRORATA_WITH_LARGEST_REMAINDER_DISTRIBUTION_ACROSS_FEE_COMPONENTS = "30_day_prorata_with_largest_remainder_distribution_across_fee_components"

Similar to THIRTY_DAY_PRORATA except rounding errors are avoided by using the largest remainder algorithm to distribute the last few cents across the component.

There is a risk of losing accuracy, for example attributing more to the membership fee than we should. But we have the guarantee the total of the components will never exceed the initial total and the inaccuracies should never exceed 1 cent.

SubscriptionPeriod dataclass

SubscriptionPeriod(*, fees)

Bases: GenericPeriod

fees instance-attribute

fees

components.premium.public.models

helper

prefix_table_args

prefix_table_args(table_args, prefix)

Return a copy of a __table_args__ tuple with all constraint/index names prefixed.

SQLAlchemy tracks named constraints and indexes in a global MetaData registry keyed by name alone, not by schema. When two concrete subclasses of the same abstract model share a __tablename__ and are loaded into the same process, SQLAlchemy raises ArgumentError on duplicate constraint names. Prefixing each name (e.g. "es_") makes them unique across subclasses.

Each Index/UniqueConstraint/CheckConstraint is reconstructed (not shallow-copied). A shallow copy would keep a reference to the same internal name object, so mutating the name to add the prefix would silently mutate the original constraint on the abstract model — making the prefix appear to work while actually renaming the source, not creating a distinct object.

Additionally, copy.copy on an Index retains the parent abstract model's table binding, which causes SQLAlchemy to raise:

"Index X is against table T, and cannot be associated with table T" when attaching it to the concrete subclass's Table. Plain dicts and unnamed entries pass through unchanged.

Only Index, UniqueConstraint, and CheckConstraint entries are prefixed. Any other SchemaItem subtype is passed through as-is without modification.

Source code in components/premium/public/models/helper.py

def prefix_table_args(
    table_args: tuple[object, ...],
    prefix: str,
) -> tuple[object, ...]:
    """
    Return a copy of a ``__table_args__`` tuple with all constraint/index names prefixed.

    SQLAlchemy tracks named constraints and indexes in a global ``MetaData`` registry keyed
    by name alone, not by schema. When two concrete subclasses of the same abstract model
    share a ``__tablename__`` and are loaded into the same process, SQLAlchemy raises
    ``ArgumentError`` on duplicate constraint names. Prefixing each name (e.g. ``"es_"``)
    makes them unique across subclasses.

    Each Index/UniqueConstraint/CheckConstraint is reconstructed (not shallow-copied).
    A shallow copy would keep a reference to the same internal ``name`` object, so mutating
    the name to add the prefix would silently mutate the original constraint on the abstract
    model — making the prefix appear to work while actually renaming the source, not creating
    a distinct object.

    Additionally, ``copy.copy`` on an ``Index`` retains the parent abstract
    model's table binding, which causes SQLAlchemy to raise:

    "Index X is against table T, and cannot be associated with table T" when attaching it to
    the concrete subclass's Table. Plain dicts and unnamed entries pass through unchanged.

    Only ``Index``, ``UniqueConstraint``, and ``CheckConstraint`` entries are prefixed.
    Any other ``SchemaItem`` subtype is passed through as-is without modification.
    """
    result: list[object] = []
    for arg in table_args:
        if isinstance(arg, Index) and arg.name:
            result.append(
                Index(
                    f"{prefix}{arg.name}",
                    *arg.expressions,
                    unique=arg.unique,
                    **dict(arg.kwargs),
                )
            )
        elif isinstance(arg, UniqueConstraint) and arg.name:
            result.append(
                UniqueConstraint(
                    *[c for c in arg._pending_colargs if c is not None],  # noqa: ALN027
                    name=f"{prefix}{arg.name}",
                    deferrable=arg.deferrable,
                    initially=arg.initially,
                    **dict(arg.kwargs),
                )
            )
        elif isinstance(arg, CheckConstraint) and arg.name:
            result.append(
                CheckConstraint(
                    arg.sqltext,
                    name=f"{prefix}{arg.name}",
                    deferrable=arg.deferrable,
                    initially=arg.initially,
                    **dict(arg.kwargs),
                )
            )
        else:
            result.append(arg)
    return tuple(result)

premium_component

PremiumComponentModel

Bases: BaseModel

Abstract base for per-country, per-product premium component tables. Each product is expected to have a subclass in their own schema.

See components/premium/README.md for how to add a new country table.

__abstract__ class-attribute instance-attribute

__abstract__ = True

__table_args__ class-attribute instance-attribute

__table_args__ = (
    UniqueConstraint(
        "enrollment_id",
        "enrollment_group_id",
        "member_id",
        "premium_entry_id",
        "coverage_type",
        "module_id",
        "beneficiary_type",
        "debtor_type",
        "period_start",
        "period_end",
        "contribution_type",
        "version",
        name="one_component_across_all_dimensions_per_version",
        postgresql_nulls_not_distinct=True,
    ),
    CheckConstraint(
        "(enrollment_id IS NOT NULL)::int + ((enrollment_group_id IS NOT NULL AND member_id IS NOT NULL AND module_id IS NOT NULL))::int = 1",
        name="exactly_one_identity_shape",
    ),
    Index(
        "ix_enrollment_id_version",
        "enrollment_id",
        "version",
    ),
    Index(
        "ix_enrollment_group_member_version",
        "enrollment_group_id",
        "member_id",
        "version",
    ),
    Index(
        "ix_period_start_end", "period_start", "period_end"
    ),
)

amount class-attribute instance-attribute

amount = mapped_column(Integer, nullable=False)

Amount for num_days of coverage from period_start to period_end. Expressed in the minor unit of the currency.

amount_before_prorata class-attribute instance-attribute

amount_before_prorata = mapped_column(
    Integer, nullable=False
)

Amount for full coverage from period_start to period_end. Expressed in the minor unit of the currency. This is the amount that would be billed if we decided not to apply prorata.

beneficiary_type class-attribute instance-attribute

beneficiary_type = mapped_column(
    Enum(
        InsuranceBeneficiary,
        create_type=False,
        values_callable=lambda x: [(value) for e in x],
    ),
    nullable=False,
)

billed_to_debtor_filter classmethod

billed_to_debtor_filter(debtor)

Returns a SQLAlchemy filter expression equivalent to PremiumEntry._is_component_billed_to_debtor.

These two must always express the same logic. Update both when the billing rules change.

Source code in components/premium/public/models/premium_component.py

@classmethod
def billed_to_debtor_filter(cls, debtor: InsuranceDebtor) -> ColumnElement[bool]:
    """
    Returns a SQLAlchemy filter expression equivalent to PremiumEntry._is_component_billed_to_debtor.

    These two must always express the same logic. Update both when the billing rules change.
    """
    match debtor:
        case InsuranceDebtor.primary:
            return and_(
                cls.debtor_type == InsuranceDebtor.primary,
                cls.collection_method == InsuranceCollectionMethod.direct_billing,
            )
        case InsuranceDebtor.company:
            return or_(
                cls.debtor_type == InsuranceDebtor.company,
                and_(
                    cls.debtor_type == InsuranceDebtor.primary,
                    cls.collection_method == InsuranceCollectionMethod.payroll,
                ),
                and_(
                    cls.debtor_type == InsuranceDebtor.primary,
                    cls.collection_method == InsuranceCollectionMethod.flexben_fund,
                ),
            )
    raise AssertionError(f"Unexpected debtor: {debtor}")

cancelled_by_entry_id class-attribute instance-attribute

cancelled_by_entry_id = mapped_column(
    UUID(as_uuid=True), nullable=True, index=True
)

cancelled_entry_id class-attribute instance-attribute

cancelled_entry_id = mapped_column(
    UUID(as_uuid=True), nullable=True, index=True
)

collection_method class-attribute instance-attribute

collection_method = mapped_column(
    Enum(
        InsuranceCollectionMethod,
        create_type=False,
        values_callable=lambda x: [(value) for e in x],
        nullable=True,
    )
)

contribution_type class-attribute instance-attribute

contribution_type = mapped_column(
    Enum(
        InsuranceContribution,
        create_type=False,
        values_callable=lambda x: [(value) for e in x],
    ),
    nullable=False,
)

coverage_type class-attribute instance-attribute

coverage_type = mapped_column(
    Enum(
        InsuranceService,
        create_type=False,
        values_callable=lambda x: [(value) for e in x],
    ),
    nullable=False,
)

currency class-attribute instance-attribute

currency = mapped_column(String, nullable=False)

ISO4217 compliant alpha code of the currency.

debtor_type class-attribute instance-attribute

debtor_type = mapped_column(
    Enum(
        InsuranceDebtor,
        name="insurancedebtor",
        create_type=False,
        values_callable=lambda x: [(value) for e in x],
    ),
    nullable=False,
)

enrollment_group_id class-attribute instance-attribute

enrollment_group_id = mapped_column(
    UUID(as_uuid=True), nullable=True
)

enrollment_id class-attribute instance-attribute

enrollment_id = mapped_column(
    UUID(as_uuid=True), nullable=True
)

Identifier of the participant, for legacy subscriptions.

finiquito_id class-attribute instance-attribute

finiquito_id = mapped_column(
    UUID(as_uuid=True), nullable=True
)

Spain-specific. A finiquito is a pay_csv generated to recap the premiums of an employee when they leave the company. In order to maintain compatibility between Spain payroll and Global premiums we need to maintain this property. However it's not required by the Global payroll system, therefore this property is temporary and will be dropped as soon as global payroll is online.

Always null in Belgium.

invoice_id class-attribute instance-attribute

invoice_id = mapped_column(nullable=True, index=True)

member_id class-attribute instance-attribute

member_id = mapped_column(String, nullable=True)

module_id class-attribute instance-attribute

module_id = mapped_column(UUID(as_uuid=True), nullable=True)

num_days class-attribute instance-attribute

num_days = mapped_column(Integer, nullable=False)

Number of days this fee covers within the billing period.

payroll_csv_id class-attribute instance-attribute

payroll_csv_id = mapped_column(
    UUID(as_uuid=True), nullable=True
)

Belgium & Spain's local payroll systems requires to keep track of which premiums are included in which file.

In order to maintain compatibility between Belgium payroll and Global premiums we need to maintain this property. However it's not required by the Global payroll system, therefore this property is temporary and will be dropped as soon as global payroll is online.

period_end class-attribute instance-attribute

period_end = mapped_column(Date, nullable=False)

period_start class-attribute instance-attribute

period_start = mapped_column(Date, nullable=False)

premium_entry_id class-attribute instance-attribute

premium_entry_id = mapped_column(
    UUID(as_uuid=True), nullable=False
)

version class-attribute instance-attribute

version = mapped_column(Integer, nullable=False)