Api reference

components.affiliation_occ_health.public.api

Public API of the Affiliation component.

affiliate

affiliate(
    *,
    affiliation_service,
    affiliable_ref,
    start_date,
    end_date=None,
    subscription_ref=None,
    commit=True
)

Affiliate an affiliable to an affiliation service.

Parameters:

Name	Type	Description	Default
affiliation_service	AffiliationService	which product / service should we affiliate to?	required
affiliable_ref	str	opaque reference describing who will be affiliated (can be a user ID, etc)	required
start_date	date	when the affiliation starts	required
end_date	date | None	if we know it at affiliation time, when it will end	None
subscription_ref	str | None	optional reference to a subscription (e.g. contract ID)	None
commit	bool	whether to commit the transaction	True

Returns: the UUID of the newly-created affiliation

Raises:

Type	Description
AffiliationAlreadyExistsOrOverlapsError	if an affiliation already exists or overlaps with the given dates

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

def affiliate(
    *,
    affiliation_service: AffiliationService,
    affiliable_ref: str,
    start_date: date,
    end_date: date | None = None,
    subscription_ref: str | None = None,
    commit: bool = True,
) -> UUID:
    """
    Affiliate an affiliable to an affiliation service.

    Args:
        affiliation_service: which product / service should we affiliate to?
        affiliable_ref: opaque reference describing who will be affiliated (can be a user ID, etc)
        start_date: when the affiliation starts
        end_date: if we know it at affiliation time, when it will end
        subscription_ref: optional reference to a subscription (e.g. contract ID)
        commit: whether to commit the transaction

    Returns: the UUID of the newly-created affiliation

    Raises:
        AffiliationAlreadyExistsOrOverlapsError: if an affiliation already exists or overlaps with the given dates
    """
    from components.affiliation_occ_health.internal.business_logic.rules.raise_if_affiliation_exists import (
        raise_if_affiliation_already_exists,
    )

    raise_if_affiliation_already_exists(
        affiliation_service=affiliation_service,
        affiliable_ref=affiliable_ref,
        start_date=start_date,
        end_date=end_date,
        subscription_ref=subscription_ref,
    )

    affiliation = OccupationalHealthAffiliation(
        affiliation_service=affiliation_service,
        start_date=start_date,
        end_date=end_date,
        affiliable_ref=affiliable_ref,
        subscription_ref=subscription_ref,
    )

    current_session.add(affiliation)
    current_session.flush()

    current_logger.info(
        f"Affiliating {affiliable_ref} to {affiliation_service} starting on {start_date} (subscription: {subscription_ref})",
        extra={
            "affiliation_service": affiliation_service,
            "affiliable_ref": affiliable_ref,
            "affiliation_id": affiliation.id,
            "start_date": start_date,
            "end_date": end_date,
        },
    )

    if commit:
        current_session.commit()

    return affiliation.id

cancel_affiliation

cancel_affiliation(
    affiliation_service, affiliation_id, commit=True
)

Cancel the affiliation.

This is different from termination in that it invalidates the affiliation entirely, rather than setting an end date. The affiliation will be considered as if it never existed.

Parameters:

Name	Type	Description	Default
affiliation_service	AffiliationService	which product / service the affiliation belongs to	required
affiliation_id	UUID	the UUID of the affiliation to cancel	required
commit	bool	whether to commit the transaction	True

Raises:

Type	Description
AffiliableNotFound	if no affiliation is found
AffiliationError	if the affiliation is already cancelled

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

def cancel_affiliation(
    affiliation_service: AffiliationService,
    affiliation_id: UUID,
    commit: bool = True,
) -> None:
    """
    Cancel the affiliation.

    This is different from termination in that it invalidates the affiliation entirely, rather than setting an end date.
    The affiliation will be considered as if it never existed.

    Args:
        affiliation_service: which product / service the affiliation belongs to
        affiliation_id: the UUID of the affiliation to cancel
        commit: whether to commit the transaction

    Raises:
        AffiliableNotFound: if no affiliation is found
        AffiliationError: if the affiliation is already cancelled
    """
    affiliation = _get_affiliation(affiliation_id, affiliation_service)

    if affiliation.is_cancelled:
        raise AffiliationError(f"Affiliation {affiliation_id} is already cancelled.")

    current_logger.info(
        f"Cancelling affiliation {affiliation_id}",
        extra={
            "affiliation_service": affiliation_service,
            "affiliable_ref": affiliation.affiliable_ref,
            "subscription_ref": affiliation.subscription_ref,
        },
    )

    affiliation.is_cancelled = True

    if commit:
        current_session.commit()

get_active_affiliation

get_active_affiliation(
    affiliable_ref,
    affiliation_service,
    subscription_ref=None,
    on_date=None,
)

Retrieve the affiliation data for a given affiliable.

If on_date is not provided, it defaults to today.

Beware: for services supporting concurrent affiliation, please provide a subscription_ref. Otherwise, the query will raise if there are multiple active affiliations.

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

def get_active_affiliation(
    affiliable_ref: str,
    affiliation_service: AffiliationService,
    subscription_ref: str | None = None,
    on_date: date | None = None,
) -> AffiliationData | None:
    """
    Retrieve the affiliation data for a given affiliable.

    If on_date is not provided, it defaults to today.

    Beware: for services supporting concurrent affiliation, please provide a subscription_ref. Otherwise, the query will raise if there are multiple active affiliations.
    """
    query = _active_affiliation_query(
        affiliable_ref, affiliation_service, subscription_ref, on_date
    )

    affiliation = query.one_or_none()

    if not affiliation:
        return None

    return _build_affiliation_data(affiliation)

get_active_affiliations_list

get_active_affiliations_list(
    affiliable_ref,
    affiliation_service,
    subscription_ref=None,
    on_date=None,
)

Retrieve the list of affiliation data for a given affiliable.

If on_date is not provided, it defaults to today.

This is mostly useful for services supporting concurrent affiliation.

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

def get_active_affiliations_list(
    affiliable_ref: str,
    affiliation_service: AffiliationService,
    subscription_ref: str | None = None,
    on_date: date | None = None,
) -> list[AffiliationData]:
    """
    Retrieve the *list* of affiliation data for a given affiliable.

    If on_date is not provided, it defaults to today.

    This is mostly useful for services supporting concurrent affiliation.
    """
    query = _active_affiliation_query(
        affiliable_ref, affiliation_service, subscription_ref, on_date
    )

    affiliations = query.all()

    return [_build_affiliation_data(affiliation) for affiliation in affiliations]

get_affiliation_list

get_affiliation_list(
    affiliation_service,
    subscription_ref=None,
    affiliable_ref=None,
    include_cancelled=False,
    only_active_on=None,
)

Return the list of absolutely all affiliation data, or filtered by subscription and/or affiliable.

Parameters:

Name	Type	Description	Default
affiliation_service	str	which service to filter on	required
subscription_ref	str | None	optional reference to a subscription (e.g. contract ID) to filter on	None
affiliable_ref	str | None	optional reference to an affiliable (e.g. profile ID) to filter on	None
include_cancelled	bool	whether to include cancelled affiliations	False
only_active_on	date | None	if provided, filter the affiliations to only include those active on that date	None

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

def get_affiliation_list(
    affiliation_service: str,
    subscription_ref: str | None = None,
    affiliable_ref: str | None = None,
    include_cancelled: bool = False,
    only_active_on: date | None = None,
) -> list[AffiliationData]:
    """
    Return the list of absolutely all affiliation data, or filtered by subscription and/or affiliable.

    Args:
        affiliation_service: which service to filter on
        subscription_ref: optional reference to a subscription (e.g. contract ID) to filter on
        affiliable_ref: optional reference to an affiliable (e.g. profile ID) to filter on
        include_cancelled: whether to include cancelled affiliations
        only_active_on: if provided, filter the affiliations to only include those active on that date
    """
    query = current_session.query(OccupationalHealthAffiliation).filter(  # noqa: ALN085
        OccupationalHealthAffiliation.affiliation_service == affiliation_service,
    )

    if subscription_ref:
        query = query.filter(
            OccupationalHealthAffiliation.subscription_ref == subscription_ref
        )

    if affiliable_ref:
        query = query.filter(
            OccupationalHealthAffiliation.affiliable_ref == affiliable_ref
        )

    if not include_cancelled:
        query = query.filter(OccupationalHealthAffiliation.is_cancelled.is_(False))

    if only_active_on:
        query = query.filter(OccupationalHealthAffiliation.is_active_on(only_active_on))

    affiliations = query.all()

    return [_build_affiliation_data(affiliation) for affiliation in affiliations]

terminate_affiliation

terminate_affiliation(
    affiliation_service,
    affiliable_ref,
    end_date,
    termination_type=None,
    subscription_ref=None,
    commit=True,
)

Terminate the affiliation.

If a subscription_ref is passed, then we require the affiliation to be tied to that subscription. This is useful for services allowing overlapping affiliation through different subscriptions (e.g. Occupational Health).

Parameters:

Name	Type	Description	Default
affiliation_service	AffiliationService	which product / service should we terminate the affiliation from?	required
affiliable_ref	str	opaque reference describing who will be unaffiliated	required
end_date	date	when the affiliation ends	required
termination_type	str | None	optional type of termination (e.g. "voluntary", "involuntary")	None
subscription_ref	str | None	optional reference to a subscription (e.g. contract ID).	None
commit	bool	whether to commit the transaction	True

Returns: the UUID of the terminated affiliation.

Raises:

Type	Description
AffiliableNotFound	if the affiliable has no affiliation, or no active affiliation at the date of termination.
AffiliationError	if the affiliation is already terminated or not active on the end date

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

def terminate_affiliation(
    affiliation_service: AffiliationService,
    affiliable_ref: str,
    end_date: date,
    termination_type: str | None = None,
    subscription_ref: str | None = None,
    commit: bool = True,
) -> UUID:
    """
    Terminate the affiliation.

    If a subscription_ref is passed, then we require the affiliation to be tied to that subscription. This is useful for services allowing overlapping
    affiliation through different subscriptions (e.g. Occupational Health).

    Args:
        affiliation_service: which product / service should we terminate the affiliation from?
        affiliable_ref: opaque reference describing who will be unaffiliated
        end_date: when the affiliation ends
        termination_type: optional type of termination (e.g. "voluntary", "involuntary")
        subscription_ref: optional reference to a subscription (e.g. contract ID).
        commit: whether to commit the transaction

    Returns: the UUID of the terminated affiliation.

    Raises:
        AffiliableNotFound: if the affiliable has no affiliation, or no active affiliation at the date of termination.
        AffiliationError: if the affiliation is already terminated or not active on the end date
    """
    affiliation = _active_affiliation_query(
        affiliable_ref,
        affiliation_service,
        subscription_ref,
        on_date=end_date,
    ).one_or_none()

    if not affiliation:
        raise AffiliableNotFound(
            f"Cannot find active affiliation on {end_date} for {affiliable_ref} to {affiliation_service} (subscription: {subscription_ref}) to terminate it."
        )

    if affiliation.end_date:
        raise AffiliationError(
            f"Affiliation of {affiliable_ref} to {affiliation_service} is already terminated (end date: {affiliation.end_date}, termination type: {affiliation.termination_type})"
        )

    current_logger.info(
        f"Terminating affiliation of {affiliable_ref} to {affiliation_service} on {end_date} with termination type {termination_type}",
        extra={
            "affiliation_service": affiliation_service,
            "affiliable_ref": affiliable_ref,
            "subscription_ref": subscription_ref,
            "affiliation_id": affiliation.id,
        },
    )

    affiliation.end_date = end_date
    affiliation.termination_type = termination_type

    if commit:
        current_session.commit()

    return affiliation.id

update_affiliation_dates

update_affiliation_dates(
    *,
    affiliation_service,
    affiliation_id,
    start_date=NOT_SET,
    end_date=NOT_SET,
    commit=True
)

Update the start date and/or end date of the affiliation. NOT_SET can be passed in either start_date or end_date to leave the date unchanged (default behavior).

Parameters:

Name	Type	Description	Default
affiliation_service	AffiliationService	which product / service should we update the affiliation from?	required
affiliation_id	UUID	the UUID of the affiliation to update	required
start_date	NotSet[date]	when the affiliation starts	NOT_SET
end_date	NotSet[date | None]	when the affiliation ends (can be None to unset the end_date)	NOT_SET
commit	bool	whether to commit the transaction	True

Returns: the UUID of the updated affiliation

Raises:

Type	Description
ValueError	if neither start_date nor end_date is provided
AffiliableNotFound	if no affiliation is found
AffiliationError	if the new dates are invalid

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

def update_affiliation_dates(
    *,
    affiliation_service: AffiliationService,
    affiliation_id: UUID,
    start_date: NotSet[date] = NOT_SET,
    end_date: NotSet[date | None] = NOT_SET,
    commit: bool = True,
) -> UUID:
    """
    Update the start date and/or end date of the affiliation.
    `NOT_SET` can be passed in either `start_date` or `end_date` to leave the date unchanged (default behavior).

    Args:
        affiliation_service: which product / service should we update the affiliation from?
        affiliation_id: the UUID of the affiliation to update
        start_date: when the affiliation starts
        end_date: when the affiliation ends (can be `None` to unset the end_date)
        commit: whether to commit the transaction

    Returns: the UUID of the updated affiliation

    Raises:
        ValueError: if neither start_date nor end_date is provided
        AffiliableNotFound: if no affiliation is found
        AffiliationError: if the new dates are invalid
    """
    from components.affiliation_occ_health.internal.business_logic.rules.raise_if_affiliation_exists import (
        raise_if_affiliation_already_exists,
    )

    if not is_set(start_date) and not is_set(end_date):
        raise ValueError("At least one of start_date or end_date must be provided")

    affiliation = _get_affiliation(affiliation_id, affiliation_service)

    logger = current_logger.bind(
        affiliation_service=affiliation_service,
        affiliation_id=affiliation_id,
        affiliable_ref=affiliation.affiliable_ref,
    )

    raise_if_affiliation_already_exists(
        affiliation_service=affiliation_service,
        affiliable_ref=affiliation.affiliable_ref,
        start_date=start_date if is_set(start_date) else affiliation.start_date,
        end_date=end_date if is_set(end_date) else affiliation.end_date,
        subscription_ref=affiliation.subscription_ref,
        # We don't want to raise if the affiliation we're updating is the one we're checking
        ignore_existing_affiliation_id=affiliation_id,
    )

    if is_set(start_date):
        # We accept that target_end_date can be None (if we're unsetting it)
        target_end_date = end_date if (is_set(end_date)) else affiliation.end_date

        if target_end_date is not None and target_end_date < start_date:
            raise AffiliationError(
                f"Cannot update affiliation {affiliation_id} to {affiliation_service} with start date {start_date} after end date {target_end_date}"
            )

        logger.info(
            f"Update affiliation {affiliation_id} to {affiliation_service} with new start date {start_date} (previously was: {affiliation.start_date})",
            extra={
                "affiliation_service": affiliation_service,
                "affiliation_id": affiliation_id,
                "affiliable_ref": affiliation.affiliable_ref,
                "previous_start_date": affiliation.start_date,
                "new_start_date": start_date,
            },
        )

        affiliation.start_date = start_date

    if is_set(end_date):
        # unlike target_end_date, here we'd already have edited affiliation.start_date
        target_start_date = affiliation.start_date
        if end_date is not None and end_date < target_start_date:
            raise AffiliationError(
                f"Cannot update affiliation {affiliation_id} to {affiliation_service} with end date {end_date} before start date {target_start_date}"
            )

        logger.info(
            f"Update affiliation {affiliation_id} to {affiliation_service} with new end date {end_date} (previously was: {affiliation.end_date})",
            extra={
                "affiliation_service": affiliation_service,
                "affiliation_id": affiliation_id,
                "affiliable_ref": affiliation.affiliable_ref,
                "previous_end_date": affiliation.end_date,
                "new_end_date": end_date,
            },
        )

        affiliation.end_date = end_date

    if commit:
        current_session.commit()

    return affiliation.id

components.affiliation_occ_health.public.entities

AffiliationData dataclass

AffiliationData(
    affiliation_id,
    start_date,
    end_date,
    is_cancelled,
    affiliable_ref,
    affiliation_service,
    subscription_ref,
)

Bases: DataHistorizable

Represent the affiliation of an affiliable to a service, over a set of dates.

affiliable_ref instance-attribute

affiliable_ref

affiliation_id instance-attribute

affiliation_id

affiliation_service instance-attribute

affiliation_service

end_date instance-attribute

end_date

is_cancelled instance-attribute

is_cancelled

start_date instance-attribute

start_date

subscription_ref instance-attribute

subscription_ref

components.affiliation_occ_health.public.enums

AffiliationService

Bases: AlanBaseEnum

The Global Centralized Affiliation component can affiliate "affiliable" to different services independently (eg Health France, Prévoyance France, etc).

OCCUPATIONAL_HEALTH class-attribute instance-attribute

OCCUPATIONAL_HEALTH = 'occupational_health'

components.affiliation_occ_health.public.exceptions

AffiliableNotFound

Bases: BaseAffiliationError

Affiliable matching criteria not found.

AffiliationAlreadyExistsOrOverlapsError

Bases: BaseAffiliationError

Enrollment already exists or overlaps with an existing one.

AffiliationError

Bases: BaseAffiliationError

Affiliation operation was not successful (eg because the affiliation was already cancelled/terminated, the new dates are invalid...)

BaseAffiliationError

Bases: Exception

Generic affiliation error.