Skip to content

Api reference

components.payroll_tool.public.api

Public API for payroll_tool component.

Re-exports types that other components (countries) need to implement their dependencies.

ActivityType

Bases: AlanBaseEnum

Activity filter applied to the raw payroll changes (framing §5.3).

all class-attribute instance-attribute 

all = 'all'

new_activity_only class-attribute instance-attribute 

new_activity_only = 'new_activity_only'

no_activity_only class-attribute instance-attribute 

no_activity_only = 'no_activity_only'

AdminCalendarRules dataclass 

AdminCalendarRules(
    max_period_weeks=7,
    min_period_weeks=2,
    max_abs_month_shift=2,
    cutoff_freeze_lead_days=0,
    label_change_permission=False,
)

Bases: DataClassJsonMixin

Validation rules applied when a company admin edits the calendar.

cutoff_freeze_lead_days class-attribute instance-attribute 

cutoff_freeze_lead_days = 0

label_change_permission class-attribute instance-attribute 

label_change_permission = False

max_abs_month_shift class-attribute instance-attribute 

max_abs_month_shift = 2

max_period_weeks class-attribute instance-attribute 

max_period_weeks = 7

min_period_weeks class-attribute instance-attribute 

min_period_weeks = 2

BaseCostBreakdown dataclass 

BaseCostBreakdown(
    employee_cost_total,
    employee_cost_taxed,
    employee_cost_untaxed,
    company_cost_total,
    company_cost_taxed,
    company_cost_untaxed,
)

Bases: DataClassJsonMixin

Country-agnostic cost breakdown stored as JSONB on PayrollChange.

Per side (employee, company), *_taxed + *_untaxed = *_total.

All amounts are integers in cents. Negative amounts are valid — they occur on cancellations and regularizations.

Country subclasses extend this with country-specific decompositions (e.g. BE's channel axis: flexben / payslip on the employee side).

company_cost_taxed instance-attribute 

company_cost_taxed

company_cost_total instance-attribute 

company_cost_total

company_cost_untaxed instance-attribute 

company_cost_untaxed

employee_cost_taxed instance-attribute 

employee_cost_taxed

employee_cost_total instance-attribute 

employee_cost_total

employee_cost_untaxed instance-attribute 

employee_cost_untaxed

BaseExemptionType

Bases: AlanBaseEnum

Base enum for exemption types.

Countries extend this with their specific exemption types. Example (FR): class FrExemptionType(BaseExemptionType): # Copy what can be found in components.fr.internal.models.enums.exemption_type.ExemptionType

BasePayload dataclass 

BasePayload()

Base class for all snapshot entry payloads.

BasePayrollChangeCategory

Bases: AlanBaseEnum

Base enum for payroll change categories.

Countries extend this with their specific categories.

BaseProfessionalCategory

Bases: AlanBaseEnum

Base enum for professional categories.

Countries extend this with their specific categories. Example (FR): class FrProfessionalCategory(BaseProfessionalCategory): CADRE = "CADRE" NON_CADRE = "NON_CADRE"

BaseRegimeType

Bases: AlanBaseEnum

Base enum for regime types.

Countries extend this with their specific regime types. Example (FR): class FrRegimeType(BaseRegimeType): GENERAL = "GENERAL" ALSACE_MOSELLE = "ALSACE_MOSELLE"

EmploymentPayload dataclass 

EmploymentPayload(external_employee_id)

Bases: BasePayload

Base payload for employment entries (base employment relationship).

Shared field

external_employee_id: External identifier for the employee (universal)

Countries extend this with their specific fields: - FR: population (FrProfessionalCategory - CADRE/NON_CADRE) - BE: No additional fields needed

external_employee_id instance-attribute 

external_employee_id

EmptyPayload dataclass 

EmptyPayload()

Bases: BasePayload

Payload for empty entries (timeline fully cancelled).

This payload has no fields and represents an empty timeline state. No country-specific extension needed.

EnrollmentPayload dataclass 

EnrollmentPayload(offer_ref=None)

Bases: BasePayload

Base payload for enrollment entries (active relationship for a service).

Shared field

offer_ref: Reference to the offer/contract (universal across countries)

Countries extend this with their specific fields: - FR: regime (FrRegimeType) - BE: coverage_module (str), is_adult_pricing (bool) - ES: TBD

offer_ref class-attribute instance-attribute 

offer_ref = None

EventType

Bases: AlanBaseEnum

Type of change detected between snapshots.

cancel class-attribute instance-attribute 

cancel = 'cancel'

end class-attribute instance-attribute 

end = 'end'

start class-attribute instance-attribute 

start = 'start'

update class-attribute instance-attribute 

update = 'update'

ExemptionPayload dataclass 

ExemptionPayload(exemption_type)

Bases: BasePayload

Base payload for exemption entries (legal or contractual exemptions).

Base field

exemption_type: Type of exemption (countries provide their enum subclass)

Usage: - FR: Use existing ExemptionType enum - BE: Not used (Belgium has no exemptions)

exemption_type instance-attribute 

exemption_type

MarmotCalendarRules dataclass 

MarmotCalendarRules(
    max_period_weeks=7,
    min_period_weeks=2,
    max_abs_month_shift=2,
    cutoff_freeze_lead_days=0,
    label_change_permission=True,
)

Bases: DataClassJsonMixin

Validation rules applied when a Marmot user edits the calendar.

cutoff_freeze_lead_days class-attribute instance-attribute 

cutoff_freeze_lead_days = 0

label_change_permission class-attribute instance-attribute 

label_change_permission = True

max_abs_month_shift class-attribute instance-attribute 

max_abs_month_shift = 2

max_period_weeks class-attribute instance-attribute 

max_period_weeks = 7

min_period_weeks class-attribute instance-attribute 

min_period_weeks = 2

MatchKey module-attribute 

MatchKey = tuple[SnapshotEntryType, ProductType, str]

OptionPayload dataclass 

OptionPayload(option_ref)

Bases: BasePayload

Base payload for option entries (top-ups on top of an enrollment).

Shared field

option_ref: Reference to the option (universal)

Usage: - FR: Supplementary option references

option_ref instance-attribute 

option_ref

PayrollCalendar

Bases: BaseModel

Rolling payroll cutoff calendar — one row per company, JSONB array of periods.

__table_args__ class-attribute instance-attribute 

__table_args__ = {'schema': PAYROLL_TOOL_SCHEMA_NAME}

__tablename__ class-attribute instance-attribute 

__tablename__ = 'payroll_calendar'

admin_rules_config class-attribute instance-attribute 

admin_rules_config = mapped_column(
    DataclassJSONB(AdminCalendarRules, none_as_null=True),
    nullable=True,
)

company_id class-attribute instance-attribute 

company_id = mapped_column(
    String(36), unique=True, nullable=False, index=True
)

country class-attribute instance-attribute 

country = mapped_column(String(2), nullable=False)

current_start_date class-attribute instance-attribute 

current_start_date = mapped_column(Date, nullable=False)

entries class-attribute instance-attribute 

entries = mapped_column(
    DataclassJSONBArray(PayrollPeriodEntry), nullable=False
)

marmot_rules_config class-attribute instance-attribute 

marmot_rules_config = mapped_column(
    DataclassJSONB(MarmotCalendarRules, none_as_null=True),
    nullable=True,
)

recurring_rule class-attribute instance-attribute 

recurring_rule = mapped_column(
    DataclassJSONB(RecurringRule),
    nullable=False,
    default=RecurringRule,
)

upcoming_cutoff_date class-attribute instance-attribute 

upcoming_cutoff_date = mapped_column(
    Date, nullable=False, index=True
)

PayrollDataRequest dataclass 

PayrollDataRequest(
    company_ids,
    period_label,
    product_types=None,
    categories=None,
    excluded_categories=None,
    excluded_categories_company_ids=None,
    activity_type=ActivityType.all,
    operational_scopes=None,
    search=None,
    sort_type=SortType.ascending,
    cursor=None,
    limit=20,
)

Runtime parameters for a Layer A query.

The template (passed alongside the request to the public entry point) contributes the what — columns, granularity, sort, filtering. The request contributes the where — which companies, which period, which scopes, runtime overrides.

The full PayrollTemplate shape is sent over the wire (rather than a template id) so admins can preview unsaved column / sort / filter edits in the UI without persisting them first. Resolved per framing discussion https://github.com/alan-eu/Topics/discussions/32979#discussioncomment-16768041 ⧉.

Framing §5.

Parameters:

Name Type Description Default
company_ids set[str]

Companies to include. Single-country is enforced structurally — one PayrollToolDependency is loaded per app, so every request is scoped to one country by construction (framing §5.7).

 required
period_label date

The payroll period anchor. Matches rows where month == period_label OR regularization_month == period_label. A date because we anticipate weekly periods (CA).

 required
product_types list[str] | None

Restrict to these product types. None = template default.

 None
categories list[str] | None

Restrict to these categories. None = template default.

 None
excluded_categories list[str] | None

Category blacklist. Applied after categories.

 None
excluded_categories_company_ids set[str] | None

When set, excluded_categories applies only to these companies (per-company scope).

 None
activity_type ActivityType

new_activity_only / no_activity_only / all.

 all
operational_scopes dict[str, dict[str, list[str]]] | None

Country-local scope spec. Passed straight to PayrollToolDependency.apply_operational_scope_filter — the country dep decides how to interpret the two-level shape (typically {scope_kind: {field: [values]}}).

 None
search str | None

Free-text search. Hits a hardcoded field allowlist (fullname, email, matricule — see PAY-1848).

 None
sort_type SortType

Row sort direction. Applied to raw Layer A columns; the what to sort on lives in the template (PTSorting).

 ascending
cursor str | None

Opaque keyset-pagination cursor. None = first page.

 None
limit int

Page size. Conservative default — admins can opt up via request override.

 20

activity_type class-attribute instance-attribute 

activity_type = all

categories class-attribute instance-attribute 

categories = None

company_ids instance-attribute 

company_ids

cursor class-attribute instance-attribute 

cursor = None

excluded_categories class-attribute instance-attribute 

excluded_categories = None

excluded_categories_company_ids class-attribute instance-attribute 

excluded_categories_company_ids = None

limit class-attribute instance-attribute 

limit = 20

operational_scopes class-attribute instance-attribute 

operational_scopes = None

period_label instance-attribute 

period_label

product_types class-attribute instance-attribute 

product_types = None

search class-attribute instance-attribute 

search = None

sort_type class-attribute instance-attribute 

sort_type = ascending

PayrollDataResponse dataclass 

PayrollDataResponse(
    rows=list(), next_cursor=None, total_rows=None
)

Response shape returned by get_payroll_data (PAY-1853).

rows is Layer-B-rendered (each row is a dict of column-key → display string). next_cursor is None when the last page has been reached. total_rows is optional — Layer A may skip it when the count query is expensive.

The response deliberately doesn't echo a template id back: the caller sent the full PayrollTemplate alongside the request, so they already know which template produced these rows.

next_cursor class-attribute instance-attribute 

next_cursor = None

rows class-attribute instance-attribute 

rows = field(default_factory=list)

total_rows class-attribute instance-attribute 

total_rows = None

PayrollPeriodSummariesResponse dataclass 

PayrollPeriodSummariesResponse(summaries=list())

summaries class-attribute instance-attribute 

summaries = field(default_factory=list)

PayrollPeriodSummary dataclass 

PayrollPeriodSummary(
    period_label,
    total_changes,
    non_no_change_changes,
    total_employee_cost_cents,
    total_company_cost_cents,
)

Summary counters for one payroll period (framing §7.1).

Parameters:

Name Type Description Default
period_label date

The period anchor (PayrollChange.month).

 required
total_changes int

Total rows in the period.

 required
non_no_change_changes int

Rows whose category is not no_change — "actual activity" count.

 required
total_employee_cost_cents int

Sum of cost_breakdown->>'employee_cost_total'.

 required
total_company_cost_cents int

Sum of cost_breakdown->>'company_cost_total'.

 required

non_no_change_changes instance-attribute 

non_no_change_changes

period_label instance-attribute 

period_label

total_changes instance-attribute 

total_changes

total_company_cost_cents instance-attribute 

total_company_cost_cents

total_employee_cost_cents instance-attribute 

total_employee_cost_cents

PayrollTemplateDataInterface

Placeholder contract for Layer B rendering.

Per-method implementations are registered by the Layer B dispatcher (PAY-1852). This class is intentionally empty until that dispatcher lands.

ProductType

Bases: AlanBaseEnum

Product domains for snapshot entries.

ani class-attribute instance-attribute 

ani = 'ani'

employment class-attribute instance-attribute 

employment = 'employment'

empty class-attribute instance-attribute 

empty = 'empty'

flex_benefits class-attribute instance-attribute 

flex_benefits = 'flex_benefits'

health class-attribute instance-attribute 

health = 'health'

prevoyance class-attribute instance-attribute 

prevoyance = 'prevoyance'

RecurringRule dataclass 

RecurringRule(cutoff_date=20, label_offset=0)

Bases: DataClassJsonMixin

Default pattern for auto-generating calendar entries.

cutoff_date class-attribute instance-attribute 

cutoff_date = 20

label_offset class-attribute instance-attribute 

label_offset = 0

SnapshotEntryData dataclass 

SnapshotEntryData(
    snapshot_date,
    country,
    primary_user_id,
    beneficiary_user_id,
    enrollment_type,
    company_id,
    product_type,
    entry_type,
    start_date,
    end_date,
    entry_data,
    link_id,
    id=None,
)

Bases: Generic[TPayload]

Dataclass representation of a snapshot entry.

This is returned by queries instead of the SnapshotEntry model. The entry_data field is generic and can be specialized by country.

Class Type Parameters:

Name Bound or Constraints Description Default
TPayload

Country-specific payload type (e.g., BeEnrollmentPayload)

 required
Example usage

Global usage with BasePayload

entries: list[SnapshotEntryData[BasePayload]] = get_company_snapshot_entries_as_of(...)

Country-specific usage (in BE component)

BeSnapshotEntryData = SnapshotEntryData[BeEnrollmentPayload | BeEmploymentPayload | BeEmptyPayload] entries: list[BeSnapshotEntryData] = get_company_snapshot_entries_as_of(...)

beneficiary_user_id instance-attribute 

beneficiary_user_id

company_id instance-attribute 

company_id

content_eq 

content_eq(other)

Compare all fields except metadata (snapshot_date, id).

Source code in components/payroll_tool/internal/business_logic/entities/snapshot_entry_data.py 
80
81
82
83
84
85
86
def content_eq(self, other: "SnapshotEntryData[TPayload]") -> bool:
    """Compare all fields except metadata (snapshot_date, id)."""
    return all(
        getattr(self, f.name) == getattr(other, f.name)
        for f in fields(self)
        if f.name not in _NON_CONTENT_FIELDS
    )

country instance-attribute 

country

end_date instance-attribute 

end_date

enrollment_type instance-attribute 

enrollment_type

entry_data instance-attribute 

entry_data

entry_type instance-attribute 

entry_type

id class-attribute instance-attribute 

id = None

link_id

match_key property 

match_key

Unique key for matching entries across timelines.

primary_user_id instance-attribute 

primary_user_id

product_type instance-attribute 

product_type

snapshot_date instance-attribute 

snapshot_date

start_date instance-attribute 

start_date

SnapshotEntryType

Bases: AlanBaseEnum

Generic entry type categories shared across all countries.

employment class-attribute instance-attribute 

employment = 'employment'

empty class-attribute instance-attribute 

empty = 'empty'

enrollment class-attribute instance-attribute 

enrollment = 'enrollment'

exemption class-attribute instance-attribute 

exemption = 'exemption'

option class-attribute instance-attribute 

option = 'option'

suspension class-attribute instance-attribute 

suspension = 'suspension'

SortType

Bases: AlanBaseEnum

Row sort direction for the final Layer A output (framing §6 Step 7).

Direction only — the what to sort on lives in PTSorting.sort_by_data_field (PAY-1842). Works for text and numeric columns alike.

ascending class-attribute instance-attribute 

ascending = 'ascending'

descending class-attribute instance-attribute 

descending = 'descending'

SuspensionPayload dataclass 

SuspensionPayload()

Bases: BasePayload

Base payload for suspension entries (unpaid leave, parental leave, etc.).

Countries extend this with their specific fields as needed.

Example (FR): @dataclass class FrSuspensionPayload(SuspensionPayload): mandatory_coverage: bool | None = None

Usage: - FR: Uses suspensions for unpaid leave tracking - BE: Not used yet

UserTimeline dataclass 

UserTimeline(
    beneficiary_user_id=None,
    company_id=None,
    _entries_by_key=dict(),
)

Immutable collection of snapshot entries keyed by (entry_type, product_type, link_id).

All mutation methods return a new instance. Equality uses content_eq to ignore metadata fields (snapshot_date, id).

__contains__ 

__contains__(key)
Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
110
111
def __contains__(self, key: object) -> bool:
    return key in self._entries_by_key

__eq__ 

__eq__(other)

Content equality: ignores snapshot_date and id on entries.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
119
120
121
122
123
124
125
126
127
128
def __eq__(self, other: object) -> bool:
    """Content equality: ignores snapshot_date and id on entries."""
    if not isinstance(other, UserTimeline):
        return NotImplemented
    if self.keys() != other.keys():
        return False
    return all(
        self._entries_by_key[k].content_eq(other.get(k))  # type: ignore[arg-type]
        for k in self._entries_by_key
    )

__hash__ 

__hash__()
Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
130
131
132
def __hash__(self) -> int:
    # frozen=True expects __hash__; delegate to id since __eq__ is custom
    return id(self)

__iter__ 

__iter__()
Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
113
114
def __iter__(self) -> Iterator[MatchKey]:
    return iter(self._entries_by_key)

__len__ 

__len__()
Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
116
117
def __len__(self) -> int:
    return len(self._entries_by_key)

beneficiary_user_id class-attribute instance-attribute 

beneficiary_user_id = None

company_id class-attribute instance-attribute 

company_id = None

empty classmethod 

empty()

Return an empty timeline (no identity — compatible with any diff).

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
55
56
57
58
@classmethod
def empty(cls) -> UserTimeline:
    """Return an empty timeline (no identity — compatible with any diff)."""
    return cls()

from_entries classmethod 

from_entries(entries)

Build a timeline from a list of entries, keyed by match_key.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def from_entries(
    cls, entries: list[SnapshotEntryData[BasePayload]]
) -> UserTimeline:
    """Build a timeline from a list of entries, keyed by match_key."""
    if not entries:
        return cls()
    first = entries[0]
    for entry in entries[1:]:
        if (
            entry.beneficiary_user_id != first.beneficiary_user_id
            or entry.company_id != first.company_id
        ):
            raise ValueError(
                f"All entries must share the same identity. "
                f"Expected ({first.beneficiary_user_id}, {first.company_id}), "
                f"got ({entry.beneficiary_user_id}, {entry.company_id})"
            )
    return cls(
        beneficiary_user_id=first.beneficiary_user_id,
        company_id=first.company_id,
        _entries_by_key={entry.match_key: entry for entry in entries},
    )

get 

get(key)

Return entry for key, or None.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
60
61
62
def get(self, key: MatchKey) -> SnapshotEntryData[BasePayload] | None:
    """Return entry for key, or None."""
    return self._entries_by_key.get(key)

has_identity 

has_identity()

True if identity fields are set (non-empty timeline).

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
90
91
92
def has_identity(self) -> bool:
    """True if identity fields are set (non-empty timeline)."""
    return self.beneficiary_user_id is not None

keys 

keys()

Return all match keys.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
64
65
66
def keys(self) -> KeysView[MatchKey]:
    """Return all match keys."""
    return self._entries_by_key.keys()

same_identity 

same_identity(other)

Check that both timelines belong to the same (beneficiary, company) pair.

Empty timelines (no identity set) are compatible with any other timeline.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def same_identity(self, other: UserTimeline) -> bool:
    """Check that both timelines belong to the same (beneficiary, company) pair.

    Empty timelines (no identity set) are compatible with any other timeline.
    """
    if not self.has_identity() or not other.has_identity():
        return True  # empty timelines are compatible with any diff
    return (
        self.beneficiary_user_id == other.beneficiary_user_id
        and self.company_id == other.company_id
    )

to_entries 

to_entries()

Return entries sorted by match_key for deterministic output.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
106
107
108
def to_entries(self) -> list[SnapshotEntryData[BasePayload]]:
    """Return entries sorted by match_key for deterministic output."""
    return sorted(self._entries_by_key.values(), key=lambda e: e.match_key)

with_entry 

with_entry(entry)

Return new timeline with entry added or replaced.

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
68
69
70
71
72
73
74
75
76
def with_entry(self, entry: SnapshotEntryData[BasePayload]) -> UserTimeline:
    """Return new timeline with entry added or replaced."""
    new_dict = dict(self._entries_by_key)
    new_dict[entry.match_key] = entry
    return UserTimeline(
        beneficiary_user_id=self.beneficiary_user_id,
        company_id=self.company_id,
        _entries_by_key=new_dict,
    )

without_key 

without_key(key)

Return new timeline with key removed (no-op if absent).

Source code in components/payroll_tool/internal/business_logic/entities/user_timeline.py 
78
79
80
81
82
83
84
85
86
87
88
def without_key(self, key: MatchKey) -> UserTimeline:
    """Return new timeline with key removed (no-op if absent)."""
    if key not in self._entries_by_key:
        return self
    new_dict = dict(self._entries_by_key)
    del new_dict[key]
    return UserTimeline(
        beneficiary_user_id=self.beneficiary_user_id,
        company_id=self.company_id,
        _entries_by_key=new_dict,
    )

fill_payroll_calendar_to_target 

fill_payroll_calendar_to_target(
    existing_entries,
    recurring_rule,
    today,
    min_period_weeks=2,
)

Extend the calendar to TARGET_ENTRY_COUNT entries using the recurring rule.

Returns the full calendar (existing + newly generated). If already at or above target, returns existing_entries unchanged.

Generates entries strictly after the last entry's cutoff, or strictly after today when the calendar is empty. If the first generated entry would be less than min_period_weeks from the seed, it is pushed to the following month.

Source code in components/payroll_tool/internal/business_logic/actions/payroll_calendar_generator.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def fill_payroll_calendar_to_target(
    existing_entries: list[PayrollPeriodEntry],
    recurring_rule: RecurringRule,
    today: date,
    min_period_weeks: int = 2,
) -> list[PayrollPeriodEntry]:
    """Extend the calendar to TARGET_ENTRY_COUNT entries using the recurring rule.

    Returns the full calendar (existing + newly generated). If already at or above
    target, returns existing_entries unchanged.

    Generates entries strictly after the last entry's cutoff, or strictly after
    today when the calendar is empty. If the first generated entry would be less
    than min_period_weeks from the seed, it is pushed to the following month.
    """
    count = TARGET_ENTRY_COUNT - len(existing_entries)
    if count <= 0:
        return existing_entries

    seed_cutoff = existing_entries[-1].cutoff_date if existing_entries else today

    return existing_entries + _generate_from_rule(
        seed_cutoff, recurring_rule, count, min_period_weeks
    )

get_company_snapshot_entries_as_of 

get_company_snapshot_entries_as_of(
    company_id, as_of, primary_user_ids=None
)

Retrieve snapshot entries for a company as of a specific date.

Returns entries from the most recent snapshot date <= as_of. Optionally filters by primary_user_ids for performance.

Parameters:

Name Type Description Default
company_id str

Company identifier

 required
as_of date

Date to retrieve snapshot as of (inclusive)

 required
primary_user_ids list[str] | None

Optional filter to only return entries for these users

 None
Source code in components/payroll_tool/internal/business_logic/queries/get_company_snapshot_entries.py 
17
18
19
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
def get_company_snapshot_entries_as_of(
    company_id: str,
    as_of: date,
    primary_user_ids: list[str] | None = None,
) -> list[SnapshotEntryData[BasePayload]]:
    """Retrieve snapshot entries for a company as of a specific date.

    Returns entries from the most recent snapshot date <= as_of.
    Optionally filters by primary_user_ids for performance.

    Args:
        company_id: Company identifier
        as_of: Date to retrieve snapshot as of (inclusive)
        primary_user_ids: Optional filter to only return entries for these users
    """
    # Find the most recent snapshot date <= as_of for this company
    latest_snapshot_date_stmt = (
        select(SnapshotEntry.snapshot_date)
        .where(
            SnapshotEntry.company_id == company_id,
            SnapshotEntry.snapshot_date <= as_of,
        )
        .order_by(SnapshotEntry.snapshot_date.desc())
        .limit(1)
    )
    result = current_session.execute(latest_snapshot_date_stmt)
    latest_snapshot_date = result.scalar_one_or_none()

    # No snapshots found for this company before the given date
    if latest_snapshot_date is None:
        return []

    # Load all entries for that snapshot date
    entries_stmt = select(SnapshotEntry).where(
        SnapshotEntry.company_id == company_id,
        SnapshotEntry.snapshot_date == latest_snapshot_date,
    )
    if primary_user_ids is not None:
        entries_stmt = entries_stmt.where(
            SnapshotEntry.primary_user_id.in_(primary_user_ids)
        )
    entries = current_session.execute(entries_stmt).scalars().all()

    # Convert models to dataclasses
    return [_model_to_dataclass(entry) for entry in entries]

get_payroll_data 

get_payroll_data(request, template, *, locale='en_US')

Public 3b entry point — 3c calls this.

Resolves the country dependency from the current Flask app and delegates to :func:run_pipeline.

Parameters:

Name Type Description Default
request PayrollDataRequest

Runtime parameters.

 required
template PayrollTemplate

Resolved template. Template storage is Part 3a's concern; during 3b dev pass a template_mocks instance.

 required
locale Locale | str

Keyword-only. Defaults to "en_US" — callers that care about French rendering should pass "fr_FR" explicitly.

 'en_US'

Returns:

Type Description
PayrollDataResponse

PayrollDataResponse.
Source code in components/payroll_tool/internal/business_logic/queries/data/pipeline.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
def get_payroll_data(
    request: PayrollDataRequest,
    template: PayrollTemplate,
    *,
    locale: Locale | str = "en_US",
) -> PayrollDataResponse:
    """
    Public 3b entry point — 3c calls this.

    Resolves the country dependency from the current Flask app and delegates
    to :func:`run_pipeline`.

    Args:
        request: Runtime parameters.
        template: Resolved template. Template storage is Part 3a's concern;
            during 3b dev pass a ``template_mocks`` instance.
        locale: Keyword-only. Defaults to ``"en_US"`` — callers that care
            about French rendering should pass ``"fr_FR"`` explicitly.

    Returns:
        ``PayrollDataResponse``.
    """
    from components.payroll_tool.public.dependencies import get_app_dependency

    return run_pipeline(request, template, get_app_dependency(), locale=locale)

get_payroll_period_summaries 

get_payroll_period_summaries(request, template=None)

Return per-period counters for a 3c landing page.

Applies every filter get_payroll_data understands except period_label — the summary spans every period available to the request's companies. operational_scopes routes through the country dep so country-local scope rules apply identically to the main pipeline.

Parameters:

Name Type Description Default
request PayrollDataRequest

Runtime parameters. period_label, cursor, limit, and search are ignored.

 required
template PayrollTemplate | None

Optional resolved template. When provided, its filtering rules (product-type / category allowlists and exclusions) are merged with request-level filters (request adds, never removes).

 None

Returns:

Type Description
PayrollPeriodSummariesResponse

PayrollPeriodSummariesResponse.
Source code in components/payroll_tool/internal/business_logic/queries/data/period_summary.py 
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
def get_payroll_period_summaries(
    request: PayrollDataRequest,
    template: PayrollTemplate | None = None,
) -> PayrollPeriodSummariesResponse:
    """
    Return per-period counters for a 3c landing page.

    Applies every filter ``get_payroll_data`` understands **except**
    ``period_label`` — the summary spans every period available to the
    request's companies. ``operational_scopes`` routes through the country
    dep so country-local scope rules apply identically to the main pipeline.

    Args:
        request: Runtime parameters. ``period_label``, ``cursor``,
            ``limit``, and ``search`` are ignored.
        template: Optional resolved template. When provided, its filtering
            rules (product-type / category allowlists and exclusions) are
            merged with request-level filters (request adds, never removes).

    Returns:
        ``PayrollPeriodSummariesResponse``.
    """
    from components.payroll_tool.public.dependencies import get_app_dependency

    return _run(request, template, get_app_dependency())

components.payroll_tool.public.categorize

Public surface for the MatchGroup categorizer.

categorize_match_group 

categorize_match_group(group, target_month, category_enum)

Apply the rule ladder. See module docstring.

Source code in components/payroll_tool/internal/business_logic/queries/categorize_match_group.py 
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 categorize_match_group(
    group: MatchGroup,
    target_month: Month,
    category_enum: type[BasePayrollChangeCategory],
) -> list[BasePayrollChangeCategory]:
    """Apply the rule ladder. See module docstring."""
    cat = category_enum.__members__  # name → enum member
    key_month, _link_id, _product_type, _service = group.key

    # Rule 1: any cancel event short-circuits.
    if any(event.event_type == EventType.cancel for event in group.events):
        return [cat["cancel"]]

    # Rule 2: collect deduped event-derived categories in event order.
    categories: list[BasePayrollChangeCategory] = []
    seen: set[BasePayrollChangeCategory] = set()
    for event in group.events:
        category_name = _EVENT_TYPE_TO_CATEGORY_NAME.get(event.event_type)
        if category_name is None:
            continue
        category = cat[category_name]
        if category in seen:
            continue
        categories.append(category)
        seen.add(category)
    if categories:
        return categories

    # Rule 3: price change for multi-line regularizations that net to 0 days
    # (only reachable when no event drove a category).
    is_regularization = key_month < target_month
    if is_regularization and len(group.entries) > 1:
        # PremiumEntry.num_days reads components[0].num_days. Safe here:
        # within a MatchGroupKey (enrollment, product, service, month), all
        # components of a PE share num_days — intra-PE splits are tax /
        # contribution shape only, not day-count.
        total_num_days = sum(entry.num_days for entry in group.entries)
        total_amount = sum(entry.total_amount() for entry in group.entries)
        if total_num_days == 0 and total_amount != 0:
            return [cat["price_change"]]

    # Rule 4: no-change fallback (typical entries-only group).
    return [cat["no_change"]]

components.payroll_tool.public.commands

components.payroll_tool.public.dependencies

COMPONENT_NAME module-attribute 

COMPONENT_NAME = 'payroll_tool'

PayrollToolDependency

Bases: ABC

PayrollToolDependency defines the interface that apps using the payroll_tool component need to implement. This interface is used to take snapshots of payroll data.

apply_operational_scope_filter 

apply_operational_scope_filter(q, operational_scopes)

Apply country-local operational scope filtering.

Default is a passthrough — countries that don't model operational scopes leave this as-is.

Parameters:

Name Type Description Default
q Select[Any]

The Layer A select to filter.

 required
operational_scopes dict[str, dict[str, list[str]]]

Country-local scope spec from PayrollDataRequest.operational_scopes. Two-level shape (typically {scope_kind: {field: [values]}}) — the country dep decides how to interpret it.

 required

Returns:

Type Description
Select[Any]

The filtered Select (or the input unchanged).
Source code in components/payroll_tool/public/dependencies.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def apply_operational_scope_filter(
    self,
    q: "Select[Any]",
    operational_scopes: dict[str, dict[str, list[str]]],  # noqa: ARG002
) -> "Select[Any]":
    """
    Apply country-local operational scope filtering.

    Default is a passthrough — countries that don't model operational
    scopes leave this as-is.

    Args:
        q: The Layer A select to filter.
        operational_scopes: Country-local scope spec from
            ``PayrollDataRequest.operational_scopes``. Two-level shape
            (typically ``{scope_kind: {field: [values]}}``) — the country
            dep decides how to interpret it.

    Returns:
        The filtered ``Select`` (or the input unchanged).
    """
    return q

can_company_edit_payroll_calendar 

can_company_edit_payroll_calendar(company_id)

Whether this company may edit its payroll cutoff calendar.

Default: always allowed. Countries with ALM delivery-group restrictions (FR) override this so only a group's designated admin company can edit.

Source code in components/payroll_tool/public/dependencies.py 
368
369
370
371
372
373
374
375
376
377
def can_company_edit_payroll_calendar(
    self,
    company_id: str,  # noqa: ARG002 — consumed by country overrides
) -> bool:
    """Whether this company may edit its payroll cutoff calendar.

    Default: always allowed. Countries with ALM delivery-group restrictions
    (FR) override this so only a group's designated admin company can edit.
    """
    return True

category_enum_class instance-attribute 

category_enum_class

Country-specific BasePayrollChangeCategory subclass. Passed to build_in_memory_payroll_changes so the categorizer resolves category names against the country's enum (BE: BePayrollChangeCategory).

chunk_size class-attribute instance-attribute 

chunk_size = 10000

compute_cost_breakdown abstractmethod 

compute_cost_breakdown(*, premium_entries)

Aggregate the given premium entries into one cost breakdown.

All entries share a (enrollment, month) group — one PayrollChange row represents that group. The returned breakdown is persisted as JSONB on PayrollChange.cost_breakdown.

Source code in components/payroll_tool/public/dependencies.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@abstractmethod
def compute_cost_breakdown(
    self,
    *,
    premium_entries: list[PremiumEntry],
) -> "BaseCostBreakdown":
    """Aggregate the given premium entries into one cost breakdown.

    All entries share a (enrollment, month) group — one PayrollChange
    row represents that group. The returned breakdown is persisted as
    JSONB on PayrollChange.cost_breakdown.
    """
    raise NotImplementedError(
        "compute_cost_breakdown must be implemented by the dependency"
    )

compute_match_key 

compute_match_key(entry)

Derive the matcher key from a PremiumEntry.

Used by generate_payroll_changes to pair PEs with events. Country-specific because the (link_id, product_type, service) mapping depends on each country's PE shape. Default raises so non-overriding countries fail loudly.

Source code in components/payroll_tool/public/dependencies.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def compute_match_key(
    self,
    entry: PremiumEntry,
) -> "MatchGroupKey":
    """Derive the matcher key from a ``PremiumEntry``.

    Used by ``generate_payroll_changes`` to pair PEs with events.
    Country-specific because the (link_id, product_type, service)
    mapping depends on each country's PE shape. Default raises so
    non-overriding countries fail loudly.
    """
    raise NotImplementedError(
        "compute_match_key must be implemented by the dependency"
    )

cost_breakdown_class instance-attribute 

cost_breakdown_class

Country-specific BaseCostBreakdown subclass. Used by InMemoryPayrollChange.from_db_payroll_change to deserialize the cost_breakdown JSONB column back into a typed dataclass (BE: BeCostBreakdown).

country instance-attribute 

country

2-letter country code persisted on each PayrollChange row. Passed to build_in_memory_payroll_changes as the country field (BE: "BE").

get_active_companies_for_snapshot abstractmethod 

get_active_companies_for_snapshot()

Returns the list of company IDs eligible for snapshot generation.

Companies are eligible if they have an active contract in the past 3 months or in the future.

Returns:

Type Description
list[str]

List of company IDs (as strings) to take snapshots for
Source code in components/payroll_tool/public/dependencies.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@abstractmethod
def get_active_companies_for_snapshot(self) -> list[str]:
    """
    Returns the list of company IDs eligible for snapshot generation.

    Companies are eligible if they have an active contract in the past 3 months
    or in the future.

    Returns:
        List of company IDs (as strings) to take snapshots for
    """
    raise NotImplementedError(
        "get_active_companies_for_snapshot must be implemented by the dependency"
    )

get_company_subquery 

get_company_subquery(company_ids)

Expose per-company metadata (display name, sector, account) for a 3b query.

Countries with no per-company UI needs may return None.

Parameters:

Name Type Description Default
company_ids set[str]

Companies to scope the subquery to.

 required
Source code in components/payroll_tool/public/dependencies.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def get_company_subquery(
    self,
    company_ids: set[str],  # noqa: ARG002 — consumed by country overrides
) -> "CompanySubquery | None":
    """
    Expose per-company metadata (display name, sector, account) for a 3b
    query.

    Countries with no per-company UI needs may return ``None``.

    Args:
        company_ids: Companies to scope the subquery to.
    """
    return None

get_created_at_lower_bound 

get_created_at_lower_bound(*, company_id, target_month)

Lower bound on created_at for events / PEs in this cycle.

Used by generate_payroll_changes: rows with created_at > bound appeared after the previous cycle was finalized and belong to the current cycle.

Implementation returns the wall-clock instant the previous cycle's pay.csv was written. None means no prior cycle exists for this company — caller treats that as "no lower bound" (include all history).

Parameters:

Name Type Description Default
company_id str

company being processed.

 required
target_month Month

the cycle being generated. Only PRIOR months count toward the bound, so a re-run picks up the previous cycle, not itself.

 required
Source code in components/payroll_tool/public/dependencies.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def get_created_at_lower_bound(
    self,
    *,
    company_id: str,  # noqa: ARG002 — consumed by country overrides
    target_month: "Month",  # noqa: ARG002 — consumed by country overrides
) -> datetime | None:
    """Lower bound on ``created_at`` for events / PEs in this cycle.

    Used by ``generate_payroll_changes``: rows with
    ``created_at > bound`` appeared after the previous cycle was
    finalized and belong to the current cycle.

    Implementation returns the wall-clock instant the previous cycle's
    pay.csv was written. ``None`` means no prior cycle exists for this
    company — caller treats that as "no lower bound" (include all
    history).

    Args:
        company_id: company being processed.
        target_month: the cycle being generated. Only PRIOR months count
            toward the bound, so a re-run picks up the previous cycle,
            not itself.
    """
    return None

get_created_at_upper_bound 

get_created_at_upper_bound(*, company_id, target_month)

Upper bound on created_at for events / PEs in this cycle.

Pinned at the wall-clock instant the cycle for target_month was finalized — proxied by target_month's pay.csv created_at. Once the paycsv exists, re-runs of the cycle reuse this bound so the run is idempotent (rows from later cycles don't leak in).

None means no paycsv exists yet for target_month (= the cycle hasn't been finalized) — caller falls back to utcnow().

Parameters:

Name Type Description Default
company_id str

company being processed.

 required
target_month Month

the cycle being generated.

 required
Source code in components/payroll_tool/public/dependencies.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def get_created_at_upper_bound(
    self,
    *,
    company_id: str,  # noqa: ARG002 — consumed by country overrides
    target_month: "Month",  # noqa: ARG002 — consumed by country overrides
) -> datetime | None:
    """Upper bound on ``created_at`` for events / PEs in this cycle.

    Pinned at the wall-clock instant the cycle for ``target_month`` was
    finalized — proxied by ``target_month``'s pay.csv ``created_at``.
    Once the paycsv exists, re-runs of the cycle reuse this bound so
    the run is idempotent (rows from later cycles don't leak in).

    ``None`` means no paycsv exists yet for ``target_month`` (= the
    cycle hasn't been finalized) — caller falls back to ``utcnow()``.

    Args:
        company_id: company being processed.
        target_month: the cycle being generated.
    """
    return None

get_current_employees abstractmethod 

get_current_employees(company_id)

Returns the list of employee user IDs for current employees at the company.

This is used as the base list for snapshot generation (section 7.1.B step 1).

Parameters:

Name Type Description Default
company_id str

The ID of the company to get employees for

 required

Returns:

Type Description
list[str]

List of employee user IDs (strings)
Source code in components/payroll_tool/public/dependencies.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@abstractmethod
def get_current_employees(self, company_id: str) -> list[str]:
    """
    Returns the list of employee user IDs for current employees at the company.

    This is used as the base list for snapshot generation (section 7.1.B step 1).

    Args:
        company_id: The ID of the company to get employees for

    Returns:
        List of employee user IDs (strings)
    """
    raise NotImplementedError(
        "get_current_employees must be implemented by the dependency"
    )

get_employment_detail_subquery 

get_employment_detail_subquery(company_ids)

Expose employment metadata (matricule, population, …) for a 3b query.

Countries with no country-local employment table return None and any template that references an employment-sourced field will skip the join (leaving those columns empty).

Parameters:

Name Type Description Default
company_ids set[str]

Companies to scope the subquery to.

 required
Source code in components/payroll_tool/public/dependencies.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get_employment_detail_subquery(
    self,
    company_ids: set[str],  # noqa: ARG002 — consumed by country overrides
) -> "EmploymentDetailSubquery | None":
    """
    Expose employment metadata (matricule, population, …) for a 3b query.

    Countries with no country-local employment table return ``None`` and
    any template that references an employment-sourced field will skip
    the join (leaving those columns empty).

    Args:
        company_ids: Companies to scope the subquery to.
    """
    return None

get_enrollment_ids_for_company 

get_enrollment_ids_for_company(*, company_id)

All enrollment ids belonging to company_id.

Used by generate_payroll_changes as the universe of link_ids the orchestrator iterates over (and as the enrollment_ids filter passed to PremiumDependency.get_entries_for_payroll).

Returns stringified ids — PayrollChange.link_id is String(36) and country-native types differ (BE: UUID; FR: int). Country overrides stringify before returning.

Default returns [] so countries not wired to the orchestrator stay bootable. BE overrides; FR/CA/ES override when they migrate.

Source code in components/payroll_tool/public/dependencies.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def get_enrollment_ids_for_company(
    self,
    *,
    company_id: str,  # noqa: ARG002 — consumed by country overrides
) -> list[str]:
    """All enrollment ids belonging to ``company_id``.

    Used by ``generate_payroll_changes`` as the universe of link_ids
    the orchestrator iterates over (and as the ``enrollment_ids`` filter
    passed to ``PremiumDependency.get_entries_for_payroll``).

    Returns stringified ids — ``PayrollChange.link_id`` is ``String(36)``
    and country-native types differ (BE: UUID; FR: int). Country
    overrides stringify before returning.

    Default returns ``[]`` so countries not wired to the orchestrator
    stay bootable. BE overrides; FR/CA/ES override when they migrate.
    """
    return []

get_extra_interface_fields 

get_extra_interface_fields()

Return country-local extensions to the 3b field registry.

Used for country-specific data_fields that don't belong in the global registry (e.g. be_contract_type). Default is empty.

Return type is list[Any] here because the concrete InterfaceFieldDefinition is defined by the field registry in the next task (PAY-1844) and importing it here would couple this public contract to an internal module. Callers that care about the concrete type can cast the return value.

Source code in components/payroll_tool/public/dependencies.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def get_extra_interface_fields(self) -> list[Any]:
    """
    Return country-local extensions to the 3b field registry.

    Used for country-specific data_fields that don't belong in the global
    registry (e.g. ``be_contract_type``). Default is empty.

    Return type is ``list[Any]`` here because the concrete
    ``InterfaceFieldDefinition`` is defined by the field registry in the
    next task (PAY-1844) and importing it here would couple this public
    contract to an internal module. Callers that care about the concrete
    type can cast the return value.
    """
    return []

get_last_pay_csv_generation_date 

get_last_pay_csv_generation_date(company_id)

Return the date of the most recent pay.csv generation for a company.

Each country derives this from its own payroll model (FR: max MonthlyPayrollInformation.upper_limit_datetime). Default returns None until the country override is wired up.

Source code in components/payroll_tool/public/dependencies.py 
214
215
216
217
218
219
220
221
222
223
224
def get_last_pay_csv_generation_date(
    self,
    company_id: str,  # noqa: ARG002 — consumed by country overrides
) -> date | None:
    """Return the date of the most recent pay.csv generation for a company.

    Each country derives this from its own payroll model
    (FR: max MonthlyPayrollInformation.upper_limit_datetime).
    Default returns None until the country override is wired up.
    """
    return None

get_legacy_shadow_pay_csvs 

get_legacy_shadow_pay_csvs(company_id, period_label)

Return the country's legacy pay-CSV(s) for a (company, period), for shadow-diff parity validation against the new global export.

Only the country knows how to produce its legacy export, so it lives on the dependency. The payroll_tool shadow-diff command (framing §12) diffs each returned CSV against the new export before BE cutover.

BE returns one entry per health_contract active in the period (legacy is per-contract). Countries with no legacy export to diff against return None — the command reports the country as unsupported.

Parameters:

Name Type Description Default
company_id str

Company to produce legacy CSV(s) for.

 required
period_label date

Payroll period (first day of the month for monthly periods).

 required
Source code in components/payroll_tool/public/dependencies.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def get_legacy_shadow_pay_csvs(
    self,
    company_id: str,  # noqa: ARG002 — consumed by country overrides
    period_label: date,  # noqa: ARG002
) -> "list[LegacyShadowPayCsv] | None":
    """
    Return the country's legacy pay-CSV(s) for a (company, period), for
    shadow-diff parity validation against the new global export.

    Only the country knows how to produce its legacy export, so it lives
    on the dependency. The ``payroll_tool shadow-diff`` command (framing
    §12) diffs each returned CSV against the new export before BE cutover.

    BE returns one entry per ``health_contract`` active in the period
    (legacy is per-contract). Countries with no legacy export to diff
    against return ``None`` — the command reports the country as
    unsupported.

    Args:
        company_id: Company to produce legacy CSV(s) for.
        period_label: Payroll period (first day of the month for monthly
            periods).
    """
    return None

get_payload_types abstractmethod 

get_payload_types()

Returns the country-specific payload type mapping.

This mapping defines which payload class to use for each SnapshotEntryType. Each country provides their own extended payload classes.

Returns:

Type Description
dict[SnapshotEntryType, type[BasePayload]]

Dictionary mapping SnapshotEntryType to country-specific payload classes
Source code in components/payroll_tool/public/dependencies.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@abstractmethod
def get_payload_types(self) -> "dict[SnapshotEntryType, type[BasePayload]]":
    """
    Returns the country-specific payload type mapping.

    This mapping defines which payload class to use for each SnapshotEntryType.
    Each country provides their own extended payload classes.

    Returns:
        Dictionary mapping SnapshotEntryType to country-specific payload classes
    """
    raise NotImplementedError(
        "get_payload_types must be implemented by the dependency"
    )

get_user_ids_by_link(link_ids, session)

Resolve link_id → (primary_user_id, beneficiary_user_id) per link.

Used by generate_payroll_changes for no_change groups — groups with no events, where PremiumEntry does not carry user_ids. Event-bearing groups source the pair from the event itself and don't call this hook. Reads must go through the passed session (the worker's transaction), never current_session.

Default returns {}. Countries override to query their own enrollment table (BE: get_user_ids_by_enrollment).

Source code in components/payroll_tool/public/dependencies.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def get_user_ids_by_link(
    self,
    link_ids: list[str],  # noqa: ARG002 — consumed by country overrides
    session: "Session",  # noqa: ARG002 — consumed by country overrides
) -> Mapping[str, tuple[str, str]]:
    """Resolve ``link_id → (primary_user_id, beneficiary_user_id)`` per link.

    Used by ``generate_payroll_changes`` for ``no_change`` groups —
    groups with no events, where ``PremiumEntry`` does not carry
    user_ids. Event-bearing groups source the pair from the event
    itself and don't call this hook. Reads must go through the passed
    ``session`` (the worker's transaction), never ``current_session``.

    Default returns ``{}``. Countries override to query their own
    enrollment table (BE: ``get_user_ids_by_enrollment``).
    """
    return {}

parse_link_ids(link_ids)

Convert stringified link_ids back to the country's native id type.

generate_payroll_changes keeps link_ids as str end-to-end (PayrollChange.link_id is String(36); events surface them as strings too). The global PE fetch — PremiumAggregationService. get_premiums_for_payroll — takes list[int | UUID]. The country dep is the only place that knows whether to call UUID(s) or int(s), so the conversion lives here.

Default raises so non-overriding countries fail loudly when the orchestrator tries to dispatch work to them.

Source code in components/payroll_tool/public/dependencies.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def parse_link_ids(self, link_ids: list[str]) -> list[int | UUID]:
    """Convert stringified link_ids back to the country's native id type.

    ``generate_payroll_changes`` keeps link_ids as ``str`` end-to-end
    (``PayrollChange.link_id`` is ``String(36)``; events surface them
    as strings too). The global PE fetch — ``PremiumAggregationService.
    get_premiums_for_payroll`` — takes ``list[int | UUID]``. The country
    dep is the only place that knows whether to call ``UUID(s)`` or
    ``int(s)``, so the conversion lives here.

    Default raises so non-overriding countries fail loudly when the
    orchestrator tries to dispatch work to them.
    """
    raise NotImplementedError(
        "parse_link_ids must be implemented by the dependency"
    )

take_snapshot abstractmethod 

take_snapshot(
    company_id, primary_user_ids, snapshot_params=None
)

Build snapshot entry data for given employees. Does NOT persist.

Parameters:

Name Type Description Default
company_id str

Company to snapshot.

 required
primary_user_ids list[str]

Primary employee user IDs.

 required
snapshot_params dict[str, Any] | None

Country-specific snapshot options interpreted by the concrete dependency implementation.

 None
Source code in components/payroll_tool/public/dependencies.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@abstractmethod
def take_snapshot(
    self,
    company_id: str,
    primary_user_ids: list[str],
    snapshot_params: dict[str, Any] | None = None,
) -> "list[SnapshotEntryData[BasePayload]]":
    """Build snapshot entry data for given employees. Does NOT persist.

    Args:
        company_id: Company to snapshot.
        primary_user_ids: Primary employee user IDs.
        snapshot_params: Country-specific snapshot options interpreted by
            the concrete dependency implementation.
    """
    raise NotImplementedError("take_snapshot must be implemented by the dependency")

get_app_dependency 

get_app_dependency()

Retrieves at runtime the payroll_tool dependency set by set_app_dependency

Source code in components/payroll_tool/public/dependencies.py 
408
409
410
411
412
413
414
415
416
417
418
def get_app_dependency() -> PayrollToolDependency:
    """Retrieves at runtime the payroll_tool dependency set by set_app_dependency"""
    from typing import cast

    from flask import current_app

    app = cast("CustomFlask", current_app)
    return cast(
        "PayrollToolDependency",
        app.get_component_dependency(COMPONENT_NAME),
    )

set_app_dependency 

set_app_dependency(dependency)

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

Source code in components/payroll_tool/public/dependencies.py 
395
396
397
398
399
400
401
402
403
404
405
def set_app_dependency(dependency: PayrollToolDependency) -> None:
    """
    Sets the payroll_tool dependency to the app so it can be accessed within this component at runtime
    """
    from typing import cast

    from flask import current_app

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

components.payroll_tool.public.entities

Public dataclasses exposed to apps using the payroll_tool component.

Holds two groups of contracts:

  • PayrollTemplateDataInterface — placeholder contract for Layer B rendering, filled by the Layer B dispatcher (PAY-1852).
  • Country-dependency subquery shapes — EmploymentDetailSubquery and CompanySubquery. A country-specific PayrollToolDependency returns these so 3b's Layer A can join against country-local tables without importing country ORM models. Concrete select / column expressions are built by the country; 3b consumes them as opaque FromClause / ColumnElement[Any].

Framing: https://github.com/alan-eu/Topics/discussions/32979 ⧉ (§5.5 dependency contract).

CompanySubquery dataclass 

CompanySubquery(
    selectable,
    company_id_col,
    display_name_col,
    sector_col=None,
    account_id_col=None,
)

Country-provided subquery exposing per-company metadata.

Joined against PayrollChange.company_id when a 3b template references any company_* data field.

Parameters:

Name Type Description Default
selectable FromClause

The FromClause to join against.

 required
company_id_col ColumnElement[Any]

Column exposing the company ID.

 required
display_name_col ColumnElement[Any]

Column exposing the company display name.

 required
sector_col ColumnElement[Any] | None

Optional column exposing the sector (template scoping). None when the country does not model sector on the company (e.g. BE).

 None
account_id_col ColumnElement[Any] | None

Optional column exposing the account ID (template scoping). None when the country does not expose one.

 None

account_id_col class-attribute instance-attribute 

account_id_col = None

company_id_col instance-attribute 

company_id_col

display_name_col instance-attribute 

display_name_col

sector_col class-attribute instance-attribute 

sector_col = None

selectable instance-attribute 

selectable

EmploymentDetailSubquery dataclass 

EmploymentDetailSubquery(
    selectable,
    user_id_col,
    company_id_col,
    external_employee_id_col,
    population_col=None,
)

Country-provided subquery exposing per-user employment metadata.

The country-specific PayrollToolDependency.get_employment_detail_subquery returns this (or None when the country has no employment detail). 3b's Layer A joins selectable against PayrollChange on user_id_col == PayrollChange.primary_user_id and company_id_col == PayrollChange.company_id.

Parameters:

Name Type Description Default
selectable FromClause

The FromClause to join against (typically a .subquery()).

 required
user_id_col ColumnElement[Any]

Column exposing the primary user ID.

 required
company_id_col ColumnElement[Any]

Column exposing the company ID.

 required
external_employee_id_col ColumnElement[Any]

Column exposing the country-local matricule. On BE this maps to BeEmployment.payroll_id.

 required
population_col ColumnElement[Any] | None

Optional column for the employee's population / sub-group — None when the country has no such concept (e.g. BE).

 None

company_id_col instance-attribute 

company_id_col

external_employee_id_col instance-attribute 

external_employee_id_col

population_col class-attribute instance-attribute 

population_col = None

selectable instance-attribute 

selectable

user_id_col instance-attribute 

user_id_col

LegacyShadowPayCsv dataclass 

LegacyShadowPayCsv(
    label, csv_text, delimiter=";", locale=None
)

One country-legacy pay-CSV, for shadow-diff parity validation.

Parameters:

Name Type Description Default
label str

Which legacy artifact this is (BE: the health_contract id) — shown in the diff report.

 required
csv_text str

The decoded legacy CSV.

 required
delimiter str

Its field delimiter (BE: ";"); each side is parsed with its own so serialization differences aren't content mismatches.

 ';'
locale str | None

Locale the legacy CSV used; the new side is rendered in it so headers/values compare like-for-like. None falls back to fr_FR.

 None

csv_text instance-attribute 

csv_text

delimiter class-attribute instance-attribute 

delimiter = ';'

label instance-attribute 

label

locale class-attribute instance-attribute 

locale = None

PayrollTemplateDataInterface

Placeholder contract for Layer B rendering.

Per-method implementations are registered by the Layer B dispatcher (PAY-1852). This class is intentionally empty until that dispatcher lands.

components.payroll_tool.public.enums

Shared enums for the payroll tool — single source of truth across models and 3b pipeline.

ColumnAggregation

Bases: AlanBaseEnum

How a column's per-group values are combined.

Used by both the ORM layer (PayrollColumnDefinition.aggregation) and the 3b data-aggregation pipeline (grouping, pagination, Layer B dispatch).

The count_rows member is named to avoid shadowing str.count on the AlanBaseEnum (str, Enum) base. The wire value stays "count".

Members:

  • sum: None treated as 0, then summed.
  • count_rows: number of rows in the group (including None values).
  • concat: sorted unique stringified non-null values joined with ", ".
  • min / max: min/max of non-null values.
  • null: pass-through — all non-null values in the group must be equal; returns that value or None; raises on inconsistency.

concat class-attribute instance-attribute 

concat = 'concat'

count_rows class-attribute instance-attribute 

count_rows = 'count'

max class-attribute instance-attribute 

max = 'max'

min class-attribute instance-attribute 

min = 'min'

null class-attribute instance-attribute 

null = 'null'

sum class-attribute instance-attribute 

sum = 'sum'

TemplateGranularity

Bases: AlanBaseEnum

Row granularity for a payroll template / Layer A output.

  • per_change — one row per PayrollChange.
  • per_enrollment — one row per (beneficiary, link_id, company).
  • per_member — one row per (beneficiary, company).
  • per_employee — one row per (primary_user, company).

per_change class-attribute instance-attribute 

per_change = 'per_change'

per_employee class-attribute instance-attribute 

per_employee = 'per_employee'

per_enrollment class-attribute instance-attribute 

per_enrollment = 'per_enrollment'

per_member class-attribute instance-attribute 

per_member = 'per_member'

components.payroll_tool.public.matching

Public surface for the event/premium-entry matcher.

MatchGroup dataclass 

MatchGroup(key, events, entries)

All events + entries sharing one MatchGroupKey.

Tuples (not lists) so the dataclass stays hashable.

Invariant: entries is non-empty. Events without a matching premium line never reach a MatchGroup — they're deferred by the matcher.

__post_init__ 

__post_init__()
Source code in components/payroll_tool/internal/business_logic/queries/entities/match_group.py 
56
57
58
59
60
61
def __post_init__(self) -> None:
    if not self.entries:
        raise ValueError(
            "MatchGroup requires at least one PremiumEntry; "
            "events-only groups are deferred upstream."
        )

entries instance-attribute 

entries

events instance-attribute 

events

key instance-attribute 

key

MatchGroupKey module-attribute 

MatchGroupKey = tuple[
    Month, str, ProductType, SnapshotEntryType
]

(period_start_month, link_id, product_type, service).

PayrollEventData dataclass 

PayrollEventData(
    event_type,
    primary_user_id,
    beneficiary_user_id,
    company_id,
    country,
    entry_type,
    product_type,
    event_date,
    effective_date,
    link_id,
    after_values,
    before_values,
    id=None,
    before_entry=None,
    after_entry=None,
)

Event produced by timeline diff. Converted to PayrollEvent ORM at persistence time.

after_entry class-attribute instance-attribute 

after_entry = None

after_values instance-attribute 

after_values

before_entry class-attribute instance-attribute 

before_entry = None

before_values instance-attribute 

before_values

beneficiary_user_id instance-attribute 

beneficiary_user_id

company_id instance-attribute 

company_id

country instance-attribute 

country

effective_date instance-attribute 

effective_date

entry_type instance-attribute 

entry_type

event_date instance-attribute 

event_date

event_type instance-attribute 

event_type

id class-attribute instance-attribute 

id = None

link_id

primary_user_id instance-attribute 

primary_user_id

product_type instance-attribute 

product_type

match_events_to_premiums 

match_events_to_premiums(
    events, keyed_entries, target_month
)

Group events + entries by MatchGroupKey.

Parameters:

Name Type Description Default
events list[PayrollEventData]

snapshot events.

 required
keyed_entries list[tuple[MatchGroupKey, PremiumEntry]]

Premium entries already keyed by the country adapter.

 required
target_month Month

The payroll cycle month being processed.

 required

Returns:

Type Description
list[MatchGroup]

Groups in deterministic order: keys first seen on the event side
list[MatchGroup]

come before keys seen only on the entry side.
Source code in components/payroll_tool/internal/business_logic/queries/match_events_to_premiums.py 
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
def match_events_to_premiums(
    events: list[PayrollEventData],
    keyed_entries: list[tuple[MatchGroupKey, PremiumEntry]],
    target_month: Month,
) -> list[MatchGroup]:
    """Group events + entries by ``MatchGroupKey``.

    Args:
        events: snapshot events.
        keyed_entries: Premium entries already keyed by the country adapter.
        target_month: The payroll cycle month being processed.

    Returns:
        Groups in deterministic order: keys first seen on the event side
        come before keys seen only on the entry side.
    """
    events_by_key: dict[MatchGroupKey, list[PayrollEventData]] = defaultdict(list)
    # Ordered set of keys (dict-as-ordered-set). Insertion order preserved
    # so callers get a stable iteration order.
    seen_keys: dict[MatchGroupKey, None] = {}

    for event in events:
        key = _event_key(event)
        events_by_key[key].append(event)
        seen_keys.setdefault(key, None)

    entries_by_key: dict[MatchGroupKey, list[PremiumEntry]] = defaultdict(list)
    for key, entry in keyed_entries:
        entries_by_key[key].append(entry)
        seen_keys.setdefault(key, None)

    groups: list[MatchGroup] = []
    for key in seen_keys:
        key_events = events_by_key.get(key, [])
        key_entries = entries_by_key.get(key, [])

        # Events with no matching premium entry produce no PayrollChange —
        # no billing line means nothing to surface. Defer (log + drop).
        # `target_month` is logged for context only; it doesn't gate the
        # decision (regardless of month, no entry = no row).
        if not key_entries:
            for ev in key_events:
                current_logger.info(
                    "Deferred PayrollEvent: no matching PremiumEntry",
                    event_id=str(ev.id) if ev.id is not None else None,
                    link_id=ev.link_id,
                    product_type=ev.product_type.value,
                    effective_date=ev.effective_date.isoformat(),
                    target_month=str(target_month),
                )
            continue

        groups.append(
            MatchGroup(
                key=key,
                events=tuple(key_events),
                entries=tuple(key_entries),
            )
        )

    return groups