adding env example

fixing the checkin
refactoring unused functions
2026-09-20 11:11:39 -04:00 · 2026-09-18 21:59:44 -04:00 · 2026-09-18 20:46:32 -04:00 · 2026-09-18 19:14:19 -04:00 · 2026-09-18 17:14:12 -04:00 · 2026-09-18 12:51:31 -04:00
42 changed files with 2702 additions and 1 deletions
@@ -0,0 +1 @@
+DATABASE_URL=postgres://{username}:{pass}@{host}/{database_name}
@@ -0,0 +1,171 @@
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+### Python Patch ###
+# Poetry local configuration file - https://python-poetry.org/docs/configuration/#local-configuration
+poetry.toml
+
+# ruff
+.ruff_cache/
+
+# LSP config files
+pyrightconfig.json
@@ -1,2 +1,429 @@
-# manager_employees_python
+# manager_employees_python — Python Backend

+> **Assignment 4: Implementing Control Structures — Python Backend**
+> **Carlos Gutierrez** · `cgutierrez44833@ucumberlands.edu` · University of the Cumberlands — MSCS — Fall 2026
+> *This repository is the **Python backend** for Assignment 4. The full assignment is implemented in contrasting languages across a 3-repo suite.*
+
+[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue)](https://www.python.org/)
+[![FastAPI](https://img.shields.io/badge/FastAPI-0.118-009688)](https://fastapi.tiangolo.com/)
+[![PostgreSQL](https://img.shields.io/badge/PostgreSQL-15%2B-336791)](https://www.postgresql.org/)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
+
+---
+
+## Author
+
+**Carlos Gutierrez** — Graduate Student, Department of Computer Science, **University of the Cumberlands**
+
+- Email: `cgutierrez44833@ucumberlands.edu`
+- Git: `CarGDev`
+- License: MIT — Copyright (c) 2026 CarGDev (see `LICENSE`)
+
+---
+
+## Repositories — Project Suite
+
+This assignment requires implementing the same scheduling application in **two contrasting languages**. This repo is the **Python backend**. The complete suite is:
+
+| Repo | Role / Language | Description | Link |
+|------|-----------------|-------------|------|
+| **manager_employees_python** — **You are here** | **Python / Backend (FastAPI + Psycopg)** | REST API, business logic, validation, and persistence. This README documents this repo. | https://git.cargdev.io/CarGDev/manager_employees_python |
+| manager_employees_rust | **Rust / Backend** | Systems-language counterpart to contrast ownership/borrowing, `match`, and zero-cost abstractions against Python’s GC and dynamic typing. | https://git.cargdev.io/CarGDev/manager_employees_rust |
+| manager_employees_gui | **GUI (Bonus)** | Graphical input/output for employee names, shift preferences, and weekly schedule rendering. Connects to the Python/Rust backends. | https://git.cargdev.io/CarGDev/manager_employees_gui |
+
+```bash
+git clone https://git.cargdev.io/CarGDev/manager_employees_python
+git clone https://git.cargdev.io/CarGDev/manager_employees_rust
+git clone https://git.cargdev.io/CarGDev/manager_employees_gui
+```
+
+> ℹ️ **Scope of this README:** Only the **Python backend** (`manager_employees_python`). For the Rust implementation or the GUI bonus, see the links above.
+
+---
+
+## Brief Story — Assignment 4: Implementing Control Structures
+
+Assignment 4 challenged us to implement **without logical errors** the core control structures — conditionals (`if`/`elif`/`else`), loops (`for`, `while`), and branching (`match`/`switch`, early returns, error handling) — in **multiple, paradigmatically different languages**.
+
+I chose **Python for the backend** because its explicit loop/conditional syntax and FastAPI’s dependency-injection make the five scheduling rules highly readable and testable, while **Rust** forces manual ownership, strict `match` exhaustiveness, and compile-time borrowing — a deliberate contrast that highlights how the same requirements map to a high-level scripting paradigm vs. a systems paradigm.
+
+For the Python backend, I structured the code so every requirement is traceable to a control-structure site: nested `for day in WEEK` / `for shift in SHIFTS` loops in `src/scheduler/manager/scheduler.py`, `if count >= 2` / `if count >= 5` guards in `src/scheduler/manager/validator.py`, and `same-day → next-day` fallback branching in `src/scheduler/manager/conflict.py`. PostgreSQL persists the data, FastAPI exposes it, and a Swagger UI provides immediate visual feedback — keeping the focus on correct control flow rather than framework magic. The optional GUI bonus lives in `manager_employees_gui`, keeping this repo focused on backend correctness.
+
+---
+
+## 1. Overview
+
+> You will implement, without logical errors, various control structures such as conditionals, loops, and branching, in multiple programming languages. This assignment will demonstrate your mastery of syntax and control flow in different paradigms.
+
+Learning objectives:
+
+- Correct use of `if`/`else`, `for`/`while`, `match`/`switch`, and branching/early-returns across languages.
+- Designing data structures that support those control structures.
+- Validating scheduling invariants without logical gaps.
+
+---
+
+## 2. Application Prompt
+
+You are tasked with developing a small application for managing employee schedules at a company. The company operates **7 days a week** and allows employees to choose shifts in **morning, afternoon, or evening**.
+
+### 2.1 Input and Storage
+
+Collect employee names and their preferred shifts (morning/afternoon/evening) for each day. Store in an appropriate data structure.
+
+**Python backend mapping:** `POST /employees` (`src/scheduler/api/employee_routes.py:25`), schemas `CreateEmployeeRequest` / `DayPreferenceRequest` / `WeeklyPreferenceRequest` (`src/scheduler/core/schemas/employee.py:6-25`), enums `WorkTime` (`src/scheduler/core/enums/work_time.py:4`) and `DayOfWeek` (`src/scheduler/core/enums/day_of_week.py:4`), persistence via `users` + `weekly_preferences` tables (`src/scheduler/database/database.py:53-104`).
+
+### 2.2 Scheduling Logic
+
+Assign shifts ensuring:
+
+- No employee works more than **one shift per day**.
+- An employee can work a maximum of **5 days per week**.
+- At least **2 employees per shift per day**. If fewer, randomly assign additional available employees who have not hit 5 days.
+
+**Mapping:** `generate_weekly_schedule()` guide (`src/scheduler/manager/scheduler.py:5-16`) → `validate_user_has_not_worked_same_day` / `validate_user_has_not_worked_more_than_five_days` / `validate_shift_has_minimum_two_employees` (`src/scheduler/manager/validator.py:5-37`), service `ScheduleService.generate_weekly_schedule()` (`src/scheduler/core/services/schedule_service.py:43`).
+
+### 2.3 Shift Conflicts
+
+Detect and resolve where preferred shift is full: assign to another available shift on the **same or next day**.
+
+**Mapping:** `detect_preferred_shift_conflict` / `resolve_conflict_same_day` / `resolve_conflict_next_day` / `resolve_missing_second_employee` (`src/scheduler/manager/conflict.py:5-32`), fallback `fill_missing_shift_coverage()` (`src/scheduler/manager/scheduler.py:26-31`).
+
+### 2.4 Output
+
+After scheduling, output the final weekly schedule in readable format indicating employee → shift → day.
+
+**Mapping:** `output_weekly_schedule()` (`src/scheduler/models/schedule.py:18-23`), `print_final_schedule()` (`src/scheduler/manager/scheduler.py:47-52`), endpoint `GET /schedules/weekly` (`src/scheduler/api/schedule_routes.py:55`) returning `WeeklyScheduleResponse` grouped by `day_of_week` → `shift`.
+
+### 2.5 Bonus (Optional)
+
+Allow priority ranking (e.g., 1st=morning, 2nd=evening) and accommodate as much as possible while meeting coverage.
+
+**Mapping:** `weekly_preferences` table with `first`/`second`/`third` (`src/scheduler/database/database.py:92-104`), `DayPreferenceRequest.first/second/third` (`src/scheduler/core/schemas/employee.py:17-21`), `PUT /employees/{id}/weekly-preferences` (`src/scheduler/api/employee_routes.py:77`), repo persists and replays preferences during generation.
+
+In addition, students may implement a **GUI** for input/output in one language — see `manager_employees_gui` (https://git.cargdev.io/CarGDev/manager_employees_gui). This Python backend exposes Swagger UI as the backend’s visual interface; full GUI is in the companion repo. TUI stubs exist at `src/scheduler/tui/` (WIP).
+
+---
+
+## 3. Programming Language Suggestions & Contrast
+
+Prompt allows any two **distinctly different** languages (e.g., not C++ and C#). Suggested: **Python, Java, C++, JavaScript, Go, Rust**.
+
+**Chosen for this suite:**
+
+| Language | Repo | Paradigm | Why it contrasts |
+|----------|------|----------|------------------|
+| **Python** (this repo) | `manager_employees_python` | Dynamic, GC, interpreted, FastAPI, Pydantic validation | Concise loops/comprehensions, runtime exceptions, DI via `Depends` (`src/scheduler/api/dependencies.py:25`) |
+| **Rust** | `manager_employees_rust` | Static, ownership/borrowing, compiled, `match` exhaustiveness | Compile-time guarantees, explicit `Result`/`Option` branching, no GC |
+| GUI | `manager_employees_gui` | Frontend/GUI layer | Visualizes the bonus requirement; consumes either backend |
+
+Contrast is visible in how the same `if count >= 5 { reject }` / `for day in days { for shift in shifts }` logic is expressed.
+
+---
+
+## Features — Python Backend
+
+- **Company CRUD** — `CompanyService` (`src/scheduler/core/services/company_service.py:5`) + `CompanyRepository` (`src/scheduler/database/company_repository.py:1`)
+- **Employee CRUD** — create/update/delete employees, assign to company, patch, preference updates (`src/scheduler/core/services/employee_service.py:5`, `src/scheduler/database/employee_repository.py:1`)
+- **Weekly Preferences** — ranked `first/second/third` per day (`PUT /employees/{id}/weekly-preferences`)
+- **Schedules** — direct create, `clock-in` / `clock-out` per day, weekly view/generate (`src/scheduler/core/services/schedule_service.py:8`, `src/scheduler/database/schedule_repository.py:1`)
+- **Health Check** — `GET /health` (`src/scheduler/api/health_routes.py:8`)
+- **OpenAPI** — auto-generated at `/api-docs/openapi.json` + `/swagger-ui/` + alias `/api-docs/openai.json` (`src/scheduler/app.py:22-41`)
+
+---
+
+## Tech Stack — Python Backend
+
+| Layer | Technology | Version | Source |
+|-------|------------|---------|--------|
+| Language | Python | `>=3.10` | `pyproject.toml:9` |
+| Framework | FastAPI | `>=0.118` | `pyproject.toml:11` |
+| Server | Uvicorn (standard) | `>=0.37` | `pyproject.toml:14` |
+| DB Driver | psycopg (binary) | `>=3.2` | `pyproject.toml:12` |
+| Env | python-dotenv | `>=1.0` | `pyproject.toml:13` |
+| DB | PostgreSQL | 15+ | `src/scheduler/database/database.py:7` |
+| Validation | Pydantic | via FastAPI | `src/scheduler/core/schemas/*` |
+| Entry | `scheduler` CLI | — | `pyproject.toml:18` `src/scheduler/__main__.py:6` |
+
+---
+
+## Architecture — Python Backend
+
+```
+Client (curl / GUI / Rust) → FastAPI app.py:create_app()
+  → APIRouter (api/*_routes.py)
+    → Service (core/services/*)  — business rules + control structures
+      → Repository (database/*_repository.py + storage.py)
+        → PostgreSQL (Database.initialize_schema())
+```
+
+**Project Structure**
+
+```
+manager_employees_python/
+├── pyproject.toml               # employee-manager, deps, scheduler entry
+├── package.json                 # mirror author metadata (Carlos Gutierrez)
+├── src/scheduler/
+│   ├── app.py                   # create_app(), lifespan, routers, CORS, OpenAPI paths
+│   ├── main.py                  # re-exports app
+│   ├── __main__.py              # uvicorn run (APP_HOST/APP_PORT)
+│   ├── api/
+│   │   ├── company_routes.py    # /companies CRUD
+│   │   ├── employee_routes.py   # /employees + preference + company assignment
+│   │   ├── schedule_routes.py   # /schedules + clock-in/out + weekly
+│   │   ├── health_routes.py     # /health
+│   │   └── dependencies.py      # get_company/employee/schedule_service (Generator DI)
+│   ├── core/
+│   │   ├── enums/ {day_of_week.py, work_time.py}
+│   │   ├── schemas/ {company.py, employee.py, schedule.py, health.py}
+│   │   └── services/ {company_service.py, employee_service.py, schedule_service.py}
+│   ├── database/
+│   │   ├── database.py          # Database class + DDL (enums, tables)
+│   │   ├── company_repository.py
+│   │   ├── employee_repository.py
+│   │   ├── schedule_repository.py
+│   │   └── storage.py           # legacy helper wrappers
+│   ├── manager/
+│   │   ├── scheduler.py         # generation guides (loops, coverage)
+│   │   ├── validator.py         # one-shift/day, five-days/week, min-2
+│   │   └── conflict.py          # same-day → next-day resolution
+│   ├── models/ {employee.py, schedule.py, shift.py}  # guide stubs
+│   └── tui/{screens,styles}/    # GUI stub (WIP, see manager_employees_gui)
+└── .env                         # DATABASE_URL
+```
+
+---
+
+## Database Schema
+
+Auto-created on startup via `Database().initialize_schema()` in `app.lifespan` (`src/scheduler/app.py:12`).
+
+**Enums** (`src/scheduler/database/database.py:22-41`):
+
+```sql
+CREATE TYPE work_time AS ENUM ('morning','afternoon','evening');
+CREATE TYPE day_of_week AS ENUM ('sunday','monday','tuesday','wednesday','thursday','friday','saturday');
+```
+
+**Tables**:
+
+- `companies(id, name UNIQUE, date_created, date_updated)` (`database.py:43-50`)
+- `users(id, name, last_name, preference work_time, date_*)` (`database.py:53-62`)
+- `company_users(id, user_id FK users, company_id FK companies, UNIQUE(user_id, company_id))` (`database.py:65-74`)
+- `schedules(id, user_id FK users, enter_time, day_of_week, out_time NULLABLE, date_*)` (`database.py:77-89`)
+- `weekly_preferences(id, user_id FK, day_of_week, first/second/third work_time, UNIQUE(user_id, day_of_week))` (`database.py:92-104`)
+
+Default URL: `postgres://carlos:password@localhost/rust_manager_employees` (`database.py:7`, `.env:1`).
+
+---
+
+## API Reference — Python Backend
+
+Base URL default: `http://127.0.0.1:5006` (`src/scheduler/__main__.py:7-9`) — Swagger at `/swagger-ui/`, OpenAPI at `/api-docs/openapi.json`.
+
+### Health
+
+| Method | Path | Description | Source |
+|--------|------|-------------|--------|
+| GET | `/health` | Liveness check `{status:"ok"}` | `api/health_routes.py:8` |
+
+### Companies
+
+| Method | Path | Description |
+|--------|------|-------------|
+| GET | `/companies` | List all companies | `api/company_routes.py:12` |
+| POST | `/companies` | Create `{name}` | `api/company_routes.py:17` |
+| GET | `/companies/{id}` | Get by id | `api/company_routes.py:22` |
+| DELETE | `/companies/{id}` | Delete by id | `api/company_routes.py:30` |
+
+### Employees
+
+| Method | Path | Description |
+|--------|------|-------------|
+| GET | `/employees` | List employees | `api/employee_routes.py:20` |
+| POST | `/employees` | Create `{company_id, name, last_name, preference}` | `api/employee_routes.py:25` |
+| PUT | `/employees/{id}` | Full update `{company_id?, name, last_name, preference}` | `api/employee_routes.py:33` |
+| PATCH | `/employees/{id}` | Patch (same as PUT) | `api/employee_routes.py:45` |
+| DELETE | `/employees/{id}` | Delete | `api/employee_routes.py:57` |
+| PUT | `/employees/{id}/preference` | Update single preference | `api/employee_routes.py:65` |
+| PUT | `/employees/{id}/weekly-preferences` | Ranked 7-day prefs `{preferences:[{day_of_week,first,second,third}]}` | `api/employee_routes.py:77` |
+| POST | `/companies/{company_id}/employees/{employee_id}` | Assign to company | `api/employee_routes.py:89` |
+| GET | `/companies/{company_name}/employees/names` | Names by company | `api/employee_routes.py:101` |
+
+### Schedules
+
+| Method | Path | Description |
+|--------|------|-------------|
+| POST | `/schedules` | Create `{user_id, enter_time, day_of_week, out_time}` | `api/schedule_routes.py:19` |
+| POST | `/employees/{id}/clock-in` | Clock in `{day_of_week, enter_time}` | `api/schedule_routes.py:27` |
+| PUT | `/employees/{id}/clock-out` | Clock out `{day_of_week, out_time}` | `api/schedule_routes.py:43` |
+| GET | `/schedules/weekly` | Weekly schedule `{schedules:[]}` | `api/schedule_routes.py:55` |
+| POST | `/schedules/weekly/generate` | Generate for company `{company_id}` | `api/schedule_routes.py:60` |
+
+Schemas: `core/schemas/company.py`, `employee.py:6-44`, `schedule.py:8-38`.
+
+---
+
+## Getting Started — Python Backend
+
+### Prerequisites
+
+- Python `>=3.10` (`pyproject.toml:9`)
+- PostgreSQL 15+ running locally
+- `psql` for DB creation (optional)
+
+### Environment
+
+Create `.env` at repo root (see `.env:1`):
+
+```env
+DATABASE_URL=postgres://carlos:password@localhost/rust_manager_employees
+APP_HOST=127.0.0.1
+APP_PORT=5006
+```
+
+Set `DATABASE_URL` to your local Postgres if different (fallback default `database.py:7` is the same).
+
+### Install
+
+```bash
+# clone this backend
+git clone https://git.cargdev.io/CarGDev/manager_employees_python
+cd manager_employees_python
+
+# venv
+python3 -m venv .venv
+source .venv/bin/activate
+
+# install (editable, uses src layout pyproject.toml:20-22)
+pip install -e .
+
+# or directly
+pip install "fastapi>=0.118" "psycopg[binary]>=3.2" "python-dotenv>=1.0" "uvicorn[standard]>=0.37"
+```
+
+### Database Setup
+
+```bash
+# create DB (name matches .env / database.py:7)
+createdb rust_manager_employees
+# or: psql -c "CREATE DATABASE rust_manager_employees;"
+
+# schema auto-creates on app startup via lifespan (app.py:12)
+# to force: python -c "from scheduler.database import Database; Database().initialize_schema()"
+```
+
+### Run — Python Backend
+
+```bash
+# via entry point (pyproject.toml:18)
+scheduler
+# → uvicorn scheduler.main:app --host 127.0.0.1 --port 5006
+
+# or directly
+python -m scheduler
+# or
+uvicorn scheduler.main:app --host 127.0.0.1 --port 5006 --reload
+
+# docs
+open http://127.0.0.1:5006/swagger-ui/
+open http://127.0.0.1:5006/api-docs/openapi.json
+```
+
+---
+
+## Usage Examples
+
+```bash
+BASE=http://127.0.0.1:5006
+
+# health
+curl $BASE/health
+
+# 1) company
+curl -X POST $BASE/companies -H "Content-Type: application/json" \
+  -d '{"name":"Acme Corp"}'
+
+# 2) employees
+curl -X POST $BASE/employees -H "Content-Type: application/json" \
+  -d '{"company_id":1,"name":"Ada","last_name":"Lovelace","preference":"morning"}'
+curl -X POST $BASE/employees -H "Content-Type: application/json" \
+  -d '{"company_id":1,"name":"Alan","last_name":"Turing","preference":"evening"}'
+curl -X POST $BASE/employees -H "Content-Type: application/json" \
+  -d '{"company_id":1,"name":"Grace","last_name":"Hopper","preference":"afternoon"}'
+
+# 3) weekly ranked preferences (bonus)
+curl -X PUT $BASE/employees/1/weekly-preferences -H "Content-Type: application/json" \
+  -d '{"preferences":[
+    {"day_of_week":"monday","first":"morning","second":"afternoon","third":"evening"},
+    {"day_of_week":"tuesday","first":"morning","second":"evening","third":"afternoon"}
+  ]}'
+
+# 4) clock in / out
+curl -X POST $BASE/employees/1/clock-in -H "Content-Type: application/json" \
+  -d '{"day_of_week":"monday","enter_time":"2026-09-18T08:00:00"}'
+curl -X PUT $BASE/employees/1/clock-out -H "Content-Type: application/json" \
+  -d '{"day_of_week":"monday","out_time":"2026-09-18T16:00:00"}'
+
+# 5) weekly schedule
+curl $BASE/schedules/weekly
+curl -X POST $BASE/schedules/weekly/generate -H "Content-Type: application/json" \
+  -d '{"company_id":1}'
+
+# 6) employee names by company
+curl $BASE/companies/Acme%20Corp/employees/names
+```
+
+---
+
+## Scheduling Logic — Control Structures in Code
+
+| Requirement | Control Structure | Code Location |
+|-------------|-------------------|---------------|
+| Loop over 7 days × 3 shifts | `for day in days: for shift in shifts:` | `manager/scheduler.py:5-16` guide, `services/schedule_service.py:43` |
+| Preference-first assignment | `if preferred == shift: assign` | `manager/scheduler.py:19-23` |
+| One-shift-per-day guard | `if exists(user,day): reject` | `manager/validator.py:5-8` |
+| ≤5-days-per-week guard | `count distinct days; if >=5: reject` | `manager/validator.py:12-16`, `database/storage.py:62-73` |
+| ≥2-per-shift coverage | `if count < 2: add employee` | `manager/validator.py:20-23`, `manager/scheduler.py:26-31` |
+| Conflict same-day fallback | `for other_shift in day: if valid: assign` | `manager/conflict.py:12-16` |
+| Conflict next-day fallback | `for shift in next_day: if valid: assign` | `manager/conflict.py:19-23` |
+| Missing-second-employee | loop over company employees, skip invalid | `manager/conflict.py:26-32` |
+
+All guides are stubbed with `pass` and comments indicating the exact `storage.*` queries and validator calls to implement — ready for Assignment 4 grading of logical correctness.
+
+---
+
+## GUI Bonus
+
+The prompt’s GUI bonus is implemented in the companion repo:
+
+- **GUI Repo:** https://git.cargdev.io/CarGDev/manager_employees_gui
+- This Python backend serves the GUI via REST; Swagger UI at `/swagger-ui/` is the backend’s immediate visual output.
+- TUI placeholders `src/scheduler/tui/screens/` and `styles/` are reserved for an in-repo TUI if needed, but the deliverable GUI lives in `manager_employees_gui`.
+
+---
+
+## Testing
+
+```bash
+# no test suite committed yet — manual via curl / Swagger
+curl http://127.0.0.1:5006/health
+curl http://127.0.0.1:5006/companies
+curl http://127.0.0.1:5006/employees
+```
+
+Add `pytest` + `httpx` and test `create_app()` (`app.py:17`) with dependency overrides for `get_db` (`api/dependencies.py:17`) when ready.
+
+---
+
+## License
+
+MIT — Copyright (c) 2026 **Carlos Gutierrez (CarGDev)** — see `LICENSE`.
+
+---
+
+## Acknowledgments
+
+- Assignment prompt: *Assignment 4 — Implementing Control Structures* (University of the Cumberlands)
+- Framework docs: FastAPI, Uvicorn, Psycopg, Pydantic
+- Repo suite: [Python Backend](https://git.cargdev.io/CarGDev/manager_employees_python) · [Rust Backend](https://git.cargdev.io/CarGDev/manager_employees_rust) · [GUI](https://git.cargdev.io/CarGDev/manager_employees_gui)
@@ -0,0 +1,34 @@
+{
+  "name": "manager_employees_python",
+  "version": "1.0.0",
+  "description": "Application to manage the employees schedule. — Python backend for Assignment 4: Implementing Control Structures",
+  "author": "Carlos Gutierrez <cgutierrez44833@ucumberlands.edu>",
+  "maintainers": [
+    {
+      "name": "Carlos Gutierrez",
+      "email": "cgutierrez44833@ucumberlands.edu"
+    }
+  ],
+  "repository": {
+    "type": "git",
+    "url": "https://git.cargdev.io/CarGDev/manager_employees_python"
+  },
+  "homepage": "https://git.cargdev.io/CarGDev/manager_employees_python",
+  "license": "MIT",
+  "keywords": [
+    "python",
+    "fastapi",
+    "backend",
+    "employee-scheduling",
+    "control-structures",
+    "assignment-4",
+    "university-of-the-cumberlands",
+    "psycopg",
+    "postgresql",
+    "uvicorn",
+    "rest-api",
+    "scheduling-logic",
+    "shift-management"
+  ],
+  "private": true
+}
@@ -0,0 +1,30 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "employee-manager"
+version = "1.0.0"
+description = "Application to manage the employees schedule."
+authors = [{name = "Carlos Gutierrez", email = "cgutierrez44833@ucumberlands.edu"}]
+maintainers = [{name = "Carlos Gutierrez", email = "cgutierrez44833@ucumberlands.edu"}]
+keywords = ["python", "fastapi", "backend", "employee-scheduling", "control-structures", "assignment-4", "university-of-the-cumberlands", "psycopg", "postgresql", "uvicorn", "rest-api", "scheduling-logic", "shift-management"]
+requires-python = ">=3.10"
+dependencies = [
+    "fastapi>=0.118",
+    "psycopg[binary]>=3.2",
+    "python-dotenv>=1.0",
+    "uvicorn[standard]>=0.37",
+    "python-multipart>=0.0.9",
+]
+
+[project.urls]
+Homepage = "https://git.cargdev.io/CarGDev/manager_employees_python"
+Repository = "https://git.cargdev.io/CarGDev/manager_employees_python"
+
+[project.scripts]
+scheduler = "scheduler.__main__:main"
+
+[tool.setuptools.packages.find]
+where=["src"]
+include = ["scheduler", "scheduler.*"]
@@ -0,0 +1,3 @@
+__version__ = "1.0.0"
+
+__all__ = ["__version__"]
@@ -0,0 +1,13 @@
+import os
+
+import uvicorn
+
+
+def main():
+    host = os.environ.get("APP_HOST", "127.0.0.1")
+    port = int(os.environ.get("APP_PORT", "5006"))
+    uvicorn.run("scheduler.main:app", host=host, port=port)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,8 @@
+from . import company_routes, employee_routes, health_routes, schedule_routes
+
+__all__ = [
+    "company_routes",
+    "employee_routes",
+    "health_routes",
+    "schedule_routes",
+]
@@ -0,0 +1,62 @@
+import codecs
+import csv
+from typing import Annotated
+
+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])
+def list_companies(service: Annotated[CompanyService, Depends(get_company_service)]):
+    return service.list_companies()
+
+
+@router.post("", response_model=Company)
+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)]
+):
+    company = service.get_company(id)
+    if company is None:
+        raise HTTPException(status_code=404, detail="Company not found")
+    return company
+
+
+@router.delete("/{id}", response_model=Company)
+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")
+    return company
@@ -0,0 +1,50 @@
+from collections.abc import Generator
+
+from dotenv import load_dotenv
+
+from scheduler.core.services.company_service import CompanyService
+from scheduler.core.services.employee_service import EmployeeService
+from scheduler.core.services.schedule_service import ScheduleService
+from scheduler.database import CompanyRepository, Database, EmployeeRepository, ScheduleRepository
+
+
+
+def get_database() -> Database:
+    load_dotenv()
+    return Database()
+
+
+def get_db() -> Generator:
+    connection = get_database().get_connection()
+    try:
+        yield connection
+    finally:
+        connection.close()
+
+
+def get_company_service() -> Generator:
+    connection = get_database().get_connection()
+    try:
+        yield CompanyService(CompanyRepository(connection))
+    finally:
+        connection.close()
+
+
+def get_employee_service() -> Generator:
+    connection = get_database().get_connection()
+    try:
+        yield EmployeeService(EmployeeRepository(connection), CompanyRepository(connection))
+    finally:
+        connection.close()
+
+
+def get_schedule_service() -> Generator:
+    connection = get_database().get_connection()
+    try:
+        yield ScheduleService(
+            ScheduleRepository(connection),
+            EmployeeRepository(connection),
+            CompanyRepository(connection),
+        )
+    finally:
+        connection.close()
@@ -0,0 +1,146 @@
+import codecs
+import csv
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, FastAPI, File, HTTPException, UploadFile
+
+from scheduler.api.dependencies import get_employee_service
+from scheduler.core.schemas.company import CompanyUser
+from scheduler.core.schemas.employee import (
+    CreateEmployeeRequest,
+    EmployeeNamesResponse,
+    EmployeeResponse,
+    RegisterPreferenceRequest,
+    UpdateEmployeeRequest,
+    WeeklyPreferenceRequest,
+)
+from scheduler.core.services.employee_service import EmployeeService
+
+router = APIRouter(tags=["employees"])
+app = FastAPI()
+
+
+@router.get("/employees", response_model=list[EmployeeResponse])
+def list_employees(service: Annotated[EmployeeService, Depends(get_employee_service)]):
+    return service.list_employees()
+
+
+@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
+    )
+    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
+    )
+    if employee is None:
+        raise HTTPException(status_code=404, detail="Employee or company not found")
+    return employee
+
+
+@router.patch("/employees/{id}", response_model=EmployeeResponse)
+def patch_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
+    )
+    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)]
+):
+    employee = service.delete_employee(id)
+    if employee is None:
+        raise HTTPException(status_code=404, detail="Employee not found")
+    return employee
+
+
+@router.put("/employees/{id}/preference", response_model=EmployeeResponse)
+def register_employee_preference(
+    id: int,
+    payload: RegisterPreferenceRequest,
+    service: Annotated[EmployeeService, Depends(get_employee_service)],
+):
+    employee = service.update_preference(id, payload.preference)
+    if employee is None:
+        raise HTTPException(status_code=404, detail="Employee not found")
+    return employee
+
+
+@router.put("/employees/{id}/weekly-preferences", response_model=EmployeeResponse)
+def register_weekly_employee_preferences(
+    id: int,
+    payload: WeeklyPreferenceRequest,
+    service: Annotated[EmployeeService, Depends(get_employee_service)],
+):
+    employee = service.save_weekly_preferences(id, payload.preferences)
+    if employee is None:
+        raise HTTPException(status_code=404, detail="Employee not found")
+    return employee
+
+
+@router.post(
+    "/companies/{company_id}/employees/{employee_id}", response_model=CompanyUser
+)
+def assign_employee_to_company(
+    company_id: int,
+    employee_id: int,
+    service: Annotated[EmployeeService, Depends(get_employee_service)],
+):
+    company_user = service.assign_to_company(company_id, employee_id)
+    if company_user is None:
+        raise HTTPException(status_code=404, detail="Company or employee not found")
+    return company_user
+
+
+@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)],
+):
+    employees = service.get_names_by_company_name(company_name)
+    if employees is None:
+        raise HTTPException(status_code=404, detail="Company not found")
+    return {"employees": employees}
@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from scheduler.core.schemas.health import HealthResponse
+
+router = APIRouter(tags=["health"])
+
+
+@router.get("/health", response_model=HealthResponse)
+def health_check():
+    return {"status": "ok"}
@@ -0,0 +1,144 @@
+import codecs
+import csv
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, FastAPI, File, HTTPException, UploadFile
+
+from scheduler.api.dependencies import get_schedule_service
+from scheduler.core.schemas.schedule import (
+    ClockInRequest,
+    ClockOutRequest,
+    CreateScheduleRequest,
+    GenerateWeeklyScheduleRequest,
+    ScheduleResponse,
+    WeeklyScheduleResponse,
+)
+from scheduler.core.services.schedule_service import (
+    ALREADY_WORKED_DAY,
+    INVALID_DAY,
+    INVALID_TIME,
+    MAX_DAYS_REACHED,
+    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,
+        payload.shift,
+    )
+    if schedule is None:
+        raise HTTPException(status_code=404, detail="Employee not found")
+    if schedule == INVALID_DAY:
+        raise HTTPException(status_code=400, detail="day_of_week does not match enter_time")
+    if schedule == INVALID_TIME:
+        raise HTTPException(status_code=400, detail="out_time must be after enter_time")
+    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:
+        try:
+            enter_time = datetime.strptime(rows["enter_time"], date_format)
+            out_time = datetime.strptime(rows["out_time"], date_format)
+        except ValueError:
+            continue
+        # only validate out_time ordering; allow UTC vs local day drift
+        if out_time < enter_time:
+            continue
+        schedule = service.create_schedule(
+            int(rows["user_id"]),
+            enter_time,
+            rows["day_of_week"],
+            None,
+            rows["shift"],
+        )
+        if schedule is None or schedule in ("invalid_day", "invalid_time"):
+            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,
+    payload: ClockInRequest,
+    service: Annotated[ScheduleService, Depends(get_schedule_service)],
+):
+    schedule = service.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 == INVALID_DAY:
+        raise HTTPException(status_code=400, detail="day_of_week does not match enter_time")
+    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
+
+
+@router.put("/employees/{id}/clock-out", response_model=ScheduleResponse)
+def clock_out(
+    id: int,
+    payload: ClockOutRequest,
+    service: Annotated[ScheduleService, Depends(get_schedule_service)],
+):
+    schedule = service.clock_out(id, payload.day_of_week, payload.out_time)
+    if schedule is None:
+        raise HTTPException(status_code=404, detail="Schedule not found")
+    if schedule == INVALID_DAY:
+        raise HTTPException(status_code=400, detail="day_of_week does not match out_time")
+    if schedule == INVALID_TIME:
+        raise HTTPException(status_code=400, detail="out_time must be after enter_time")
+    return schedule
+
+
+@router.get("/schedules/weekly", response_model=WeeklyScheduleResponse)
+def get_weekly_schedule(
+    service: Annotated[ScheduleService, Depends(get_schedule_service)],
+):
+    return service.get_weekly_schedule()
+
+
+@router.post("/schedules/weekly/generate", response_model=WeeklyScheduleResponse)
+def generate_weekly_schedule(
+    payload: GenerateWeeklyScheduleRequest,
+    service: Annotated[ScheduleService, Depends(get_schedule_service)],
+):
+    schedule = service.generate_weekly_schedule(payload.company_id)
+    if schedule is None:
+        raise HTTPException(status_code=404, detail="Company not found")
+    return schedule
@@ -0,0 +1,46 @@
+from contextlib import asynccontextmanager
+
+from dotenv import load_dotenv
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from scheduler.api import company_routes, employee_routes, health_routes, schedule_routes
+from scheduler.database import Database
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    Database().initialize_schema()
+    yield
+
+
+def create_app() -> FastAPI:
+    load_dotenv()
+
+    app = FastAPI(
+        title="manager_employees_python",
+        lifespan=lifespan,
+        openapi_url="/api-docs/openapi.json",
+        docs_url="/swagger-ui/",
+    )
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    app.include_router(health_routes.router)
+    app.include_router(company_routes.router)
+    app.include_router(employee_routes.router)
+    app.include_router(schedule_routes.router)
+
+    @app.get("/api-docs/openai.json", include_in_schema=False)
+    def openai_json_alias():
+        return app.openapi()
+
+    return app
+
+
+app = create_app()
@@ -0,0 +1,4 @@
+from scheduler.core.enums.day_of_week import DayOfWeek
+from scheduler.core.enums.work_time import WorkTime
+
+__all__ = ["DayOfWeek", "WorkTime"]
@@ -0,0 +1,11 @@
+from enum import Enum
+
+
+class DayOfWeek(str, Enum):
+    SUNDAY = "sunday"
+    MONDAY = "monday"
+    TUESDAY = "tuesday"
+    WEDNESDAY = "wednesday"
+    THURSDAY = "thursday"
+    FRIDAY = "friday"
+    SATURDAY = "saturday"
@@ -0,0 +1,7 @@
+from enum import Enum
+
+
+class WorkTime(str, Enum):
+    MORNING = "morning"
+    AFTERNOON = "afternoon"
+    EVENING = "evening"
@@ -0,0 +1,22 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class CreateCompanyRequest(BaseModel):
+    name: str
+
+
+class Company(BaseModel):
+    id: int
+    name: str
+    date_created: datetime
+    date_updated: datetime
+
+
+class CompanyUser(BaseModel):
+    id: int
+    user_id: int
+    company_id: int
+    date_created: datetime
+    date_updated: datetime
@@ -0,0 +1,44 @@
+from pydantic import BaseModel
+
+from scheduler.core.enums import DayOfWeek, WorkTime
+
+
+class CreateEmployeeRequest(BaseModel):
+    company_id: int
+    name: str
+    last_name: str
+    preference: WorkTime
+
+
+class RegisterPreferenceRequest(BaseModel):
+    preference: WorkTime
+
+
+class DayPreferenceRequest(BaseModel):
+    day_of_week: DayOfWeek
+    first: WorkTime
+    second: WorkTime
+    third: WorkTime
+
+
+class WeeklyPreferenceRequest(BaseModel):
+    preferences: list[DayPreferenceRequest]
+
+
+class EmployeeNamesResponse(BaseModel):
+    employees: list[str]
+
+
+class UpdateEmployeeRequest(BaseModel):
+    company_id: int | None = None
+    name: str
+    last_name: str
+    preference: WorkTime
+
+
+class EmployeeResponse(BaseModel):
+    id: int
+    company_id: int
+    name: str
+    last_name: str
+    preference: WorkTime
@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class HealthResponse(BaseModel):
+    status: str
@@ -0,0 +1,57 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from scheduler.core.enums import DayOfWeek, WorkTime
+
+
+class CreateScheduleRequest(BaseModel):
+    user_id: int
+    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):
+    day_of_week: DayOfWeek
+    out_time: datetime
+
+
+class GenerateWeeklyScheduleRequest(BaseModel):
+    company_id: int
+
+
+class ScheduleResponse(BaseModel):
+    id: int
+    user_id: int
+    enter_time: datetime
+    day_of_week: DayOfWeek
+    out_time: datetime | None = None
+    shift: WorkTime
+
+
+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
+    shift: WorkTime
+
+
+class WeeklyScheduleResponse(BaseModel):
+    schedules: list[WeeklyScheduleItemResponse]
@@ -0,0 +1,18 @@
+from scheduler.database.company_repository import CompanyRepository
+
+
+class CompanyService:
+    def __init__(self, repository: CompanyRepository):
+        self.repository = repository
+
+    def list_companies(self):
+        return self.repository.list()
+
+    def create_company(self, name: str):
+        return self.repository.create(name)
+
+    def get_company(self, company_id: int):
+        return self.repository.get(company_id)
+
+    def delete_company(self, company_id: int):
+        return self.repository.delete(company_id)
@@ -0,0 +1,57 @@
+from scheduler.database.company_repository import CompanyRepository
+from scheduler.database.employee_repository import EmployeeRepository
+
+
+class EmployeeService:
+    def __init__(self, employee_repository: EmployeeRepository, company_repository: CompanyRepository):
+        self.employee_repository = employee_repository
+        self.company_repository = company_repository
+
+    def list_employees(self):
+        return self.employee_repository.list()
+
+    def create_employee(self, company_id: int, name: str, last_name: str, preference: str):
+        if self.company_repository.get(company_id) is None:
+            return None
+        return self.employee_repository.create(company_id, name, last_name, preference)
+
+    def delete_employee(self, employee_id: int):
+        return self.employee_repository.delete(employee_id)
+
+    def update_preference(self, employee_id: int, preference: str):
+        return self.employee_repository.update_preference(employee_id, preference)
+
+    def update_employee(
+        self,
+        employee_id: int,
+        name: str,
+        last_name: str,
+        preference: str,
+        company_id: int | None,
+    ):
+        if self.employee_repository.get(employee_id) is None:
+            return None
+        if company_id is not None and self.company_repository.get(company_id) is None:
+            return None
+        updated = self.employee_repository.update(employee_id, name, last_name, preference)
+        if company_id is not None:
+            updated = self.employee_repository.update_company(employee_id, company_id)
+        return updated
+
+    def assign_to_company(self, company_id: int, employee_id: int):
+        if self.company_repository.get(company_id) is None:
+            return None
+        if self.employee_repository.get(employee_id) is None:
+            return None
+        return self.employee_repository.assign_to_company(company_id, employee_id)
+
+    def get_names_by_company_name(self, company_name: str):
+        company = self.company_repository.get_by_name(company_name)
+        if company is None:
+            return None
+        return self.employee_repository.get_names_by_company(company["id"])
+
+    def save_weekly_preferences(self, employee_id: int, preferences: list):
+        if self.employee_repository.get(employee_id) is None:
+            return None
+        return self.employee_repository.save_weekly_preferences(employee_id, preferences)
@@ -0,0 +1,132 @@
+from datetime import datetime, timezone
+
+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"
+INVALID_DAY = "invalid_day"
+INVALID_TIME = "invalid_time"
+
+
+def _to_naive(dt: datetime) -> datetime:
+    if dt is None:
+        return None
+    if dt.tzinfo is not None:
+        return dt.astimezone(timezone.utc).replace(tzinfo=None)
+    return dt
+
+
+def _day_for_datetime(dt: datetime) -> str:
+    mapping = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
+    return mapping[_to_naive(dt).weekday()]
+
+
+class ScheduleService:
+    def __init__(
+        self,
+        schedule_repository: ScheduleRepository,
+        employee_repository: EmployeeRepository,
+        company_repository: CompanyRepository,
+    ):
+        self.schedule_repository = schedule_repository
+        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,
+        shift: str,
+    ):
+        if self.employee_repository.get(user_id) is None:
+            return None
+        enter_time = _to_naive(enter_time)
+        out_time = _to_naive(out_time)
+        # day_of_week is provided as local day (e.g. Friday) while enter_time is UTC (e.g. 2026-09-19T01:39:00Z -> Saturday UTC)
+        # strict UTC check would reject valid local times, so only validate out_time ordering
+        if out_time is not None and out_time < enter_time:
+            return INVALID_TIME
+        return self.schedule_repository.create(
+            user_id, enter_time, day_of_week, out_time, shift
+        )
+
+    def _week_bounds(self, reference: datetime):
+        from datetime import timedelta
+
+        reference = _to_naive(reference)
+        days_from_sunday = (reference.weekday() + 1) % 7
+        start = reference.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(
+            days=days_from_sunday
+        )
+        return start, start + timedelta(days=7)
+
+    def clock_in(
+        self,
+        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
+
+        enter_time = _to_naive(enter_time)
+        week_start, week_end = self._week_bounds(enter_time)
+
+        if (
+            self.schedule_repository.get_by_user_and_day(
+                user_id, day_of_week, week_start, week_end
+            )
+            is not None
+        ):
+            return ALREADY_WORKED_DAY
+
+        if (
+            self.schedule_repository.count_distinct_days_by_user(
+                user_id, week_start, week_end
+            )
+            >= 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)
+
+    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
+        out_time = _to_naive(out_time)
+        # allow UTC vs local day drift for out_time as well
+        existing = self.schedule_repository.get_by_user_and_day(user_id, day_of_week)
+        if existing is not None:
+            existing_enter = _to_naive(existing["enter_time"])
+            if out_time < existing_enter:
+                return INVALID_TIME
+        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)}
+
+    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)
@@ -0,0 +1,6 @@
+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
+
+__all__ = ["CompanyRepository", "Database", "EmployeeRepository", "ScheduleRepository"]
@@ -0,0 +1,66 @@
+class CompanyRepository:
+    def __init__(self, connection):
+        self.connection = connection
+
+    def list(self):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT id, name, date_created, date_updated
+                FROM companies
+                ORDER BY id
+                """
+            )
+            return cursor.fetchall()
+
+    def create(self, name: str):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO companies (name)
+                VALUES (%s)
+                RETURNING id, name, date_created, date_updated
+                """,
+                (name,),
+            )
+            company = cursor.fetchone()
+        self.connection.commit()
+        return company
+
+    def get(self, company_id: int):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT id, name, date_created, date_updated
+                FROM companies
+                WHERE id = %s
+                """,
+                (company_id,),
+            )
+            return cursor.fetchone()
+
+    def get_by_name(self, name: str):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT id, name, date_created, date_updated
+                FROM companies
+                WHERE name = %s
+                """,
+                (name,),
+            )
+            return cursor.fetchone()
+
+    def delete(self, company_id: int):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                DELETE FROM companies
+                WHERE id = %s
+                RETURNING id, name, date_created, date_updated
+                """,
+                (company_id,),
+            )
+            company = cursor.fetchone()
+        self.connection.commit()
+        return company
@@ -0,0 +1,109 @@
+import os
+
+import psycopg
+from psycopg.rows import dict_row
+
+
+DEFAULT_DATABASE_URL = "postgres://carlos:password@localhost/rust_manager_employees"
+
+
+class Database:
+    def __init__(self, database_url: str | None = None):
+        self.database_url = database_url or os.environ.get("DATABASE_URL", DEFAULT_DATABASE_URL)
+
+    def get_connection(self):
+        return psycopg.connect(self.database_url, row_factory=dict_row)
+
+    def initialize_schema(self):
+        with self.get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    DO $$
+                    BEGIN
+                        CREATE TYPE work_time AS ENUM ('morning', 'afternoon', 'evening');
+                    EXCEPTION WHEN duplicate_object THEN NULL;
+                    END $$;
+                    """
+                )
+                cursor.execute("ALTER TYPE work_time ADD VALUE IF NOT EXISTS 'afternoon';")
+                cursor.execute(
+                    """
+                    DO $$
+                    BEGIN
+                        CREATE TYPE day_of_week AS ENUM (
+                            'sunday', 'monday', 'tuesday', 'wednesday',
+                            'thursday', 'friday', 'saturday'
+                        );
+                    EXCEPTION WHEN duplicate_object THEN NULL;
+                    END $$;
+                    """
+                )
+                cursor.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS companies (
+                        id SERIAL PRIMARY KEY,
+                        name TEXT NOT NULL UNIQUE,
+                        date_created TIMESTAMP NOT NULL DEFAULT NOW(),
+                        date_updated TIMESTAMP NOT NULL DEFAULT NOW()
+                    );
+                    """
+                )
+                cursor.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS users (
+                        id SERIAL PRIMARY KEY,
+                        name TEXT NOT NULL,
+                        last_name TEXT NOT NULL,
+                        preference work_time NOT NULL,
+                        date_created TIMESTAMP NOT NULL DEFAULT NOW(),
+                        date_updated TIMESTAMP NOT NULL DEFAULT NOW()
+                    );
+                    """
+                )
+                cursor.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS company_users (
+                        id SERIAL PRIMARY KEY,
+                        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                        company_id INTEGER NOT NULL REFERENCES companies(id) ON DELETE CASCADE,
+                        date_created TIMESTAMP NOT NULL DEFAULT NOW(),
+                        date_updated TIMESTAMP NOT NULL DEFAULT NOW(),
+                        UNIQUE (user_id, company_id)
+                    );
+                    """
+                )
+                cursor.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS schedules (
+                        id SERIAL PRIMARY KEY,
+                        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                        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 (
+                        id SERIAL PRIMARY KEY,
+                        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                        day_of_week day_of_week NOT NULL,
+                        first work_time NOT NULL,
+                        second work_time NOT NULL,
+                        third work_time NOT NULL,
+                        date_created TIMESTAMP NOT NULL DEFAULT NOW(),
+                        date_updated TIMESTAMP NOT NULL DEFAULT NOW(),
+                        UNIQUE (user_id, day_of_week)
+                    );
+                    """
+                )
+            connection.commit()
@@ -0,0 +1,228 @@
+class EmployeeRepository:
+    def __init__(self, connection):
+        self.connection = connection
+
+    def list(self):
+        with self.connection.cursor() as cursor:
+            cursor.execute("""
+                SELECT users.id, company_users.company_id, users.name, users.last_name, users.preference
+                FROM users
+                INNER JOIN company_users ON company_users.user_id = users.id
+                ORDER BY users.id
+                """)
+            return cursor.fetchall()
+
+    def create(self, company_id: int, name: str, last_name: str, preference: str):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO users (name, last_name, preference)
+                VALUES (%s, %s, %s)
+                RETURNING id, name, last_name, preference
+                """,
+                (name, last_name, preference),
+            )
+            user = cursor.fetchone()
+            cursor.execute(
+                """
+                INSERT INTO company_users (user_id, company_id)
+                VALUES (%s, %s)
+                RETURNING company_id
+                """,
+                (user["id"], company_id),
+            )
+            company_user = cursor.fetchone()
+        self.connection.commit()
+        return {**user, "company_id": company_user["company_id"]}
+
+    def get(self, employee_id: int):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT users.id, company_users.company_id, users.name, users.last_name, users.preference
+                FROM users
+                INNER JOIN company_users ON company_users.user_id = users.id
+                WHERE users.id = %s
+                """,
+                (employee_id,),
+            )
+            return cursor.fetchone()
+
+    def delete(self, employee_id: int):
+        employee = self.get(employee_id)
+        if employee is None:
+            return None
+        with self.connection.cursor() as cursor:
+            cursor.execute("DELETE FROM users WHERE id = %s", (employee_id,))
+        self.connection.commit()
+        return employee
+
+    def update_preference(self, employee_id: int, preference: str):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE users
+                SET preference = %s, date_updated = NOW()
+                WHERE id = %s
+                """,
+                (preference, employee_id),
+            )
+        self.connection.commit()
+        return self.get(employee_id)
+
+    def update(
+        self,
+        employee_id: int,
+        name: str | None,
+        last_name: str | None,
+        preference: str | None,
+    ):
+        updates: list[str] = []
+        params: list[object] = []
+        if name is not None:
+            updates.append("name = %s")
+            params.append(name)
+        if last_name is not None:
+            updates.append("last_name = %s")
+            params.append(last_name)
+        if preference is not None:
+            updates.append("preference = %s")
+            params.append(preference)
+        if updates:
+            updates.append("date_updated = NOW()")
+            params.append(employee_id)
+            with self.connection.cursor() as cursor:
+                cursor.execute(
+                    f"""
+                    UPDATE users
+                    SET {", ".join(updates)}
+                    WHERE id = %s
+                    """,
+                    tuple(params),
+                )
+            self.connection.commit()
+        return self.get(employee_id)
+
+    def update_company(self, employee_id: int, company_id: int):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                "DELETE FROM company_users WHERE user_id = %s", (employee_id,)
+            )
+            cursor.execute(
+                """
+                INSERT INTO company_users (user_id, company_id)
+                VALUES (%s, %s)
+                ON CONFLICT (user_id, company_id) DO UPDATE SET date_updated = NOW()
+                """,
+                (employee_id, company_id),
+            )
+        self.connection.commit()
+        return self.get(employee_id)
+
+    def assign_to_company(self, company_id: int, employee_id: int):
+        employee = self.get(employee_id)
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO company_users (user_id, company_id)
+                VALUES (%s, %s)
+                ON CONFLICT (user_id)
+                DO UPDATE SET
+                    company_id = EXCLUDED.company_id,
+                    date_updated = NOW()
+                RETURNING id, user_id, company_id, date_created, date_updated
+                """,
+                (employee_id, company_id),
+            )
+            company_user = cursor.fetchone()
+
+        self.connection.commit()
+        return company_user
+
+    def get_names_by_company(self, company_id: int):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT users.name, users.last_name
+                FROM users
+                INNER JOIN company_users ON company_users.user_id = users.id
+                WHERE company_users.company_id = %s
+                ORDER BY users.name, users.last_name
+                """,
+                (company_id,),
+            )
+            return [f"{row['name']} {row['last_name']}" for row in cursor.fetchall()]
+
+    def save_weekly_preferences(self, employee_id: int, preferences: list):
+        with self.connection.cursor() as cursor:
+            for preference in preferences:
+                cursor.execute(
+                    """
+                    INSERT INTO weekly_preferences (user_id, day_of_week, first, second, third)
+                    VALUES (%s, %s, %s, %s, %s)
+                    ON CONFLICT (user_id, day_of_week) DO UPDATE SET
+                        first = EXCLUDED.first,
+                        second = EXCLUDED.second,
+                        third = EXCLUDED.third,
+                        date_updated = NOW()
+                    """,
+                    (
+                        employee_id,
+                        preference.day_of_week,
+                        preference.first,
+                        preference.second,
+                        preference.third,
+                    ),
+                )
+        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()
@@ -0,0 +1,204 @@
+from datetime import datetime
+
+
+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,
+        shift: str | None,
+    ):
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                """
+                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, shift),
+            )
+            schedule = cursor.fetchone()
+        self.connection.commit()
+        return schedule
+
+    def get_by_user_and_day(self, user_id: int, day_of_week: str, week_start=None, week_end=None):
+        if week_start is not None and week_end is not None:
+            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
+                      AND enter_time >= %s AND enter_time < %s
+                    ORDER BY date_created DESC
+                    LIMIT 1
+                    """,
+                    (user_id, day_of_week, week_start, week_end),
+                )
+                return cursor.fetchone()
+        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, week_start=None, week_end=None):
+        if week_start is not None and week_end is not None:
+            with self.connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    SELECT COUNT(DISTINCT day_of_week) AS days_worked
+                    FROM schedules
+                    WHERE user_id = %s
+                      AND enter_time >= %s AND enter_time < %s
+                    """,
+                    (user_id, week_start, week_end),
+                )
+                row = cursor.fetchone()
+                return row["days_worked"]
+        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 = ""
+        if company_id is not None:
+            where = "WHERE company_users.company_id = %s"
+            params = (company_id,)
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT schedules.id, schedules.enter_time, schedules.day_of_week,
+                       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
+                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,
+            )
+            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"],
+                "shift": schedule["shift"],
+            }
+            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:
+            cursor.execute(
+                """
+                UPDATE schedules
+                SET out_time = %s, date_updated = NOW()
+                WHERE id = (
+                    SELECT id
+                    FROM schedules
+                    WHERE user_id = %s AND day_of_week = %s
+                    ORDER BY date_created DESC
+                    LIMIT 1
+                )
+                RETURNING id, user_id, enter_time, day_of_week, out_time, shift
+                """,
+                (out_time, user_id, day_of_week),
+            )
+            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
+
+    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"]
@@ -0,0 +1,165 @@
+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
+    )
@@ -0,0 +1,3 @@
+from scheduler.app import app, create_app
+
+__all__ = ["app", "create_app"]
@@ -0,0 +1,3 @@
+from . import conflict, scheduler, validator
+
+__all__ = ["conflict", "scheduler", "validator"]
@@ -0,0 +1,137 @@
+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 _datetime_for_day(day_of_week: str, reference: datetime) -> datetime:
+    from datetime import timedelta
+
+    days_from_sunday = (reference.weekday() + 1) % 7
+    week_start = reference.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(
+        days=days_from_sunday
+    )
+    offset = {"sunday": 0, "monday": 1, "tuesday": 2, "wednesday": 3, "thursday": 4, "friday": 5, "saturday": 6}[
+        _value(day_of_week).lower()
+    ]
+    return week_start + timedelta(days=offset, hours=9)
+
+
+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
+    now = datetime.now()
+    enter_time = _datetime_for_day(_value(day_of_week), now)
+    if not validate_employee_can_join_shift(
+        connection, user_id, _value(day_of_week), _value(shift), enter_time
+    ):
+        return None
+    return storage.add_schedule(
+        connection,
+        user_id,
+        enter_time,
+        _value(day_of_week),
+        None,
+        _value(shift),
+    )
+
+
+def detect_preferred_shift_conflict(connection, user_id, day_of_week, preferred_shift):
+    company_id = _get_user_company_id(connection, user_id)
+    now = datetime.now()
+    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), now
+        )
+        or not validate_user_has_not_worked_more_than_five_days(connection, user_id, now)
+    )
+
+
+def resolve_conflict_same_day(connection, user_id, day_of_week):
+    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):
+    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):
+    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
@@ -0,0 +1,106 @@
+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 _datetime_for_day(day_of_week: str, reference: datetime) -> datetime:
+    from datetime import timedelta
+
+    days_from_sunday = (reference.weekday() + 1) % 7
+    week_start = reference.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(
+        days=days_from_sunday
+    )
+    offset = {"sunday": 0, "monday": 1, "tuesday": 2, "wednesday": 3, "thursday": 4, "friday": 5, "saturday": 6}[
+        _value(day_of_week).lower()
+    ]
+    return week_start + timedelta(days=offset, hours=9)
+
+
+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):
+    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):
+    day = _value(day_of_week)
+    shift = _value(preferred_shift)
+    now = datetime.now()
+    enter_time = _datetime_for_day(day, now)
+    if not validate_employee_can_join_shift(connection, user_id, day, shift, enter_time):
+        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, enter_time, day, None, shift)
+
+
+def fill_missing_shift_coverage(connection, company_id, day_of_week, shift):
+    return conflict.resolve_missing_second_employee(
+        connection, company_id, _value(day_of_week), _value(shift)
+    )
@@ -0,0 +1,27 @@
+from scheduler.database import storage
+
+
+def validate_user_has_not_worked_same_day(connection, user_id, day_of_week, reference=None):
+    return storage.get_user_enter_time_by_day(connection, user_id, day_of_week, reference) is None
+
+
+def validate_user_has_not_worked_more_than_five_days(connection, user_id, reference=None):
+    return len(storage.get_days_worked_by_user(connection, user_id, reference)) < 5
+
+
+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, reference=None):
+    return validate_user_has_not_worked_same_day(
+        connection, user_id, day_of_week, reference
+    ) and validate_user_has_not_worked_more_than_five_days(connection, user_id, reference)
+
+
+def validate_company_has_enough_employees(connection, company_id):
+    return len(storage.get_company_employee_ids(connection, company_id)) >= 2
@@ -0,0 +1,3 @@
+from . import employee, schedule, shift
+
+__all__ = ["employee", "schedule", "shift"]
@@ -0,0 +1,32 @@
+from scheduler.core.enums import WorkTime
+from scheduler.core.schemas.employee import DayPreferenceRequest
+from scheduler.database import storage
+
+
+def define_employee_fields():
+    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):
+    return storage.update_user_preference(connection, user_id, preference)
+
+
+def get_employee_names_by_company(connection, company_name):
+    company = storage.get_company_by_name(connection, company_name)
+    if company is None:
+        return None
+    return storage.get_employee_names_by_company(connection, company["id"])
+
+
+def register_user_into_company(connection, company_id, name, last_name, preference):
+    user = storage.add_user(connection, name, last_name, preference)
+    storage.add_company_user(connection, company_id, user["id"])
+    return user
@@ -0,0 +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():
+    return {
+        "user_id": int,
+        "company_id": int,
+        "day_of_week": DayOfWeek,
+        "shift": WorkTime,
+        "enter_time": datetime,
+        "out_time": datetime | None,
+    }
+
+
+def _value(value):
+    return value.value if hasattr(value, "value") else value
+
+
+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()
@@ -0,0 +1,30 @@
+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():
+    return tuple(s.value for s in WorkTime)
+
+
+def define_day_values():
+    return tuple(d.value for d in DayOfWeek)
+
+
+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,
+    }
Author	SHA1	Message	Date
CarGDev	26253f3945	adding env example	2026-09-20 11:11:39 -04:00
CarGDev	db0a2a3042	fixing the checkin	2026-09-18 21:59:44 -04:00
CarGDev	e26eb302d0	refactoring unused functions	2026-09-18 20:46:32 -04:00
CarGDev	b10f345d21	fixing the company updates	2026-09-18 19:14:19 -04:00
CarGDev	6bfdd25cd8	fixing the company and schedule endpoints	2026-09-18 17:14:12 -04:00
CarGDev	e7d887203e	adding documentation	2026-09-18 12:51:31 -04:00
CarGDev	f38a178fcd	adding endpoints	2026-09-18 12:26:25 -04:00
CarGDev	d5c341d1fc	Merge branch 'master' of https://git.cargdev.io/CarGDev/manager_employees_python	2026-09-17 18:49:25 -04:00
CarGDev	e7c989496d	initial commit	2026-09-17 18:49:19 -04:00
				`@@ -0,0 +1 @@`
				`DATABASE_URL=postgres://{username}:{pass}@{host}/{database_name}`