Skip to content

Api reference

components.income.public.actions

parse_and_create_income_declaration 

parse_and_create_income_declaration(
    user_id,
    effective_start_date,
    selected_bracket_data,
    income_brackets_data,
    declaration_type,
    subscription_id=None,
    commit=False,
)

Create income declaration from raw dictionary data with Pydantic validation. Args: user_id: User identifier effective_start_date: When the income declaration becomes effective selected_bracket_data: User's selected income bracket (dict with min/max keys) income_brackets_data: Available income bracket options (list of dicts) declaration_type: Type of declaration to create subscription_id: Subscription identifier commit: Whether to commit the database transaction

Source code in components/income/internal/actions/create_income_declaration.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def parse_and_create_income_declaration(
    user_id: str,
    effective_start_date: date,
    selected_bracket_data: Optional[dict[str, int | None]],
    income_brackets_data: Optional[list[dict[str, int | None]]],
    declaration_type: IncomeDeclarationType,
    subscription_id: str | None = None,
    commit: bool = False,
) -> None:
    """
    Create income declaration from raw dictionary data with Pydantic validation.
    Args:
        user_id: User identifier
        effective_start_date: When the income declaration becomes effective
        selected_bracket_data: User's selected income bracket (dict with min/max keys)
        income_brackets_data: Available income bracket options (list of dicts)
        declaration_type: Type of declaration to create
        subscription_id: Subscription identifier
        commit: Whether to commit the database transaction
    """
    if income_brackets_data is None or len(income_brackets_data) == 0:
        return

    income_brackets, selected_income_raw = _parse_income_declaration_creation_params(
        income_brackets_data, selected_bracket_data
    )
    member_type = mandatory(
        get_member_type_for_user_on(user_id=user_id, on_date=effective_start_date),
        "Member type should be defined before creating an income declaration.",
    )
    create_income_declaration(
        user_id=user_id,
        min_income_in_cents=(selected_income_raw.min if selected_income_raw else None),
        max_income_in_cents=(selected_income_raw.max if selected_income_raw else None),
        effective_start_date=effective_start_date,
        income_brackets=income_brackets,
        member_type=member_type,
        submission_date=utctoday(),
        declaration_type=declaration_type,
        subscription_id=subscription_id,
        commit=commit,
    )

send_unpaid_leave_start_email 

send_unpaid_leave_start_email(user_ref)
Source code in components/income/internal/actions/income_unpaid_leave_start_mailer.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def send_unpaid_leave_start_email(user_ref: str) -> LangTemplateMailerParams | None:
    from components.income.internal.mailers.income_unpaid_leave_start_email import (
        send_income_unpaid_leave_start_email_with_args,
    )

    repository = IncomeDeclarationRepository()
    declarations = repository.get_income_declarations(user_ref)
    income_declaration = declarations[-1] if declarations else None

    if not income_declaration:
        return None

    if not is_price_depending_on_income_brackets(income_declaration):
        return None

    profile_email = get_user_email(user_ref)

    return (
        send_income_unpaid_leave_start_email_with_args(user_email=profile_email)
        if profile_email
        else None
    )

components.income.public.dependencies

IncomeDependency

IncomeDependency defines the interface that apps using the income component need to implement

get_async_mailer 

get_async_mailer(
    category,
    priority,
    email_queue_name,
    delivery=DEFAULT_DELIVERY,
    job_timeout=None,
    enqueue=True,
    redacted_args=None,
)

Get the local async_mailer

Source code in components/income/internal/dependencies/income.py 
40
41
42
43
44
45
46
47
48
49
50
51
def get_async_mailer(
    self,
    category: EmailCategory,
    priority: EmailPriority,
    email_queue_name: Optional[str],
    delivery: Union[CustomerIODelivery, SendgridDelivery] = DEFAULT_DELIVERY,
    job_timeout: Optional[int] = None,
    enqueue: bool = True,
    redacted_args: set[str] | None = None,
) -> Callable[[_MailerCallable], _MailerCallable]:
    """Get the local async_mailer"""
    raise NotImplementedError()

get_eligibility_requests_for_user 

get_eligibility_requests_for_user(user_id)

Returns income eligibility requests for a user.

Source code in components/income/internal/dependencies/income.py 
93
94
95
96
97
98
99
def get_eligibility_requests_for_user(
    self, user_id: str
) -> list[IncomeEligibilityRequest]:
    """
    Returns income eligibility requests for a user.
    """
    raise NotImplementedError()

get_income_brackets_for_subscription_on 

get_income_brackets_for_subscription_on(
    subscription_id, on_date, for_target=None
)

Returns income brackets for a given subscription on a specific date.

Source code in components/income/internal/dependencies/income.py 
64
65
66
67
68
69
70
71
def get_income_brackets_for_subscription_on(
    self,
    subscription_id: str,
    on_date: date,
    for_target: Optional[IncomeBracketTarget] = None,
) -> list[IncomeBracketEntity]:
    """Returns income brackets for a given subscription on a specific date."""
    raise NotImplementedError()

get_income_brackets_for_subscriptions_on 

get_income_brackets_for_subscriptions_on(
    subscription_ids, on_date, for_target=None
)

Returns income brackets for multiple subscriptions on a specific date.

Parameters:

Name Type Description Default
subscription_ids list[str]

List of subscription IDs to get income brackets for

 required
on_date date

The date to check the income brackets against

 required
for_target Optional[IncomeBracketTarget]

Optional target for which the income brackets are requested

 None

Returns:

Type Description
dict[str, list[IncomeBracketEntity]]

dict[str, list[IncomeBracketEntity]]: Mapping of subscription_id to income brackets
dict[str, list[IncomeBracketEntity]]

Only includes subscriptions that have income brackets
Source code in components/income/internal/dependencies/income.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def get_income_brackets_for_subscriptions_on(
    self,
    subscription_ids: list[str],
    on_date: date,
    for_target: Optional[IncomeBracketTarget] = None,
) -> dict[str, list[IncomeBracketEntity]]:
    """
    Returns income brackets for multiple subscriptions on a specific date.

    Args:
        subscription_ids: List of subscription IDs to get income brackets for
        on_date: The date to check the income brackets against
        for_target: Optional target for which the income brackets are requested

    Returns:
        dict[str, list[IncomeBracketEntity]]: Mapping of subscription_id to income brackets
        Only includes subscriptions that have income brackets
    """
    raise NotImplementedError()

get_member_type_for_user_on 

get_member_type_for_user_on(user_id, on_date)
Source code in components/income/internal/dependencies/income.py 
59
60
61
62
def get_member_type_for_user_on(
    self, user_id: str, on_date: date
) -> Optional[MemberType]:
    raise NotImplementedError()

get_subscription_id_for_user_on 

get_subscription_id_for_user_on(user_id, on_date)

Returns user subscription id linked to the user on a given on_date

Source code in components/income/internal/dependencies/income.py 
53
54
55
56
57
def get_subscription_id_for_user_on(
    self, user_id: str, on_date: date
) -> Optional[str]:
    """Returns user subscription id linked to the user on a given on_date"""
    raise NotImplementedError()

get_subscription_ids_for_users_on 

get_subscription_ids_for_users_on(user_ids, on_date)

Get user subscription IDs active at a specific date.

Parameters:

Name Type Description Default
user_ids list[str]

List of user IDs to get subscription IDs for

 required
on_date date

The date to check for active subscriptions

 required

Returns:

Type Description
dict[str, str]

dict[str, str]: Mapping of user_id to their subscription_id. Only includes users with active subscriptions.
Source code in components/income/internal/dependencies/income.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def get_subscription_ids_for_users_on(
    self, user_ids: list[str], on_date: date
) -> dict[str, str]:
    """
    Get user subscription IDs active at a specific date.

    Args:
        user_ids: List of user IDs to get subscription IDs for
        on_date: The date to check for active subscriptions

    Returns:
        dict[str, str]: Mapping of user_id to their subscription_id.
            Only includes users with active subscriptions.
    """
    raise NotImplementedError()

get_subscription_info_for_subscriptions_on 

get_subscription_info_for_subscriptions_on(
    subscription_ids, on_date
)

Returns subscription info for a list of subscription IDs at a specific date.

Parameters:

Name Type Description Default
subscription_ids list[str]

List of subscription IDs to get info for

 required
on_date date

The date to get subscription info at

 required

Returns:

Type Description
list[SubscriptionInfoEntity]

list[SubscriptionInfoEntity]: List of subscription info entities
Source code in components/income/internal/dependencies/income.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def get_subscription_info_for_subscriptions_on(
    self, subscription_ids: list[str], on_date: date
) -> list[SubscriptionInfoEntity]:
    """
    Returns subscription info for a list of subscription IDs at a specific date.

    Args:
        subscription_ids: List of subscription IDs to get info for
        on_date: The date to get subscription info at

    Returns:
        list[SubscriptionInfoEntity]: List of subscription info entities
    """
    raise NotImplementedError()

get_user_income_sources_from_dsn 

get_user_income_sources_from_dsn(user_id, months)

Fetches the salary for a user from the DSN for given months.

Source code in components/income/internal/dependencies/income.py 
101
102
103
104
105
def get_user_income_sources_from_dsn(
    self, user_id: str, months: Sequence[Month]
) -> dict[Month, int]:
    """Fetches the salary for a user from the DSN for given months."""
    raise NotImplementedError()

has_enrollment_depending_on_income_brackets_on 

has_enrollment_depending_on_income_brackets_on(
    user_id, on_date
)

Checks if the user has an enrollment depending on income brackets on a given date.

Source code in components/income/internal/dependencies/income.py 
113
114
115
116
117
118
119
def has_enrollment_depending_on_income_brackets_on(
    self, user_id: str, on_date: date
) -> bool:
    """
    Checks if the user has an enrollment depending on income brackets on a given date.
    """
    raise NotImplementedError()

is_subscription_linked_to_income_brackets 

is_subscription_linked_to_income_brackets(subscription_id)

Checks if the subscription is linked to income brackets.

Source code in components/income/internal/dependencies/income.py 
107
108
109
110
111
def is_subscription_linked_to_income_brackets(self, subscription_id: str) -> bool:
    """
    Checks if the subscription is linked to income brackets.
    """
    raise NotImplementedError()

set_app_dependency 

set_app_dependency(dependency)

Sets the income dependency to the app so it can be accessed within this component at runtime

Source code in components/income/internal/dependencies/income.py 
138
139
140
141
142
143
144
def set_app_dependency(dependency: IncomeDependency) -> None:
    """Sets the income dependency to the app so it can be accessed within this component at runtime"""
    from flask import current_app

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

components.income.public.entities

IncomeBracketEntity dataclass 

IncomeBracketEntity(min, max)

Bases: DataClassJsonMixin

__eq__ 

__eq__(other)
Source code in components/income/internal/entities/income_bracket.py 
19
20
21
22
def __eq__(self, other: object) -> bool:
    if not isinstance(other, IncomeBracketEntity):
        return False
    return self.min == other.min and self.max == other.max

__post_init__ 

__post_init__()
Source code in components/income/internal/entities/income_bracket.py 
13
14
15
16
17
def __post_init__(self) -> None:
    if self.min < 0 or (self.max is not None and self.max < 0):
        raise ValueError("Income bracket values must be non-negative.")
    if self.max is not None and self.min > self.max:
        raise ValueError("Minimum value cannot be greater than maximum value.")

build_list_from_mins classmethod 

build_list_from_mins(min_edges)
Source code in components/income/internal/entities/income_bracket.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
@classmethod
def build_list_from_mins(cls, min_edges: list[int]) -> list["IncomeBracketEntity"]:
    income_brackets: list[IncomeBracketEntity] = []
    for index, min_edge in enumerate(min_edges):
        income_brackets.append(
            cls(
                min=min_edge,
                max=(
                    min_edges[index + 1] - 1 if index + 1 < len(min_edges) else None
                ),
            )
        )
    return income_brackets

is_upper_bracket_of 

is_upper_bracket_of(income_brackets)
Source code in components/income/internal/entities/income_bracket.py 
24
25
26
27
28
29
30
def is_upper_bracket_of(self, income_brackets: list["IncomeBracketEntity"]) -> bool:
    if len(income_brackets) == 0:
        return False

    sorted_brackets = sorted(income_brackets, key=lambda bracket: bracket.min)
    upper_bracket = sorted_brackets[-1]
    return self == upper_bracket

max instance-attribute 

max

min instance-attribute 

min

IncomeBracketTarget

Bases: AlanBaseEnum

Status of an income declaration.

The target can be for a specific context such as 'option' or 'dependent'. This enum is used to determine the context for which income brackets are requested. Possible values include 'option' or 'dependent'.

dependent class-attribute instance-attribute 

dependent = 'dependent'

option class-attribute instance-attribute 

option = 'option'

IncomeDeclarationEntity dataclass 

IncomeDeclarationEntity(
    id,
    min_declared_monthly_income,
    max_declared_monthly_income,
    min_effective_monthly_income,
    max_effective_monthly_income,
    user_ref,
    submission_date,
    affiliation_start_date,
    status,
    member_type,
    declaration_type,
    sources,
    income_brackets,
    subscription_id,
    proof_requested_on_date=None,
    invalidation_reason=None,
    validated_on_date=None,
)

Entity representing an income declaration model.

affiliation_start_date instance-attribute 

affiliation_start_date

declaration_type instance-attribute 

declaration_type

dsn_validating_months property 

dsn_validating_months

Get the months for which DSN validation is applicable. This includes the first full month after affiliation start date and all subsequent months until the end of the year.

Returns:

Type Description
list[Month]

list[Month]: List of months for DSN validation

find_highest_bracket 

find_highest_bracket()

Find the highest income bracket available.

Returns:

Name Type Description
IncomeBracketEntity IncomeBracketEntity

The highest income bracket (with the highest minimum threshold)

Raises:

Type Description
ValueError

If no brackets are available
Source code in components/income/internal/entities/income_declaration.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def find_highest_bracket(self) -> IncomeBracketEntity:
    """
    Find the highest income bracket available.

    Returns:
        IncomeBracketEntity: The highest income bracket (with the highest minimum threshold)

    Raises:
        ValueError: If no brackets are available
    """
    if not self.income_brackets:
        raise ValueError("No income brackets available")

    highest_bracket = max(self.income_brackets, key=lambda bracket: bracket.min)

    return highest_bracket

find_matching_bracket_for_income 

find_matching_bracket_for_income(income_in_cents)

Find the appropriate income bracket for a given income amount.

Parameters:

Name Type Description Default
income_in_cents int

The income amount in cents

 required

Returns:

Name Type Description
IncomeBracketEntity IncomeBracketEntity

The appropriate income bracket

Raises:

Type Description
ValueError

If no appropriate bracket is found
Source code in components/income/internal/entities/income_declaration.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def find_matching_bracket_for_income(
    self, income_in_cents: int
) -> IncomeBracketEntity:
    """
    Find the appropriate income bracket for a given income amount.

    Args:
        income_in_cents: The income amount in cents

    Returns:
        IncomeBracketEntity: The appropriate income bracket

    Raises:
        ValueError: If no appropriate bracket is found
    """
    sorted_brackets = sorted(self.income_brackets, key=lambda bracket: bracket.min)

    for bracket in sorted_brackets:
        if bracket.min <= income_in_cents and (
            bracket.max is None or income_in_cents <= bracket.max
        ):
            return bracket

    # This should not happen if brackets are properly defined
    raise ValueError(f"No appropriate bracket found for income: {income_in_cents}")

first_full_month property 

first_full_month

Get the first full month after affiliation start date. If the affiliation start date the first day of a month, we take that month, otherwise we take the next month.

Returns:

Name Type Description
Month Month

first full month after affiliation start date

get_admin_override_sources 

get_admin_override_sources()

Get all admin override sources from this declaration.

Returns:

Type Description
list[IncomeSourceEntity]

list[IncomeSourceEntity]: List of admin override sources
Source code in components/income/internal/entities/income_declaration.py 
132
133
134
135
136
137
138
139
140
141
142
143
def get_admin_override_sources(self) -> list[IncomeSourceEntity]:
    """
    Get all admin override sources from this declaration.

    Returns:
        list[IncomeSourceEntity]: List of admin override sources
    """
    return [
        source
        for source in self.sources
        if source.source_type == IncomeSourceType.admin_override
    ]

get_invalidating_dsn_sources 

get_invalidating_dsn_sources()

Get all DSN sources that invalidate the declared income (monthly_income > max_declared_monthly_income).

Returns:

Type Description
list[IncomeSourceEntity]

list[IncomeSourceEntity]: List of invalidating DSN sources
Source code in components/income/internal/entities/income_declaration.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def get_invalidating_dsn_sources(self) -> list[IncomeSourceEntity]:
    """
    Get all DSN sources that invalidate the declared income (monthly_income > max_declared_monthly_income).

    Returns:
        list[IncomeSourceEntity]: List of invalidating DSN sources
    """
    if self.max_declared_monthly_income is None:
        return []

    # Filter for DSN sources first, then apply invalidation logic
    dsn_sources = [
        source
        for source in self.sources
        if source.source_type == IncomeSourceType.dsn
    ]
    return filter_invalidating_dsn_sources(
        dsn_sources, self.max_declared_monthly_income
    )

get_proof_document_sources 

get_proof_document_sources()

Get all proof document sources from this declaration.

Returns:

Type Description
list[IncomeSourceEntity]

list[IncomeSourceEntity]: List of proof document sources
Source code in components/income/internal/entities/income_declaration.py 
145
146
147
148
149
150
151
152
153
154
155
156
def get_proof_document_sources(self) -> list[IncomeSourceEntity]:
    """
    Get all proof document sources from this declaration.

    Returns:
        list[IncomeSourceEntity]: List of proof document sources
    """
    return [
        source
        for source in self.sources
        if source.source_type == IncomeSourceType.proof_document
    ]

get_validating_dsn_sources 

get_validating_dsn_sources()

Get all DSN sources that validate the declared income (monthly_income <= max_declared_monthly_income).

Returns:

Type Description
list[IncomeSourceEntity]

list[IncomeSourceEntity]: List of validating DSN sources
Source code in components/income/internal/entities/income_declaration.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def get_validating_dsn_sources(self) -> list[IncomeSourceEntity]:
    """
    Get all DSN sources that validate the declared income (monthly_income <= max_declared_monthly_income).

    Returns:
        list[IncomeSourceEntity]: List of validating DSN sources
    """
    if self.max_declared_monthly_income is None:
        return []

    # Filter for DSN sources first, then apply validation logic
    dsn_sources = [
        source
        for source in self.sources
        if source.source_type == IncomeSourceType.dsn
    ]
    return filter_validating_dsn_sources(
        dsn_sources, self.max_declared_monthly_income
    )

id instance-attribute 

id

income_brackets instance-attribute 

income_brackets

invalidation_reason class-attribute instance-attribute 

invalidation_reason = None

max_declared_monthly_income instance-attribute 

max_declared_monthly_income

max_effective_monthly_income instance-attribute 

max_effective_monthly_income

member_type instance-attribute 

member_type

min_declared_monthly_income instance-attribute 

min_declared_monthly_income

min_effective_monthly_income instance-attribute 

min_effective_monthly_income

proof_requested_on_date class-attribute instance-attribute 

proof_requested_on_date = None

sources instance-attribute 

sources

status instance-attribute 

status

submission_date instance-attribute 

submission_date

subscription_id instance-attribute 

subscription_id

user_ref instance-attribute 

user_ref

validated_on_date class-attribute instance-attribute 

validated_on_date = None

IncomeDeclarationStatus

Bases: AlanBaseEnum

Status of an income declaration.

The declaration can be pending (waiting for validation), validated (approved), invalidated (rejected), or cancelled.

cancelled class-attribute instance-attribute 

cancelled = 'cancelled'

invalidated class-attribute instance-attribute 

invalidated = 'invalidated'

pending class-attribute instance-attribute 

pending = 'pending'

validated class-attribute instance-attribute 

validated = 'validated'

IncomeDeclarationType

Bases: AlanBaseEnum

Type of an income declaration.

The declaration can be: - declaration: Initial declaration made by the user - renewal: Renewal of an existing declaration - amendment: Amendment to an existing declaration

amendment class-attribute instance-attribute 

amendment = 'amendment'

declaration class-attribute instance-attribute 

declaration = 'declaration'

renewal class-attribute instance-attribute 

renewal = 'renewal'

IncomeEligibilityRequest dataclass 

IncomeEligibilityRequest(
    id, status, created_at, eligibility_type
)

Local representation of an eligibility request for the income component.

created_at instance-attribute 

created_at

eligibility_type instance-attribute 

eligibility_type

id instance-attribute 

id

status instance-attribute 

status

IncomeEligibilityRequestStatus

Bases: AlanBaseEnum

Enum representing the status of an income eligibility request.

Attributes:

Name Type Description
uploaded

The request has been uploaded.
parsing

The request is currently being parsed.
parsed

The request has been parsed.
to_manual_review

The request is waiting for manual review.
rejected

The request has been rejected.
validated

The request has been validated.
not_eligible

The request is not eligible.

not_eligible class-attribute instance-attribute 

not_eligible = 'not_eligible'

parsed class-attribute instance-attribute 

parsed = 'parsed'

parsing class-attribute instance-attribute 

parsing = 'parsing'

rejected class-attribute instance-attribute 

rejected = 'rejected'

to_manual_review class-attribute instance-attribute 

to_manual_review = 'to_manual_review'

uploaded class-attribute instance-attribute 

uploaded = 'uploaded'

validated class-attribute instance-attribute 

validated = 'validated'

IncomePeriod dataclass 

IncomePeriod(validity_period, monthly_income)

Bases: WithValidityPeriod

do_overlap 

do_overlap(other)
Source code in components/income/internal/entities/income_period.py 
15
16
def do_overlap(self, other: "WithValidityPeriod") -> bool:
    return self.validity_period.do_overlap(other.validity_period)

monthly_income instance-attribute 

monthly_income

validity_period instance-attribute 

validity_period

IncomeSourceEntity dataclass 

IncomeSourceEntity(
    id,
    monthly_income,
    user_ref,
    start_month,
    end_month,
    submission_date,
    source_type,
    source_ref=None,
    income_declaration_id=None,
)

Entity representing an income source model.

end_month instance-attribute 

end_month

id instance-attribute 

id

income_declaration_id class-attribute instance-attribute 

income_declaration_id = None

monthly_income instance-attribute 

monthly_income

source_ref class-attribute instance-attribute 

source_ref = None

source_type instance-attribute 

source_type

start_month instance-attribute 

start_month

submission_date instance-attribute 

submission_date

user_ref instance-attribute 

user_ref

MemberType

Bases: AlanBaseEnum

Enum representing the type of member for income declarations.

employee class-attribute instance-attribute 

employee = 'employee'

retiree class-attribute instance-attribute 

retiree = 'retiree'

SubscriptionInfoEntity dataclass 

SubscriptionInfoEntity(subscription_id, renewal_date)

Entity representing subscription information for income renewal processing.

renewal_date instance-attribute 

renewal_date

subscription_id instance-attribute 

subscription_id

components.income.public.fixtures

create_income_declaration 

create_income_declaration(
    user_id,
    min_income_in_cents,
    max_income_in_cents,
    effective_start_date,
    income_brackets_mins,
    submission_date=None,
)
Source code in components/income/internal/fixtures/create_income_declaration.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def create_income_declaration(
    user_id: str,
    min_income_in_cents: int | None,
    max_income_in_cents: int | None,
    effective_start_date: date,
    income_brackets_mins: list[int],
    submission_date: date | None = None,
) -> UUID | None:
    return _create_income_declaration(
        user_id=user_id,
        min_income_in_cents=min_income_in_cents,
        max_income_in_cents=max_income_in_cents,
        effective_start_date=effective_start_date,
        income_brackets=IncomeBracketEntity.build_list_from_mins(income_brackets_mins),
        submission_date=submission_date if submission_date else utctoday(),
        member_type=MemberType.employee,
    )

expire_income_declaration 

expire_income_declaration(
    income_declaration_id,
    processing_date=None,
    commit=False,
)

Update the income declaration if it is in those 3 cases: - has not received any DSN source after the first full month after their affiliation start date: their status should be updated to proof requested and an email should be sent to the user. - pending after 3 full months after their affiliation start date: they should be validated/invalidated based on the DSN sources we already have, even if less than 3. - declarations that are pending and with the status proof requested for over a month: they should be invalidated and an email should be sent to the user

Source code in components/income/internal/actions/expire_income_declaration.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def expire_income_declaration(
    income_declaration_id: UUID,
    processing_date: date | None = None,
    commit: bool = False,
) -> None:
    """
    Update the income declaration if it is in those 3 cases:
    - has not received any DSN source after the first full month
    after their affiliation start date: their status should be updated to proof requested
    and an email should be sent to the user.
    - pending after 3 full months after their affiliation
    start date: they should be validated/invalidated based on the DSN sources we already have,
    even if less than 3.
    - declarations that are pending and with the status proof requested for over a month: they
    should be invalidated and an email should be sent to the user
    """
    income_declaration = mandatory(
        IncomeDeclarationRepository().get_income_declaration(
            declaration_id=income_declaration_id
        )
    )
    _processing_date: date = processing_date or utctoday()

    if (
        income_declaration.status != IncomeDeclarationStatus.pending
        or income_declaration.proof_requested_on_date is not None
        and not _has_passed_delay_to_receive_income_proof(
            income_declaration=income_declaration, processing_date=_processing_date
        )
    ):
        current_logger.warning(
            "Income declaration is not pending or already has proof requested recently",
            income_declaration_id=income_declaration_id,
        )
        return

    if income_declaration.member_type == MemberType.retiree:
        current_logger.warning(
            "An income declaration for a retiree should not expired and should automatically be validated",
            income_declaration_id=income_declaration_id,
        )
        return

    if (
        _has_passed_delay_to_receive_three_dsn_sources(
            income_declaration=income_declaration, processing_date=_processing_date
        )
        and income_declaration.sources
        and income_declaration.proof_requested_on_date is None
    ):
        validate_income_declaration_from_dsn_sources(
            income_declaration_id=income_declaration.id,
            dsn_sources=income_declaration.sources,
            on_date=_processing_date,
        )
    elif (
        _has_passed_delay_to_receive_first_dsn_source(
            income_declaration=income_declaration, processing_date=_processing_date
        )
        and not income_declaration.sources
        and income_declaration.proof_requested_on_date is None
    ):
        notify_proof_requested(
            income_declaration_id=income_declaration.id,
            request_date=_processing_date,
        )
    elif _has_passed_delay_to_receive_income_proof(
        income_declaration=income_declaration, processing_date=_processing_date
    ):
        invalidate_and_notify_proof_missing(
            income_declaration_id=income_declaration.id,
            on_date=_processing_date,
        )
    else:
        current_logger.info(
            "Income declaration does not require expiring action",
            income_declaration_id=income_declaration_id,
        )

    if commit:
        current_session.commit()

components.income.public.message_types

IncomeChanged dataclass 

IncomeChanged(user_ref)

Bases: Message

Represents a change in income for a user. This message is used to notify other components (mostly billing) about the change in income.

user_ref instance-attribute 

user_ref

components.income.public.queries

ProofUploadDeadline dataclass 

ProofUploadDeadline(deadline, member_type)

Dataclass containing the deadline and member type for proof upload.

deadline instance-attribute 

deadline

member_type instance-attribute 

member_type

get_proof_upload_deadline 

get_proof_upload_deadline(user_ref)

Returns the deadline date and member type for uploading proof for a user's income declaration.

This function checks if: 1. A proof was requested for any income declaration 2. For each proof request date, it checks if there are any non-rejected eligibility requests for income type that were created after that date 3. Returns the deadline date (proof_request_date + time_frame) and member type as soon as we find a proof request date that has no valid eligibility requests after it

Parameters:

Name Type Description Default
user_ref str

The user ref to check

 required

Returns:

Type Description
ProofUploadDeadline | None

The ProofUploadDeadline containing deadline date and member type, or None if no proof was requested or all proofs have been provided.
Source code in components/income/internal/queries/needs_proof_for_income.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def get_proof_upload_deadline(user_ref: str) -> ProofUploadDeadline | None:
    """
    Returns the deadline date and member type for uploading proof for a user's income declaration.

    This function checks if:
    1. A proof was requested for any income declaration
    2. For each proof request date, it checks if there are any non-rejected eligibility requests
       for income type that were created after that date
    3. Returns the deadline date (proof_request_date + time_frame) and member type as soon as we find a proof request date
       that has no valid eligibility requests after it

    Args:
        user_ref: The user ref to check

    Returns:
        The ProofUploadDeadline containing deadline date and member type, or None if no proof was requested or all proofs have been provided.
    """

    from components.income.internal.repository.income_declaration_repository import (
        IncomeDeclarationRepository,
    )

    income_declaration_repository = IncomeDeclarationRepository()
    proof_requested_declarations = (
        income_declaration_repository.get_pending_income_declarations(
            user_id=user_ref, with_requested_proof=True
        )
    )

    if not proof_requested_declarations:
        # No proof was requested
        return None

    # Get all proof request dates with their corresponding declarations
    proof_request_data: list[tuple[date, MemberType]] = [
        (declaration.proof_requested_on_date, declaration.member_type)
        for declaration in proof_requested_declarations
        if declaration.proof_requested_on_date is not None
    ]

    # Sort proof request dates to check the most urgent deadline first
    proof_request_data.sort(key=lambda x: x[0])

    # Query eligibility requests for this user through the dependency
    income_dependency = get_app_dependency()
    eligibility_requests = income_dependency.get_eligibility_requests_for_user(user_ref)
    time_frame = get_time_frame_to_upload_proof()

    # For each proof request date, check if there are any valid eligibility requests created after it
    for proof_request_date, member_type in proof_request_data:
        has_valid_request_after = False

        for request in eligibility_requests:
            # Check if it was created after the proof was requested
            request_created_at = request.created_at.date()

            if request_created_at < proof_request_date:
                continue

            # Check if it's not rejected
            if request.status == IncomeEligibilityRequestStatus.rejected:
                continue

            # Found a valid request created after this proof request date
            has_valid_request_after = True
            break

        # If no valid eligibility request was found after this proof request date,
        # we need proof for this date - return the deadline and member type
        if not has_valid_request_after:
            return ProofUploadDeadline(
                deadline=proof_request_date + time_frame, member_type=member_type
            )

    # All proof request dates have valid eligibility requests created after them
    return None

get_user_income_on 

get_user_income_on(user_id, on_date)

Returns the IncomeEntity for a user on a specific date.

Parameters:

Name Type Description Default
user_id str

The user ID to get income brackets for

 required
on_date date

The date to check the income against

 required

Returns:

Type Description
IncomeEntity | None

IncomeEntity or None if there is no income for the user on the given date.
Source code in components/income/internal/queries/get_user_income.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def get_user_income_on(
    user_id: str,
    on_date: date,
) -> IncomeEntity | None:
    """
    Returns the IncomeEntity for a user on a specific date.

    Args:
        user_id: The user ID to get income brackets for
        on_date: The date to check the income against

    Returns:
        IncomeEntity or None if there is no income for the user on the given date.
    """
    user_incomes = IncomeRepository().get_user_incomes(user_id)

    if len(user_incomes) == 0:
        return None

    on_date_month = Month(on_date)

    income_in_timeframe = first_or_none(
        user_incomes,
        lambda income: income.start_month <= on_date_month
        and income.end_month >= on_date_month,
    )

    return income_in_timeframe

get_user_income_timeline 

get_user_income_timeline(user_ref)

Returns a timeline of IncomePeriod for a user. The timeline is sorted but not necessarily continuous. IncomePeriods can't overlap.

Parameters:

Name Type Description Default
user_id

The user ID to get income brackets for

 required

Returns:

Type Description
Timeline[IncomePeriod]

A timeline of IncomePeriod for the user.
Source code in components/income/internal/queries/get_user_income.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_user_income_timeline(
    user_ref: str,
) -> Timeline[IncomePeriod]:
    """
    Returns a timeline of IncomePeriod for a user.
    The timeline is sorted but not necessarily continuous.
    IncomePeriods can't overlap.

    Args:
        user_id: The user ID to get income brackets for

    Returns:
        A timeline of IncomePeriod for the user.
    """
    return get_user_income_timelines([user_ref]).get(user_ref, Timeline(periods=()))

has_subscription_price_depending_on_income_brackets 

has_subscription_price_depending_on_income_brackets(
    user_id, on_date
)
Source code in components/income/internal/queries/has_price_depending_on_income_brackets.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
def has_subscription_price_depending_on_income_brackets(
    user_id: str,
    on_date: date,
) -> bool:
    dependencies = get_app_dependency()
    subscription_id = dependencies.get_subscription_id_for_user_on(user_id, on_date)
    has_income_brackets = (
        dependencies.is_subscription_linked_to_income_brackets(subscription_id)
        if subscription_id
        else False
    )

    return bool(has_income_brackets)

is_subscription_price_linked_to_income_brackets 

is_subscription_price_linked_to_income_brackets(
    subscription_id,
)
Source code in components/income/internal/queries/has_price_depending_on_income_brackets.py 
26
27
28
29
30
31
32
33
34
def is_subscription_price_linked_to_income_brackets(
    subscription_id: str,
) -> bool:
    dependencies = get_app_dependency()
    has_income_brackets = dependencies.is_subscription_linked_to_income_brackets(
        subscription_id
    )

    return bool(has_income_brackets)

preload_user_income_timelines 

preload_user_income_timelines(user_refs)

This function precomputes calls to get_user_income_timelines, caches them and returns a new function get_cached_user_income_timeline that can be used instead of the original implementation.

Source code in components/income/internal/queries/get_user_income.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def preload_user_income_timelines(
    user_refs: list[str],
) -> Callable[[str], Timeline[IncomePeriod]]:
    """
    This function precomputes calls to get_user_income_timelines, caches
    them and returns a new function `get_cached_user_income_timeline` that can be
    used instead of the original implementation.
    """
    cached_user_income_timelines = get_user_income_timelines(user_refs)

    def get_cached_user_income_timeline(
        user_ref: str,
    ) -> Timeline[IncomePeriod]:
        try:
            return cached_user_income_timelines[user_ref]
        except KeyError:
            # Fallback to the uncached implementation
            return get_user_income_timeline(
                user_ref=user_ref,
            )

    return get_cached_user_income_timeline