Api reference

components.healthy_benefits.public.affiliation ¶

EnrollArgs `dataclass` ¶

EnrollArgs(
    benefit_type_ref,
    profile_ref,
    start_date,
    transfer_history_id,
)

Arguments to create an enrollment.

benefit_type_ref `instance-attribute` ¶

benefit_type_ref

profile_ref `instance-attribute` ¶

profile_ref

start_date `instance-attribute` ¶

start_date

transfer_history_id `instance-attribute` ¶

transfer_history_id

Enrollment `dataclass` ¶

Enrollment(
    id,
    benefit_type,
    subscription_id,
    transfer_history_id,
    profile_ref,
    start_date=isodate_field(),
    end_date=optional_isodate_field(default=None),
    is_cancelled=False,
)

Represents a Healthy Benefits enrollment.

benefit_type `instance-attribute` ¶

benefit_type

end_date `class-attribute` `instance-attribute` ¶

end_date = optional_isodate_field(default=None)

id `instance-attribute` ¶

id

is_active_at ¶

is_active_at(date)

True if the enrollment is active on the given date.

Source code in components/healthy_benefits/subcomponents/affiliation/protected/entities.py

def is_active_at(self, date: date) -> bool:
    """True if the enrollment is active on the given date."""
    return self.start_date <= date and (
        self.end_date is None or self.end_date >= date
    )

is_cancelled `class-attribute` `instance-attribute` ¶

is_cancelled = False

is_ended_at ¶

is_ended_at(date)

True if the enrollment ended before the given date.

Source code in components/healthy_benefits/subcomponents/affiliation/protected/entities.py

def is_ended_at(self, date: date) -> bool:
    """True if the enrollment ended before the given date."""
    return self.end_date is not None and self.end_date < date

profile_ref `instance-attribute` ¶

profile_ref

start_date `class-attribute` `instance-attribute` ¶

start_date = isodate_field()

subscription_id `instance-attribute` ¶

subscription_id

transfer_history_id `instance-attribute` ¶

transfer_history_id

EnrollmentId `module-attribute` ¶

EnrollmentId = NewType('EnrollmentId', UUID)

EnrollmentModelBroker ¶

Bases: BaseModelBroker

cancel_enrollment `classmethod` ¶

cancel_enrollment(enrollment_id)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def cancel_enrollment(cls, enrollment_id: UUID) -> None:
    enrollment = get_or_raise_missing_resource(cls.model, enrollment_id)
    enrollment.is_cancelled = True

cancel_enrollments `classmethod` ¶

cancel_enrollments(enrollment_ids)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def cancel_enrollments(cls, enrollment_ids: list[UUID]) -> None:
    cls.query().where(cls.model.id.in_(enrollment_ids)).update(
        {cls.model.is_cancelled: True}
    )

create_enrollment `classmethod` ¶

create_enrollment(**data)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def create_enrollment(cls, **data) -> Enrollment:  # type: ignore[no-untyped-def]
    enrollment = cls.model(**data)
    current_session.add(enrollment)
    current_session.flush()

    return enrollment

get_active_or_ended_benefit_enrollment `classmethod` ¶

get_active_or_ended_benefit_enrollment(
    profile_ref,
    at_date,
    benefit_type,
    subscription_ref=None,
)

Retrieve the most recent enrollment based on the given profile reference, date, and benefit type, optionally filtering by subscription reference.

This method will return the most recent benefit enrollment that is either active or ended as of a specified date for a given benefit type.

Parameters:

Name	Type	Description	Default
`cls`	`type`	The class on which this method is called.	required
`profile_ref`	`str`	The profile reference to filter enrollments.	required
`at_date`	`date`	The date at which the enrollment status is checked.	required
`benefit_type`	`BenefitType`	The type of benefit to filter enrollments.	required
`subscription_ref`	`str \| None`	Optional; A subscription reference to further filter the enrollments.	`None`

Returns:

Type	Description
`HealthyBenefitsEnrollment \| None`	Enrollment \| None: The most recent Enrollment object that meets the criteria, or None if no enrollment is found.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_active_or_ended_benefit_enrollment(
    cls,
    profile_ref: str,
    at_date: date,
    benefit_type: BenefitType,
    subscription_ref: str | None = None,
) -> Enrollment | None:
    """
    Retrieve the most recent enrollment based on the given profile reference, date, and benefit type,
    optionally filtering by subscription reference.

    This method will return the most recent benefit enrollment that is either active or ended
    as of a specified date for a given benefit type.

    Parameters:
        cls (type): The class on which this method is called.
        profile_ref (str): The profile reference to filter enrollments.
        at_date (date): The date at which the enrollment status is checked.
        benefit_type (BenefitType): The type of benefit to filter enrollments.
        subscription_ref (str | None): Optional; A subscription reference to further filter the enrollments.

    Returns:
        Enrollment | None: The most recent Enrollment object that meets the criteria, or None if no enrollment is found.
    """
    query = (
        cls.query()
        .where(
            cls.model.profile_ref == profile_ref,
            or_(
                cls.model.is_active_on(at_date),
                cls.model.is_ended_on(at_date),
            ),
            cls.model.benefit_type == benefit_type,
        )
        .order_by(cls.model.start_date.desc())
    )
    if subscription_ref:
        query = query.where(cls.model.subscription_ref == subscription_ref)

    return cast("Enrollment | None", query.first())

get_ended_enrollments `classmethod` ¶

get_ended_enrollments(
    at_date=None,
    profile_refs=None,
    benefit_types=None,
    subscription_refs=None,
)

Retrieve all ended enrollments as of a specific date.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_ended_enrollments(
    cls,
    at_date: date | None = None,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_refs: list[str] | None = None,
) -> list[Enrollment]:
    """
    Retrieve all ended enrollments as of a specific date.
    """
    _at_date = at_date or utctoday()
    query = current_session.query(cls.model).filter(cls.model.is_ended_on(_at_date))  # noqa: ALN085

    if profile_refs:
        query = query.filter(cls.model.profile_ref.in_(profile_refs))

    if benefit_types:
        query = query.filter(cls.model.benefit_type.in_(benefit_types))

    if subscription_refs:
        query = query.filter(cls.model.subscription_ref.in_(subscription_refs))

    return cast("list[Enrollment]", query.all())  # type: ignore[redundant-cast]

get_enrollment `classmethod` ¶

get_enrollment(id)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollment(cls, id: UUID) -> Enrollment:
    return get_or_raise_missing_resource(cls.model, id)

get_enrollment_by_ids `classmethod` ¶

get_enrollment_by_ids(ids)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollment_by_ids(cls, ids: list[EnrollmentId]) -> list[Enrollment]:
    return cls.query().where(cls.model.id.in_(ids)).all()

get_enrollment_by_transfer_history_ref `classmethod` ¶

get_enrollment_by_transfer_history_ref(
    transfer_history_ref, active_on
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollment_by_transfer_history_ref(
    cls, transfer_history_ref: str, active_on: datetime
) -> Enrollment | None:
    return (
        current_session.query(cls.model)  # noqa: ALN085
        .filter(cls.model.is_active_on(active_on))
        .filter(cls.model.transfer_history_ref == transfer_history_ref)
        .one_or_none()
    )

get_enrollments `classmethod` ¶

get_enrollments(
    active_on=None,
    ended_on=None,
    active_on_or_after=None,
    profile_refs=None,
    benefit_types=None,
    subscription_refs=None,
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollments(
    cls,
    active_on: datetime | None = None,
    ended_on: date | None = None,
    active_on_or_after: date | None = None,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_refs: list[str] | None = None,
) -> list[Enrollment]:
    query = cls.query()
    if active_on:
        query = query.filter(cls.model.is_active_on(active_on))
    if profile_refs:
        query = query.filter(cls.model.profile_ref.in_(profile_refs))

    if benefit_types:
        query = query.filter(cls.model.benefit_type.in_(benefit_types))

    if subscription_refs:
        query = query.filter(cls.model.subscription_ref.in_(subscription_refs))

    if ended_on:
        query = query.filter(cls.model.is_ended_on(ended_on))

    if active_on_or_after:
        query = query.filter(
            or_(
                cls.model.is_active_on(active_on_or_after),
                cls.model.start_date >= active_on_or_after,
            )
        )

    return query.all()

get_enrollments_active_after `classmethod` ¶

get_enrollments_active_after(
    active_after,
    profile_refs=None,
    benefit_types=None,
    subscription_refs=None,
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollments_active_after(
    cls,
    active_after: datetime,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_refs: list[str] | None = None,
) -> list[Enrollment]:
    query = current_session.query(cls.model).filter(  # noqa: ALN085
        cls.model.start_date >= active_after.date(),
        cls.model.is_cancelled.is_(False),
    )

    if profile_refs:
        query = query.filter(cls.model.profile_ref.in_(profile_refs))

    if benefit_types:
        query = query.filter(cls.model.benefit_type.in_(benefit_types))

    if subscription_refs:
        query = query.filter(cls.model.subscription_ref.in_(subscription_refs))

    return query.all()

get_enrollments_active_in_period `classmethod` ¶

get_enrollments_active_in_period(
    period,
    profile_refs=None,
    benefit_types=None,
    subscription_refs=None,
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollments_active_in_period(
    cls,
    period: ValidityPeriod,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_refs: list[str] | None = None,
) -> list[Enrollment]:
    query = current_session.query(cls.model).filter(  # noqa: ALN085
        cls.model.is_ever_active_between(period.start_date, period.end_date)
    )

    if profile_refs:
        query = query.filter(cls.model.profile_ref.in_(profile_refs))

    if benefit_types:
        query = query.filter(cls.model.benefit_type.in_(benefit_types))

    if subscription_refs:
        query = query.filter(cls.model.subscription_ref.in_(subscription_refs))

    return query.all()

get_enrollments_active_on `classmethod` ¶

get_enrollments_active_on(
    active_on,
    profile_refs=None,
    benefit_types=None,
    subscription_refs=None,
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def get_enrollments_active_on(
    cls,
    active_on: datetime,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_refs: list[str] | None = None,
) -> list[Enrollment]:
    query = current_session.query(cls.model).filter(  # noqa: ALN085
        cls.model.is_active_on(active_on)
    )

    if profile_refs:
        query = query.filter(cls.model.profile_ref.in_(profile_refs))

    if benefit_types:
        query = query.filter(cls.model.benefit_type.in_(benefit_types))

    if subscription_refs:
        query = query.filter(cls.model.subscription_ref.in_(subscription_refs))

    query = query.order_by(cls.model.created_at.desc())

    return query.all()

model `class-attribute` `instance-attribute` ¶

model = HealthyBenefitsEnrollment

terminate_enrollment `classmethod` ¶

terminate_enrollment(id, end_date)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def terminate_enrollment(cls, id: UUID, end_date: date) -> Enrollment:
    from components.healthy_benefits.subcomponents.affiliation.protected.exceptions import (
        InvalidEnrollmentDatesException,
    )

    enrollment = get_or_raise_missing_resource(cls.model, id)

    if enrollment.start_date and enrollment.start_date > end_date:
        raise InvalidEnrollmentDatesException(
            f"Enrollment end date must be on or after enrollment start date. "
            f"Enrollment start date: {enrollment.start_date}, "
            f"Enrollment end date: {end_date}"
        )
    enrollment.end_date = end_date

    return enrollment

update_enrollment `classmethod` ¶

update_enrollment(id, **data)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/enrollment.py

@classmethod
def update_enrollment(cls, id: UUID, **data) -> Enrollment:  # type: ignore[no-untyped-def]
    from components.healthy_benefits.subcomponents.affiliation.protected.exceptions import (
        InvalidEnrollmentDatesException,
    )

    enrollment = get_or_raise_missing_resource(cls.model, id)
    end_date = data.get("end_date") or enrollment.end_date
    start_date = data.get("start_date") or enrollment.start_date

    if start_date and end_date and start_date > end_date:
        raise InvalidEnrollmentDatesException(
            f"Enrollment end date must be on or after enrollment start date. "
            f"Enrollment start date: {start_date}, "
            f"Enrollment end date: {end_date}"
        )
    enrollment.assign(data)

    return enrollment

TemporaryLeave `dataclass` ¶

TemporaryLeave(
    id,
    enrollment_id,
    start_date,
    end_date,
    is_cancelled,
    status,
)

Represents a temporary leave for a healthy benefit enrollment.

end_date `instance-attribute` ¶

end_date

enrollment_id `instance-attribute` ¶

enrollment_id

id `instance-attribute` ¶

id

is_cancelled `instance-attribute` ¶

is_cancelled

start_date `instance-attribute` ¶

start_date

status `instance-attribute` ¶

status

TemporaryLeaveId `module-attribute` ¶

TemporaryLeaveId = NewType('TemporaryLeaveId', UUID)

TemporaryLeaveModelBroker ¶

Bases: BaseModelBroker

cancel_temporary_leaves `classmethod` ¶

cancel_temporary_leaves(temporary_leave_ids)

Cancels a list of temporary leaves.

Parameters:

Name	Type	Description	Default
`temporary_leave_ids`	`list[TemporaryLeaveId]`	The temporary leave ids.	required

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/temporary_leave.py

@classmethod
def cancel_temporary_leaves(
    cls, temporary_leave_ids: list[TemporaryLeaveId]
) -> None:
    """Cancels a list of temporary leaves.

    Args:
        temporary_leave_ids(list[TemporaryLeaveId]): The temporary leave ids.
    """
    stmt = (
        update(cls.model)
        .where(cls.model.id.in_(temporary_leave_ids))
        .values(is_cancelled=True)
    )
    current_session.execute(stmt)
    current_session.flush()

create_temporary_leave `classmethod` ¶

create_temporary_leave(
    enrollment_id, start_date, end_date=None
)

Create a temporary leave for a given enrollment.

Parameters:

Name	Type	Description	Default
`enrollment_id`	`EnrollmentId`	The enrollment id.	required
`start_date`	`date`	The start date of the temporary leave.	required
`end_date`	`date \| None`	The end date of the temporary leave. Defaults to None.	`None`

Returns:

Name	Type	Description
`TemporaryLeave`	`TemporaryLeave`	The created temporary leave.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/temporary_leave.py

@classmethod
def create_temporary_leave(
    cls,
    enrollment_id: EnrollmentId,
    start_date: date,
    end_date: date | None = None,
) -> TemporaryLeave:
    """Create a temporary leave for a given enrollment.

    Args:
        enrollment_id(EnrollmentId): The enrollment id.
        start_date(date): The start date of the temporary leave.
        end_date(date | None): The end date of the temporary leave. Defaults to None.

    Returns:
        TemporaryLeave: The created temporary leave.
    """
    temporary_leave = cls.model(
        enrollment_id=enrollment_id,
        start_date=start_date,
        end_date=end_date,
        is_cancelled=False,
    )
    current_session.add(temporary_leave)
    current_session.flush()
    return temporary_leave

get_temporary_leaves `classmethod` ¶

get_temporary_leaves(
    temporary_leave_ids, with_cancelled=False
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/temporary_leave.py

@classmethod
def get_temporary_leaves(
    cls,
    temporary_leave_ids: list[TemporaryLeaveId],
    with_cancelled: bool = False,
) -> list[TemporaryLeave]:
    stmt = select(cls.model).where(cls.model.id.in_(temporary_leave_ids))

    if with_cancelled:
        # Include all leaves
        pass
    else:
        # Only non-cancelled leaves (default)
        stmt = stmt.where(cls.model.is_cancelled.is_(False))

    stmt = stmt.order_by(cls.model.created_at.desc())
    return list(current_session.execute(stmt).scalars().all())

get_temporary_leaves_for_enrollments `classmethod` ¶

get_temporary_leaves_for_enrollments(
    enrollment_ids, with_cancelled=False
)

Get the temporary leaves for a list of enrollments.

Parameters:

Name	Type	Description	Default
`enrollment_ids`	`list[EnrollmentId]`	The enrollment ids.	required
`with_cancelled`	`bool`	Whether to include cancelled temporary leaves. Defaults to False.	`False`

Returns:

Type	Description
`list[TemporaryLeave]`	list[TemporaryLeave]: The temporary leaves for the given enrollments.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/temporary_leave.py

@classmethod
def get_temporary_leaves_for_enrollments(
    cls,
    enrollment_ids: list[EnrollmentId],
    with_cancelled: bool = False,
) -> list[TemporaryLeave]:
    """Get the temporary leaves for a list of enrollments.

    Args:
        enrollment_ids(list[EnrollmentId]): The enrollment ids.
        with_cancelled(bool, optional): Whether to include cancelled temporary leaves. Defaults to False.

    Returns:
        list[TemporaryLeave]: The temporary leaves for the given enrollments.
    """
    stmt = select(cls.model).where(cls.model.enrollment_id.in_(enrollment_ids))
    if with_cancelled:
        # Include all leaves
        pass
    else:
        # Only non-cancelled leaves (default)
        stmt = stmt.where(cls.model.is_cancelled.is_(False))

    stmt = stmt.order_by(cls.model.created_at.desc())
    return list(current_session.execute(stmt).scalars().all())

model `class-attribute` `instance-attribute` ¶

model = TemporaryLeave

update_temporary_leaves `classmethod` ¶

update_temporary_leaves(
    temporary_leave_ids,
    *,
    start_date=None,
    end_date=NOT_SET
)

Updates the start date and/or end date of a list of temporary leaves. As end_date can be None, we need to use a sentinel value to differentiate between None and not_set.

Parameters:

Name	Type	Description	Default
`temporary_leave_ids`	`list[TemporaryLeaveId]`	The temporary leave ids.	required
`start_date`	`date \| None`	The start date of the temporary leave.	`None`
`end_date`	`date \| None`	The end date of the temporary leave.	`NOT_SET`

Source code in components/healthy_benefits/subcomponents/affiliation/internal/models/brokers/temporary_leave.py

@classmethod
def update_temporary_leaves(
    cls,
    temporary_leave_ids: list[TemporaryLeaveId],
    *,
    start_date: date | None = None,
    end_date: NotSet[date | None] = NOT_SET,
) -> None:
    """Updates the start date and/or end date of a list of temporary leaves.
    As end_date can be None, we need to use a sentinel value to differentiate between None and not_set.

    Args:
        temporary_leave_ids(list[TemporaryLeaveId]): The temporary leave ids.
        start_date(date | None): The start date of the temporary leave.
        end_date(date | None): The end date of the temporary leave.
    """
    values: dict[str, date | None] = {}

    if start_date:
        values["start_date"] = start_date
    if is_set(end_date):
        values["end_date"] = end_date

    stmt = (
        update(cls.model)
        .where(cls.model.id.in_(temporary_leave_ids))
        .values(**values)
    )
    current_session.execute(stmt)
    current_session.flush()

TemporaryLeaveStatus ¶

Bases: AlanBaseEnum

Represents the status of a temporary leave.

ACTIVE `class-attribute` `instance-attribute` ¶

ACTIVE = 'active'

CANCELLED `class-attribute` `instance-attribute` ¶

CANCELLED = 'cancelled'

ENDED `class-attribute` `instance-attribute` ¶

ENDED = 'ended'

SCHEDULED `class-attribute` `instance-attribute` ¶

SCHEDULED = 'scheduled'

bulk_cancel_enrollments ¶

bulk_cancel_enrollments(enrollments_ids, commit=True)

Cancel provided list of enrollments Ignores enrollments that do not exist

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def bulk_cancel_enrollments(
    enrollments_ids: list[EnrollmentId], commit: bool = True
) -> None:
    """
    Cancel provided list of enrollments
    Ignores enrollments that do not exist
    """
    EnrollmentModelBroker.cancel_enrollments(
        enrollment_ids=[enrollment_id for enrollment_id in enrollments_ids]
    )

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

bulk_enroll ¶

bulk_enroll(enrollments_args, subscription_id, commit=True)

Enroll list of profiles to a given subscription. Will raise an error if the given subscription is not actively subscribed to this benefit at this date

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def bulk_enroll(
    enrollments_args: list[EnrollArgs],
    subscription_id: SubscriptionId,
    commit: bool = True,
) -> list[Enrollment]:
    """
    Enroll list of profiles to a given subscription.
    Will raise an error if the given subscription is not actively subscribed to this benefit at this date
    """
    from components.healthy_benefits.subcomponents.affiliation.internal.business_logic.enrollment import (
        bulk_create_enrollments as bulk_create_enrollments_action,
    )

    return bulk_create_enrollments_action(
        enrollments_args=enrollments_args,
        subscription_id=subscription_id,
        commit=commit,
    )

bulk_terminate_enrollments ¶

bulk_terminate_enrollments(
    enrollments_ids, end_date=None, commit=True
)

End a bulk of enrollments at the given end date (default to today if not set)

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def bulk_terminate_enrollments(
    enrollments_ids: list[EnrollmentId],
    end_date: date | None = None,
    commit: bool = True,
) -> list[Enrollment]:
    """End a bulk of enrollments at the given end date (default to today if not set)"""
    from components.healthy_benefits.subcomponents.affiliation.internal.business_logic.enrollment import (
        bulk_terminate_enrollments as bulk_terminate_enrollments_action,
    )

    return bulk_terminate_enrollments_action(
        enrollments_ids=enrollments_ids, end_date=end_date, commit=commit
    )

cancel_enrollment ¶

cancel_enrollment(enrollment_id, commit=True)

Cancel provided enrollment. Raise an error if the enrollment does not exist

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def cancel_enrollment(enrollment_id: EnrollmentId, commit: bool = True) -> None:
    """
    Cancel provided enrollment.
    Raise an error if the enrollment does not exist
    """
    EnrollmentModelBroker.cancel_enrollment(enrollment_id=enrollment_id)
    if commit:
        current_session.commit()
    else:
        current_session.flush()

get_active_or_future_enrollments ¶

get_active_or_future_enrollments(
    at_date, subscription_id, profile_ref=None
)

Retrieve active enrollments at a given date, or the next future enrollments if none are active.

Parameters:

Name	Type	Description	Default
`at_date`	`date`	The reference date to check for active enrollments.	required
`subscription_id`	`SubscriptionId`	The subscription to filter enrollments by.	required
`profile_ref`	`Optional[str]`	Unique reference identifier of the user profile. If None, enrollments are fetched without filtering by profile.	`None`

Returns:

Type	Description
`list[Enrollment]`	list[Enrollment]: A list of active enrollments at the given date, or the
`list[Enrollment]`	next available future enrollments if none are active.

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def get_active_or_future_enrollments(
    at_date: date,
    subscription_id: SubscriptionId,
    profile_ref: Optional[str] = None,
) -> list[Enrollment]:
    """
    Retrieve active enrollments at a given date, or the next future
    enrollments if none are active.

    Args:
        at_date (date): The reference date to check for active enrollments.
        subscription_id (SubscriptionId): The subscription to filter enrollments by.
        profile_ref (Optional[str]): Unique reference identifier of the user profile.
            If None, enrollments are fetched without filtering by profile.

    Returns:
        list[Enrollment]: A list of active enrollments at the given date, or the
        next available future enrollments if none are active.
    """
    enrollments = get_profile_enrollments_active_on(
        profile_ref=profile_ref,
        at_date=at_date,
        subscription_id=subscription_id,
    )
    if len(enrollments) == 0:
        enrollments = get_profile_enrollments_active_after(
            profile_ref=profile_ref,
            at_date=at_date,
            subscription_id=subscription_id,
        )
    return enrollments

get_active_or_future_or_last_ended_benefit_enrollment ¶

get_active_or_future_or_last_ended_benefit_enrollment(
    profile_ref, benefit_type, at_date, subscription_id=None
)

Retrieve the most suitable enrollment based on a user's profile reference, benefit type, and a specific date, with an optional subscription ID filter. This method prioritizes active, then future, and finally the most recent ended enrollment.

Parameters:

Name	Type	Description	Default
`profile_ref`	`str`	The profile reference to filter enrollments.	required
`benefit_type`	`BenefitType`	The type of benefit to filter enrollments.	required
`at_date`	`date`	The date used to determine the relevant active or ended enrollment status.	required
`subscription_id`	`SubscriptionId \| None`	Optional; The subscription ID to further filter the enrollments.	`None`

Returns:

Type	Description
`Enrollment \| None`	Enrollment \| None: The most relevant Enrollment object if found; otherwise, None.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_active_or_future_or_last_ended_benefit_enrollment(
    profile_ref: str,
    benefit_type: BenefitType,
    at_date: date,
    subscription_id: SubscriptionId | None = None,
) -> Enrollment | None:
    """
    Retrieve the most suitable enrollment based on a user's profile reference, benefit type,
    and a specific date, with an optional subscription ID filter. This method prioritizes active,
    then future, and finally the most recent ended enrollment.

    Parameters:
        profile_ref (str): The profile reference to filter enrollments.
        benefit_type (BenefitType): The type of benefit to filter enrollments.
        at_date (date): The date used to determine the relevant active or ended enrollment status.
        subscription_id (SubscriptionId | None): Optional; The subscription ID to further filter the enrollments.

    Returns:
        Enrollment | None: The most relevant Enrollment object if found; otherwise, None.
    """
    enrollment = get_active_or_last_ended_user_enrollment(
        profile_ref=profile_ref,
        benefit_type=benefit_type,
        at_date=at_date,
        subscription_id=subscription_id,
    )

    if enrollment is None:
        enrollments = get_enrollments_starting_after(
            active_after=date_to_datetime(at_date),
            profile_refs=[profile_ref],
            benefit_types=[benefit_type],
            subscription_ids=[subscription_id] if subscription_id else None,
        )
        enrollment = enrollments[0] if enrollments else None

    return enrollment

get_active_or_last_ended_user_enrollment ¶

get_active_or_last_ended_user_enrollment(
    profile_ref, benefit_type, at_date, subscription_id=None
)

Get the active or last ended enrollment for a given user profile and benefit type at a given date. Optionally filter by subscription_id.

Returns None if no active or ended enrollment is found.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_active_or_last_ended_user_enrollment(
    profile_ref: str,
    benefit_type: BenefitType,
    at_date: date,
    subscription_id: SubscriptionId | None = None,
) -> Enrollment | None:
    """
    Get the active or last ended enrollment for a given user profile and benefit type at a given date.
    Optionally filter by subscription_id.

    Returns None if no active or ended enrollment is found.
    """
    active_or_ended_enrollment = (
        EnrollmentModelBroker.get_active_or_ended_benefit_enrollment(
            profile_ref=profile_ref,
            at_date=at_date,
            benefit_type=benefit_type,
            subscription_ref=str(subscription_id) if subscription_id else None,
        )
    )

    return (
        active_or_ended_enrollment.to_dataclass()
        if active_or_ended_enrollment
        else None
    )

get_ended_enrollments ¶

get_ended_enrollments(
    at_date=None,
    benefit_types=None,
    profile_refs=None,
    subscription_ids=None,
)

Retrieves a list of ended enrollments based on provided filters.

Parameters:

Name	Type	Description	Default
`at_date`	`date \| None`	Optional date to filter enrollments that ended before this date.	`None`
`benefit_types`	`list[BenefitType] \| None`	Optional list of benefit types to filter enrollments by.	`None`
`profile_refs`	`list[str] \| None`	Optional list of profile reference IDs to filter enrollments by.	`None`
`subscription_ids`	`list[SubscriptionId] \| None`	Optional list of subscription IDs to filter enrollments by.	`None`

Returns:

Type	Description
`list[Enrollment]`	List of Enrollment dataclass instances representing ended enrollments.

Raises:

Type	Description
`ValueError`	If no filter parameters (profile_refs, benefit_types, subscription_ids) are provided.

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_ended_enrollments(
    at_date: date | None = None,
    benefit_types: list[BenefitType] | None = None,
    profile_refs: list[str] | None = None,
    subscription_ids: list[SubscriptionId] | None = None,
) -> list[Enrollment]:
    """
    Retrieves a list of ended enrollments based on provided filters.

    Args:
        at_date: Optional date to filter enrollments that ended before this date.
        benefit_types: Optional list of benefit types to filter enrollments by.
        profile_refs: Optional list of profile reference IDs to filter enrollments by.
        subscription_ids: Optional list of subscription IDs to filter enrollments by.

    Returns:
        List of Enrollment dataclass instances representing ended enrollments.

    Raises:
        ValueError: If no filter parameters (profile_refs, benefit_types, subscription_ids) are provided.
    """
    if not profile_refs and not benefit_types and not subscription_ids:
        raise ValueError(
            "At least one of profile_refs, benefit_type_refs or subscription_ids must be provided"
        )

    active_enrollments = EnrollmentModelBroker.get_ended_enrollments(
        benefit_types=benefit_types,
        profile_refs=profile_refs,
        at_date=at_date,
        subscription_refs=[str(subscription_id) for subscription_id in subscription_ids]
        if subscription_ids
        else None,
    )

    return [enrollment.to_dataclass() for enrollment in active_enrollments]

get_enrollment ¶

get_enrollment(enrollment_id)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_enrollment(enrollment_id: EnrollmentId) -> Enrollment:
    enrollment = EnrollmentModelBroker.get_enrollment(id=enrollment_id)

    return enrollment.to_dataclass()

get_enrollment_by_transfer_history_id ¶

get_enrollment_by_transfer_history_id(
    transfer_history_id, active_on
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_enrollment_by_transfer_history_id(
    transfer_history_id: TransferHistoryId,
    active_on: datetime,
) -> Enrollment | None:
    enrollment = EnrollmentModelBroker.get_enrollment_by_transfer_history_ref(
        transfer_history_ref=str(transfer_history_id), active_on=active_on
    )

    return enrollment.to_dataclass() if enrollment else None

get_enrollments_active_in_period ¶

get_enrollments_active_in_period(
    period,
    benefit_types=None,
    profile_refs=None,
    subscription_ids=None,
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_enrollments_active_in_period(
    period: ValidityPeriod,
    benefit_types: list[BenefitType] | None = None,
    profile_refs: list[str] | None = None,
    subscription_ids: list[SubscriptionId] | None = None,
) -> list[Enrollment]:
    if not profile_refs and not benefit_types and not subscription_ids:
        raise ValueError(
            "At least one of profile_refs, benefit_type_refs or subscription_ids must be provided"
        )

    active_enrollments = EnrollmentModelBroker.get_enrollments_active_in_period(
        benefit_types=benefit_types,
        profile_refs=profile_refs,
        period=period,
        subscription_refs=[str(subscription_id) for subscription_id in subscription_ids]
        if subscription_ids
        else None,
    )

    return [enrollment.to_dataclass() for enrollment in active_enrollments]

get_enrollments_active_on ¶

get_enrollments_active_on(
    active_on,
    profile_refs=None,
    benefit_types=None,
    subscription_ids=None,
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_enrollments_active_on(
    active_on: datetime,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_ids: list[SubscriptionId] | None = None,
) -> list[Enrollment]:
    if not profile_refs and not benefit_types and not subscription_ids:
        raise ValueError(
            "At least one of profile_refs, benefit_type_refs or subscription_ids must be provided"
        )

    active_enrollments = EnrollmentModelBroker.get_enrollments_active_on(
        benefit_types=benefit_types,
        active_on=active_on,
        profile_refs=profile_refs,
        subscription_refs=[str(subscription_id) for subscription_id in subscription_ids]
        if subscription_ids
        else None,
    )

    return [enrollment.to_dataclass() for enrollment in active_enrollments]

get_enrollments_by_ids ¶

get_enrollments_by_ids(enrollment_ids)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_enrollments_by_ids(enrollment_ids: list[EnrollmentId]) -> list[Enrollment]:
    enrollments = EnrollmentModelBroker.get_enrollment_by_ids(ids=enrollment_ids)

    return [enrollment.to_dataclass() for enrollment in enrollments]

get_enrollments_on_subscription ¶

get_enrollments_on_subscription(
    subscription_ref,
    active_on=None,
    ended_on=None,
    active_on_or_after=None,
    profile_ref=None,
)

Retrieves a list of enrollments for a given subscription.

Parameters:

Name	Type	Description	Default
`subscription_ref`	`str`	The reference ID of the subscription to fetch enrollments for	required
`active_on`	`datetime \| None`	If provided, only returns enrollments that are active on this date	`None`
`profile_ref`	`str \| None`	If provided, filters enrollments to only those matching this profile reference	`None`
`active_on_or_after`	`date \| None`	If provided, only returns enrollments that are active on or after this date	`None`
`ended_on`	`date \| None`	If provided, only returns enrollments that ended on this date	`None`

Returns:

Type	Description
`list[Enrollment]`	list[Enrollment]: List of Enrollment dataclass objects representing the matching enrollments

The function queries the enrollment database and converts the results to dataclass objects. Enrollments can be filtered by: - Active status on a specific date - Profile reference - Subscription reference (required)

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def get_enrollments_on_subscription(
    subscription_ref: str,
    active_on: datetime | None = None,
    ended_on: date | None = None,
    active_on_or_after: date | None = None,
    profile_ref: str | None = None,
) -> list[Enrollment]:
    """
    Retrieves a list of enrollments for a given subscription.

    Args:
        subscription_ref (str): The reference ID of the subscription to fetch enrollments for
        active_on (datetime | None): If provided, only returns enrollments that are active on this date
        profile_ref (str | None): If provided, filters enrollments to only those matching this profile reference
        active_on_or_after (date | None): If provided, only returns enrollments that are active on or after this date
        ended_on (date | None): If provided, only returns enrollments that ended on this date

    Returns:
        list[Enrollment]: List of Enrollment dataclass objects representing the matching enrollments

    The function queries the enrollment database and converts the results to dataclass objects.
    Enrollments can be filtered by:
    - Active status on a specific date
    - Profile reference
    - Subscription reference (required)
    """
    enrollments = EnrollmentModelBroker.get_enrollments(
        subscription_refs=[subscription_ref],
        active_on=active_on,
        profile_refs=[profile_ref] if profile_ref else None,
        active_on_or_after=active_on_or_after,
        ended_on=ended_on,
    )
    return [enrollment.to_dataclass() for enrollment in enrollments]

get_enrollments_starting_after ¶

get_enrollments_starting_after(
    active_after,
    profile_refs=None,
    benefit_types=None,
    subscription_ids=None,
)

Get all enrollments that starts AFTER the provided date Warning: This method does not return enrollments that start before the provided date even if active on it

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def get_enrollments_starting_after(
    active_after: datetime,
    profile_refs: list[str] | None = None,
    benefit_types: list[BenefitType] | None = None,
    subscription_ids: list[SubscriptionId] | None = None,
) -> list[Enrollment]:
    """
    Get all enrollments that starts AFTER the provided date
    Warning: This method does not return enrollments that start before the provided date even if active on it
    """
    if not profile_refs and not benefit_types and not subscription_ids:
        raise ValueError(
            "At least one of profile_refs, benefit_type_refs or subscription_ids must be provided"
        )

    active_enrollments = EnrollmentModelBroker.get_enrollments_active_after(
        benefit_types=benefit_types,
        active_after=active_after,
        profile_refs=profile_refs,
        subscription_refs=[str(subscription_id) for subscription_id in subscription_ids]
        if subscription_ids
        else None,
    )

    return [enrollment.to_dataclass() for enrollment in active_enrollments]

get_profile_enrollments_active_on ¶

get_profile_enrollments_active_on(
    at_date, subscription_id, profile_ref=None
)

Return all active Enrollments at date, for profile and subscription

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def get_profile_enrollments_active_on(
    at_date: date,
    subscription_id: SubscriptionId,
    profile_ref: Optional[str] = None,
) -> list[Enrollment]:
    """
    Return all active Enrollments at date, for profile and subscription
    """
    from components.contracting.subcomponents.subscription.public.entities import (
        Subscription,
    )
    from components.healthy_benefits.subcomponents.subscription.protected.entities import (
        SubscriptionVersion,
    )
    from components.healthy_benefits.subcomponents.subscription.protected.subscription import (
        get_subscription_by_id,
    )

    subscription = cast(
        "Subscription[SubscriptionVersion]",
        get_subscription_by_id(subscription_id),
    )
    if not subscription:
        current_logger.info("No subscription found", subscription_id=subscription_id)
        return []

    ongoing_subscription_version = subscription.versions.get_ongoing_period(
        on_date=at_date
    )
    if ongoing_subscription_version is None:
        current_logger.info(
            "No active subscription found", subscription_id=subscription_id
        )
        return []

    benefits = mandatory(ongoing_subscription_version).payload.benefits

    enrollments = get_enrollments_active_on(
        active_on=datetime.combine(at_date, datetime.min.time()),
        profile_refs=[profile_ref] if profile_ref else None,
        benefit_types=[benefit.type_ref for benefit in benefits],
        subscription_ids=[subscription_id],
    )

    return enrollments

get_subscription_active_enrollments_on ¶

get_subscription_active_enrollments_on(
    at_date, subscription_id
)

Retrieves a list of active enrollments for a given subscription on a specific date.

Parameters:

Name	Type	Description	Default
`at_date`	`date`	The date for which to check active enrollments.	required
`subscription_id`	`SubscriptionId`	The reference ID of the subscription.	required

Returns:

Type	Description
`list[Enrollment]`	A list of Enrollment objects that are active on the specified date for the given subscription.
`list[Enrollment]`	Returns an empty list if there is no active subscription on that date.

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def get_subscription_active_enrollments_on(
    at_date: date,
    subscription_id: SubscriptionId,
) -> list[Enrollment]:
    """
    Retrieves a list of active enrollments for a given subscription on a
    specific date.

    Args:
        at_date: The date for which to check active enrollments.
        subscription_id: The reference ID of the subscription.

    Returns:
        A list of Enrollment objects that are active on the specified date for the given subscription.
        Returns an empty list if there is no active subscription on that date.
    """
    from shared.helpers.time.date import date_to_datetime

    return get_enrollments_on_subscription(
        subscription_ref=str(subscription_id),
        active_on=date_to_datetime(at_date),
    )

is_profile_enrolled ¶

is_profile_enrolled(at_date, profile_ref, subscription_id)

Determine whether a user profile is enrolled in a given subscription at a specific date.

Parameters:

Name	Type	Description	Default
`at_date`	`datetime`	The date at which enrollment status should be checked.	required
`profile_ref`	`str`	Unique reference identifier of the user profile.	required
`subscription_id`	`SubscriptionId`	The subscription to check enrollment for.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the profile is enrolled at the given date, False otherwise.

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def is_profile_enrolled(
    at_date: datetime,
    profile_ref: str,
    subscription_id: SubscriptionId,
) -> bool:
    """
    Determine whether a user profile is enrolled in a given subscription
    at a specific date.

    Args:
        at_date (datetime): The date at which enrollment status should be checked.
        profile_ref (str): Unique reference identifier of the user profile.
        subscription_id (SubscriptionId): The subscription to check enrollment for.

    Returns:
        bool: True if the profile is enrolled at the given date, False otherwise.
    """
    from components.healthy_benefits.subcomponents.affiliation.internal.business_logic.enrollment import (
        is_profile_enrolled as is_profile_enrolled_query,
    )

    return is_profile_enrolled_query(
        at_date=at_date,
        profile_ref=profile_ref,
        subscription_id=subscription_id,
    )

terminate_enrollment ¶

terminate_enrollment(
    enrollment_id, end_date=None, commit=True
)

End an enrollment at the given end date (default to today if not set)

Source code in components/healthy_benefits/subcomponents/affiliation/protected/enrollment.py

def terminate_enrollment(
    enrollment_id: EnrollmentId,
    end_date: date | None = None,
    commit: bool = True,
) -> Enrollment:
    """End an enrollment at the given end date (default to today if not set)"""
    from components.healthy_benefits.subcomponents.affiliation.internal.business_logic.enrollment import (
        terminate_enrollment as terminate_enrollment_action,
    )

    return terminate_enrollment_action(
        enrollment_id=enrollment_id,
        end_date=end_date,
        commit=commit,
    )

update_enrollment ¶

update_enrollment(
    enrollment_id, start_date, end_date=None, commit=True
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def update_enrollment(
    enrollment_id: EnrollmentId,
    start_date: date,
    end_date: date | None = None,
    commit: bool = True,
) -> Enrollment:
    enrollment = EnrollmentModelBroker.update_enrollment(
        id=enrollment_id, start_date=start_date, end_date=end_date
    )

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

    return enrollment.to_dataclass()

update_enrollment_transfer_history_id ¶

update_enrollment_transfer_history_id(
    enrollment_id, transfer_history_id, *, commit=True
)

Source code in components/healthy_benefits/subcomponents/affiliation/internal/business_logic/enrollment.py

def update_enrollment_transfer_history_id(
    enrollment_id: EnrollmentId,
    transfer_history_id: TransferHistoryId,
    *,
    commit: bool = True,
) -> Enrollment:
    from components.healthy_benefits.subcomponents.affiliation.internal.models.brokers.enrollment import (
        EnrollmentModelBroker,
    )

    enrollment = EnrollmentModelBroker.update_enrollment(
        id=enrollment_id, transfer_history_ref=str(transfer_history_id)
    )

    if commit:
        current_session.commit()

    return enrollment.to_dataclass()

components.healthy_benefits.public.commands ¶

app_group ¶

healthy_benefits_commands `module-attribute` ¶

healthy_benefits_commands = AppGroup(
    "global_healthy_benefits"
)

register_healthy_benefits_commands ¶

register_healthy_benefits_commands()

Source code in components/healthy_benefits/public/commands/app_group.py

def register_healthy_benefits_commands() -> None:  # noqa: D103
    from components.healthy_benefits.subcomponents.subscription.protected.blueprint import (
        register_subscription_commands,
    )

    register_subscription_commands()

components.healthy_benefits.public.entities ¶

BenefitType ¶

Bases: AlanBaseEnum

Healthy benefit types.

food `class-attribute` `instance-attribute` ¶

food = 'food'

is_eligible_for_card_payment ¶

is_eligible_for_card_payment()

Determines if a given BenefitType is eligible for card payment. As of today, only food and transport are.

Source code in components/healthy_benefits/public/entities.py

def is_eligible_for_card_payment(self) -> bool:
    """Determines if a given BenefitType is eligible for card payment. As of today, only food and transport are."""
    card_eligible_types = {self.food, self.transport}
    return self in card_eligible_types

learning `class-attribute` `instance-attribute` ¶

learning = 'learning'

nursery `class-attribute` `instance-attribute` ¶

nursery = 'nursery'

transport `class-attribute` `instance-attribute` ¶

transport = 'transport'

components.healthy_benefits.public.pricing ¶

DiscountId `module-attribute` ¶

DiscountId = NewType('DiscountId', UUID)

SubscriptionDiscount `dataclass` ¶

SubscriptionDiscount(
    id,
    subscription_id,
    start_month,
    end_month,
    price,
    reason,
)

Discount for a subscription and a given period.

end_month `instance-attribute` ¶

end_month

id `instance-attribute` ¶

id

price `instance-attribute` ¶

price

reason `instance-attribute` ¶

reason

start_month `instance-attribute` ¶

start_month

subscription_id `instance-attribute` ¶

subscription_id

create_subscription_discount ¶

create_subscription_discount(
    subscription_id, start_month, end_month, price, reason
)

Create a discount for a subscription.

Parameters:

Name	Type	Description	Default
`subscription_id`	`SubscriptionId`	The subscription id.	required
`start_month`	`Month`	The start month of the discount.	required
`end_month`	`Month`	The end month of the discount.	required
`price`	`int`	The price of the discount.	required
`reason`	`str`	The reason for the discount.	required

Source code in components/healthy_benefits/subcomponents/pricing/protected/discount.py

def create_subscription_discount(
    subscription_id: SubscriptionId,
    start_month: Month,
    end_month: Month,
    price: int,
    reason: str,
) -> None:
    """Create a discount for a subscription.

    Args:
        subscription_id: The subscription id.
        start_month: The start month of the discount.
        end_month: The end month of the discount.
        price: The price of the discount.
        reason: The reason for the discount.
    """
    DiscountModelBroker.create_discount(
        subscription_id=subscription_id,
        start_month=start_month,
        end_month=end_month,
        price=price,
        reason=reason,
    )

get_baseline_subscription_price ¶

get_baseline_subscription_price(subscription_id, on_date)

Get the baseline subscription price on a given date, excluding discounts.

Parameters:

Name	Type	Description	Default
`subscription_id`	`SubscriptionId`	The subscription id.	required
`on_date`	`date`	The date for which to get the subscription price.	required

Source code in components/healthy_benefits/subcomponents/pricing/protected/discount.py

def get_baseline_subscription_price(
    subscription_id: SubscriptionId, on_date: date
) -> int:
    """Get the baseline subscription price on a given date, excluding discounts.

    Args:
        subscription_id: The subscription id.
        on_date: The date for which to get the subscription price.
    """
    subscription_version = get_ongoing_or_upcoming_subscription_by_id(
        subscription_id=subscription_id, on_date=on_date
    )
    if subscription_version is None:
        raise ValueError(
            f"Subscription {subscription_id} has no active version at {on_date}"
        )
    return subscription_version.payload.price

get_current_subscription_discount ¶

get_current_subscription_discount(subscription_id, on_date)

Get the discount for a subscription on a given date.

Parameters:

Name	Type	Description	Default
`subscription_id`	`SubscriptionId`	The subscription id.	required
`on_date`	`date`	The date for which to get the discount.	required

Source code in components/healthy_benefits/subcomponents/pricing/protected/discount.py

def get_current_subscription_discount(
    subscription_id: SubscriptionId,
    on_date: date,
) -> SubscriptionDiscount | None:
    """Get the discount for a subscription on a given date.

    Args:
        subscription_id: The subscription id.
        on_date: The date for which to get the discount.
    """
    discount = DiscountModelBroker.get_discount_for_subscription_on_date(
        subscription_id=subscription_id,
        on_date=on_date,
    )
    return _to_subscription_discount(discount) if discount else None

get_current_subscription_price ¶

get_current_subscription_price(subscription_id, on_date)

Get the current subscription price on a given date.

This will take into account any ongoing discounts.

Parameters:

Name	Type	Description	Default
`subscription_id`	`SubscriptionId`	The subscription id.	required
`on_date`	`date`	The date for which to get the subscription price.	required

Source code in components/healthy_benefits/subcomponents/pricing/protected/discount.py

def get_current_subscription_price(
    subscription_id: SubscriptionId, on_date: date
) -> int:
    """Get the current subscription price on a given date.

    This will take into account any ongoing discounts.

    Args:
        subscription_id: The subscription id.
        on_date: The date for which to get the subscription price.
    """
    # Active discounts take precedence over baseline subscription price
    discount = get_current_subscription_discount(subscription_id, on_date)
    if discount:
        return discount.price

    return get_baseline_subscription_price(subscription_id, on_date)

get_subscription_discounts ¶

get_subscription_discounts(subscription_id)

Get all discounts for a subscription.

Parameters:

Name	Type	Description	Default
`subscription_id`	`SubscriptionId`	The subscription id.	required

Returns:

Type	Description
`list[SubscriptionDiscount]`	List of all discounts for the subscription, ordered by start_month descending.

Source code in components/healthy_benefits/subcomponents/pricing/protected/discount.py

def get_subscription_discounts(
    subscription_id: SubscriptionId,
) -> list[SubscriptionDiscount]:
    """Get all discounts for a subscription.

    Args:
        subscription_id: The subscription id.

    Returns:
        List of all discounts for the subscription, ordered by start_month descending.
    """
    discounts = DiscountModelBroker.get_discounts_for_subscription(
        subscription_id=subscription_id
    )
    return [
        _to_subscription_discount(discount)
        for discount in sorted(discounts, key=lambda d: d.start_month, reverse=True)
    ]

update_discount_end_date ¶

update_discount_end_date(
    subscription_id, discount_id, new_end_month
)

Update the end date of a discount.

Parameters:

Name	Type	Description	Default
`subscription_id`	`SubscriptionId`	The subscription id (for validation).	required
`discount_id`	`DiscountId`	The discount id to update.	required
`new_end_month`	`Month`	The new end month for the discount.	required

Raises:

Type	Description
`ValueError`	If the discount doesn't belong to the subscription or if the new end date is invalid.

Source code in components/healthy_benefits/subcomponents/pricing/protected/discount.py

def update_discount_end_date(
    subscription_id: SubscriptionId,
    discount_id: DiscountId,
    new_end_month: Month,
) -> None:
    """Update the end date of a discount.

    Args:
        subscription_id: The subscription id (for validation).
        discount_id: The discount id to update.
        new_end_month: The new end month for the discount.

    Raises:
        ValueError: If the discount doesn't belong to the subscription or if the
                   new end date is invalid.
    """
    discount = DiscountModelBroker.get_discount(discount_id)

    if discount.subscription_id != subscription_id:
        raise ValueError(
            f"Discount {discount_id} does not belong to subscription {subscription_id}"
        )

    if new_end_month < discount.start_month:
        raise ValueError(
            f"New end month {new_end_month} cannot be before start month {discount.start_month}"
        )

    # Check if new end date would create overlaps with other discounts
    existing_discounts = DiscountModelBroker.get_discounts_for_subscription(
        subscription_id=subscription_id
    )

    for existing_discount in existing_discounts:
        if existing_discount.id == discount.id:
            continue

        if (
            discount.start_month <= existing_discount.end_month
            and new_end_month >= existing_discount.start_month
        ):
            raise ValueError(
                f"Updated discount would overlap with existing discount from "
                f"{existing_discount.start_month} to {existing_discount.end_month}"
            )

    DiscountModelBroker.update_discount_end_date(
        discount_id=discount_id,
        new_end_month=new_end_month,
    )

components.healthy_benefits.public.subscription ¶

Benefit `dataclass` ¶

Benefit(id, population_id, type_ref, allowance)

Benefit definition within a Healthy Benefits configuration.

allowance `instance-attribute` ¶

allowance

id `instance-attribute` ¶

id

population_id `instance-attribute` ¶

population_id

type_ref `instance-attribute` ¶

type_ref

BenefitArgs `dataclass` ¶

BenefitArgs(type_ref, population_id, allowance)

Arguments to create a benefit entry.

allowance `instance-attribute` ¶

allowance

population_id `instance-attribute` ¶

population_id

type_ref `instance-attribute` ¶

type_ref

BenefitId `module-attribute` ¶

BenefitId = NewType('BenefitId', UUID)

HealthyBenefitsSignedDocument `dataclass` ¶

HealthyBenefitsSignedDocument(
    id,
    signed_bundle_id,
    type,
    uploader_id,
    subscription_id,
    signed_at,
)

Metadata for a signed document in Healthy Benefits.

id `instance-attribute` ¶

id

signed_at `instance-attribute` ¶

signed_at

signed_bundle_id `instance-attribute` ¶

signed_bundle_id

subscription_id `instance-attribute` ¶

subscription_id

type `instance-attribute` ¶

type

uploader_id `instance-attribute` ¶

uploader_id

HealthyBenefitsSignedDocumentType ¶

Bases: AlanBaseEnum

Types of signed documents linked to a Healthy Benefits subscription.

general_conditions `class-attribute` `instance-attribute` ¶

general_conditions = 'general_conditions'

particular_conditions `class-attribute` `instance-attribute` ¶

particular_conditions = 'particular_conditions'

sepa_mandate `class-attribute` `instance-attribute` ¶

sepa_mandate = 'sepa_mandate'

HealthyBenefitsSubscription `module-attribute` ¶

HealthyBenefitsSubscription = Subscription[
    SubscriptionVersion
]

Population `dataclass` ¶

Population(id, name, entity_ref)

Target population for a benefit.

entity_ref `instance-attribute` ¶

entity_ref

id `instance-attribute` ¶

id

name `instance-attribute` ¶

name

PopulationId `module-attribute` ¶

PopulationId = NewType('PopulationId', UUID)

SubscriptionConfigurationArgs `dataclass` ¶

SubscriptionConfigurationArgs(
    benefits, price, validityPeriod, transfer_history_id
)

Arguments to configure a healthy benefits subscription.

benefits `instance-attribute` ¶

benefits

price `instance-attribute` ¶

price

transfer_history_id `instance-attribute` ¶

transfer_history_id

validityPeriod `instance-attribute` ¶

validityPeriod

SubscriptionId `module-attribute` ¶

SubscriptionId = NewType('SubscriptionId', UUID)

SubscriptionVersion `dataclass` ¶

SubscriptionVersion(
    subscription_id, validity_period, payload
)

Bases: WithValidityPeriod

A version of a subscription with a validity period and payload.

do_overlap ¶

do_overlap(other)

Return True if both versions belong have the same id and overlap.

Source code in components/healthy_benefits/subcomponents/subscription/protected/entities.py

def do_overlap(self, other: WithValidityPeriod) -> bool:
    """Return True if both versions belong have the same id and overlap."""
    if not isinstance(other, SubscriptionVersion):
        raise ValueError(
            f"Impossible comparison between a {self.__class__} and a {other.__class__}"
        )
    return (
        self.subscription_id == other.subscription_id
        and self.validity_period.do_overlap(other.validity_period)
    )

payload `instance-attribute` ¶

payload

subscription_id `instance-attribute` ¶

subscription_id

validity_period `instance-attribute` ¶

validity_period

create_benefit ¶

create_benefit(
    payload_id,
    population_id,
    type_ref,
    allowance,
    commit=True,
)

Create a benefit for a given subscription payload and population.

Parameters:

Name	Type	Description	Default
`payload_id`	`SubscriptionPayloadId`	Target subscription payload identifier.	required
`population_id`	`PopulationId`	Population that will receive the benefit.	required
`type_ref`	`str`	Benefit type reference (string key).	required
`allowance`	`int`	Allowance amount (in cents).	required
`commit`	`bool`	Whether to commit the change immediately.	`True`

Returns:

Type	Description
`Benefit`	The created Benefit entity.

Source code in components/healthy_benefits/subcomponents/subscription/protected/benefit.py

def create_benefit(
    payload_id: SubscriptionPayloadId,
    population_id: PopulationId,
    type_ref: str,
    allowance: int,
    commit: bool = True,
) -> Benefit:
    """Create a benefit for a given subscription payload and population.

    Args:
        payload_id: Target subscription payload identifier.
        population_id: Population that will receive the benefit.
        type_ref: Benefit type reference (string key).
        allowance: Allowance amount (in cents).
        commit: Whether to commit the change immediately.

    Returns:
        The created Benefit entity.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.benefit import (
        create_benefit as create_benefit_action,
    )

    return create_benefit_action(
        payload_id=payload_id,
        population_id=population_id,
        type_ref=type_ref,
        allowance=allowance,
        commit=commit,
    )

create_population ¶

create_population(entity_ref, name, commit=True)

Create a new population for a given entity_ref.

Parameters:

Name	Type	Description	Default
`entity_ref`	`str`	Reference identifier of the entity the population belongs to.	required
`name`	`str`	The display name of the population.	required
`commit`	`bool`	Whether to persist changes immediately. Defaults to True.	`True`

Returns:

Name	Type	Description
`Population`	`Population`	The newly created population instance.

Source code in components/healthy_benefits/subcomponents/subscription/protected/population.py

def create_population(
    entity_ref: str,
    name: str,
    commit: bool = True,
) -> Population:
    """
    Create a new population for a given entity_ref.

    Args:
        entity_ref (str): Reference identifier of the entity the population belongs to.
        name (str): The display name of the population.
        commit (bool, optional): Whether to persist changes immediately. Defaults to True.

    Returns:
        Population: The newly created population instance.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.population import (
        create_population as create_population_action,
    )

    return create_population_action(entity_ref=entity_ref, name=name, commit=commit)

create_subscription_payload ¶

create_subscription_payload(price, commit=True)

Source code in components/healthy_benefits/subcomponents/subscription/protected/payload.py

def create_subscription_payload(price: int, commit: bool = True) -> SubscriptionPayload:  # noqa: D103
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.payload import (
        create_subscription_payload as create_subscription_payload_action,
    )

    return create_subscription_payload_action(
        price=price,
        transfer_history_id=None,
        commit=commit,
    )

end_company_subscription ¶

end_company_subscription(
    subscription_id, end_date, commit=True
)

Ends a healthy benefits subscription

Source code in components/healthy_benefits/subcomponents/subscription/internal/business_logic/subscription.py

def end_company_subscription(
    subscription_id: SubscriptionId, end_date: date, commit: bool = True
) -> None:
    """Ends a healthy benefits subscription"""
    from components.contracting.subcomponents.subscription.public.actions import (
        record_subscription_updates,
    )

    subscription = get_subscription_by_id(subscription_id=subscription_id)
    subscription = mandatory_type(Subscription, subscription)

    # To set an end_date to your version, you need to create a subscription update starting the following date with
    # is_deletion set to True
    record_subscription_updates(
        subscription_scope=SubscriptionScope.healthy_benefits,
        subscription_ref=subscription.id,
        updates=[
            SubscriptionUpdateRequest(
                validity_period=ValidityPeriod(
                    start_date=end_date + timedelta(days=1), end_date=None
                ),
                payload_ref=None,
                operation_ref=None,
                is_deletion=True,
            )
        ],
        commit=commit,
    )

get_company_subscriptions_in_period ¶

get_company_subscriptions_in_period(company_ref, period)

Source code in components/healthy_benefits/subcomponents/subscription/internal/business_logic/subscription.py

def get_company_subscriptions_in_period(
    company_ref: str,
    period: ValidityPeriod,
) -> list[HealthyBenefitsSubscription]:
    subscriptions = get_subscriptions_for(
        company_ref=company_ref,
        period=period,
    )

    subscriptions = sorted(
        subscriptions, key=lambda s: s.validity_period.start_date, reverse=True
    )

    if not subscriptions:
        raise BaseErrorCode.missing_resource(
            message="No healthy benefits subscription found for company in period",
            company_ref=company_ref,
            period=period,
        )

    return subscriptions

get_ongoing_or_future_or_last_subscription_or_none ¶

get_ongoing_or_future_or_last_subscription_or_none(
    company_ref, at_date
)

Return ongoing/future subscription at date, else the most recent past, or None.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_ongoing_or_future_or_last_subscription_or_none(
    company_ref: str,
    at_date: date,
) -> HealthyBenefitsSubscription | None:
    """Return ongoing/future subscription at date, else the most recent past, or None."""
    ongoing_or_future = get_ongoing_or_future_subscription_or_none(company_ref, at_date)
    if ongoing_or_future:
        return ongoing_or_future

    subscriptions = get_subscriptions_for(
        company_ref=company_ref,
    )
    return first_or_none(
        sorted(subscriptions, key=lambda s: s.validity_period.start_date, reverse=True)
    )

get_ongoing_or_future_subscription ¶

get_ongoing_or_future_subscription(company_ref, at_date)

Return the ongoing or future subscription for a company at a date.

Raises:

Type	Description
`BaseErrorCode`	If none exists.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_ongoing_or_future_subscription(
    company_ref: str,
    at_date: date,
) -> HealthyBenefitsSubscription:
    """Return the ongoing or future subscription for a company at a date.

    Raises:
        BaseErrorCode: If none exists.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        get_ongoing_or_future_subscription as get_ongoing_or_future_subscription_query,
    )

    return get_ongoing_or_future_subscription_query(
        company_ref=company_ref, at_date=at_date
    )

get_ongoing_or_future_subscription_or_none ¶

get_ongoing_or_future_subscription_or_none(
    company_ref, at_date
)

Return ongoing/future subscription for a company, or None if absent.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_ongoing_or_future_subscription_or_none(
    company_ref: str,
    at_date: date,
) -> HealthyBenefitsSubscription | None:
    """Return ongoing/future subscription for a company, or None if absent."""
    try:
        return get_ongoing_or_future_subscription(
            company_ref=company_ref, at_date=at_date
        )
    except BaseErrorCode:
        return None

get_or_download_document ¶

get_or_download_document(
    document_id, download_for_owner_ref
)

Downloads a document if the requester is the owner of the subscription.

Parameters:

Name	Type	Description	Default
`document_id`	`UUID`	The ID of the document to download.	required
`download_for_owner_ref`	`str`	The owner_ref of the user requesting the download.	required

Raises:

Type	Description
`missing_resource`	If the document does not exist.
`integrity_error`	If the subscription attached to the document is not found.
`unauthorized`	If the requester is not the owner of the subscription.

Returns:

Type	Description
`tuple[str, IO[bytes]]`	tuple[str, IO[bytes]]: A tuple with the file name and the file content.

Source code in components/healthy_benefits/subcomponents/subscription/protected/signed_documents.py

def get_or_download_document(
    document_id: UUID, download_for_owner_ref: str
) -> tuple[str, IO[bytes]]:
    """Downloads a document if the requester is the owner of the subscription.

    Args:
        document_id (UUID): The ID of the document to download.
        download_for_owner_ref (str): The owner_ref of the user requesting the download.

    Raises:
        BaseErrorCode.missing_resource: If the document does not exist.
        BaseErrorCode.integrity_error: If the subscription attached to the document is not found.
        BaseErrorCode.unauthorized: If the requester is not the owner of the subscription.

    Returns:
        tuple[str, IO[bytes]]: A tuple with the file name and the file content.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        get_subscription_by_id,
    )

    document = SignedDocumentModelBroker.get_document(document_id)
    if document is None:
        raise BaseErrorCode.missing_resource(
            message="Document does not exist", document_id=document_id
        )

    subscription = get_subscription_by_id(
        SubscriptionId(document.signed_bundle.subscription_id)
    )
    if subscription is None:
        raise BaseErrorCode.integrity_error(
            message="Subscription attached to document is not found",
            document_id=document_id,
            subscription_id=document.signed_bundle.subscription_id,
        )

    if subscription.owner_ref != download_for_owner_ref:
        raise BaseErrorCode.unauthorized(
            message="Owner is not allowed to download this document",
            download_for_owner_ref=download_for_owner_ref,
            owner_ref=subscription.owner_ref,
        )

    file_name = f"{document.type}.pdf"

    return file_name, document.get_or_download_file()

get_population ¶

get_population(population_id)

Retrieve a single population by its unique identifier.

Parameters:

Name	Type	Description	Default
`population_id`	`PopulationId`	The unique identifier of the population to fetch.	required

Returns:

Name	Type	Description
`Population`	`Population`	The population instance corresponding to the provided ID.

Source code in components/healthy_benefits/subcomponents/subscription/protected/population.py

def get_population(
    population_id: PopulationId,
) -> Population:
    """
    Retrieve a single population by its unique identifier.

    Args:
        population_id (PopulationId): The unique identifier of the population to fetch.

    Returns:
        Population: The population instance corresponding to the provided ID.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.population import (
        get_population as get_population_action,
    )

    return get_population_action(population_id)

get_populations ¶

get_populations(entity_ref)

Retrieve all populations associated with a specific entity.

Parameters:

Name	Type	Description	Default
`entity_ref`	`str`	Reference identifier of the entity whose populations should be fetched.	required

Returns:

Type	Description
`list[Population]`	list[Population]: A list of populations associated with the entity.

Source code in components/healthy_benefits/subcomponents/subscription/protected/population.py

def get_populations(entity_ref: str) -> list[Population]:
    """
    Retrieve all populations associated with a specific entity.

    Args:
        entity_ref (str): Reference identifier of the entity whose populations should be fetched.

    Returns:
        list[Population]: A list of populations associated with the entity.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.population import (
        get_populations as get_populations_action,
    )

    return get_populations_action(entity_ref)

get_signed_documents_for_subscription ¶

get_signed_documents_for_subscription(subscription_id)

Retrieves all signed documents for a given subscription.

Parameters:

Name	Type	Description	Default
`subscription_id`	`UUID`	The ID of the subscription.	required

Returns:

Type	Description
`list[HealthyBenefitsSignedDocument]`	list[HealthyBenefitsSignedDocument]: A list of signed documents.

Source code in components/healthy_benefits/subcomponents/subscription/protected/signed_documents.py

def get_signed_documents_for_subscription(
    subscription_id: UUID,
) -> list[HealthyBenefitsSignedDocument]:
    """Retrieves all signed documents for a given subscription.

    Args:
        subscription_id (UUID): The ID of the subscription.

    Returns:
        list[HealthyBenefitsSignedDocument]: A list of signed documents.
    """
    documents = SignedDocumentModelBroker.get_documents_for_subscription(
        subscription_id
    )

    return [
        HealthyBenefitsSignedDocument(
            id=document.id,
            subscription_id=document.signed_bundle.subscription_id,
            signed_bundle_id=document.signed_bundle_id,
            uploader_id=document.uploader_id,
            type=document.type,
            signed_at=document.signed_at,
        )
        for document in documents
    ]

get_subscription ¶

get_subscription(company_ref, at_date)

Return the subscription active for a company at a date.

Raises:

Type	Description
`BaseErrorCode`	If none exists.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_subscription(
    company_ref: str,
    at_date: date,
) -> HealthyBenefitsSubscription:
    """Return the subscription active for a company at a date.

    Raises:
        BaseErrorCode: If none exists.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        get_subscription as get_subscription_query,
    )

    return get_subscription_query(company_ref=company_ref, at_date=at_date)

get_subscription_benefit ¶

get_subscription_benefit(
    subscription_id, benefit_type, population_id, at_date
)

Return a specific benefit (by type and population) at a date, or None.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_subscription_benefit(
    subscription_id: SubscriptionId,
    benefit_type: str,
    population_id: PopulationId,
    at_date: date,
) -> Benefit | None:
    """Return a specific benefit (by type and population) at a date, or None."""
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        get_subscription_benefit as get_subscription_benefit_query,
    )

    return get_subscription_benefit_query(
        subscription_id=subscription_id,
        benefit_type=benefit_type,
        population_id=population_id,
        at_date=at_date,
    )

get_subscription_benefits ¶

get_subscription_benefits(
    subscription_id, population_id, at_date
)

Return all benefits for a population at a date.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_subscription_benefits(
    subscription_id: SubscriptionId,
    population_id: PopulationId,
    at_date: date,
) -> list[Benefit]:
    """Return all benefits for a population at a date."""
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        get_subscription_benefits as get_subscription_benefits_query,
    )

    return get_subscription_benefits_query(
        subscription_id=subscription_id,
        population_id=population_id,
        at_date=at_date,
    )

get_subscription_by_id ¶

get_subscription_by_id(subscription_id)

Fetch a subscription (any type) by its identifier.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def get_subscription_by_id(
    subscription_id: SubscriptionId,
) -> BaseSubscription | Subscription[SubscriptionVersion]:
    """Fetch a subscription (any type) by its identifier."""
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        get_subscription_by_id as get_subscription_by_id_query,
    )

    return get_subscription_by_id_query(subscription_id=subscription_id)

get_subscriptions ¶

get_subscriptions(period=None)

Retrieves all healthy benefits subscriptions filtered by an optional time period.

Parameters:

Name	Type	Description	Default
`period`	`ValidityPeriod \| None`	Optional time period to filter subscriptions. When None, returns all subscriptions regardless of their validity period.	`None`

Returns:

Type	Description
`dict[str, HealthyBenefitsSubscription]`	dict[str, HealthyBenefitsSubscription]: Dictionary mapping owner reference to their corresponding healthy benefits subscription. Only returns owners that have at least one subscription.

Note

Uses the external subscription service to fetch raw subscription data
Filters for healthy benefits scope specifically
For owners with multiple subscriptions, only returns the first subscription
Owners without any subscriptions are excluded from the result

Source code in components/healthy_benefits/subcomponents/subscription/internal/business_logic/subscription.py

def get_subscriptions(
    period: ValidityPeriod | None = None,
) -> dict[str, HealthyBenefitsSubscription]:
    """
    Retrieves all healthy benefits subscriptions filtered by an optional time period.

    Args:
        period (ValidityPeriod | None): Optional time period to filter subscriptions.
            When None, returns all subscriptions regardless of their validity period.

    Returns:
        dict[str, HealthyBenefitsSubscription]: Dictionary mapping owner reference to their
            corresponding healthy benefits subscription. Only returns owners that have at least
            one subscription.

    Note:
        - Uses the external subscription service to fetch raw subscription data
        - Filters for healthy benefits scope specifically
        - For owners with multiple subscriptions, only returns the first subscription
        - Owners without any subscriptions are excluded from the result
    """
    from components.contracting.public.subscription.subscription import (
        get_subscriptions as external_get_subscriptions,
    )

    subscriptions_by_owner_ref = external_get_subscriptions(
        subscription_scope=SubscriptionScope.healthy_benefits,
        subscription_version_serializer=_SubscriptionVersionSerializer(),
        period=period,
    )

    return {
        owner_ref: get_subscription_result.subscriptions[0]
        for owner_ref, get_subscription_result in subscriptions_by_owner_ref.items()
        if len(get_subscription_result.subscriptions) > 0
    }

get_subscriptions_for ¶

get_subscriptions_for(company_ref, period=None)

Get all subscriptions to healthy benefits for a given company. :param company_ref: The company reference :param period: if set, only subscriptions overlapping this period are returned :return: list[HealthyBenefitsSubscription]

Source code in components/healthy_benefits/subcomponents/subscription/internal/business_logic/subscription.py

def get_subscriptions_for(
    company_ref: str, period: ValidityPeriod | None = None
) -> list[HealthyBenefitsSubscription]:
    """
    Get all subscriptions to healthy benefits for a given company.
    :param company_ref: The company reference
    :param period: if set, only subscriptions overlapping this period are returned
    :return: list[HealthyBenefitsSubscription]
    """
    from components.contracting.public.subscription.subscription import (
        get_subscriptions_for as external_get_subscription_for,
    )

    return external_get_subscription_for(
        owner_ref=company_ref,
        subscription_scope=SubscriptionScope.healthy_benefits,
        subscription_version_serializer=_SubscriptionVersionSerializer(),
        period=period,
    ).subscriptions

subscribe(
    subscription_configuration_args,
    company_ref,
    commit=True,
)

Create a subscription for a company.

Source code in components/healthy_benefits/subcomponents/subscription/protected/subscription.py

def subscribe(
    subscription_configuration_args: SubscriptionConfigurationArgs,
    company_ref: str,
    commit: bool = True,
) -> Subscription[SubscriptionVersion]:
    """Create a subscription for a company."""
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.subscription import (
        subscribe as subscribe_action,
    )

    return subscribe_action(
        subscription_configuration_args=subscription_configuration_args,
        company_ref=company_ref,
        commit=commit,
    )

update_subscription ¶

update_subscription(
    subscription,
    subscription_configuration_args,
    commit=True,
)

Source code in components/healthy_benefits/subcomponents/subscription/internal/business_logic/subscription.py

def update_subscription(
    subscription: Subscription[SubscriptionVersion] | BaseSubscription,
    subscription_configuration_args: SubscriptionConfigurationArgs,
    commit: bool = True,
) -> None:
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.benefit import (
        create_benefit,
    )
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.payload import (
        create_subscription_payload,
    )

    payload = create_subscription_payload(
        price=subscription_configuration_args.price,
        transfer_history_id=subscription_configuration_args.transfer_history_id,
        commit=False,
    )

    for benefit in subscription_configuration_args.benefits:
        create_benefit(
            population_id=benefit.population_id,
            type_ref=benefit.type_ref,
            allowance=benefit.allowance,
            payload_id=payload.id,
            commit=False,
        )

    update_subscription_configuration(
        subscription=subscription,
        payload_id=payload.id,
        validity_period=subscription_configuration_args.validityPeriod,
        commit=False,
    )

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

upload_signed_contract ¶

upload_signed_contract(
    subscription,
    document,
    uploader_id,
    document_type,
    signer,
    commit=True,
)

Uploads a signed contract and adds general conditions for a subscription.

Parameters:

Name	Type	Description	Default
`subscription`	`Subscription[SubscriptionVersion]`	The subscription to upload the contract for.	required
`document`	`BytesIO`	The signed contract file.	required
`uploader_id`	`UUID`	The ID of the user uploading the document.	required
`document_type`	`HealthyBenefitsSignedDocumentType`	The type of the document.	required
`signer`	`Signer`	The signer of the document.	required
`commit`	`bool`	Whether to commit the session. Defaults to True.	`True`

Returns:

Name	Type	Description
`HealthyBenefitsSignedDocument`	`HealthyBenefitsSignedDocument`	The uploaded signed document.

Source code in components/healthy_benefits/subcomponents/subscription/protected/signed_documents.py

def upload_signed_contract(
    subscription: Subscription[SubscriptionVersion],
    document: BytesIO,
    uploader_id: UUID,
    document_type: HealthyBenefitsSignedDocumentType,
    signer: Signer,
    commit: bool = True,
) -> HealthyBenefitsSignedDocument:
    """Uploads a signed contract and adds general conditions for a subscription.

    Args:
        subscription (Subscription[SubscriptionVersion]): The subscription to upload the contract for.
        document (BytesIO): The signed contract file.
        uploader_id (UUID): The ID of the user uploading the document.
        document_type (HealthyBenefitsSignedDocumentType): The type of the document.
        signer (Signer): The signer of the document.
        commit (bool): Whether to commit the session. Defaults to True.

    Returns:
        HealthyBenefitsSignedDocument: The uploaded signed document.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.signed_documents import (
        add_general_conditions_for_subscription,
        upload_signed_document,
    )

    add_general_conditions_for_subscription(
        subscription_id=subscription.id, uploader_id=uploader_id, signer=signer
    )

    signed_document = upload_signed_document(
        subscription_id=subscription.id,
        document=document,
        uploader_id=uploader_id,
        document_type=document_type,
        signer=signer,
    )

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

    return signed_document

upload_signed_document ¶

upload_signed_document(
    subscription,
    document,
    uploader_id,
    document_type,
    signer,
    commit=True,
)

Uploads a signed document for a subscription.

Parameters:

Name	Type	Description	Default
`subscription`	`Subscription[SubscriptionVersion]`	The subscription to upload the document for.	required
`document`	`BytesIO`	The signed document file.	required
`uploader_id`	`UUID`	The ID of the user uploading the document.	required
`document_type`	`HealthyBenefitsSignedDocumentType`	The type of the document.	required
`signer`	`Signer`	The signer of the document.	required
`commit`	`bool`	Whether to commit the session. Defaults to True.	`True`

Returns:

Name	Type	Description
`HealthyBenefitsSignedDocument`	`HealthyBenefitsSignedDocument`	The uploaded signed document.

Source code in components/healthy_benefits/subcomponents/subscription/protected/signed_documents.py

def upload_signed_document(
    subscription: Subscription[SubscriptionVersion],
    document: BytesIO,
    uploader_id: UUID,
    document_type: HealthyBenefitsSignedDocumentType,
    signer: Signer,
    commit: bool = True,
) -> HealthyBenefitsSignedDocument:
    """Uploads a signed document for a subscription.

    Args:
        subscription (Subscription[SubscriptionVersion]): The subscription to upload the document for.
        document (BytesIO): The signed document file.
        uploader_id (UUID): The ID of the user uploading the document.
        document_type (HealthyBenefitsSignedDocumentType): The type of the document.
        signer (Signer): The signer of the document.
        commit (bool): Whether to commit the session. Defaults to True.

    Returns:
        HealthyBenefitsSignedDocument: The uploaded signed document.
    """
    from components.healthy_benefits.subcomponents.subscription.internal.business_logic.signed_documents import (
        upload_signed_document as upload_signed_document_logic,
    )

    signed_document = upload_signed_document_logic(
        subscription_id=subscription.id,
        document=document,
        uploader_id=uploader_id,
        document_type=document_type,
        signer=signer,
    )

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

    return signed_document

Api reference

components.healthy_benefits.public.affiliation ¶

EnrollArgs dataclass ¶

benefit_type_ref instance-attribute ¶

profile_ref instance-attribute ¶

start_date instance-attribute ¶

transfer_history_id instance-attribute ¶

Enrollment dataclass ¶

benefit_type instance-attribute ¶

end_date class-attribute instance-attribute ¶

id instance-attribute ¶

is_active_at ¶

is_cancelled class-attribute instance-attribute ¶

is_ended_at ¶

profile_ref instance-attribute ¶

start_date class-attribute instance-attribute ¶

subscription_id instance-attribute ¶

transfer_history_id instance-attribute ¶

EnrollmentId module-attribute ¶

EnrollmentModelBroker ¶

cancel_enrollment classmethod ¶

cancel_enrollments classmethod ¶

create_enrollment classmethod ¶

get_active_or_ended_benefit_enrollment classmethod ¶

get_ended_enrollments classmethod ¶

get_enrollment classmethod ¶

get_enrollment_by_ids classmethod ¶

get_enrollment_by_transfer_history_ref classmethod ¶

get_enrollments classmethod ¶

get_enrollments_active_after classmethod ¶

get_enrollments_active_in_period classmethod ¶

get_enrollments_active_on classmethod ¶

model class-attribute instance-attribute ¶

terminate_enrollment classmethod ¶

update_enrollment classmethod ¶

TemporaryLeave dataclass ¶

end_date instance-attribute ¶

enrollment_id instance-attribute ¶

id instance-attribute ¶

is_cancelled instance-attribute ¶

start_date instance-attribute ¶

status instance-attribute ¶

TemporaryLeaveId module-attribute ¶

TemporaryLeaveModelBroker ¶

cancel_temporary_leaves classmethod ¶

create_temporary_leave classmethod ¶

get_temporary_leaves classmethod ¶

get_temporary_leaves_for_enrollments classmethod ¶

model class-attribute instance-attribute ¶

update_temporary_leaves classmethod ¶

TemporaryLeaveStatus ¶

ACTIVE class-attribute instance-attribute ¶

CANCELLED class-attribute instance-attribute ¶

ENDED class-attribute instance-attribute ¶

SCHEDULED class-attribute instance-attribute ¶

bulk_cancel_enrollments ¶

bulk_enroll ¶

bulk_terminate_enrollments ¶

cancel_enrollment ¶

get_active_or_future_enrollments ¶

get_active_or_future_or_last_ended_benefit_enrollment ¶

get_active_or_last_ended_user_enrollment ¶

get_ended_enrollments ¶

get_enrollment ¶

get_enrollment_by_transfer_history_id ¶

get_enrollments_active_in_period ¶

get_enrollments_active_on ¶

get_enrollments_by_ids ¶

get_enrollments_on_subscription ¶

get_enrollments_starting_after ¶

get_profile_enrollments_active_on ¶

get_subscription_active_enrollments_on ¶

is_profile_enrolled ¶

terminate_enrollment ¶

update_enrollment ¶

update_enrollment_transfer_history_id ¶

components.healthy_benefits.public.commands ¶

app_group ¶

healthy_benefits_commands module-attribute ¶

register_healthy_benefits_commands ¶

EnrollArgs `dataclass` ¶

benefit_type_ref `instance-attribute` ¶

profile_ref `instance-attribute` ¶

start_date `instance-attribute` ¶

transfer_history_id `instance-attribute` ¶

Enrollment `dataclass` ¶

benefit_type `instance-attribute` ¶

end_date `class-attribute` `instance-attribute` ¶

id `instance-attribute` ¶

is_cancelled `class-attribute` `instance-attribute` ¶

profile_ref `instance-attribute` ¶

start_date `class-attribute` `instance-attribute` ¶

subscription_id `instance-attribute` ¶

transfer_history_id `instance-attribute` ¶

EnrollmentId `module-attribute` ¶

cancel_enrollment `classmethod` ¶

cancel_enrollments `classmethod` ¶

create_enrollment `classmethod` ¶

get_active_or_ended_benefit_enrollment `classmethod` ¶

get_ended_enrollments `classmethod` ¶

get_enrollment `classmethod` ¶

get_enrollment_by_ids `classmethod` ¶

get_enrollment_by_transfer_history_ref `classmethod` ¶

get_enrollments `classmethod` ¶

get_enrollments_active_after `classmethod` ¶

get_enrollments_active_in_period `classmethod` ¶

get_enrollments_active_on `classmethod` ¶

model `class-attribute` `instance-attribute` ¶

terminate_enrollment `classmethod` ¶

update_enrollment `classmethod` ¶

TemporaryLeave `dataclass` ¶

end_date `instance-attribute` ¶

enrollment_id `instance-attribute` ¶

id `instance-attribute` ¶

is_cancelled `instance-attribute` ¶

start_date `instance-attribute` ¶

status `instance-attribute` ¶

TemporaryLeaveId `module-attribute` ¶

cancel_temporary_leaves `classmethod` ¶

create_temporary_leave `classmethod` ¶

get_temporary_leaves `classmethod` ¶

get_temporary_leaves_for_enrollments `classmethod` ¶

model `class-attribute` `instance-attribute` ¶

update_temporary_leaves `classmethod` ¶

ACTIVE `class-attribute` `instance-attribute` ¶

CANCELLED `class-attribute` `instance-attribute` ¶

ENDED `class-attribute` `instance-attribute` ¶

SCHEDULED `class-attribute` `instance-attribute` ¶

healthy_benefits_commands `module-attribute` ¶

food `class-attribute` `instance-attribute` ¶

learning `class-attribute` `instance-attribute` ¶

nursery `class-attribute` `instance-attribute` ¶

transport `class-attribute` `instance-attribute` ¶

DiscountId `module-attribute` ¶

SubscriptionDiscount `dataclass` ¶

end_month `instance-attribute` ¶

id `instance-attribute` ¶

price `instance-attribute` ¶

reason `instance-attribute` ¶

start_month `instance-attribute` ¶

subscription_id `instance-attribute` ¶

Benefit `dataclass` ¶

allowance `instance-attribute` ¶

id `instance-attribute` ¶

population_id `instance-attribute` ¶

type_ref `instance-attribute` ¶

BenefitArgs `dataclass` ¶

allowance `instance-attribute` ¶

population_id `instance-attribute` ¶

type_ref `instance-attribute` ¶

BenefitId `module-attribute` ¶

HealthyBenefitsSignedDocument `dataclass` ¶

id `instance-attribute` ¶

signed_at `instance-attribute` ¶

signed_bundle_id `instance-attribute` ¶

subscription_id `instance-attribute` ¶

type `instance-attribute` ¶

uploader_id `instance-attribute` ¶

general_conditions `class-attribute` `instance-attribute` ¶

particular_conditions `class-attribute` `instance-attribute` ¶