166 lines
5.2 KiB
Python
166 lines
5.2 KiB
Python
from scheduler.database.company_repository import CompanyRepository
|
|
from scheduler.database.database import Database
|
|
from scheduler.database.employee_repository import EmployeeRepository
|
|
from scheduler.database.schedule_repository import ScheduleRepository
|
|
|
|
|
|
def get_connection():
|
|
return Database().get_connection()
|
|
|
|
|
|
def initialize_schema(connection=None):
|
|
Database().initialize_schema()
|
|
|
|
|
|
def add_company(connection, name):
|
|
return CompanyRepository(connection).create(name)
|
|
|
|
|
|
def get_company_by_name(connection, name):
|
|
return CompanyRepository(connection).get_by_name(name)
|
|
|
|
|
|
def remove_company(connection, company_id):
|
|
return CompanyRepository(connection).delete(company_id)
|
|
|
|
|
|
def get_employee_names_by_company(connection, company_id):
|
|
return EmployeeRepository(connection).get_names_by_company(company_id)
|
|
|
|
|
|
def add_user(connection, name, last_name, preference):
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO users (name, last_name, preference)
|
|
VALUES (%s, %s, %s)
|
|
RETURNING id, name, last_name, preference, date_created, date_updated
|
|
""",
|
|
(name, last_name, preference),
|
|
)
|
|
user = cursor.fetchone()
|
|
connection.commit()
|
|
return user
|
|
|
|
|
|
def update_user_preference(connection, user_id, preference):
|
|
return EmployeeRepository(connection).update_preference(user_id, preference)
|
|
|
|
|
|
def add_company_user(connection, company_id, user_id):
|
|
return EmployeeRepository(connection).assign_to_company(company_id, user_id)
|
|
|
|
|
|
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, shift):
|
|
return ScheduleRepository(connection).create(
|
|
user_id, enter_time, day_of_week, out_time, shift
|
|
)
|
|
|
|
|
|
def _week_bounds(reference):
|
|
from datetime import timedelta, timezone
|
|
|
|
# Convert aware to naive UTC for DB comparison
|
|
if getattr(reference, "tzinfo", None) is not None:
|
|
reference = reference.astimezone(timezone.utc).replace(tzinfo=None)
|
|
# Sunday start
|
|
days_from_sunday = (reference.weekday() + 1) % 7
|
|
start_date = (reference - timedelta(days=days_from_sunday)).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
)
|
|
return start_date, start_date + timedelta(days=7)
|
|
|
|
|
|
def get_days_worked_by_user(connection, user_id, reference=None):
|
|
if reference is not None:
|
|
week_start, week_end = _week_bounds(reference)
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT DISTINCT day_of_week
|
|
FROM schedules
|
|
WHERE user_id = %s
|
|
AND enter_time >= %s AND enter_time < %s
|
|
ORDER BY day_of_week
|
|
""",
|
|
(user_id, week_start, week_end),
|
|
)
|
|
return [row["day_of_week"] for row in cursor.fetchall()]
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT DISTINCT day_of_week
|
|
FROM schedules
|
|
WHERE user_id = %s
|
|
ORDER BY day_of_week
|
|
""",
|
|
(user_id,),
|
|
)
|
|
return [row["day_of_week"] for row in cursor.fetchall()]
|
|
|
|
|
|
def get_user_enter_time_by_day(connection, user_id, day_of_week, reference=None):
|
|
if reference is not None:
|
|
week_start, week_end = _week_bounds(reference)
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT enter_time
|
|
FROM schedules
|
|
WHERE user_id = %s AND day_of_week = %s
|
|
AND enter_time >= %s AND enter_time < %s
|
|
ORDER BY date_created DESC
|
|
LIMIT 1
|
|
""",
|
|
(user_id, day_of_week, week_start, week_end),
|
|
)
|
|
row = cursor.fetchone()
|
|
return None if row is None else row["enter_time"]
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT enter_time
|
|
FROM schedules
|
|
WHERE user_id = %s AND day_of_week = %s
|
|
ORDER BY date_created DESC
|
|
LIMIT 1
|
|
""",
|
|
(user_id, day_of_week),
|
|
)
|
|
row = cursor.fetchone()
|
|
return None if row is None else row["enter_time"]
|
|
|
|
|
|
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
|
|
)
|
|
|
|
|
|
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
|
|
)
|