Skip to content

API Reference

REST endpoints

Full endpoint documentation: On-Call Groups on ReDoc ⧉

Method Path Description
GET /oncall-groups/ List all active on-call groups (with slack_handle set)
POST /oncall-groups/ Create a new on-call group
GET /oncall-groups/<id> Get a single on-call group
PATCH /oncall-groups/<id> Update an on-call group
GET /oncall-groups/<id>/shifts List shifts (query params: start_date, end_date)
POST /oncall-groups/<id>/shifts/update-request Trigger schedule update
PUT /oncall-groups/<id>/shifts/<date>/<alaner_id> Add an alaner to a shift
DELETE /oncall-groups/<id>/shifts/<date>/<alaner_id> Remove an alaner from a shift
PATCH /oncall-groups/<id>/shifts/<date> Set load estimation for a date
PUT /oncall-groups/<id>/roster/<alaner_id> Add an alaner to the roster
DELETE /oncall-groups/<id>/roster/<alaner_id> Remove an alaner from the roster
GET /oncall-groups/<id>/pending-members Get pending roster members (requested + approved but not yet synced)

Code reference

apps.eu_tools.alan_home.models.oncall_group.OncallGroup

Bases: BaseAlanGroup

__alan_home_path__ class-attribute instance-attribute 

__alan_home_path__ = 'on-call'

__name_suffix__ class-attribute instance-attribute 

__name_suffix__ = 'on-call'

__notion_database_id__ class-attribute instance-attribute 

__notion_database_id__ = (
    "53254a52-13b2-4ee6-9bf3-5a0ac183bcef"
)

__tablename__ class-attribute instance-attribute 

__tablename__ = 'oncall_group'

area class-attribute instance-attribute 

area = relationship(
    Area,
    foreign_keys=[area_id],
    uselist=False,
    back_populates="oncall_groups",
)

area_id class-attribute instance-attribute 

area_id = mapped_column(
    UUID(as_uuid=True),
    ForeignKey(id),
    index=True,
    nullable=True,
)

assembled_integration class-attribute instance-attribute 

assembled_integration = mapped_column(
    Boolean, default=False
)

bank_holiday_countries class-attribute instance-attribute 

bank_holiday_countries = mapped_column(
    JSONB(none_as_null=True)
)

calendar_id class-attribute instance-attribute 

calendar_id = mapped_column(Text)

calendar_invite_description class-attribute instance-attribute 

calendar_invite_description = mapped_column(Text)

calendar_next_sync_token class-attribute instance-attribute 

calendar_next_sync_token = mapped_column(Text)

calendar_watch_channel_id class-attribute instance-attribute 

calendar_watch_channel_id = mapped_column(Text)

calendar_watch_resource_id class-attribute instance-attribute 

calendar_watch_resource_id = mapped_column(Text)

disabled class-attribute instance-attribute 

disabled = mapped_column(
    Boolean,
    default=False,
    server_default=false(),
    nullable=False,
)

dust_assistant class-attribute instance-attribute 

dust_assistant = mapped_column(String(255))

external_users class-attribute instance-attribute 

external_users = relationship(
    "ExternalUser",
    secondary="oncall_group_shift",
    primaryjoin="and_(OncallGroup.id == OncallGroupShift.oncall_group_id, cast(OncallGroupShift.date, Date) == func.current_date(), OncallGroupShift.declined_on.is_(None))",
    secondaryjoin="and_(OncallGroupShift.external_user_id == ExternalUser.id, not_(ExternalUser.is_ended))",
    back_populates="current_oncall_groups",
    uselist=True,
    order_by="ExternalUser.email",
    viewonly=True,
)

external_users_roster class-attribute instance-attribute 

external_users_roster = relationship(
    "ExternalUser",
    secondary="oncall_group_membership",
    primaryjoin="and_(OncallGroup.id == OncallGroupMembership.oncall_group_id, OncallGroupMembership.is_active)",
    secondaryjoin="OncallGroupMembership.external_user_id == ExternalUser.id",
    back_populates="oncall_groups",
    uselist=True,
    order_by="ExternalUser.email",
)

grace_period_after_long_ooo class-attribute instance-attribute 

grace_period_after_long_ooo = mapped_column(
    Boolean,
    default=True,
    server_default=true(),
    nullable=False,
)

handover_message class-attribute instance-attribute 

handover_message = mapped_column(Text)

has_load_estimation class-attribute instance-attribute 

has_load_estimation = mapped_column(Boolean)

included_days class-attribute instance-attribute 

included_days = mapped_column(
    JSONB(none_as_null=True), default=[1, 2, 3, 4, 5]
)

included_days_validator 

included_days_validator(key, included_days)
Source code in apps/eu_tools/alan_home/models/oncall_group.py 
177
178
179
180
181
@validates("included_days")
def included_days_validator(self, key, included_days):  # type: ignore[no-untyped-def]  # noqa: ARG002
    if isinstance(included_days, str):
        return sorted(loads(included_days))
    return sorted(included_days)

is_active 

is_active()
Source code in apps/eu_tools/alan_home/models/oncall_group.py 
191
192
193
@hybrid_property
def is_active(self) -> bool:
    return super().is_active and not self.disabled

member_role_id property 

member_role_id

members class-attribute instance-attribute 

members = relationship(
    "Alaner",
    secondary="oncall_group_shift",
    primaryjoin="and_(OncallGroup.id == OncallGroupShift.oncall_group_id, cast(OncallGroupShift.date, Date) == func.current_date(), OncallGroupShift.declined_on.is_(None))",
    secondaryjoin="and_(OncallGroupShift.alaner_id == Alaner.id, not_(Alaner.is_ended))",
    back_populates="current_oncall_groups",
    uselist=True,
    order_by="Alaner.slack_handle",
    viewonly=True,
)

notion_sync_payload property 

notion_sync_payload

ownership class-attribute instance-attribute 

ownership = relationship(
    "Ownership",
    foreign_keys="Ownership.oncall_group_id",
    primaryjoin="Ownership.oncall_group_id == OncallGroup.id",
    back_populates="oncall_group",
    uselist=True,
    order_by="Ownership.created_at.asc()",
)

roster class-attribute instance-attribute 

roster = relationship(
    "Alaner",
    secondary="oncall_group_membership",
    primaryjoin="and_(OncallGroup.id == OncallGroupMembership.oncall_group_id, OncallGroupMembership.is_active)",
    secondaryjoin="OncallGroupMembership.alaner_id == Alaner.id",
    back_populates="oncall_groups",
    uselist=True,
    order_by="Alaner.slack_handle",
)

roster_smart_group_id class-attribute instance-attribute 

roster_smart_group_id = mapped_column(
    UUID(as_uuid=True), ForeignKey(id), index=True
)

rotation_size class-attribute instance-attribute 

rotation_size = mapped_column(Integer, default=1)

rotation_type class-attribute instance-attribute 

rotation_type = mapped_column(String(255), default='daily')

scheduling_strategy_id class-attribute instance-attribute 

scheduling_strategy_id = mapped_column(Text)

shift_hours_end class-attribute instance-attribute 

shift_hours_end = mapped_column(Time)

shift_hours_start class-attribute instance-attribute 

shift_hours_start = mapped_column(Time)

shifts class-attribute instance-attribute 

shifts = relationship(
    "OncallGroupShift",
    primaryjoin="OncallGroup.id == foreign(OncallGroupShift.oncall_group_id)",
    back_populates="oncall_group",
    uselist=True,
    lazy="dynamic",
    order_by="OncallGroupShift.date",
    viewonly=True,
)

slack_channel_id class-attribute instance-attribute 

slack_channel_id = mapped_column(Text)

slack_channel_name class-attribute instance-attribute 

slack_channel_name = mapped_column(Text)

smart_roster class-attribute instance-attribute 

smart_roster = relationship(
    "Alaner",
    secondary="smart_group_membership",
    primaryjoin="and_(OncallGroup.roster_smart_group_id == SmartGroupMembership.smart_group_id, SmartGroupMembership.is_active)",
    secondaryjoin="SmartGroupMembership.alaner_id == Alaner.id",
    back_populates="smart_oncall_groups",
    uselist=True,
    order_by="Alaner.slack_handle",
    viewonly=True,
)

use_calendar_invite class-attribute instance-attribute 

use_calendar_invite = mapped_column(
    Boolean,
    default=True,
    server_default=true(),
    nullable=False,
)

validate_rotation_type class-attribute instance-attribute 

validate_rotation_type = create_validator('rotation_type')

validate_slack_channel_name 

validate_slack_channel_name(key, slack_channel_name)
Source code in apps/eu_tools/alan_home/models/oncall_group.py 
183
184
185
186
187
188
189
@validates("slack_channel_name")
def validate_slack_channel_name(self, key, slack_channel_name: str | None):  # type: ignore[no-untyped-def]  # noqa: ARG002
    if slack_channel_name:
        if not slack_channel_name.startswith("#"):
            slack_channel_name = f"#{slack_channel_name}"
        return slack_channel_name.strip().lower()
    return None

wrap_up_additional_message class-attribute instance-attribute 

wrap_up_additional_message = mapped_column(Text)

apps.eu_tools.alan_home.models.oncall_group.OncallRotationType

Bases: AlanBaseEnum

daily class-attribute instance-attribute 

daily = 'daily'

sprint class-attribute instance-attribute 

sprint = 'sprint'

triduum class-attribute instance-attribute 

triduum = 'triduum'

weekly class-attribute instance-attribute 

weekly = 'weekly'

apps.eu_tools.alan_home.models.oncall_group_shift.OncallGroupShift

Bases: BaseModel

__table_args__ class-attribute instance-attribute 

__table_args__ = (
    UniqueConstraint(
        "oncall_group_id",
        "date",
        "alaner_id",
        name="oncall_group_shift_oncall_group_id_date_alaner_id_key",
    ),
    UniqueConstraint(
        "oncall_group_id",
        "date",
        "external_user_id",
        name="oncall_group_shift_oncall_group_id_date_external_user_id_key",
    ),
    CheckConstraint(
        "(alaner_id IS NULL) != (external_user_id IS NULL)",
        name="oncall_group_shift_user_is_not_null",
    ),
)

__tablename__ class-attribute instance-attribute 

__tablename__ = 'oncall_group_shift'

alaner class-attribute instance-attribute 

alaner = relationship(
    "Alaner",
    foreign_keys=alaner_id,
    primaryjoin="Alaner.id == OncallGroupShift.alaner_id",
    uselist=False,
)

alaner_id class-attribute instance-attribute 

alaner_id = mapped_column(
    Integer, nullable=True, index=True
)

calendar_event_id class-attribute instance-attribute 

calendar_event_id = mapped_column(Text)

date class-attribute instance-attribute 

date = mapped_column(Date, nullable=False)

declined_on class-attribute instance-attribute 

declined_on = mapped_column(Date)

duty_person property 

duty_person

external_user class-attribute instance-attribute 

external_user = relationship(
    "ExternalUser",
    foreign_keys=external_user_id,
    primaryjoin="ExternalUser.id == OncallGroupShift.external_user_id",
    uselist=False,
)

external_user_id class-attribute instance-attribute 

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

load class-attribute instance-attribute 

load = mapped_column(Float)

oncall_group class-attribute instance-attribute 

oncall_group = relationship(
    "OncallGroup",
    foreign_keys=[oncall_group_id],
    back_populates="shifts",
    uselist=False,
)

oncall_group_id class-attribute instance-attribute 

oncall_group_id = mapped_column(
    UUID(as_uuid=True),
    ForeignKey(id),
    nullable=False,
    index=True,
)

apps.eu_tools.alan_home.models.oncall_group_membership.OncallGroupMembership

Bases: HistorizableAlanerRelation, HistorizableExternalUserRelation

__table_args__ class-attribute instance-attribute 

__table_args__ = (
    CheckConstraint(
        "(alaner_id IS NULL) != (external_user_id IS NULL)",
        name="oncall_group_membership_user_is_not_null",
    ),
    ExcludeConstraint(
        ("alaner_id", "="),
        ("external_user_id", "="),
        ("oncall_group_id", "="),
        (
            text("daterange(start_date, end_date, '[]')"),
            "&&",
        ),
        where="is_cancelled = false",
        name="non_overlapping_active_membership_per_alaner_and_oncall_group",
    ),
    *(__table_args__),
)

__tablename__ class-attribute instance-attribute 

__tablename__ = 'oncall_group_membership'

oncall_group class-attribute instance-attribute 

oncall_group = relationship(
    "OncallGroup",
    foreign_keys=[oncall_group_id],
    uselist=False,
    viewonly=True,
)

oncall_group_id class-attribute instance-attribute 

oncall_group_id = mapped_column(
    UUID(as_uuid=True),
    ForeignKey(id),
    nullable=False,
    index=True,
)

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.ASSEMBLED_EMAIL module-attribute 

ASSEMBLED_EMAIL = 'alan@assembledhq.com'

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.DAYS_BACKWARD module-attribute 

DAYS_BACKWARD = 90

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.DAYS_FORWARD module-attribute 

DAYS_FORWARD = 60

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.TRIDUUM_SHIFT_GENESIS module-attribute 

TRIDUUM_SHIFT_GENESIS = date(2023, 1, 2)

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.create_or_update_schedule 

create_or_update_schedule(
    oncall_group_id, slack_monitor=None
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/scheduling.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def create_or_update_schedule(
    oncall_group_id: UUID,
    slack_monitor: SlackMonitor | None = None,
) -> None:
    try:
        oncall_group = get_or_404(OncallGroup, oncall_group_id)
        current_logger.info(
            f"Updating on-call schedule for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}",
            slack_handle=oncall_group.slack_handle,
            alan_home_url=oncall_group.alan_home_url,
        )
        _clean_roster(oncall_group)
        _clean_schedule(oncall_group)
        _fill_schedule(oncall_group)
        if is_production_mode():
            _update_calendar(oncall_group)
    except Exception as e:
        current_logger.exception(
            f"Could not update schedule for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}"
        )
        if slack_monitor is not None:
            slack_monitor.add_context(
                f"Could not update schedule for `{oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}`: ```{str(e)}```"
            )

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.find_shift_dates 

find_shift_dates(oncall_group, oncall_date)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/scheduling.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def find_shift_dates(oncall_group: OncallGroup, oncall_date: date) -> tuple[date, date]:  # type: ignore[return]
    if oncall_group.rotation_type == OncallRotationType.daily:
        return (oncall_date, oncall_date)
    elif oncall_group.rotation_type == OncallRotationType.triduum:
        return _find_triduum_shift_dates(oncall_date=oncall_date)
    elif oncall_group.rotation_type == OncallRotationType.weekly:
        return (
            oncall_date + relativedelta(weekday=MO(-1)),
            oncall_date + relativedelta(weekday=FR),
        )
    elif oncall_group.rotation_type == OncallRotationType.sprint:
        return (
            oncall_date + relativedelta(weekday=WE(-1)),
            oncall_date + relativedelta(weekday=TU),
        )

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.find_shift_for_swap 

find_shift_for_swap(shift)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/scheduling.py 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
def find_shift_for_swap(shift: OncallGroupShift) -> OncallGroupShift | None:
    replacing_shift_alias = aliased(OncallGroupShift)
    original_shift_alias = aliased(OncallGroupShift)
    office_presence_alias = aliased(OfficePresence)
    return (
        current_session.execute(
            select(OncallGroupShift)
            .options(
                joinedload(OncallGroupShift.alaner),
                joinedload(OncallGroupShift.external_user),
                joinedload(OncallGroupShift.oncall_group),
            )
            .join(
                OfficePresence,
                and_(
                    OfficePresence.alaner_id == OncallGroupShift.alaner_id,
                    OfficePresence.date == shift.date,
                    OfficePresence.type.notin_([OfficePresenceType.remote]),
                ),
                isouter=True,
            )
            .join(
                replacing_shift_alias,
                and_(
                    replacing_shift_alias.oncall_group_id == shift.oncall_group_id,
                    replacing_shift_alias.date == OncallGroupShift.date,
                    or_(
                        and_(
                            replacing_shift_alias.alaner_id.isnot(None),
                            replacing_shift_alias.alaner_id == shift.alaner_id,
                        ),
                        and_(
                            replacing_shift_alias.external_user_id.isnot(None),
                            replacing_shift_alias.external_user_id
                            == shift.external_user_id,
                        ),
                    ),
                    replacing_shift_alias.declined_on.isnot(None),
                ),
                isouter=True,
            )
            .join(
                original_shift_alias,
                and_(
                    original_shift_alias.oncall_group_id == shift.oncall_group_id,
                    original_shift_alias.date == shift.date,
                    or_(
                        and_(
                            original_shift_alias.alaner_id.isnot(None),
                            original_shift_alias.alaner_id
                            == OncallGroupShift.alaner_id,
                        ),
                        and_(
                            original_shift_alias.external_user_id.isnot(None),
                            original_shift_alias.external_user_id
                            == OncallGroupShift.external_user_id,
                        ),
                    ),
                    original_shift_alias.declined_on.isnot(None),
                ),
                isouter=True,
            )
            .join(
                office_presence_alias,
                and_(
                    office_presence_alias.alaner_id == shift.alaner_id,
                    office_presence_alias.date == OncallGroupShift.date,
                    office_presence_alias.type.notin_([OfficePresenceType.remote]),
                ),
                isouter=True,
            )
            .filter(
                OncallGroupShift.oncall_group_id == shift.oncall_group_id,
                or_(
                    OncallGroupShift.external_user_id != shift.external_user_id,
                    OncallGroupShift.alaner_id != shift.alaner_id,
                ),
                OncallGroupShift.declined_on.is_(None),
                OncallGroupShift.date > max(shift.date, utctoday() + timedelta(days=7)),
                OfficePresence.id.is_(None),
                replacing_shift_alias.id.is_(None),
                original_shift_alias.id.is_(None),
                office_presence_alias.id.is_(None),
            )
            .order_by(OncallGroupShift.date)
        )
        .scalars()
        .unique()
        .first()
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.get_scheduling_strategy 

get_scheduling_strategy(oncall_group)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/scheduling.py 
260
261
262
263
264
265
266
267
268
269
270
def get_scheduling_strategy(
    oncall_group: OncallGroup,
) -> type[OncallSchedulingStrategy]:
    if oncall_group.scheduling_strategy_id is None:
        return DefaultOncallSchedulingStrategy

    return next(
        scheduling_strategy
        for scheduling_strategy in OncallSchedulingStrategy.__subclasses__()
        if scheduling_strategy.strategy_id == oncall_group.scheduling_strategy_id
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.get_shift_dates 

get_shift_dates(oncall_group, start_date, end_date)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/scheduling.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def get_shift_dates(oncall_group: OncallGroup, start_date: date, end_date: date):  # type: ignore[no-untyped-def]
    from pandas import date_range

    bank_holidays = (
        {
            holiday.date
            for holiday in current_session.execute(
                select(BankHoliday).filter(
                    BankHoliday.country.in_(oncall_group.bank_holiday_countries),
                    BankHoliday.date >= utctoday(),
                )
            )
            .scalars()
            .unique()
            .all()
        }
        if oncall_group.bank_holiday_countries
        else set()
    )
    return [
        schedule_date.date()
        for schedule_date in date_range(start_date, end_date, freq="B")
        if (
            schedule_date.date().isoweekday() in oncall_group.included_days  # type: ignore[operator]
            and (
                oncall_group.start_date is None
                or schedule_date.date() >= oncall_group.start_date
            )
            and schedule_date.date() not in bank_holidays
        )
    ]

apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling.get_shifts_following_long_ooo 

get_shifts_following_long_ooo(
    oncall_group, days_threshold=3
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/scheduling.py 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
def get_shifts_following_long_ooo(
    oncall_group: OncallGroup, days_threshold: int = 3
) -> list[OncallGroupShift]:
    long_ooo_subquery = (
        current_session.query(OfficePresence)  # noqa: ALN085
        .with_entities(
            OfficePresence.alaner_id,
            func.max(OfficePresence.date).label("last_long_ooo_day"),
        )
        .filter(
            OfficePresence.date >= utctoday(),
            OfficePresence.type != OfficePresenceType.remote,
            OfficePresence.span == OfficePresenceSpan.full,
            func.extract("isodow", OfficePresence.date).notin_([6, 7]),
        )
        .group_by(OfficePresence.alaner_id, OfficePresence.event_id)
        .having(func.count(distinct(OfficePresence.date)) >= days_threshold)
        .subquery()
    )

    return cast(
        "list[OncallGroupShift]",
        current_session.execute(
            select(OncallGroupShift)
            .join(
                long_ooo_subquery,
                OncallGroupShift.alaner_id == long_ooo_subquery.c.alaner_id,
            )
            .filter(
                OncallGroupShift.oncall_group_id == oncall_group.id,
                case(
                    (
                        func.extract(
                            "isodow", long_ooo_subquery.c.last_long_ooo_day
                        ).in_([5, 6, 7]),
                        long_ooo_subquery.c.last_long_ooo_day
                        + timedelta(days=1)
                        * (
                            8
                            - func.extract(
                                "isodow", long_ooo_subquery.c.last_long_ooo_day
                            )
                        )
                        == OncallGroupShift.date,
                    ),
                    else_=long_ooo_subquery.c.last_long_ooo_day + timedelta(days=1)
                    == OncallGroupShift.date,
                ),
            )
        )
        .scalars()
        .unique()
        .all(),
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.OncallSchedulingStrategy

Bases: ABC

description instance-attribute 

description

fill_schedule_gap abstractmethod classmethod 

fill_schedule_gap(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)

Fill gaps in a shift

Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
24
25
26
27
28
29
30
31
32
33
34
@classmethod
@abstractmethod
def fill_schedule_gap(
    cls,
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> None:
    """Fill gaps in a shift"""

strategy_id instance-attribute 

strategy_id

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.build_candidate_pool 

build_candidate_pool(
    oncall_group,
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    all_excluded_users,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
def build_candidate_pool(
    oncall_group: OncallGroup,
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    all_excluded_users: set[AlanUser],
) -> list[AlanUser]:
    candidates = _get_candidates(
        oncall_date,
        shift_start_date,
        oncall_group,
        schedule,
        all_excluded_users,
    )

    candidate_pool = []
    # If anyone from the start of the shift is available, we take them in priority
    if candidates:
        candidate_pool = _filter_available_candidates(
            candidates,
            shift_start_date,
            shift_end_date,
            oncall_date,
            all_excluded_users,
        )

    # If we don't have enough candidates, we take more from the full roster, sorted by last shift date
    if len(candidate_pool) < oncall_group.rotation_size:  # type: ignore[operator]
        candidate_pool = _get_additional_candidates_from_roster(
            oncall_group,
            schedule,
            oncall_date,
            shift_start_date,
            shift_end_date,
            list(candidate_pool),
            all_excluded_users,
        )

    return sorted(
        candidate_pool,
        key=lambda c: get_number_of_days_to_closest_shift_date_for_duty_person(
            c,
            oncall_group,
            shift_start_date,
            pending_schedule=schedule,
        ),
        reverse=True,
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.get_alaners_back_from_long_ooo 

get_alaners_back_from_long_ooo(
    oncall_date, days_threshold=3
)

Returns list of Alaners who are on their first day back after being OOO for at least days_threshold consecutive days

Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def get_alaners_back_from_long_ooo(
    oncall_date: date, days_threshold: int = 3
) -> list[Alaner]:
    """
    Returns list of Alaners who are on their first day back after being OOO
    for at least days_threshold consecutive days
    """
    return cast(
        "list[Alaner]",
        current_session.execute(
            select(Alaner)
            .join(OfficePresence, OfficePresence.alaner_id == Alaner.id)
            .filter(
                OfficePresence.date >= oncall_date - timedelta(days=days_threshold),
                OfficePresence.date <= oncall_date,
                OfficePresence.type != OfficePresenceType.remote,
                OfficePresence.span == OfficePresenceSpan.full,
            )
            .group_by(
                Alaner.id,
                OfficePresence.event_id,
                OfficePresence.type,
                OfficePresence.span,
            )
            .having(
                and_(
                    func.count(func.distinct(OfficePresence.date)) >= days_threshold,
                    func.max(OfficePresence.date) == oncall_date - timedelta(days=1),
                )
            )
        )
        .scalars()
        .unique()
        .all(),
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.get_candidates_pool_excluding_other_shifts 

get_candidates_pool_excluding_other_shifts(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def get_candidates_pool_excluding_other_shifts(
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> list[AlanUser]:
    # Get users who are on other oncalls on the same date
    other_oncall_shifts = current_session.query(OncallGroupShift).filter(  # noqa: ALN085
        OncallGroupShift.date == oncall_date,
        OncallGroupShift.oncall_group_id != oncall_group.id,
        OncallGroupShift.declined_on.is_(None),
    )

    all_excluded_users = {
        shift.duty_person
        for shift in other_oncall_shifts
        if shift.duty_person is not None and shift.duty_person in oncall_group.roster
    }

    return build_candidate_pool(
        oncall_group,
        schedule,
        oncall_date,
        shift_start_date,
        shift_end_date,
        all_excluded_users,
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.get_default_sorted_candidates_pool 

get_default_sorted_candidates_pool(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def get_default_sorted_candidates_pool(
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> list[AlanUser]:
    # Get shifts from other oncall groups using "distinct" strategy.
    # This avoids assigning users who are already in distinct strategy groups.
    other_oncall_shifts = (
        current_session.query(OncallGroupShift)  # noqa: ALN085
        .join(OncallGroup, OncallGroupShift.oncall_group_id != OncallGroup.id)
        .filter(
            OncallGroupShift.date == oncall_date,
            OncallGroupShift.oncall_group_id != oncall_group.id,
            OncallGroupShift.declined_on.is_(None),
            OncallGroup.scheduling_strategy_id == "distinct",
        )
    )

    all_excluded_users = {
        shift.duty_person
        for shift in other_oncall_shifts
        if shift.duty_person is not None and shift.duty_person in oncall_group.roster
    }

    return build_candidate_pool(
        oncall_group,
        schedule,
        oncall_date,
        shift_start_date,
        shift_end_date,
        all_excluded_users,
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.get_last_shift_date_for_duty_person 

get_last_shift_date_for_duty_person(
    duty_person, oncall_group, before, pending_schedule
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
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
def get_last_shift_date_for_duty_person(
    duty_person: AlanUser,
    oncall_group: OncallGroup,
    before: date,
    pending_schedule: dict[date, list[AlanUser]],
) -> date:
    last_shift = (
        current_session.execute(
            select(OncallGroupShift)
            .filter(
                *(
                    OncallGroupShift.alaner_id == duty_person.id
                    if duty_person.type == AlanAccountType.alaner
                    else OncallGroupShift.external_user_id == duty_person.id,
                ),
                OncallGroupShift.oncall_group_id == oncall_group.id,
                OncallGroupShift.date < before,
                OncallGroupShift.declined_on.is_(None),
            )
            .order_by(OncallGroupShift.date.desc())
        )
        .scalars()
        .unique()
        .first()
    )
    last_shift_date = last_shift.date if last_shift else date(1970, 1, 1)
    last_pending_shift_date = max(
        (
            shift_date
            for shift_date in pending_schedule.keys()
            if shift_date < before and duty_person in pending_schedule[shift_date]
        ),
        default=date(1970, 1, 1),
    )
    return max(last_shift_date, last_pending_shift_date)

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.get_next_shift_date_for_duty_person 

get_next_shift_date_for_duty_person(
    duty_person, oncall_group, after, pending_schedule
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
 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
def get_next_shift_date_for_duty_person(
    duty_person: AlanUser,
    oncall_group: OncallGroup,
    after: date,
    pending_schedule: dict[date, list[AlanUser]],
) -> date:
    next_shift = (
        current_session.execute(
            select(OncallGroupShift)
            .filter(
                *(
                    OncallGroupShift.alaner_id == duty_person.id
                    if duty_person.type == AlanAccountType.alaner
                    else OncallGroupShift.external_user_id == duty_person.id,
                ),
                OncallGroupShift.oncall_group_id == oncall_group.id,
                OncallGroupShift.date > after,
                OncallGroupShift.declined_on.is_(None),
            )
            .order_by(OncallGroupShift.date.asc())
        )
        .scalars()
        .unique()
        .first()
    )
    next_shift_date = next_shift.date if next_shift else date(9999, 12, 31)
    next_pending_shift_date = min(
        (
            shift_date
            for shift_date in pending_schedule.keys()
            if shift_date > after and duty_person in pending_schedule[shift_date]
        ),
        default=date(9999, 12, 31),
    )
    return min(next_shift_date, next_pending_shift_date)

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.get_number_of_days_to_closest_shift_date_for_duty_person 

get_number_of_days_to_closest_shift_date_for_duty_person(
    duty_person,
    oncall_group,
    current_date,
    pending_schedule,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_number_of_days_to_closest_shift_date_for_duty_person(
    duty_person: AlanUser,
    oncall_group: OncallGroup,
    current_date: date,
    pending_schedule: dict[date, list[AlanUser]],
) -> int:
    next_shift_date = get_next_shift_date_for_duty_person(
        duty_person, oncall_group, current_date, pending_schedule
    )
    last_shift_date = get_last_shift_date_for_duty_person(
        duty_person, oncall_group, current_date, pending_schedule
    )
    return min(next_shift_date - current_date, current_date - last_shift_date).days

apps.eu_tools.alan_home.business_logic.oncall_groups.helpers.has_refused_shift_on 

has_refused_shift_on(duty_person, date)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/helpers.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def has_refused_shift_on(duty_person: AlanUser, date: date) -> bool:
    return (
        current_session.query(OncallGroupShift)  # noqa: ALN085
        .filter(
            *(
                OncallGroupShift.alaner_id == duty_person.id
                if duty_person.type == AlanAccountType.alaner
                else OncallGroupShift.external_user_id == duty_person.id,
            ),
            OncallGroupShift.date == date,
            OncallGroupShift.declined_on.isnot(None),
        )
        .count()
        > 0
    )

apps.eu_tools.alan_home.business_logic.oncall_groups.strategies.default.DefaultOncallSchedulingStrategy

Bases: OncallSchedulingStrategy

description class-attribute instance-attribute 

description = "Randomly select available duty persons prioritizing those who have been on-call the least recently"

fill_schedule_gap classmethod 

fill_schedule_gap(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/strategies/default.py 
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
@classmethod
def fill_schedule_gap(
    cls,
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> None:
    candidate_pool = get_default_sorted_candidates_pool(
        schedule,
        oncall_date,
        shift_start_date,
        shift_end_date,
        oncall_group,
    )

    # Apply long OOO filtering only if the flag is enabled
    filtered_candidate_pool = candidate_pool
    if oncall_group.grace_period_after_long_ooo:
        alaners_back_from_long_ooo = set(
            cls._get_alaners_back_from_long_ooo(oncall_date, days_threshold=3)
        )
        filtered_candidate_pool = [
            candidate
            for candidate in candidate_pool
            if candidate not in alaners_back_from_long_ooo
        ]

    # Fill the slot with the candidates, if we have enough
    selected_candidates = filtered_candidate_pool[
        : oncall_group.rotation_size  # type: ignore[operator]
        - len(schedule[oncall_date])
    ]
    schedule[oncall_date].extend(selected_candidates)
    current_logger.info(
        f"Added {len(selected_candidates)} on-call on {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}. Candidates: {', '.join([candidate.full_name for candidate in selected_candidates])}"
    )

strategy_id class-attribute instance-attribute 

strategy_id = 'default'

apps.eu_tools.alan_home.business_logic.oncall_groups.strategies.distinct.DistinctOncallSchedulingStrategy

Bases: OncallSchedulingStrategy

description class-attribute instance-attribute 

description = "Randomly select available duty persons, prioritizing those who have been on-call least recently. The stategy ensures that Alaners will not be assigned to multiple shifts simultaneously."

fill_schedule_gap classmethod 

fill_schedule_gap(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/strategies/distinct.py 
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
@classmethod
def fill_schedule_gap(
    cls,
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> None:
    candidate_pool = get_candidates_pool_excluding_other_shifts(
        schedule,
        oncall_date,
        shift_start_date,
        shift_end_date,
        oncall_group,
    )

    # Apply long OOO filtering only if the flag is enabled
    filtered_candidate_pool = candidate_pool
    if oncall_group.grace_period_after_long_ooo:
        alaners_back_from_long_ooo = set(
            cls._get_alaners_back_from_long_ooo(oncall_date, days_threshold=3)
        )
        filtered_candidate_pool = [
            candidate
            for candidate in candidate_pool
            if candidate not in alaners_back_from_long_ooo
        ]

    # Fill the slot with the candidates, if we have enough
    selected_candidates = filtered_candidate_pool[
        : oncall_group.rotation_size  # type: ignore[operator]
        - len(schedule[oncall_date])
    ]
    schedule[oncall_date].extend(selected_candidates)
    current_logger.info(
        f"Added {len(selected_candidates)} on-call on {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}. Candidates: {', '.join([candidate.full_name for candidate in selected_candidates])}"
    )

strategy_id class-attribute instance-attribute 

strategy_id = 'distinct'

apps.eu_tools.alan_home.business_logic.oncall_groups.strategies.manual.ManualOncallSchedulingStrategy

Bases: OncallSchedulingStrategy

description class-attribute instance-attribute 

description = (
    "Manual scheduling - system does not auto-fill gaps"
)

fill_schedule_gap classmethod 

fill_schedule_gap(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)

Manual strategy: do not fill gaps automatically

Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/strategies/manual.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
@classmethod
def fill_schedule_gap(
    cls,
    schedule: dict[date, list[AlanUser]],  # noqa: ARG003
    oncall_date: date,
    shift_start_date: date,  # noqa: ARG003
    shift_end_date: date,  # noqa: ARG003
    oncall_group: OncallGroup,
) -> None:
    """Manual strategy: do not fill gaps automatically"""
    current_logger.info(
        f"Skipping auto-fill for {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name} (manual scheduling strategy)"
    )

strategy_id class-attribute instance-attribute 

strategy_id = 'manual'

apps.eu_tools.alan_home.business_logic.oncall_groups.strategies.part_time.PartTimeOncallSchedulingStrategy

Bases: OncallSchedulingStrategy

description class-attribute instance-attribute 

description = "Randomly select available duty persons prioritizing those who have been on-call the least recently and replacing part time persons if needed"

fill_schedule_gap classmethod 

fill_schedule_gap(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/strategies/part_time.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
96
97
@classmethod
def fill_schedule_gap(
    cls,
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> None:
    from apps.eu_tools.alan_home.business_logic.oncall_groups.scheduling import (
        find_shift_dates,
    )

    # 1. We compute the candidate pool for the current shift at the beginning of the shift
    # including part time candidate (They might not be available on `oncall_date` but we deal
    # with that later) but ignoring any already stored candidate. Indeed, a stored candidate might
    # be a replacement of a part time candidate who was not available on the shift start date.
    candidate_pool_including_part_time_for_the_shift = (
        get_sorted_candidates_pool_for_the_shift(
            schedule,
            oncall_date,
            shift_start_date,
            shift_end_date,
            oncall_group,
        )
    )
    # 2. We keep only the required candidates size
    # We don't filter out unavailable part time candidates because we don't want to include
    # candidates who are next in the pool to replace the part time candidates.
    # Indeed, if we do that, candidates who are next in the pool will have a very short shift
    # (ie. 1 day for 80% part time replacement).
    selected_candidates_including_part_time_for_the_shift = candidate_pool_including_part_time_for_the_shift[
        : oncall_group.rotation_size  # type: ignore[operator]
        - len(schedule[oncall_date])
    ]
    # 3. We filter out part time candidates who are not available on `oncall_date`
    selected_candidates = set(
        candidate
        for candidate in selected_candidates_including_part_time_for_the_shift
        if not candidate.is_ooo_on(oncall_date)
    )
    schedule[oncall_date].extend(selected_candidates)
    already_selected_candidate = set(schedule[oncall_date])

    # 4. We fill the gap with candidates from the last shift (ie. we extend their shift)
    # if we don't have enough candidates (ie. there was a part time candidate in
    # `candidate_pool_including_part_time_for_the_shift` or the pool was not big enough).
    selected_candidates_from_last_shift = []
    if oncall_group.rotation_size - len(already_selected_candidate) > 0:  # type: ignore[operator]
        current_logger.info(
            "Not enough candidates for the shift, extending the shift with the latest shift candidates"
        )
        last_shift_start_date, last_shift_end_date = find_shift_dates(
            oncall_group, shift_start_date - relativedelta(days=3)
        )
        available_candidates_from_last_shift = [
            candidate
            # Same query as step 1. but for last shift then exclude candidates who are
            # already in the schedule or unavailable on `oncall_date`
            for candidate in get_sorted_candidates_pool_for_the_shift(
                schedule,
                oncall_date,
                last_shift_start_date,
                last_shift_end_date,
                oncall_group,
            )
            if not candidate.is_ooo_on(oncall_date)
            and candidate not in already_selected_candidate
        ]
        selected_candidates_from_last_shift = available_candidates_from_last_shift[
            : oncall_group.rotation_size  # type: ignore[operator]
            - len(already_selected_candidate)
        ]
        schedule[oncall_date].extend(selected_candidates_from_last_shift)

    current_logger.info(
        f"Added {len(selected_candidates) + len(selected_candidates_from_last_shift)} on-call on {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}"
    )

strategy_id class-attribute instance-attribute 

strategy_id = 'part_time'

apps.eu_tools.alan_home.business_logic.oncall_groups.strategies.part_time.get_sorted_candidates_pool_for_the_shift 

get_sorted_candidates_pool_for_the_shift(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/strategies/part_time.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def get_sorted_candidates_pool_for_the_shift(
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> list[AlanUser]:
    candidate_pool = sorted(
        [
            candidate
            for candidate in (
                oncall_group.smart_roster
                or (oncall_group.roster + oncall_group.external_users_roster)
            )
            if (
                candidate.type == AlanAccountType.external
                or not candidate.is_ooo_between(
                    shift_start_date,
                    shift_end_date,
                    ignore_part_time=True,
                )
            )
            and not has_refused_shift_on(candidate, oncall_date)
            and candidate.is_active_on(oncall_date)
        ],
        key=lambda c: get_last_shift_date_for_duty_person(
            c,
            oncall_group,
            shift_start_date,
            pending_schedule=schedule,
        ),
    )

    return candidate_pool

apps.eu_tools.alan_home.business_logic.oncall_groups.strategies.office.OfficeOncallSchedulingStrategy

Bases: OncallSchedulingStrategy

description class-attribute instance-attribute 

description = 'Custom strategy for office_oncall'

fill_schedule_gap classmethod 

fill_schedule_gap(
    schedule,
    oncall_date,
    shift_start_date,
    shift_end_date,
    oncall_group,
)
Source code in apps/eu_tools/alan_home/business_logic/oncall_groups/strategies/office.py 
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@classmethod
def fill_schedule_gap(
    cls,
    schedule: dict[date, list[AlanUser]],
    oncall_date: date,
    shift_start_date: date,
    shift_end_date: date,
    oncall_group: OncallGroup,
) -> None:
    # Add external users based on a static schedule
    day_of_week = oncall_date.isoweekday()
    external_duty_emails = [
        "nathalie.semedolopes@alan.eu",
        "jules.pacquette@alan.eu",
    ]
    assigned_external_duty_persons = (
        current_session.execute(
            select(ExternalUser).filter(
                ExternalUser.email.in_(external_duty_emails),  # type: ignore[attr-defined]
                ExternalUser.is_active_on(oncall_date),
            )
        )
        .scalars()
        .unique()
        .all()
    )

    for assigned_external_duty_person in assigned_external_duty_persons:
        if assigned_external_duty_person not in schedule[oncall_date]:
            current_logger.info(
                f"Added {assigned_external_duty_person.email} as fixed on-call on {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}"
            )
            schedule[oncall_date].append(assigned_external_duty_person)

    # Try to use fixed schedule if the assigned duty person is not OOO
    # It's super scrappy but good enough for now
    if len(schedule[oncall_date]) < oncall_group.rotation_size:  # type: ignore[operator]
        assigned_duty_email = (
            "eve.pariente@alan.eu"
            if day_of_week in (1, 4)
            else "pauline.deleuze@alan.eu"
            if day_of_week in (2, 5)
            else "lucie.paladino@alan.eu"
        )
        assigned_duty_person = (
            current_session.execute(
                select(Alaner)
                .join(
                    OfficePresence,
                    and_(
                        OfficePresence.alaner_id == Alaner.id,
                        OfficePresence.date == oncall_date,
                        OfficePresence.span != OfficePresenceSpan.morning,
                    ),
                    isouter=True,
                )
                .filter(
                    Alaner.email == assigned_duty_email,
                    OfficePresence.id.is_(None),
                    Alaner.is_active_on(oncall_date),
                )
            )
            .scalars()
            .unique()
            .one_or_none()
        )
        if assigned_duty_person is not None:
            current_logger.info(
                f"Added {assigned_duty_person.email} as fixed on-call on {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}"
            )
            schedule[oncall_date].append(assigned_duty_person)

    if len(schedule[oncall_date]) < oncall_group.rotation_size:  # type: ignore[operator]
        # Fill the slot if needed
        backup_candidate_pool = [
            c
            for c in get_default_sorted_candidates_pool(
                schedule,
                oncall_date,
                shift_start_date,
                shift_end_date,
                oncall_group,
            )
            if c.type != AlanAccountType.external
            and assigned_duty_person not in schedule[oncall_date]
        ]
        selected_backup_candidates = backup_candidate_pool[
            : oncall_group.rotation_size  # type: ignore[operator]
            - len(schedule[oncall_date])
        ]
        schedule[oncall_date].extend(selected_backup_candidates)
        current_logger.info(
            f"Added {len(selected_backup_candidates)} on-call on {oncall_date.isoformat()} for {oncall_group.slack_handle if oncall_group.slack_handle else oncall_group.name}"
        )

strategy_id class-attribute instance-attribute 

strategy_id = 'office'