CarGDev/manager_employees_python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactoring unused functions

Browse Source
This commit is contained in:
Carlos
2026-09-18 20:46:32 -04:00
parent b10f345d21
commit e26eb302d0
13 changed files with 489 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/scheduler/api/schedule_routes.py
+10 -1
View File
@@ -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
src/scheduler/core/schemas/schedule.py
+4
View File
@@ -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):
src/scheduler/core/services/schedule_service.py
+21 -1
View File
@@ -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)
src/scheduler/database/database.py
+4
View File
@@ -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 (
src/scheduler/database/employee_repository.py
+50
View File
@@ -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()
src/scheduler/database/schedule_repository.py
+60 -2
View File
@@ -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"]
src/scheduler/database/storage.py
+31 -3
View File
@@ -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
)
src/scheduler/manager/conflict.py
+109 -20
View File
@@ -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
src/scheduler/manager/scheduler.py
+82 -43
View File
@@ -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)
)
src/scheduler/manager/validator.py
+13 -24
View File
@@ -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
src/scheduler/models/employee.py
+11 -6
View File
@@ -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):
src/scheduler/models/schedule.py
+72 -15
View File
@@ -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()
src/scheduler/models/shift.py
+22 -14
View File
@@ -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,
}