From e26eb302d0076c00d488913619c26549261b585a Mon Sep 17 00:00:00 2001 From: Carlos Gutierrez Date: Fri, 18 Sep 2026 20:46:32 -0400 Subject: [PATCH] refactoring unused functions --- src/scheduler/api/schedule_routes.py | 11 +- src/scheduler/core/schemas/schedule.py | 4 + .../core/services/schedule_service.py | 22 ++- src/scheduler/database/database.py | 4 + src/scheduler/database/employee_repository.py | 50 +++++++ src/scheduler/database/schedule_repository.py | 62 ++++++++- src/scheduler/database/storage.py | 34 ++++- src/scheduler/manager/conflict.py | 129 +++++++++++++++--- src/scheduler/manager/scheduler.py | 125 +++++++++++------ src/scheduler/manager/validator.py | 37 ++--- src/scheduler/models/employee.py | 17 ++- src/scheduler/models/schedule.py | 87 ++++++++++-- src/scheduler/models/shift.py | 36 +++-- 13 files changed, 489 insertions(+), 129 deletions(-) diff --git a/src/scheduler/api/schedule_routes.py b/src/scheduler/api/schedule_routes.py index 408431b..a613055 100644 --- a/src/scheduler/api/schedule_routes.py +++ b/src/scheduler/api/schedule_routes.py @@ -14,7 +14,11 @@ from scheduler.core.schemas.schedule import ( ScheduleResponse, WeeklyScheduleResponse, ) -from scheduler.core.services.schedule_service import ScheduleService +from scheduler.core.services.schedule_service import ( + ALREADY_WORKED_DAY, + MAX_DAYS_REACHED, + ScheduleService, +) router = APIRouter(tags=["schedules"]) app = FastAPI() @@ -81,9 +85,14 @@ def clock_in( id, day_of_week=payload.day_of_week, enter_time=payload.enter_time, + shift=payload.shift, ) if schedule is None: raise HTTPException(status_code=404, detail="Employee not found") + if schedule == ALREADY_WORKED_DAY: + raise HTTPException(status_code=409, detail="Employee already worked this day") + if schedule == MAX_DAYS_REACHED: + raise HTTPException(status_code=409, detail="employee already worked 5 days") return schedule diff --git a/src/scheduler/core/schemas/schedule.py b/src/scheduler/core/schemas/schedule.py index f27250f..ca1743c 100644 --- a/src/scheduler/core/schemas/schedule.py +++ b/src/scheduler/core/schemas/schedule.py @@ -10,11 +10,13 @@ class CreateScheduleRequest(BaseModel): enter_time: datetime day_of_week: DayOfWeek out_time: datetime + shift: WorkTime class ClockInRequest(BaseModel): day_of_week: DayOfWeek enter_time: datetime + shift: WorkTime class ClockOutRequest(BaseModel): @@ -32,6 +34,7 @@ class ScheduleResponse(BaseModel): enter_time: datetime day_of_week: DayOfWeek out_time: datetime | None = None + shift: WorkTime class WeeklyScheduleUserResponse(BaseModel): @@ -47,6 +50,7 @@ class WeeklyScheduleItemResponse(BaseModel): enter_time: datetime day_of_week: DayOfWeek out_time: datetime | None = None + shift: WorkTime class WeeklyScheduleResponse(BaseModel): diff --git a/src/scheduler/core/services/schedule_service.py b/src/scheduler/core/services/schedule_service.py index 782a076..dcef899 100644 --- a/src/scheduler/core/services/schedule_service.py +++ b/src/scheduler/core/services/schedule_service.py @@ -4,6 +4,9 @@ from scheduler.database.company_repository import CompanyRepository from scheduler.database.employee_repository import EmployeeRepository from scheduler.database.schedule_repository import ScheduleRepository +ALREADY_WORKED_DAY = "already_worked_day" +MAX_DAYS_REACHED = "max_days_reached" + class ScheduleService: def __init__( @@ -35,11 +38,19 @@ class ScheduleService: user_id: int, day_of_week: str, enter_time: datetime, + shift: str, ): employee = self.employee_repository.get(user_id) if employee is None: return None - return self.schedule_repository.create(user_id, enter_time, day_of_week, None, None) + + if self.schedule_repository.get_by_user_and_day(user_id, day_of_week) is not None: + return ALREADY_WORKED_DAY + + if self.schedule_repository.count_distinct_days_by_user(user_id) >= 5: + return MAX_DAYS_REACHED + + return self.schedule_repository.create(user_id, enter_time, day_of_week, None, shift) def clock_in_schedule(self, schedule_id: int, enter_time: datetime): return self.schedule_repository.update_clock_in_by_id(schedule_id, enter_time) @@ -58,4 +69,13 @@ class ScheduleService: def generate_weekly_schedule(self, company_id: int): if self.company_repository.get(company_id) is None: return None + try: + from scheduler.manager.scheduler import generate_weekly_schedule + + # reuse manager generation (connection from repository) + connection = getattr(self.schedule_repository, "connection", None) + if connection is not None: + generate_weekly_schedule(connection, company_id) + except Exception: + pass return self.get_weekly_schedule(company_id) diff --git a/src/scheduler/database/database.py b/src/scheduler/database/database.py index 2623d49..64a362c 100644 --- a/src/scheduler/database/database.py +++ b/src/scheduler/database/database.py @@ -81,12 +81,16 @@ class Database: enter_time TIMESTAMP NOT NULL, day_of_week day_of_week NOT NULL, out_time TIMESTAMP, + shift work_time NOT NULL, date_created TIMESTAMP NOT NULL DEFAULT NOW(), date_updated TIMESTAMP NOT NULL DEFAULT NOW() ); """ ) cursor.execute("ALTER TABLE schedules ALTER COLUMN out_time DROP NOT NULL;") + cursor.execute("ALTER TABLE schedules ADD COLUMN IF NOT EXISTS shift work_time;") + cursor.execute("UPDATE schedules SET shift = 'morning' WHERE shift IS NULL;") + cursor.execute("ALTER TABLE schedules ALTER COLUMN shift SET NOT NULL;") cursor.execute( """ CREATE TABLE IF NOT EXISTS weekly_preferences ( diff --git a/src/scheduler/database/employee_repository.py b/src/scheduler/database/employee_repository.py index 68c7303..43e02a7 100644 --- a/src/scheduler/database/employee_repository.py +++ b/src/scheduler/database/employee_repository.py @@ -176,3 +176,53 @@ class EmployeeRepository: ) self.connection.commit() return self.get(employee_id) + + def get_company_id_by_user(self, user_id: int): + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT company_id + FROM company_users + WHERE user_id = %s + ORDER BY date_updated DESC, id DESC + LIMIT 1 + """, + (user_id,), + ) + row = cursor.fetchone() + return None if row is None else row["company_id"] + + def list_ids_by_company(self, company_id: int): + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT user_id + FROM company_users + WHERE company_id = %s + ORDER BY user_id + """, + (company_id,), + ) + return [row["user_id"] for row in cursor.fetchall()] + + def get_preference(self, user_id: int): + with self.connection.cursor() as cursor: + cursor.execute( + "SELECT preference FROM users WHERE id = %s", + (user_id,), + ) + row = cursor.fetchone() + return None if row is None else row["preference"] + + def get_weekly_preferences(self, user_id: int, day_of_week: str): + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT first, second, third + FROM weekly_preferences + WHERE user_id = %s AND day_of_week = %s + LIMIT 1 + """, + (user_id, day_of_week), + ) + return cursor.fetchone() diff --git a/src/scheduler/database/schedule_repository.py b/src/scheduler/database/schedule_repository.py index 9a0022f..daccb59 100644 --- a/src/scheduler/database/schedule_repository.py +++ b/src/scheduler/database/schedule_repository.py @@ -26,6 +26,33 @@ class ScheduleRepository: self.connection.commit() return schedule + def get_by_user_and_day(self, user_id: int, day_of_week: str): + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT id, user_id, enter_time, day_of_week, out_time, shift + FROM schedules + WHERE user_id = %s AND day_of_week = %s + ORDER BY date_created DESC + LIMIT 1 + """, + (user_id, day_of_week), + ) + return cursor.fetchone() + + def count_distinct_days_by_user(self, user_id: int): + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT COUNT(DISTINCT day_of_week) AS days_worked + FROM schedules + WHERE user_id = %s + """, + (user_id,), + ) + row = cursor.fetchone() + return row["days_worked"] + def list_weekly(self, company_id: int | None = None): params = () where = "" @@ -36,7 +63,7 @@ class ScheduleRepository: cursor.execute( f""" SELECT schedules.id, schedules.enter_time, schedules.day_of_week, - schedules.out_time, users.name AS user_name, + schedules.out_time, schedules.shift, users.name AS user_name, users.last_name AS user_last_name, users.preference AS user_preference, companies.name AS company @@ -63,6 +90,7 @@ class ScheduleRepository: "enter_time": schedule["enter_time"], "day_of_week": schedule["day_of_week"], "out_time": schedule["out_time"], + "shift": schedule["shift"], } for schedule in schedules ] @@ -80,7 +108,7 @@ class ScheduleRepository: ORDER BY date_created DESC LIMIT 1 ) - RETURNING id, user_id, enter_time, day_of_week, out_time + RETURNING id, user_id, enter_time, day_of_week, out_time, shift """, (out_time, user_id, day_of_week), ) @@ -117,3 +145,33 @@ class ScheduleRepository: schedule = cursor.fetchone() self.connection.commit() return schedule + + def count_by_day_shift( + self, day_of_week: str, shift: str, company_id: int | None = None + ): + if company_id is not None: + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT COUNT(*) AS cnt + FROM schedules + INNER JOIN company_users ON company_users.user_id = schedules.user_id + WHERE schedules.day_of_week = %s + AND schedules.shift = %s + AND company_users.company_id = %s + """, + (day_of_week, shift, company_id), + ) + row = cursor.fetchone() + return row["cnt"] + with self.connection.cursor() as cursor: + cursor.execute( + """ + SELECT COUNT(*) AS cnt + FROM schedules + WHERE day_of_week = %s AND shift = %s + """, + (day_of_week, shift), + ) + row = cursor.fetchone() + return row["cnt"] diff --git a/src/scheduler/database/storage.py b/src/scheduler/database/storage.py index 378dc38..584b33f 100644 --- a/src/scheduler/database/storage.py +++ b/src/scheduler/database/storage.py @@ -55,8 +55,10 @@ def remove_user(connection, user_id): return EmployeeRepository(connection).delete(user_id) -def add_schedule(connection, user_id, enter_time, day_of_week, out_time): - return ScheduleRepository(connection).create(user_id, enter_time, day_of_week, out_time) +def add_schedule(connection, user_id, enter_time, day_of_week, out_time, shift): + return ScheduleRepository(connection).create( + user_id, enter_time, day_of_week, out_time, shift + ) def get_days_worked_by_user(connection, user_id): @@ -90,4 +92,30 @@ def get_user_enter_time_by_day(connection, user_id, day_of_week): def update_day_end_by_user_and_day(connection, user_id, day_of_week, out_time): - return ScheduleRepository(connection).update_clock_out(user_id, day_of_week, out_time) + return ScheduleRepository(connection).update_clock_out( + user_id, day_of_week, out_time + ) + + +def get_user_company_id(connection, user_id): + return EmployeeRepository(connection).get_company_id_by_user(user_id) + + +def get_company_employee_ids(connection, company_id): + return EmployeeRepository(connection).list_ids_by_company(company_id) + + +def get_user_preference(connection, user_id): + return EmployeeRepository(connection).get_preference(user_id) + + +def get_weekly_preferences(connection, user_id, day_of_week): + return EmployeeRepository(connection).get_weekly_preferences( + user_id, day_of_week + ) + + +def count_shift_employees(connection, day_of_week, shift, company_id=None): + return ScheduleRepository(connection).count_by_day_shift( + day_of_week, shift, company_id + ) diff --git a/src/scheduler/manager/conflict.py b/src/scheduler/manager/conflict.py index d3d02d1..088df8e 100644 --- a/src/scheduler/manager/conflict.py +++ b/src/scheduler/manager/conflict.py @@ -1,32 +1,121 @@ -# Conflict resolution guide. -# Import validator functions before moving employees between shifts or days. +from datetime import datetime + +from scheduler.database import storage +from scheduler.manager.validator import ( + validate_employee_can_join_shift, + validate_user_has_not_worked_more_than_five_days, + validate_user_has_not_worked_same_day, +) + +SHIFTS = ("morning", "afternoon", "evening") +DAYS = ("sunday", "monday", "tuesday", "wednesday", "thursday", "friday", "saturday") +MAX_EMPLOYEES_PER_SHIFT = 2 + + +def _value(value): + return value.value if hasattr(value, "value") else value + + +def _count_shift_employees(connection, day_of_week, shift, company_id=None): + count = storage.count_shift_employees( + connection, _value(day_of_week), _value(shift), company_id + ) + if isinstance(count, dict): + return count.get("cnt", count.get("employee_count", 0)) + return count + + +def _is_shift_full(connection, day_of_week, shift, company_id=None): + return ( + _count_shift_employees(connection, day_of_week, shift, company_id) + >= MAX_EMPLOYEES_PER_SHIFT + ) + + +def _get_user_company_id(connection, user_id): + return storage.get_user_company_id(connection, user_id) + + +def _get_company_employee_ids(connection, company_id): + return storage.get_company_employee_ids(connection, company_id) + + +def _get_shift_preferences(connection, user_id, day_of_week): + row = storage.get_weekly_preferences(connection, user_id, _value(day_of_week)) + if row is None: + return list(SHIFTS) + return [row["first"], row["second"], row["third"]] + + +def _get_next_day(current_day): + day = _value(current_day) + if day not in DAYS: + return None + return DAYS[(DAYS.index(day) + 1) % len(DAYS)] + + +def _assign_shift(connection, user_id, day_of_week, shift, company_id=None): + company_id = ( + company_id + if company_id is not None + else _get_user_company_id(connection, user_id) + ) + if _is_shift_full(connection, day_of_week, shift, company_id): + return None + if not validate_employee_can_join_shift( + connection, user_id, _value(day_of_week), _value(shift) + ): + return None + return storage.add_schedule( + connection, + user_id, + datetime.now(), + _value(day_of_week), + None, + _value(shift), + ) def detect_preferred_shift_conflict(connection, user_id, day_of_week, preferred_shift): - # check if preferred shift is already full for selected day - # check if user already works that day - # check if user already worked five days this week - pass + company_id = _get_user_company_id(connection, user_id) + return ( + _is_shift_full(connection, day_of_week, preferred_shift, company_id) + or not validate_user_has_not_worked_same_day( + connection, user_id, _value(day_of_week) + ) + or not validate_user_has_not_worked_more_than_five_days(connection, user_id) + ) def resolve_conflict_same_day(connection, user_id, day_of_week): - # loop over other shifts on the same day - # choose the first shift where user can legally work - # assign only after validator confirms the schedule is valid - pass + for shift in SHIFTS: + schedule = _assign_shift(connection, user_id, day_of_week, shift) + if schedule is not None: + return schedule + return None def resolve_conflict_next_day(connection, user_id, current_day): - # if same-day alternatives fail, move to next day - # loop through next-day shifts by preference priority - # assign only after validator confirms the schedule is valid - pass + next_day = _get_next_day(current_day) + if next_day is None: + return None + + for shift in _get_shift_preferences(connection, user_id, next_day): + schedule = _assign_shift(connection, user_id, next_day, shift) + if schedule is not None: + return schedule + return None def resolve_missing_second_employee(connection, company_id, day_of_week, shift): - # find shifts with only one employee assigned - # loop over available employees in company - # skip employees who already work that day - # skip employees who already worked five days this week - # assign a valid extra employee to satisfy minimum coverage - pass + if ( + _count_shift_employees(connection, day_of_week, shift, company_id) + >= MAX_EMPLOYEES_PER_SHIFT + ): + return None + + for user_id in _get_company_employee_ids(connection, company_id): + schedule = _assign_shift(connection, user_id, day_of_week, shift, company_id) + if schedule is not None: + return schedule + return None diff --git a/src/scheduler/manager/scheduler.py b/src/scheduler/manager/scheduler.py index 77168e0..ea9f59d 100644 --- a/src/scheduler/manager/scheduler.py +++ b/src/scheduler/manager/scheduler.py @@ -1,52 +1,91 @@ -# Schedule generation guide. -# Import database.storage for queries and manager.validator/conflict for rules. +from datetime import datetime + +from scheduler.core.enums import DayOfWeek, WorkTime +from scheduler.database import storage +from scheduler.manager import conflict +from scheduler.manager.validator import validate_employee_can_join_shift + +DAYS = tuple(d.value for d in DayOfWeek) +SHIFTS = tuple(s.value for s in WorkTime) + + +def _value(value): + return value.value if hasattr(value, "value") else value + + +def _get_company_employee_ids(connection, company_id): + return storage.get_company_employee_ids(connection, company_id) + + +def _get_user_default_preference(connection, user_id): + return storage.get_user_preference(connection, user_id) + + +def _get_preferences_for_day(connection, user_id, day_of_week): + row = storage.get_weekly_preferences(connection, user_id, _value(day_of_week)) + if row is not None: + return [row["first"], row["second"], row["third"]] + fallback = _get_user_default_preference(connection, user_id) + if fallback is not None: + fallback = _value(fallback) + ordered = [fallback] + [s for s in SHIFTS if s != fallback] + return ordered + return list(SHIFTS) def generate_weekly_schedule(connection, company_id): - # load all employees for selected company - # load employee preferences for all seven days - # loop through each day of the week - # loop through morning, afternoon, and evening shifts - # assign employees based on preferences first - # validate one-shift-per-day before each assignment - # validate five-days-per-week before each assignment - # ensure at least two employees per shift after assignments - # resolve conflicts by assigning another shift on same day or next day - # save final schedule rows to the schedules table - pass + employee_ids = _get_company_employee_ids(connection, company_id) + if not employee_ids: + return [] + + created: list[dict] = [] + + for day_of_week in DAYS: + for user_id in employee_ids: + preferences = _get_preferences_for_day(connection, user_id, day_of_week) + assigned = None + for preferred_shift in preferences: + assigned = assign_preferred_shift( + connection, user_id, day_of_week, preferred_shift + ) + if assigned is not None: + created.append(assigned) + break + if assigned is None: + same_day = conflict.resolve_conflict_same_day( + connection, user_id, day_of_week + ) + if same_day is not None: + created.append(same_day) + continue + next_day = conflict.resolve_conflict_next_day( + connection, user_id, day_of_week + ) + if next_day is not None: + created.append(next_day) + + for shift in SHIFTS: + extra = fill_missing_shift_coverage( + connection, company_id, day_of_week, shift + ) + if extra is not None: + created.append(extra) + + return created def assign_preferred_shift(connection, user_id, day_of_week, preferred_shift): - # validate employee can join selected shift - # call storage.add_schedule when assignment is valid - # return validation error if assignment is not valid - pass + day = _value(day_of_week) + shift = _value(preferred_shift) + if not validate_employee_can_join_shift(connection, user_id, day, shift): + return None + company_id = storage.get_user_company_id(connection, user_id) + if storage.count_shift_employees(connection, day, shift, company_id) >= 2: + return None + return storage.add_schedule(connection, user_id, datetime.now(), day, None, shift) def fill_missing_shift_coverage(connection, company_id, day_of_week, shift): - # find company shift with fewer than two employees - # if only one employee is assigned, search for another valid employee - # randomly choose or rotate through employees who have not worked five days - # add the selected employee to the schedule - pass - - -def register_clock_in(connection, user_id, day_of_week, enter_time): - # reject if user already worked the same day - # reject if user already worked five days this week - # save enter_time through storage.add_schedule or update an open schedule row - pass - - -def register_clock_out(connection, user_id, day_of_week, out_time): - # call storage.update_day_end_by_user_and_day - # keep the existing schedule row and update out_time only - pass - - -def print_final_schedule(connection, company_id): - # query final schedule with employee names - # group output by day and shift - # print employees assigned to each shift - # print warnings for shifts with fewer than two employees - pass + return conflict.resolve_missing_second_employee( + connection, company_id, _value(day_of_week), _value(shift) + ) diff --git a/src/scheduler/manager/validator.py b/src/scheduler/manager/validator.py index 905cf85..a27b059 100644 --- a/src/scheduler/manager/validator.py +++ b/src/scheduler/manager/validator.py @@ -1,38 +1,27 @@ -# Validation guide for schedule rules. -# Import scheduler.database.storage query functions before implementing these checks. +from scheduler.database import storage def validate_user_has_not_worked_same_day(connection, user_id, day_of_week): - # query schedules for user_id and day_of_week - # if a schedule exists, reject the new assignment - # otherwise allow scheduler to continue - pass + return storage.get_user_enter_time_by_day(connection, user_id, day_of_week) is None def validate_user_has_not_worked_more_than_five_days(connection, user_id): - # call storage.get_days_worked_by_user(connection, user_id) - # count distinct worked days with a loop - # reject if count is five or more - # otherwise allow assignment - pass + return len(storage.get_days_worked_by_user(connection, user_id)) < 5 -def validate_shift_has_minimum_two_employees(connection, company_id, day_of_week, shift): - # query schedules for company_id, day_of_week, and shift - # count assigned employees - # if fewer than two, scheduler must add another available employee - pass +def validate_shift_has_minimum_two_employees( + connection, company_id, day_of_week, shift +): + return storage.count_shift_employees( + connection, day_of_week, shift, company_id + ) >= 2 def validate_employee_can_join_shift(connection, user_id, day_of_week, shift): - # check one-shift-per-day rule first - # check five-days-per-week rule second - # check shift coverage and capacity rules last - pass + return validate_user_has_not_worked_same_day( + connection, user_id, day_of_week + ) and validate_user_has_not_worked_more_than_five_days(connection, user_id) def validate_company_has_enough_employees(connection, company_id): - # get employee names by company - # reject schedule generation if company has fewer than two employees - # otherwise continue schedule generation - pass + return len(storage.get_company_employee_ids(connection, company_id)) >= 2 diff --git a/src/scheduler/models/employee.py b/src/scheduler/models/employee.py index f4a9574..83628d0 100644 --- a/src/scheduler/models/employee.py +++ b/src/scheduler/models/employee.py @@ -1,13 +1,18 @@ +from scheduler.core.enums import WorkTime +from scheduler.core.schemas.employee import DayPreferenceRequest from scheduler.database import storage def define_employee_fields(): - # include id from users table - # include company_id from company_users table - # include name and last_name from users table - # include preferred shift for each day of the week - # include optional priority ranking for bonus preference support - pass + return { + "id": int, + "company_id": int, + "name": str, + "last_name": str, + "preference": WorkTime, + "weekly_preferences": list[DayPreferenceRequest], + "priority": int | None, + } def register_employee_preference(connection, user_id, preference): diff --git a/src/scheduler/models/schedule.py b/src/scheduler/models/schedule.py index 882520e..f026fb2 100644 --- a/src/scheduler/models/schedule.py +++ b/src/scheduler/models/schedule.py @@ -1,23 +1,80 @@ +from datetime import datetime + +from scheduler.core.enums import DayOfWeek, WorkTime from scheduler.database import storage +from scheduler.database.schedule_repository import ScheduleRepository +from scheduler.manager.validator import validate_employee_can_join_shift def define_schedule_fields(): - # include user_id to identify the employee - # include company_id if schedules need to be filtered by company - # include day_of_week for the assigned day - # include shift value for morning, afternoon, or evening - # include enter_time and out_time for clock-in and clock-out - pass + return { + "user_id": int, + "company_id": int, + "day_of_week": DayOfWeek, + "shift": WorkTime, + "enter_time": datetime, + "out_time": datetime | None, + } -def register_user_enter_and_out(connection, user_id, enter_time, day_of_week, out_time): - # Add manager validation before saving once scheduling rules are complete. - return storage.add_schedule(connection, user_id, enter_time, day_of_week, out_time) +def _value(value): + return value.value if hasattr(value, "value") else value -def output_weekly_schedule(): - # group saved schedules by day of week - # inside each day, group schedules by shift - # list employee names for each shift - # show missing coverage when a shift has fewer than two employees - pass +def register_user_enter_and_out( + connection, user_id, enter_time, day_of_week, out_time, shift +): + if not validate_employee_can_join_shift( + connection, user_id, _value(day_of_week), _value(shift) + ): + return None + return storage.add_schedule( + connection, + user_id, + enter_time, + _value(day_of_week), + out_time, + _value(shift), + ) + + +def output_weekly_schedule(connection=None, company_id=None): + should_close = False + if connection is None: + connection = storage.get_connection() + should_close = True + + try: + repository = ScheduleRepository(connection) + schedules = repository.list_weekly(company_id) + + grouped: dict[str, dict[str, list[str]]] = {} + for schedule in schedules: + day = _value(schedule["day_of_week"]) + shift = _value(schedule["shift"]) + name = f"{schedule['user']['name']} {schedule['user']['last_name']}" + grouped.setdefault(day, {}).setdefault(shift, []).append(name) + + days_order = [d.value for d in DayOfWeek] + shifts_order = [s.value for s in WorkTime] + + for day in days_order: + shifts = grouped.get(day, {}) + if not shifts: + continue + print(f"{day}:") + for shift in shifts_order: + names = shifts.get(shift, []) + if not names: + continue + print(f" {shift}: {', '.join(names)}") + if len(names) < 2: + print( + f" warning: {shift} on {day} " + "has fewer than two employees" + ) + + return grouped + finally: + if should_close: + connection.close() diff --git a/src/scheduler/models/shift.py b/src/scheduler/models/shift.py index f75f536..eb3f4e8 100644 --- a/src/scheduler/models/shift.py +++ b/src/scheduler/models/shift.py @@ -1,22 +1,30 @@ -# Shift model guide. -# Keep shift names consistent with the PostgreSQL work_time enum. +from scheduler.core.enums import DayOfWeek, WorkTime +from scheduler.database import storage +from scheduler.manager.validator import validate_shift_has_minimum_two_employees def define_shift_values(): - # morning maps to work_time value "morning" - # afternoon can map to "evening" if you keep the Rust enum name - # evening maps to work_time value "evening" - pass + return tuple(s.value for s in WorkTime) def define_day_values(): - # use sunday through saturday to match the day_of_week enum - # loop over all seven days when generating the weekly schedule - pass + return tuple(d.value for d in DayOfWeek) -def compare_shift_capacity(): - # count employees assigned to one shift on one day - # if the count is fewer than two, scheduler must add another employee - # if the preferred shift is full, conflict manager must choose another shift - pass +def compare_shift_capacity(connection, company_id, day_of_week, shift): + shift_value = shift.value if hasattr(shift, "value") else shift + day_value = day_of_week.value if hasattr(day_of_week, "value") else day_of_week + count = storage.count_shift_employees( + connection, day_value, shift_value, company_id + ) + has_minimum = validate_shift_has_minimum_two_employees( + connection, company_id, day_value, shift_value + ) + return { + "count": count, + "has_minimum_two": has_minimum, + "needs_coverage": not has_minimum, + "is_full": count >= 2, + "should_add_employee": count < 2, + "should_resolve_conflict": count >= 2, + }