Extract determine_shift_type into shared layer

Eliminates duplicated weekend day/night boundary logic between
the ring scheduler and Slack bot Lambdas.
This commit is contained in:
Adam Moussa 2026-05-12 17:38:09 -04:00
parent 889a1b307c
commit e14eeaa17f
3 changed files with 30 additions and 20 deletions

View file

@ -11,21 +11,18 @@ from datetime import datetime
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
from shared.ring_scheduler import update_queue_routing from shared.ring_scheduler import update_queue_routing
from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule from shared.schedule import (
FALLBACK_EXTENSION,
WEEKEND_DAYS,
ShiftSchedule,
determine_shift_type,
)
from shared.secrets import get_secret from shared.secrets import get_secret
logger = logging.getLogger() logger = logging.getLogger()
logger.setLevel(logging.INFO) logger.setLevel(logging.INFO)
EASTERN = ZoneInfo("America/New_York") EASTERN = ZoneInfo("America/New_York")
WEEKEND_DAYS = {"Saturday", "Sunday"}
def determine_shift_type(now: datetime) -> str:
day_name = now.strftime("%A")
if day_name in WEEKEND_DAYS and 8 <= now.hour < 17:
return "day"
return "night"
def handler(event, context): def handler(event, context):

View file

@ -8,14 +8,30 @@ Single-table design:
""" """
import os import os
from datetime import datetime
from zoneinfo import ZoneInfo
import boto3 import boto3
from boto3.dynamodb.conditions import Key from boto3.dynamodb.conditions import Key
EASTERN = ZoneInfo("America/New_York")
WEEKEND_DAYS = {"Saturday", "Sunday"}
FALLBACK_EXTENSION = "100" FALLBACK_EXTENSION = "100"
def determine_shift_type(now: datetime | None = None) -> str:
"""Return the currently active shift type: 'day' or 'night'.
Weekend 8am-5pm is the day shift; all other times are night.
"""
if now is None:
now = datetime.now(EASTERN)
day_name = now.strftime("%A")
if day_name in WEEKEND_DAYS and 8 <= now.hour < 17:
return "day"
return "night"
class ShiftSchedule: class ShiftSchedule:
def __init__(self, table_name: str | None = None): def __init__(self, table_name: str | None = None):
dynamodb = boto3.resource("dynamodb") dynamodb = boto3.resource("dynamodb")

View file

@ -16,7 +16,12 @@ from shared.blocks import (
build_week_schedule, build_week_schedule,
) )
from shared.ring_scheduler import update_queue_routing from shared.ring_scheduler import update_queue_routing
from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule from shared.schedule import (
FALLBACK_EXTENSION,
WEEKEND_DAYS,
ShiftSchedule,
determine_shift_type,
)
from shared.secrets import get_secret from shared.secrets import get_secret
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -88,16 +93,8 @@ def is_today(date_str: str) -> bool:
return date_str == datetime.now(EASTERN).strftime("%Y-%m-%d") return date_str == datetime.now(EASTERN).strftime("%Y-%m-%d")
WEEKEND_DAYS = {"Saturday", "Sunday"}
def _is_active_shift_type(shift_type: str) -> bool: def _is_active_shift_type(shift_type: str) -> bool:
"""Check if the given shift type is the currently active one.""" return determine_shift_type() == shift_type
now = datetime.now(EASTERN)
day_name = now.strftime("%A")
if day_name in WEEKEND_DAYS and 8 <= now.hour < 17:
return shift_type == "day"
return shift_type == "night"
def _find_employee_shift(schedule, date_str, day_name, employee_ext): def _find_employee_shift(schedule, date_str, day_name, employee_ext):