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

fixing the company and schedule endpoints

Browse Source
This commit is contained in:
Carlos
2026-09-18 17:14:12 -04:00
parent e7d887203e
commit 6bfdd25cd8
7 changed files with 231 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -15,6 +15,7 @@ dependencies = [
"psycopg[binary]>=3.2",
"python-dotenv>=1.0",
"uvicorn[standard]>=0.37",
"python-multipart>=0.0.9",
]
[project.urls]
src/scheduler/api/company_routes.py
+31 -4
View File
@@ -1,12 +1,15 @@
import codecs
import csv
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, FastAPI, File, HTTPException, UploadFile
from scheduler.api.dependencies import get_company_service
from scheduler.core.schemas.company import Company, CreateCompanyRequest
from scheduler.core.services.company_service import CompanyService
router = APIRouter(prefix="/companies", tags=["companies"])
app = FastAPI()
@router.get("", response_model=list[Company])
@@ -15,12 +18,34 @@ def list_companies(service: Annotated[CompanyService, Depends(get_company_servic
@router.post("", response_model=Company)
def create_company(payload: CreateCompanyRequest, service: Annotated[CompanyService, Depends(get_company_service)]):
def create_company(
payload: CreateCompanyRequest,
service: Annotated[CompanyService, Depends(get_company_service)],
):
return service.create_company(payload.name)
@router.post("/upload")
def upload(
service: Annotated[CompanyService, Depends(get_company_service)],
file: UploadFile = File(...),
):
csvReader = csv.DictReader(codecs.iterdecode(file.file, "utf-8"))
data = {}
idx = 0
for rows in csvReader:
service.create_company(rows["name"])
data[idx] = rows["name"]
idx += 1
file.file.close()
return data
@router.get("/{id}", response_model=Company)
def get_company(id: int, service: Annotated[CompanyService, Depends(get_company_service)]):
def get_company(
id: int, service: Annotated[CompanyService, Depends(get_company_service)]
):
company = service.get_company(id)
if company is None:
raise HTTPException(status_code=404, detail="Company not found")
@@ -28,7 +53,9 @@ def get_company(id: int, service: Annotated[CompanyService, Depends(get_company_
@router.delete("/{id}", response_model=Company)
def delete_company(id: int, service: Annotated[CompanyService, Depends(get_company_service)]):
def delete_company(
id: int, service: Annotated[CompanyService, Depends(get_company_service)]
):
company = service.delete_company(id)
if company is None:
raise HTTPException(status_code=404, detail="Company not found")
src/scheduler/api/employee_routes.py
+45 -8
View File
@@ -1,6 +1,8 @@
import codecs
import csv
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, FastAPI, File, HTTPException, UploadFile
from scheduler.api.dependencies import get_employee_service
from scheduler.core.schemas.company import CompanyUser
@@ -15,6 +17,7 @@ from scheduler.core.schemas.employee import (
from scheduler.core.services.employee_service import EmployeeService
router = APIRouter(tags=["employees"])
app = FastAPI()
@router.get("/employees", response_model=list[EmployeeResponse])
@@ -23,20 +26,46 @@ def list_employees(service: Annotated[EmployeeService, Depends(get_employee_serv
@router.post("/employees", response_model=EmployeeResponse)
def create_employee(payload: CreateEmployeeRequest, service: Annotated[EmployeeService, Depends(get_employee_service)]):
employee = service.create_employee(payload.company_id, payload.name, payload.last_name, payload.preference)
def create_employee(
payload: CreateEmployeeRequest,
service: Annotated[EmployeeService, Depends(get_employee_service)],
):
employee = service.create_employee(
payload.company_id, payload.name, payload.last_name, payload.preference
)
if employee is None:
raise HTTPException(status_code=404, detail="Company not found")
return employee
@router.post("/employees/upload")
def upload(
service: Annotated[EmployeeService, Depends(get_employee_service)],
file: UploadFile = File(...),
):
csvReader = csv.DictReader(codecs.iterdecode(file.file, "utf-8"))
data = {}
idx = 0
for rows in csvReader:
service.create_employee(
int(rows["company_id"]), rows["name"], rows["last_name"], rows["preference"]
)
data[idx] = rows["name"]
idx += 1
file.file.close()
return data
@router.put("/employees/{id}", response_model=EmployeeResponse)
def update_employee(
id: int,
payload: UpdateEmployeeRequest,
service: Annotated[EmployeeService, Depends(get_employee_service)],
):
employee = service.update_employee(id, payload.name, payload.last_name, payload.preference, payload.company_id)
employee = service.update_employee(
id, payload.name, payload.last_name, payload.preference, payload.company_id
)
if employee is None:
raise HTTPException(status_code=404, detail="Employee or company not found")
return employee
@@ -48,14 +77,18 @@ def patch_employee(
payload: UpdateEmployeeRequest,
service: Annotated[EmployeeService, Depends(get_employee_service)],
):
employee = service.update_employee(id, payload.name, payload.last_name, payload.preference, payload.company_id)
employee = service.update_employee(
id, payload.name, payload.last_name, payload.preference, payload.company_id
)
if employee is None:
raise HTTPException(status_code=404, detail="Employee or company not found")
return employee
@router.delete("/employees/{id}", response_model=EmployeeResponse)
def delete_employee(id: int, service: Annotated[EmployeeService, Depends(get_employee_service)]):
def delete_employee(
id: int, service: Annotated[EmployeeService, Depends(get_employee_service)]
):
employee = service.delete_employee(id)
if employee is None:
raise HTTPException(status_code=404, detail="Employee not found")
@@ -86,7 +119,9 @@ def register_weekly_employee_preferences(
return employee
@router.post("/companies/{company_id}/employees/{employee_id}", response_model=CompanyUser)
@router.post(
"/companies/{company_id}/employees/{employee_id}", response_model=CompanyUser
)
def assign_employee_to_company(
company_id: int,
employee_id: int,
@@ -98,7 +133,9 @@ def assign_employee_to_company(
return company_user
@router.get("/companies/{company_name}/employees/names", response_model=EmployeeNamesResponse)
@router.get(
"/companies/{company_name}/employees/names", response_model=EmployeeNamesResponse
)
def get_employee_names_by_company(
company_name: str,
service: Annotated[EmployeeService, Depends(get_employee_service)],
src/scheduler/api/schedule_routes.py
+53 -4
View File
@@ -1,6 +1,9 @@
import codecs
import csv
from datetime import datetime
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, FastAPI, File, HTTPException, UploadFile
from scheduler.api.dependencies import get_schedule_service
from scheduler.core.schemas.schedule import (
@@ -14,16 +17,60 @@ from scheduler.core.schemas.schedule import (
from scheduler.core.services.schedule_service import ScheduleService
router = APIRouter(tags=["schedules"])
app = FastAPI()
@router.post("/schedules", response_model=ScheduleResponse)
def create_schedule(payload: CreateScheduleRequest, service: Annotated[ScheduleService, Depends(get_schedule_service)]):
schedule = service.create_schedule(payload.user_id, payload.enter_time, payload.day_of_week, payload.out_time)
def create_schedule(
payload: CreateScheduleRequest,
service: Annotated[ScheduleService, Depends(get_schedule_service)],
):
schedule = service.create_schedule(
payload.user_id,
payload.enter_time,
payload.day_of_week,
payload.out_time,
payload.shift,
)
if schedule is None:
raise HTTPException(status_code=404, detail="Employee not found")
return schedule
@router.post("/schedules/upload")
def upload(
service: Annotated[ScheduleService, Depends(get_schedule_service)],
file: UploadFile = File(...),
):
csvReader = csv.DictReader(codecs.iterdecode(file.file, "utf-8-sig"))
data = {}
idx = 0
date_format = "%Y-%m-%d %H:%M:%S"
for rows in csvReader:
enter_time = datetime.strptime(rows["enter_time"], date_format)
out_time = datetime.strptime(rows["out_time"], date_format)
schedule = service.create_schedule(
int(rows["user_id"]),
enter_time,
rows["day_of_week"],
None,
rows["shift"],
)
if schedule is None:
continue
schedule = service.clock_in_schedule(
schedule["id"],
enter_time,
)
schedule = service.clock_out_schedule(schedule["id"], out_time)
data[idx] = schedule
idx += 1
file.file.close()
return data
@router.post("/employees/{id}/clock-in", response_model=ScheduleResponse)
def clock_in(
id: int,
@@ -53,7 +100,9 @@ def clock_out(
@router.get("/schedules/weekly", response_model=WeeklyScheduleResponse)
def get_weekly_schedule(service: Annotated[ScheduleService, Depends(get_schedule_service)]):
def get_weekly_schedule(
service: Annotated[ScheduleService, Depends(get_schedule_service)],
):
return service.get_weekly_schedule()
src/scheduler/core/schemas/schedule.py
+17 -2
View File
@@ -2,7 +2,7 @@ from datetime import datetime
from pydantic import BaseModel
from scheduler.core.enums import DayOfWeek
from scheduler.core.enums import DayOfWeek, WorkTime
class CreateScheduleRequest(BaseModel):
@@ -34,5 +34,20 @@ class ScheduleResponse(BaseModel):
out_time: datetime | None = None
class WeeklyScheduleUserResponse(BaseModel):
name: str
last_name: str
preference: WorkTime
class WeeklyScheduleItemResponse(BaseModel):
id: int
user: WeeklyScheduleUserResponse
company: str
enter_time: datetime
day_of_week: DayOfWeek
out_time: datetime | None = None
class WeeklyScheduleResponse(BaseModel):
schedules: list[ScheduleResponse]
schedules: list[WeeklyScheduleItemResponse]
src/scheduler/core/services/schedule_service.py
+18 -3
View File
@@ -16,10 +16,19 @@ class ScheduleService:
self.employee_repository = employee_repository
self.company_repository = company_repository
def create_schedule(self, user_id: int, enter_time: datetime, day_of_week: str, out_time: datetime):
def create_schedule(
self,
user_id: int,
enter_time: datetime,
day_of_week: str,
out_time: datetime,
shift: str,
):
if self.employee_repository.get(user_id) is None:
return None
return self.schedule_repository.create(user_id, enter_time, day_of_week, out_time)
return self.schedule_repository.create(
user_id, enter_time, day_of_week, out_time, shift
)
def clock_in(
self,
@@ -30,13 +39,19 @@ class ScheduleService:
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)
return self.schedule_repository.create(user_id, enter_time, day_of_week, None, None)
def clock_in_schedule(self, schedule_id: int, enter_time: datetime):
return self.schedule_repository.update_clock_in_by_id(schedule_id, enter_time)
def clock_out(self, user_id: int, day_of_week: str, out_time: datetime):
if self.employee_repository.get(user_id) is None:
return None
return self.schedule_repository.update_clock_out(user_id, day_of_week, out_time)
def clock_out_schedule(self, schedule_id: int, out_time: datetime):
return self.schedule_repository.update_clock_out_by_id(schedule_id, out_time)
def get_weekly_schedule(self, company_id: int | None = None):
return {"schedules": self.schedule_repository.list_weekly(company_id)}
src/scheduler/database/schedule_repository.py
+66 -8
View File
@@ -5,15 +5,22 @@ class ScheduleRepository:
def __init__(self, connection):
self.connection = connection
def create(self, user_id: int, enter_time: datetime, day_of_week: str, out_time: datetime | None):
def create(
self,
user_id: int,
enter_time: datetime,
day_of_week: str,
out_time: datetime | None,
shift: str | None,
):
with self.connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO schedules (user_id, enter_time, day_of_week, out_time)
VALUES (%s, %s, %s, %s)
RETURNING id, user_id, enter_time, day_of_week, out_time
INSERT INTO schedules (user_id, enter_time, day_of_week, out_time, shift)
VALUES (%s, %s, %s, %s, %s)
RETURNING id, user_id, enter_time, day_of_week, out_time, shift
""",
(user_id, enter_time, day_of_week, out_time),
(user_id, enter_time, day_of_week, out_time, shift),
)
schedule = cursor.fetchone()
self.connection.commit()
@@ -28,16 +35,37 @@ class ScheduleRepository:
with self.connection.cursor() as cursor:
cursor.execute(
f"""
SELECT schedules.id, schedules.user_id, schedules.enter_time,
schedules.day_of_week, schedules.out_time
SELECT schedules.id, schedules.enter_time, schedules.day_of_week,
schedules.out_time, users.name AS user_name,
users.last_name AS user_last_name,
users.preference AS user_preference,
companies.name AS company
FROM schedules
INNER JOIN company_users ON company_users.user_id = schedules.user_id
INNER JOIN users ON users.id = schedules.user_id
INNER JOIN companies ON companies.id = company_users.company_id
{where}
ORDER BY schedules.day_of_week, schedules.enter_time, schedules.id
""",
params,
)
return cursor.fetchall()
schedules = cursor.fetchall()
return [
{
"id": schedule["id"],
"user": {
"name": schedule["user_name"],
"last_name": schedule["user_last_name"],
"preference": schedule["user_preference"],
},
"company": schedule["company"],
"enter_time": schedule["enter_time"],
"day_of_week": schedule["day_of_week"],
"out_time": schedule["out_time"],
}
for schedule in schedules
]
def update_clock_out(self, user_id: int, day_of_week: str, out_time: datetime):
with self.connection.cursor() as cursor:
@@ -59,3 +87,33 @@ class ScheduleRepository:
schedule = cursor.fetchone()
self.connection.commit()
return schedule
def update_clock_in_by_id(self, schedule_id: int, enter_time: datetime):
with self.connection.cursor() as cursor:
cursor.execute(
"""
UPDATE schedules
SET enter_time = %s, date_updated = NOW()
WHERE id = %s
RETURNING id, user_id, enter_time, day_of_week, out_time, shift
""",
(enter_time, schedule_id),
)
schedule = cursor.fetchone()
self.connection.commit()
return schedule
def update_clock_out_by_id(self, schedule_id: int, out_time: datetime):
with self.connection.cursor() as cursor:
cursor.execute(
"""
UPDATE schedules
SET out_time = %s, date_updated = NOW()
WHERE id = %s
RETURNING id, user_id, enter_time, day_of_week, out_time, shift
""",
(out_time, schedule_id),
)
schedule = cursor.fetchone()
self.connection.commit()
return schedule