from datetime import datetime, time, date, timedelta
from sqlalchemy import func, and_, or_, case, extract
from src.models import DatabaseContextManager
from src.models.models import (
NotificationPreference, TeachingSession, TutorAvailability,
Tutor, Supervisor, tutor_course_association, Course, AvailabilityOpeningSession,
SupervisorAvailability
)
from flask import current_app
from src.utils import (
ApiABC,
custom_response,
send_email
)
import uuid
import json
from typing import Dict, List, Optional, Any
from enum import Enum
from sqlalchemy.orm import joinedload
class AvailabilityConflictType(Enum):
"""Types of availability conflicts"""
OVERLAP = "overlap"
OUTSIDE_WORKING_HOURS = "outside_working_hours"
EXCEEDS_MAX_HOURS = "exceeds_max_hours"
LEAVE_CONFLICT = "leave_conflict"
SESSION_CONFLICT = "session_conflict"
class TutorAvailabilityManager(ApiABC):
def __init__(self):
self.table = TutorAvailability
def create(self, tutor_id: str, payload: Dict) -> Dict:
"""
Create new availability slots for a tutor.
Validates for conflicts before creating.
Args:
tutor_id: ID of the tutor
payload: {
'availabilities': [{
'day_of_week': int (0-6, Monday-Sunday),
'start_time': str ('HH:MM'),
'end_time': str ('HH:MM'),
'is_recurring': bool,
'valid_from': str (date, optional),
'valid_to': str (date, optional),
'availability_type': str,
'location': str (optional),
'notes': str (optional)
}]
}
Returns:
Response with created availabilities or error
"""
with DatabaseContextManager() as ctx:
# Verify tutor exists and is active
tutor = ctx.session.query(Tutor).filter(
Tutor.id == tutor_id,
Tutor.is_active == True
).first()
if not tutor:
return custom_response(
success=False,
data="Tutor not found or inactive",
status_code=404
)
# Validate payload structure
if 'availabilities' not in payload or not isinstance(payload['availabilities'], list):
return custom_response(
success=False,
data="Invalid payload structure. Expected 'availabilities' array",
status_code=400
)
# Check for conflicts
conflict_check = self._check_availability_conflicts(ctx, tutor_id, payload['availabilities'])
if conflict_check['has_conflicts']:
return custom_response(
success=False,
data={
'message': 'Availability conflicts detected',
'conflicts': conflict_check['conflicts']
},
status_code=409
)
created_availabilities = []
for avail_data in payload['availabilities']:
# Parse times
try:
start_time = datetime.strptime(avail_data['start_time'], '%H:%M').time()
end_time = datetime.strptime(avail_data['end_time'], '%H:%M').time()
if end_time <= start_time:
return custom_response(
success=False,
data=f"End time must be after start time for day {avail_data['day_of_week']}",
status_code=400
)
# Create availability record
availability = TutorAvailability(
id=str(uuid.uuid4()),
tutor_id=tutor_id,
day_of_week=avail_data['day_of_week'],
start_time=start_time,
end_time=end_time,
is_recurring=avail_data.get('is_recurring', True),
valid_from=datetime.strptime(avail_data['valid_from'], '%Y-%m-%d').date() if avail_data.get('valid_from') else None,
valid_to=datetime.strptime(avail_data['valid_to'], '%Y-%m-%d').date() if avail_data.get('valid_to') else None,
availability_type=avail_data.get('availability_type', 'teaching'),
location=avail_data.get('location'),
notes=avail_data.get('notes'),
is_approved=False # Needs supervisor approval by default
)
ctx.session.add(availability)
created_availabilities.append(availability)
except ValueError as e:
return custom_response(
success=False,
data=f"Invalid time format: {str(e)}",
status_code=400
)
ctx.session.commit()
# Notify supervisor if tutor has one
if tutor.supervisor_id:
self._notify_supervisor_of_pending_approval(tutor, created_availabilities)
return custom_response(
success=True,
data={
'message': 'Availability created successfully. Pending supervisor approval.',
'availabilities': [self._availability_to_dict(avail) for avail in created_availabilities]
},
status_code=201
)
def get(self, availability_id: str) -> Dict:
"""
Get a specific availability slot by ID
Args:
availability_id: ID of the availability slot
Returns:
Response with availability details or error
"""
with DatabaseContextManager() as ctx:
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability not found",
status_code=404
)
return custom_response(
success=True,
data=self._availability_to_dict(availability),
status_code=200
)
def fetchAll(self, user_id: str) -> Dict:
"""
Fetch all tutor availabilities. This method handles both cases:
1. When called by a supervisor (user_id is supervisor_id) - returns tutor availabilities for that supervisor
2. When called by a tutor (user_id is tutor_id) - returns the tutor's own availabilities
Args:
user_id: ID of the supervisor or tutor
Returns:
Response with grouped tutor availabilities or error
"""
with DatabaseContextManager() as ctx:
# First, check if this is a supervisor
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == user_id,
Supervisor.is_active == True
).first()
if supervisor:
# This is a supervisor - get tutor availabilities for this supervisor
# Get all tutors associated with this supervisor through courses
tutors = ctx.session.query(Tutor).join(
tutor_course_association,
Tutor.id == tutor_course_association.c.tutor_id
).join(
Course,
tutor_course_association.c.course_id == Course.id
).filter(
Course.supervisor_id == user_id,
Tutor.is_active == True
).distinct().all()
# If no tutors are found, return empty result instead of error
if not tutors:
return custom_response(
success=True,
data=[],
status_code=200
)
# Get all availabilities for these tutors
tutor_ids = [t.id for t in tutors]
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id.in_(tutor_ids)
).order_by(
TutorAvailability.tutor_id,
TutorAvailability.day_of_week,
TutorAvailability.start_time
).all()
# Group by tutor and day for better organization
result = []
for tutor in tutors:
tutor_availabilities = [avail for avail in availabilities if avail.tutor_id == tutor.id]
if not tutor_availabilities:
continue
# Group by day of week
days = {day: [] for day in range(7)} # 0-6 for Monday-Sunday
for avail in tutor_availabilities:
days[avail.day_of_week].append(self._availability_to_dict(avail))
# Format days for response
formatted_days = []
day_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
for day_num in range(7):
if days[day_num]:
formatted_days.append({
'day_number': day_num,
'day_name': day_names[day_num],
'slots': days[day_num]
})
result.append({
'tutor': {
'id': tutor.id,
"first_name": tutor.first_name,
"last_name": tutor.last_name,
'name': f"{tutor.first_name} {tutor.last_name}",
'email': tutor.email,
'departments': [item.to_json() for item in tutor.departments] if hasattr(tutor, 'departments') and tutor.departments else [],
'is_full_time': tutor.is_full_time,
'max_teaching_hours': tutor.max_teaching_hours
},
'availabilities': formatted_days,
'total_weekly_hours': self._calculate_weekly_teaching_hours(ctx, tutor.id)
})
return custom_response(
success=True,
data=result,
status_code=200
)
# If not a supervisor, check if it's a tutor
tutor = ctx.session.query(Tutor).filter(
Tutor.id == user_id,
Tutor.is_active == True
).first()
if tutor:
# This is a tutor - get their own availabilities
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == user_id
).order_by(
TutorAvailability.day_of_week,
TutorAvailability.start_time
).all()
# Convert to the expected format
availability_list = []
for avail in availabilities:
availability_list.append(self._availability_to_dict(avail))
return custom_response(
success=True,
data=availability_list,
status_code=200
)
# If neither supervisor nor tutor found
return custom_response(
success=False,
data="User not found or inactive",
status_code=404
)
def get_tutor_availabilities(self, tutor_id: str, include_pending: bool = False) -> Dict:
"""
Get all availabilities for a tutor
Args:
tutor_id: ID of the tutor
include_pending: Whether to include pending approval slots
Returns:
Response with list of availabilities
"""
with DatabaseContextManager() as ctx:
query = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id
)
if not include_pending:
query = query.filter(TutorAvailability.is_approved == True)
availabilities = query.order_by(
TutorAvailability.day_of_week,
TutorAvailability.start_time
).all()
return custom_response(
success=True,
data=[self._availability_to_dict(avail) for avail in availabilities],
status_code=200
)
def get_approved_availabilities(self, tutor_id: str) -> Dict:
"""
Get only approved availabilities for a tutor
Args:
tutor_id: ID of the tutor
Returns:
Response with list of approved availabilities
"""
return self.get_tutor_availabilities(tutor_id, include_pending=False)
def update(self, availability_id: str, payload: Dict) -> Dict:
"""
Update an availability slot. Requires re-approval if changing core fields.
Args:
availability_id: ID of the availability slot
payload: Update data (same structure as create but for single availability)
Returns:
Response with updated availability or error
"""
with DatabaseContextManager() as ctx:
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability not found",
status_code=404
)
# Check if core fields are being changed (requires re-approval)
core_fields = ['day_of_week', 'start_time', 'end_time', 'is_recurring', 'valid_from', 'valid_to']
needs_reapproval = any(field in payload for field in core_fields)
# For time fields, parse first
time_updates = {}
if 'start_time' in payload:
try:
time_updates['start_time'] = datetime.strptime(payload['start_time'], '%H:%M').time()
except ValueError:
return custom_response(
success=False,
data="Invalid start_time format. Use HH:MM",
status_code=400
)
if 'end_time' in payload:
try:
time_updates['end_time'] = datetime.strptime(payload['end_time'], '%H:%M').time()
except ValueError:
return custom_response(
success=False,
data="Invalid end_time format. Use HH:MM",
status_code=400
)
# Check time validity if both are being updated
if 'start_time' in payload and 'end_time' in payload:
if time_updates['end_time'] <= time_updates['start_time']:
return custom_response(
success=False,
data="End time must be after start time",
status_code=400
)
# Check for conflicts if changing time/day
if 'day_of_week' in payload or 'start_time' in payload or 'end_time' in payload:
# Create a test availability with proposed changes
test_availability = {
'day_of_week': payload.get('day_of_week', availability.day_of_week),
'start_time': payload.get('start_time', availability.start_time.strftime('%H:%M')),
'end_time': payload.get('end_time', availability.end_time.strftime('%H:%M')),
'is_recurring': payload.get('is_recurring', availability.is_recurring),
'valid_from': payload.get('valid_from', str(availability.valid_from) if availability.valid_from else None),
'valid_to': payload.get('valid_to', str(availability.valid_to) if availability.valid_to else None)
}
conflict_check = self._check_availability_conflicts(
ctx,
availability.tutor_id,
[test_availability],
exclude_availability_id=availability_id
)
if conflict_check['has_conflicts']:
return custom_response(
success=False,
data={
'message': 'Availability conflicts detected with proposed changes',
'conflicts': conflict_check['conflicts']
},
status_code=409
)
# Apply updates
if 'day_of_week' in payload:
availability.day_of_week = payload['day_of_week']
if 'start_time' in payload:
availability.start_time = time_updates['start_time']
if 'end_time' in payload:
availability.end_time = time_updates['end_time']
if 'is_recurring' in payload:
availability.is_recurring = payload['is_recurring']
if 'valid_from' in payload:
availability.valid_from = datetime.strptime(payload['valid_from'], '%Y-%m-%d').date() if payload['valid_from'] else None
if 'valid_to' in payload:
availability.valid_to = datetime.strptime(payload['valid_to'], '%Y-%m-%d').date() if payload['valid_to'] else None
# Update optional fields
if 'availability_type' in payload:
availability.availability_type = payload['availability_type']
if 'location' in payload:
availability.location = payload['location']
if 'notes' in payload:
availability.notes = payload['notes']
# If core fields changed, reset approval status
if needs_reapproval:
availability.is_approved = False
availability.approval_date = None
availability.approved_by = None
# Notify supervisor
tutor = ctx.session.query(Tutor).filter(Tutor.id == availability.tutor_id).first()
if tutor and tutor.supervisor_id:
self._notify_supervisor_of_pending_approval(tutor, [availability])
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Availability updated successfully' + (' - requires re-approval' if needs_reapproval else ''),
'availability': self._availability_to_dict(availability)
},
status_code=200
)
def delete(self, availability_id: str) -> Dict:
"""
Delete an availability slot
Args:
availability_id: ID of the availability slot
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability not found",
status_code=404
)
ctx.session.delete(availability)
ctx.session.commit()
return custom_response(
success=True,
data="Availability deleted successfully",
status_code=200
)
def approve_availability(self, supervisor_id: str, availability_id: str) -> Dict:
"""
Approve a tutor's availability slot (by supervisor)
Args:
supervisor_id: ID of the approving supervisor
availability_id: ID of the availability slot
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability not found",
status_code=404
)
# Check if already approved
if availability.is_approved:
return custom_response(
success=False,
data="Availability already approved",
status_code=400
)
# Verify this supervisor can approve this tutor's availability
tutor = ctx.session.query(Tutor).filter(
Tutor.id == availability.tutor_id,
).first()
if not tutor:
return custom_response(
success=False,
data="Not authorized to approve this availability",
status_code=403
)
# Check for conflicts (in case something changed since request)
test_availability = {
'day_of_week': availability.day_of_week,
'start_time': availability.start_time.strftime('%H:%M'),
'end_time': availability.end_time.strftime('%H:%M'),
'is_recurring': availability.is_recurring,
'valid_from': str(availability.valid_from) if availability.valid_from else None,
'valid_to': str(availability.valid_to) if availability.valid_to else None
}
conflict_check = self._check_availability_conflicts(
ctx,
availability.tutor_id,
[test_availability],
exclude_availability_id=availability_id
)
if conflict_check['has_conflicts']:
return custom_response(
success=False,
data={
'message': 'Cannot approve - availability conflicts detected',
'conflicts': conflict_check['conflicts']
},
status_code=409
)
# Approve the availability
availability.is_approved = True
availability.approval_date = datetime.utcnow()
availability.approved_by = supervisor_id
ctx.session.commit()
# Notify tutor
self._notify_tutor_of_approval(tutor, availability)
return custom_response(
success=True,
data="Availability approved successfully",
status_code=200
)
def bulk_approve(self, supervisor_id: str, availability_ids: List[str]) -> Dict:
"""
Bulk approve multiple availability slots
Args:
supervisor_id: ID of the approving supervisor
availability_ids: List of availability IDs to approve
Returns:
Response with approval results
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
results = {
'approved': [],
'failed': []
}
for avail_id in availability_ids:
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == avail_id
).first()
if not availability:
results['failed'].append({
'availability_id': avail_id,
'reason': 'Not found'
})
continue
# Check if already approved
if availability.is_approved:
results['failed'].append({
'availability_id': avail_id,
'reason': 'Already approved'
})
continue
# Verify authorization
tutor = ctx.session.query(Tutor).filter(
Tutor.id == availability.tutor_id,
Tutor.supervisor_id == supervisor_id
).first()
if not tutor:
results['failed'].append({
'availability_id': avail_id,
'reason': 'Not authorized'
})
continue
# Check for conflicts
test_availability = {
'day_of_week': availability.day_of_week,
'start_time': availability.start_time.strftime('%H:%M'),
'end_time': availability.end_time.strftime('%H:%M'),
'is_recurring': availability.is_recurring,
'valid_from': str(availability.valid_from) if availability.valid_from else None,
'valid_to': str(availability.valid_to) if availability.valid_to else None
}
conflict_check = self._check_availability_conflicts(
ctx,
availability.tutor_id,
[test_availability],
exclude_availability_id=avail_id
)
if conflict_check['has_conflicts']:
results['failed'].append({
'availability_id': avail_id,
'reason': 'Conflicts detected',
'conflicts': conflict_check['conflicts']
})
continue
# Approve this availability
availability.is_approved = True
availability.approval_date = datetime.utcnow()
availability.approved_by = supervisor_id
results['approved'].append(avail_id)
# Notify tutor (could be batched outside the loop for better performance)
self._notify_tutor_of_approval(tutor, availability)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Bulk approval completed',
'results': results
},
status_code=200
)
def reject_availability(self, supervisor_id: str, availability_id: str, reason: str) -> Dict:
"""
Reject a tutor's availability request
Args:
supervisor_id: ID of the rejecting supervisor
availability_id: ID of the availability slot
reason: Reason for rejection
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability not found",
status_code=404
)
# Check if already approved
if availability.is_approved:
return custom_response(
success=False,
data="Cannot reject - availability already approved",
status_code=400
)
# Verify this supervisor can reject this tutor's availability
tutor = ctx.session.query(Tutor).filter(
Tutor.id == availability.tutor_id,
Tutor.supervisor_id == supervisor_id
).first()
if not tutor:
return custom_response(
success=False,
data="Not authorized to reject this availability",
status_code=403
)
# Delete the availability (or could mark as rejected with reason)
ctx.session.delete(availability)
ctx.session.commit()
# Notify tutor
self._notify_tutor_of_rejection(tutor, availability, reason)
return custom_response(
success=True,
data="Availability rejected and deleted successfully",
status_code=200
)
def get_pending_approvals(self, supervisor_id: str) -> Dict:
"""
Get all availability slots pending approval for tutors supervised by this supervisor
Args:
supervisor_id: ID of the supervisor
Returns:
Response with list of pending availabilities
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Get all tutors supervised by this supervisor
tutors = ctx.session.query(Tutor).filter(
Tutor.supervisor_id == supervisor_id,
Tutor.is_active == True
).all()
if not tutors:
return custom_response(
success=True,
data=[],
status_code=200
)
# Get pending availabilities for these tutors
tutor_ids = [t.id for t in tutors]
pending_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id.in_(tutor_ids),
TutorAvailability.is_approved == False
).order_by(
TutorAvailability.day_of_week,
TutorAvailability.start_time
).all()
# Group by tutor for easier review
grouped_availabilities = {}
for avail in pending_availabilities:
if avail.tutor_id not in grouped_availabilities:
tutor = ctx.session.query(Tutor).filter(Tutor.id == avail.tutor_id).first()
grouped_availabilities[avail.tutor_id] = {
'tutor': {
'id': tutor.id,
'name': f"{tutor.first_name} {tutor.last_name}",
'email': tutor.email,
'department': tutor.department
},
'availabilities': []
}
grouped_availabilities[avail.tutor_id]['availabilities'].append(
self._availability_to_dict(avail)
)
return custom_response(
success=True,
data=list(grouped_availabilities.values()),
status_code=200
)
def check_availability(self, tutor_id: str, start_datetime: datetime, end_datetime: datetime) -> Dict:
"""
Check if a tutor is available at a specific date/time range
Args:
tutor_id: ID of the tutor
start_datetime: Start datetime to check
end_datetime: End datetime to check
Returns:
Response with availability status and details
"""
with DatabaseContextManager() as ctx:
# Verify tutor exists
tutor = ctx.session.query(Tutor).filter(
Tutor.id == tutor_id,
Tutor.is_active == True
).first()
if not tutor:
return custom_response(
success=False,
data="Tutor not found or inactive",
status_code=404
)
# Check if tutor is on leave
if tutor.is_on_leave and tutor.leave_start_date and tutor.leave_end_date:
leave_start = tutor.leave_start_date
leave_end = tutor.leave_end_date
requested_date = start_datetime.date()
if leave_start <= requested_date <= leave_end:
return custom_response(
success=True,
data={
'is_available': False,
'reason': 'Tutor is on leave during this period',
'leave_details': {
'start_date': str(leave_start),
'end_date': str(leave_end),
'reason': tutor.leave_reason
}
},
status_code=200
)
# Get day of week (0-6, Monday-Sunday)
day_of_week = start_datetime.weekday()
start_time = start_datetime.time()
end_time = end_datetime.time()
# Check approved availabilities for this day
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.day_of_week == day_of_week,
TutorAvailability.is_approved == True,
or_(
TutorAvailability.is_recurring == True,
and_(
TutorAvailability.is_recurring == False,
TutorAvailability.valid_from <= start_datetime.date(),
TutorAvailability.valid_to >= start_datetime.date()
)
)
).all()
# Check if any availability slot covers the requested time
for avail in availabilities:
if avail.start_time <= start_time and avail.end_time >= end_time:
return custom_response(
success=True,
data={
'is_available': True,
'availability_slot': self._availability_to_dict(avail),
'conflicts': []
},
status_code=200
)
# If we get here, no matching availability found - check for conflicts
conflicts = self._find_specific_conflicts(ctx, tutor_id, start_datetime, end_datetime)
return custom_response(
success=True,
data={
'is_available': False,
'reason': 'No matching availability slot found',
'conflicts': conflicts
},
status_code=200
)
def get_weekly_availability(self, tutor_id: str, week_start: date = None) -> Dict:
"""
Get tutor's availability for a specific week
Args:
tutor_id: ID of the tutor
week_start: Optional start date of the week (defaults to current week)
Returns:
Response with weekly availability schedule
"""
with DatabaseContextManager() as ctx:
# Verify tutor exists
tutor = ctx.session.query(Tutor).filter(
Tutor.id == tutor_id,
Tutor.is_active == True
).first()
if not tutor:
return custom_response(
success=False,
data="Tutor not found or inactive",
status_code=404
)
# Determine week start (Monday)
if not week_start:
today = date.today()
week_start = today - timedelta(days=today.weekday())
week_end = week_start + timedelta(days=6)
# Get all approved availabilities that apply to this week
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.is_approved == True,
or_(
TutorAvailability.is_recurring == True,
and_(
TutorAvailability.is_recurring == False,
TutorAvailability.valid_from <= week_end,
TutorAvailability.valid_to >= week_start
)
)
).order_by(
TutorAvailability.day_of_week,
TutorAvailability.start_time
).all()
# Organize by day
weekly_availability = {day: [] for day in range(7)} # 0-6 for Monday-Sunday
for avail in availabilities:
weekly_availability[avail.day_of_week].append(self._availability_to_dict(avail))
# Format for response
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
formatted_availability = []
for day_num in range(7):
current_date = week_start + timedelta(days=day_num)
formatted_availability.append({
'day_name': days[day_num],
'date': str(current_date),
'slots': weekly_availability[day_num]
})
return custom_response(
success=True,
data={
'week_start': str(week_start),
'week_end': str(week_end),
'tutor_id': tutor_id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'availability': formatted_availability
},
status_code=200
)
def _availability_to_dict(self, availability: TutorAvailability) -> Dict:
"""Convert TutorAvailability model to dictionary"""
return {
'id': availability.id,
'tutor_id': availability.tutor_id,
'day_of_week': availability.day_of_week,
'day_name': ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'][availability.day_of_week],
'start_time': availability.start_time.strftime('%H:%M'),
'end_time': availability.end_time.strftime('%H:%M'),
'duration_minutes': (datetime.combine(date.today(), availability.end_time) -
datetime.combine(date.today(), availability.start_time)).seconds // 60,
'is_recurring': availability.is_recurring,
'valid_from': str(availability.valid_from) if availability.valid_from else None,
'valid_to': str(availability.valid_to) if availability.valid_to else None,
'is_approved': availability.is_approved,
'approval_date': availability.approval_date.isoformat() if availability.approval_date else None,
'approved_by': availability.approved_by,
'availability_type': availability.availability_type,
'location': availability.location,
'notes': availability.notes,
'created_at': availability.created_at.isoformat() if availability.created_at else None
}
def _check_availability_conflicts(self, ctx, tutor_id: str, new_availabilities: List[Dict],
exclude_availability_id: str = None) -> Dict:
"""
Check for conflicts with existing availabilities, sessions, or tutor constraints
Args:
ctx: Database context
tutor_id: ID of the tutor
new_availabilities: List of new availability slots to check
exclude_availability_id: Optional ID of availability to exclude from conflict checks (for updates)
Returns:
Dictionary with conflict information
"""
conflicts = []
# Get tutor details
tutor = ctx.session.query(Tutor).filter(Tutor.id == tutor_id).first()
if not tutor:
return {'has_conflicts': True, 'conflicts': [{'type': 'invalid_tutor', 'message': 'Tutor not found'}]}
# Check each new availability
for new_avail in new_availabilities:
# Parse times
try:
start_time = datetime.strptime(new_avail['start_time'], '%H:%M').time()
end_time = datetime.strptime(new_avail['end_time'], '%H:%M').time()
day_of_week = new_avail['day_of_week']
if end_time <= start_time:
conflicts.append({
'type': AvailabilityConflictType.OVERLAP.value,
'message': 'End time must be after start time',
'day_of_week': day_of_week,
'start_time': new_avail['start_time'],
'end_time': new_avail['end_time']
})
continue
# Check against standard working hours (8am-6pm by default)
if start_time < time(8, 0) or end_time > time(18, 0):
conflicts.append({
'type': AvailabilityConflictType.OUTSIDE_WORKING_HOURS.value,
'message': 'Availability outside standard working hours (8am-6pm)',
'day_of_week': day_of_week,
'start_time': new_avail['start_time'],
'end_time': new_avail['end_time']
})
# Check for overlapping with existing approved availabilities
existing_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.day_of_week == day_of_week,
TutorAvailability.is_approved == True,
TutorAvailability.id != exclude_availability_id if exclude_availability_id else True
).all()
for existing in existing_availabilities:
if not (end_time <= existing.start_time or start_time >= existing.end_time):
conflicts.append({
'type': AvailabilityConflictType.OVERLAP.value,
'message': f"Overlaps with existing availability ({existing.start_time.strftime('%H:%M')}-{existing.end_time.strftime('%H:%M')})",
'day_of_week': day_of_week,
'start_time': new_avail['start_time'],
'end_time': new_avail['end_time'],
'conflicting_id': existing.id,
'conflicting_start': existing.start_time.strftime('%H:%M'),
'conflicting_end': existing.end_time.strftime('%H:%M')
})
# Check tutor's max teaching hours if this is a teaching availability
if new_avail.get('availability_type', 'teaching') == 'teaching':
# Calculate weekly teaching hours if this availability is added
weekly_hours = self._calculate_weekly_teaching_hours(ctx, tutor_id, new_avail)
if tutor.max_teaching_hours and weekly_hours > tutor.max_teaching_hours:
conflicts.append({
'type': AvailabilityConflictType.EXCEEDS_MAX_HOURS.value,
'message': f"Would exceed tutor's max teaching hours ({tutor.max_teaching_hours}h/week)",
'day_of_week': day_of_week,
'start_time': new_avail['start_time'],
'end_time': new_avail['end_time'],
'current_weekly_hours': weekly_hours - self._calculate_availability_hours(new_avail),
'projected_weekly_hours': weekly_hours,
'max_allowed': tutor.max_teaching_hours
})
# Check for leave conflicts if this is a specific date range
if not new_avail.get('is_recurring', True) and tutor.is_on_leave:
valid_from = datetime.strptime(new_avail['valid_from'], '%Y-%m-%d').date() if new_avail.get('valid_from') else None
valid_to = datetime.strptime(new_avail['valid_to'], '%Y-%m-%d').date() if new_avail.get('valid_to') else None
if valid_from and valid_to and tutor.leave_start_date and tutor.leave_end_date:
if not (valid_to < tutor.leave_start_date or valid_from > tutor.leave_end_date):
conflicts.append({
'type': AvailabilityConflictType.LEAVE_CONFLICT.value,
'message': 'Conflicts with tutor leave period',
'day_of_week': day_of_week,
'start_time': new_avail['start_time'],
'end_time': new_avail['end_time'],
'leave_start': str(tutor.leave_start_date),
'leave_end': str(tutor.leave_end_date),
'valid_from': new_avail.get('valid_from'),
'valid_to': new_avail.get('valid_to')
})
# Check for scheduled sessions conflicts
if not new_avail.get('is_recurring', True):
valid_from = datetime.strptime(new_avail['valid_from'], '%Y-%m-%d').date() if new_avail.get('valid_from') else None
valid_to = datetime.strptime(new_avail['valid_to'], '%Y-%m-%d').date() if new_avail.get('valid_to') else None
if valid_from and valid_to:
# Find any scheduled sessions that would conflict with this availability
conflicting_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.tutor_id == tutor_id,
TeachingSession.status == 'scheduled',
extract('dow', TeachingSession.start_time) == day_of_week,
or_(
and_(
func.time(TeachingSession.start_time) < end_time,
func.time(TeachingSession.end_time) > start_time
),
and_(
TeachingSession.start_time.date() >= valid_from,
TeachingSession.start_time.date() <= valid_to
),
and_(
TeachingSession.end_time.date() >= valid_from,
TeachingSession.end_time.date() <= valid_to
)
)
).all()
for session in conflicting_sessions:
conflicts.append({
'type': AvailabilityConflictType.SESSION_CONFLICT.value,
'message': f"Conflicts with scheduled session: {session.title}",
'day_of_week': day_of_week,
'start_time': new_avail['start_time'],
'end_time': new_avail['end_time'],
'session_id': session.id,
'session_title': session.title,
'session_start': session.start_time.isoformat(),
'session_end': session.end_time.isoformat(),
'course_code': session.course.code if session.course else None
})
except ValueError as e:
conflicts.append({
'type': 'invalid_data',
'message': str(e),
'availability': new_avail
})
continue
return {
'has_conflicts': len(conflicts) > 0,
'conflicts': conflicts
}
def _calculate_weekly_teaching_hours(self, ctx, tutor_id: str, new_availability: Dict = None) -> float:
"""
Calculate total weekly teaching hours for a tutor, optionally including a new availability
Args:
ctx: Database context
tutor_id: ID of the tutor
new_availability: Optional new availability to include in calculation
Returns:
Total weekly teaching hours in hours
"""
# Get all approved teaching availabilities
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.is_approved == True,
TutorAvailability.availability_type == 'teaching'
).all()
# Convert to list of dicts for easier processing
availability_dicts = [self._availability_to_dict(avail) for avail in availabilities]
# Add the new availability if provided
if new_availability:
availability_dicts.append(new_availability)
# Group by day of week and calculate total hours per day
day_hours = {day: 0.0 for day in range(7)}
for avail in availability_dicts:
duration = (datetime.combine(date.today(), datetime.strptime(avail['end_time'], '%H:%M').time()) -
datetime.combine(date.today(), datetime.strptime(avail['start_time'], '%H:%M').time()))
day_hours[avail['day_of_week']] += duration.total_seconds() / 3600
# Sum up weekly hours
weekly_hours = sum(day_hours.values())
return weekly_hours
def _calculate_availability_hours(self, availability: Dict) -> float:
"""Calculate duration of an availability slot in hours"""
start = datetime.strptime(availability['start_time'], '%H:%M').time()
end = datetime.strptime(availability['end_time'], '%H:%M').time()
duration = datetime.combine(date.today(), end) - datetime.combine(date.today(), start)
return duration.total_seconds() / 3600
def _find_specific_conflicts(self, ctx, tutor_id: str, start_datetime: datetime, end_datetime: datetime) -> List[Dict]:
"""
Find specific conflicts for a date/time range
Args:
ctx: Database context
tutor_id: ID of the tutor
start_datetime: Start datetime to check
end_datetime: End datetime to check
Returns:
List of conflict dictionaries
"""
conflicts = []
# Check if tutor is on leave
tutor = ctx.session.query(Tutor).filter(Tutor.id == tutor_id).first()
if tutor and tutor.is_on_leave and tutor.leave_start_date and tutor.leave_end_date:
if tutor.leave_start_date <= start_datetime.date() <= tutor.leave_end_date:
conflicts.append({
'type': AvailabilityConflictType.LEAVE_CONFLICT.value,
'message': 'Tutor is on leave during this period',
'leave_start': str(tutor.leave_start_date),
'leave_end': str(tutor.leave_end_date),
'reason': tutor.leave_reason
})
# Check for scheduled sessions
conflicting_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.tutor_id == tutor_id,
TeachingSession.status == 'scheduled',
or_(
and_(
TeachingSession.start_time < end_datetime,
TeachingSession.end_time > start_datetime
)
)
).all()
for session in conflicting_sessions:
conflicts.append({
'type': AvailabilityConflictType.SESSION_CONFLICT.value,
'message': 'Conflicts with scheduled teaching session',
'session_id': session.id,
'session_title': session.title,
'session_start': session.start_time.isoformat(),
'session_end': session.end_time.isoformat(),
'course_code': session.course.code if session.course else None
})
return conflicts
def _notify_supervisor_of_pending_approval(self, tutor: Tutor, availabilities: List[TutorAvailability]) -> None:
"""
Send notification to supervisor about pending availability approvals
Args:
tutor: Tutor model instance
availabilities: List of TutorAvailability instances needing approval
"""
with DatabaseContextManager() as ctx:
supervisor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.supervisor.id
).first()
if not supervisor_prefs or not supervisor_prefs.receive_email:
return
# Prepare email
subject = f"Pending Availability Approvals for {tutor.first_name} {tutor.last_name}"
# Group availabilities by day
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
grouped_availabilities = {day: [] for day in days}
for avail in availabilities:
day_name = days[avail.day_of_week]
grouped_availabilities[day_name].append({
'time_range': f"{avail.start_time.strftime('%H:%M')} - {avail.end_time.strftime('%H:%M')}",
'type': avail.availability_type,
'location': avail.location or 'Not specified'
})
# Most readable approach using string concatenation:
availability_text = ""
for day, slots in grouped_availabilities.items():
if slots:
availability_text += f"\n{day}\n"
for slot in slots:
availability_text += f"* {slot['time_range']} ({slot['type']}) - {slot['location']}\n"
message = f"""Pending Availability Approvals
Hello {tutor.supervisor.first_name},
Your tutor {tutor.first_name} {tutor.last_name} has submitted new availability slots requiring your approval:
{availability_text}
Review Availabilities
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.supervisor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send availability approval notification: {str(e)}")
def _notify_tutor_of_approval(self, tutor: Tutor, availability: TutorAvailability) -> None:
"""
Send notification to tutor about approved availability
Args:
tutor: Tutor model instance
availability: Approved TutorAvailability instance
"""
with DatabaseContextManager() as ctx:
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
return
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_name = days[availability.day_of_week]
subject = f"Availability Approved: {day_name} {availability.start_time.strftime('%H:%M')}-{availability.end_time.strftime('%H:%M')}"
message = f"""
Availability Approved
Hello {tutor.first_name},
Your supervisor has approved the following availability slot:
Day: {day_name}
Time: {availability.start_time.strftime('%H:%M')} - {availability.end_time.strftime('%H:%M')}
Type: {availability.availability_type}
Location: {availability.location or 'Not specified'}
Recurring: {'Yes' if availability.is_recurring else 'No'}
{f"
Valid From: {availability.valid_from.strftime('%Y-%m-%d')}
" if availability.valid_from else ''}
{f"
Valid To: {availability.valid_to.strftime('%Y-%m-%d')}
" if availability.valid_to else ''}
This slot is now active and can be used for scheduling.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send availability approval notification: {str(e)}")
def _notify_tutor_of_rejection(self, tutor: Tutor, availability: TutorAvailability, reason: str) -> None:
"""
Send notification to tutor about rejected availability
Args:
tutor: Tutor model instance
availability: Rejected TutorAvailability instance
reason: Reason for rejection
"""
with DatabaseContextManager() as ctx:
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
return
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_name = days[availability.day_of_week]
subject = f"Availability Rejected: {day_name} {availability.start_time.strftime('%H:%M')}-{availability.end_time.strftime('%H:%M')}"
message = f"""
Availability Rejected
Hello {tutor.first_name},
Your supervisor has rejected the following availability slot:
Day: {day_name}
Time: {availability.start_time.strftime('%H:%M')} - {availability.end_time.strftime('%H:%M')}
Reason: {reason or 'Not specified'}
Please review and submit a new availability request if needed.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send availability rejection notification: {str(e)}")
# Opening Sessions Methods
def get_opening_sessions(self) -> Dict:
"""Get all opening sessions"""
with DatabaseContextManager() as ctx:
sessions = ctx.session.query(AvailabilityOpeningSession).order_by(
AvailabilityOpeningSession.created_at.desc()
).all()
return custom_response(
success=True,
data=[self._opening_session_to_dict(session) for session in sessions],
status_code=200
)
def get_opening_session(self, session_id: str) -> Dict:
"""Get a specific opening session"""
with DatabaseContextManager() as ctx:
session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.id == session_id
).first()
if not session:
return custom_response(
success=False,
data="Opening session not found",
status_code=404
)
return custom_response(
success=True,
data=self._opening_session_to_dict(session),
status_code=200
)
def create_opening_session(self, payload: Dict) -> Dict:
"""Create a new opening session"""
with DatabaseContextManager() as ctx:
# Validate required fields
required_fields = ['name', 'start_datetime', 'end_datetime', 'created_by']
for field in required_fields:
if field not in payload:
return custom_response(
success=False,
data=f"Missing required field: {field}",
status_code=400
)
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == payload['created_by'],
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Parse datetime strings
try:
start_datetime = datetime.fromisoformat(payload['start_datetime'].replace('Z', '+00:00'))
end_datetime = datetime.fromisoformat(payload['end_datetime'].replace('Z', '+00:00'))
except ValueError:
return custom_response(
success=False,
data="Invalid datetime format. Use ISO format (YYYY-MM-DDTHH:MM:SS)",
status_code=400
)
if end_datetime <= start_datetime:
return custom_response(
success=False,
data="End datetime must be after start datetime",
status_code=400
)
# Create opening session
session = AvailabilityOpeningSession(
id=str(uuid.uuid4()),
name=payload['name'],
description=payload.get('description'),
start_datetime=start_datetime,
end_datetime=end_datetime,
created_by=payload['created_by'],
max_slots_per_tutor=payload.get('max_slots_per_tutor', 10),
allowed_availability_types=json.dumps(payload.get('allowed_availability_types', ['teaching', 'office_hours', 'research'])),
allowed_days=json.dumps(payload.get('allowed_days', [0, 1, 2, 3, 4])),
time_range_start=datetime.strptime(payload.get('time_range_start', '08:00'), '%H:%M').time(),
time_range_end=datetime.strptime(payload.get('time_range_end', '18:00'), '%H:%M').time(),
min_slot_duration=payload.get('min_slot_duration', 60),
max_slot_duration=payload.get('max_slot_duration', 240),
notify_tutors=payload.get('notify_tutors', True),
notification_message=payload.get('notification_message')
)
ctx.session.add(session)
ctx.session.commit()
# Send notifications to tutors if enabled
if session.notify_tutors:
self._notify_tutors_of_opening_session(ctx, session)
return custom_response(
success=True,
data=self._opening_session_to_dict(session),
status_code=201
)
def update_opening_session(self, session_id: str, payload: Dict) -> Dict:
"""Update an opening session"""
with DatabaseContextManager() as ctx:
session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.id == session_id
).first()
if not session:
return custom_response(
success=False,
data="Opening session not found",
status_code=404
)
# Update fields
if 'name' in payload:
session.name = payload['name']
if 'description' in payload:
session.description = payload['description']
if 'start_datetime' in payload:
try:
session.start_time = datetime.fromisoformat(payload['start_datetime'].replace('Z', '+00:00'))
except ValueError:
return custom_response(
success=False,
data="Invalid start_datetime format",
status_code=400
)
if 'end_datetime' in payload:
try:
session.end_time = datetime.fromisoformat(payload['end_datetime'].replace('Z', '+00:00'))
except ValueError:
return custom_response(
success=False,
data="Invalid end_datetime format",
status_code=400
)
if 'max_slots_per_tutor' in payload:
session.max_slots_per_tutor = payload['max_slots_per_tutor']
if 'allowed_availability_types' in payload:
session.allowed_availability_types = json.dumps(payload['allowed_availability_types'])
if 'allowed_days' in payload:
session.allowed_days = json.dumps(payload['allowed_days'])
if 'time_range_start' in payload:
session.time_range_start = datetime.strptime(payload['time_range_start'], '%H:%M').time()
if 'time_range_end' in payload:
session.time_range_end = datetime.strptime(payload['time_range_end'], '%H:%M').time()
if 'min_slot_duration' in payload:
session.min_slot_duration = payload['min_slot_duration']
if 'max_slot_duration' in payload:
session.max_slot_duration = payload['max_slot_duration']
if 'notify_tutors' in payload:
session.notify_tutors = payload['notify_tutors']
if 'notification_message' in payload:
session.notification_message = payload['notification_message']
session.updated_at = datetime.utcnow()
ctx.session.commit()
return custom_response(
success=True,
data=self._opening_session_to_dict(session),
status_code=200
)
def delete_opening_session(self, session_id: str) -> Dict:
"""Delete an opening session"""
with DatabaseContextManager() as ctx:
session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.id == session_id
).first()
if not session:
return custom_response(
success=False,
data="Opening session not found",
status_code=404
)
# Check if there are any availabilities linked to this session
linked_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.opening_session_id == session_id
).count()
if linked_availabilities > 0:
return custom_response(
success=False,
data=f"Cannot delete opening session. {linked_availabilities} availability slots are linked to this session.",
status_code=400
)
ctx.session.delete(session)
ctx.session.commit()
return custom_response(
success=True,
data="Opening session deleted successfully",
status_code=200
)
def update_opening_session_status(self, session_id: str, payload: Dict) -> Dict:
"""Update opening session status"""
with DatabaseContextManager() as ctx:
session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.id == session_id
).first()
if not session:
return custom_response(
success=False,
data="Opening session not found",
status_code=404
)
new_status = payload.get('status')
if new_status not in ['scheduled', 'active', 'closed', 'cancelled']:
return custom_response(
success=False,
data="Invalid status. Must be one of: scheduled, active, closed, cancelled",
status_code=400
)
session.status = new_status
session.updated_at = datetime.utcnow()
ctx.session.commit()
# Send notifications if status is changed to active
if new_status == 'active' and session.notify_tutors:
self._notify_tutors_of_opening_session(ctx, session)
return custom_response(
success=True,
data=self._opening_session_to_dict(session),
status_code=200
)
def _opening_session_to_dict(self, session: AvailabilityOpeningSession) -> Dict:
"""Convert opening session to dictionary"""
return {
'id': session.id,
'name': session.name,
'description': session.description,
'start_datetime': session.start_datetime,
'end_datetime': session.end_datetime,
'is_active': session.is_active,
'status': str(session.status), # Convert to string to ensure consistency
'created_by': session.created_by,
'created_at': session.created_at.isoformat(),
'updated_at': session.updated_at.isoformat(),
'max_slots_per_tutor': session.max_slots_per_tutor,
'allowed_availability_types': json.loads(session.allowed_availability_types) if session.allowed_availability_types else [],
'allowed_days': json.loads(session.allowed_days) if session.allowed_days else [],
'time_range_start': session.time_range_start.strftime('%H:%M') if session.time_range_start else '08:00',
'time_range_end': session.time_range_end.strftime('%H:%M') if session.time_range_end else '18:00',
'min_slot_duration': session.min_slot_duration,
'max_slot_duration': session.max_slot_duration,
'notify_tutors': session.notify_tutors,
'notification_message': session.notification_message,
'reminder_sent': session.reminder_sent,
'reminder_sent_at': session.reminder_sent_at.isoformat() if session.reminder_sent_at else None,
'creator': {
'id': session.creator.id,
'first_name': session.creator.first_name,
'last_name': session.creator.last_name,
'email': session.creator.email
} if session.creator else None
}
def _notify_tutors_of_opening_session(self, ctx, session: AvailabilityOpeningSession) -> None:
"""Send notifications to tutors about the opening session"""
# Get all tutors supervised by the session creator
tutors = ctx.session.query(Tutor).filter(
Tutor.supervisor_id == session.created_by,
Tutor.is_active == True
).all()
if not tutors:
return
subject = f"New Availability Opening Session: {session.name}"
message = f"""
New Availability Opening Session
Hello,
A new availability opening session has been created by your supervisor:
{session.name}
Description: {session.description or 'No description provided'}
Start Date: {session.start_time.strftime('%Y-%m-%d %H:%M')}
End Date: {session.end_time.strftime('%Y-%m-%d %H:%M')}
Max Slots per Tutor: {session.max_slots_per_tutor}
Time Range: {session.time_range_start.strftime('%H:%M')} - {session.time_range_end.strftime('%H:%M')}
Min Slot Duration: {session.min_slot_duration} minutes
Max Slot Duration: {session.max_slot_duration} minutes
{f"Message from Supervisor: {session.notification_message}
" if session.notification_message else ''}
You can now create your availability slots during this session period.
Best regards,
{current_app.config.get('APP_NAME', 'KISNAP Training Institute')} Team
"""
for tutor in tutors:
try:
# Check if tutor has email notifications enabled
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
continue
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send opening session notification to {tutor.email}: {str(e)}")
def auto_generate_availability(self, tutor_id: str, payload: Dict = None) -> Dict:
"""
Auto-generate availability slots for a tutor (Monday-Friday, 8AM-6PM)
Args:
tutor_id: ID of the tutor
payload: Optional configuration for auto-generation
{
'start_time': '08:00' (default),
'end_time': '18:00' (default),
'days': [0,1,2,3,4] (Monday-Friday, default),
'availability_type': 'teaching' (default),
'location': 'Main Campus' (default),
'notes': 'Auto-generated availability' (default),
'opening_session_id': 'uuid' (optional, if not provided will find active session)
}
Returns:
Response with created availabilities or error
"""
with DatabaseContextManager() as ctx:
# Verify tutor exists and is active
tutor = ctx.session.query(Tutor).filter(
Tutor.id == tutor_id,
Tutor.is_active == True
).first()
if not tutor:
return custom_response(
success=False,
data="Tutor not found or inactive",
status_code=404
)
# Get opening session - either from payload or find active one
opening_session = None
if payload and payload.get('opening_session_id'):
# Use the specified opening session
session_id = payload['opening_session_id']
current_app.logger.info(f"Looking for opening session with ID: {session_id}")
# Debug: List all opening sessions in the database
all_sessions = ctx.session.query(AvailabilityOpeningSession).all()
current_app.logger.info(f"All opening sessions in database: {[(s.id, s.name, s.is_active, s.status) for s in all_sessions]}")
# First, try to find the session without filters to see if it exists
opening_session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.id == session_id
).first()
if not opening_session:
current_app.logger.error(f"Opening session with ID {session_id} not found in database")
return custom_response(
success=False,
data=f"Opening session with ID {session_id} not found",
status_code=404
)
current_app.logger.info(f"Found opening session: {opening_session.name}, is_active: {opening_session.is_active}, status: {opening_session.status}")
# Now check if it's active and has the right status
if not opening_session.is_active:
current_app.logger.warning(f"Opening session '{opening_session.name}' is not active (is_active: {opening_session.is_active})")
return custom_response(
success=False,
data=f"Opening session '{opening_session.name}' is not active",
status_code=400
)
if opening_session.status != 'active':
current_app.logger.warning(f"Opening session '{opening_session.name}' has status '{opening_session.status}', but needs to be 'active'")
return custom_response(
success=False,
data=f"Opening session '{opening_session.name}' has status '{opening_session.status}', but needs to be 'active'",
status_code=400
)
current_app.logger.info(f"Using specified opening session: {opening_session.name} (ID: {opening_session.id})")
else:
# Find active opening session (fallback behavior)
current_app.logger.info("No opening_session_id in payload, looking for fallback active session")
opening_session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.is_active == True,
AvailabilityOpeningSession.status == 'active',
AvailabilityOpeningsession.start_time <= datetime.utcnow(),
AvailabilityOpeningsession.end_time >= datetime.utcnow()
).first()
if not opening_session:
current_app.logger.warning("No fallback active opening session found")
return custom_response(
success=False,
data="No active opening session found. Please contact your supervisor to create and activate an opening session.",
status_code=400
)
current_app.logger.info(f"Using fallback active opening session: {opening_session.name} (ID: {opening_session.id})")
# Double-check that we have a valid opening session
if not opening_session:
current_app.logger.error("opening_session is None after all checks")
return custom_response(
success=False,
data="Failed to retrieve opening session",
status_code=500
)
current_app.logger.info(f"Final opening session: {opening_session.name} (ID: {opening_session.id})")
# Check if tutor has reached the maximum slots limit
existing_slots = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.opening_session_id == opening_session.id
).count()
if existing_slots >= opening_session.max_slots_per_tutor:
return custom_response(
success=False,
data=f"You have reached the maximum limit of {opening_session.max_slots_per_tutor} availability slots for this opening session.",
status_code=400
)
# Default configuration
config = {
'start_time': '08:00',
'end_time': '18:00',
'days': [0, 1, 2, 3, 4], # Monday to Friday
'availability_type': 'teaching',
'location': 'Main Campus',
'notes': 'Auto-generated availability'
}
# Override with payload if provided
if payload:
config.update(payload)
# Generate availability slots
generated_availabilities = []
for day in config['days']:
if 0 <= day <= 6: # Valid day range
# Check if slot already exists for this day
existing_slot = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.day_of_week == day,
TutorAvailability.opening_session_id == opening_session.id
).first()
if not existing_slot:
# Create new availability slot
availability = TutorAvailability(
id=str(uuid.uuid4()),
tutor_id=tutor_id,
opening_session_id=opening_session.id,
day_of_week=day,
start_time=datetime.strptime(config['start_time'], '%H:%M').time(),
end_time=datetime.strptime(config['end_time'], '%H:%M').time(),
is_recurring=True,
availability_type=config['availability_type'],
location=config['location'],
notes=config['notes'],
is_approved=True, # Auto-approved since supervisor is generating it
approval_date=datetime.utcnow(), # Set approval date to now
approved_by=payload.get('created_by', 'system'), # Set who approved it
created_at=datetime.utcnow()
)
ctx.session.add(availability)
generated_availabilities.append(availability)
if not generated_availabilities:
return custom_response(
success=False,
data="No new availability slots could be generated. You may already have slots for the specified days.",
status_code=400
)
try:
ctx.session.commit()
# No need to notify supervisor since availability is auto-approved
# self._notify_supervisor_of_pending_approval(tutor, generated_availabilities)
return custom_response(
success=True,
data={
'message': f'Successfully generated and approved {len(generated_availabilities)} availability slots',
'availabilities': [self._availability_to_dict(avail) for avail in generated_availabilities],
'total_slots': existing_slots + len(generated_availabilities),
'max_slots': opening_session.max_slots_per_tutor,
'auto_approved': True
},
status_code=201
)
except Exception as e:
ctx.session.rollback()
return custom_response(
success=False,
data=f"Failed to generate availability slots: {str(e)}",
status_code=500
)
class SupervisorAvailabilityManager(ApiABC):
"""Manager for supervisor availability operations"""
def __init__(self):
super().__init__(SupervisorAvailability)
def get(self, availability_id: str) -> Dict[str, Any]:
"""
Get a specific supervisor availability slot by ID
Args:
availability_id: ID of the availability slot
Returns:
Response with availability details or error
"""
with DatabaseContextManager() as ctx:
availability = ctx.session.query(SupervisorAvailability).options(
joinedload(SupervisorAvailability.supervisor)
).filter(
SupervisorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Supervisor availability not found",
status_code=404
)
return custom_response(
success=True,
data=self._supervisor_availability_to_dict(availability),
status_code=200
)
def fetchAll(self) -> Dict[str, Any]:
"""
Fetch all supervisor availabilities (for compatibility with ApiABC)
Returns:
Response with all supervisor availabilities
"""
with DatabaseContextManager() as ctx:
availabilities = ctx.session.query(SupervisorAvailability).options(
joinedload(SupervisorAvailability.supervisor)
).filter(
SupervisorAvailability.is_approved == True,
SupervisorAvailability.is_cancelled == False
).all()
return custom_response(
success=True,
data=[self._supervisor_availability_to_dict(avail) for avail in availabilities],
status_code=200
)
def create(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Create new availability slots for a supervisor (default implementation)
Args:
payload: Availability data
Returns:
Response with created availabilities or error
"""
# This method requires supervisor_id, so we'll use a different approach
# The actual implementation is in the create method with supervisor_id parameter
return custom_response(
success=False,
data="Please use create(supervisor_id, payload) method instead",
status_code=400
)
def update(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Update a supervisor availability slot (default implementation)
Args:
payload: Updated availability data
Returns:
Response with updated availability or error
"""
# This method requires availability_id, so we'll use a different approach
# The actual implementation is in the update method with availability_id parameter
return custom_response(
success=False,
data="Please use update(availability_id, payload) method instead",
status_code=400
)
def delete(self, availability_id: str) -> Dict[str, Any]:
"""
Delete a supervisor availability slot (default implementation)
Args:
availability_id: ID of the availability slot
Returns:
Response with success/error message
"""
# This method requires availability_id, so we'll use a different approach
# The actual implementation is in the delete method with availability_id parameter
return custom_response(
success=False,
data="Please use delete(availability_id) method instead",
status_code=400
)
def create_supervisor_availability(self, supervisor_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Create new availability slots for a supervisor
Args:
supervisor_id: ID of the supervisor
payload: Availability data
Returns:
Response with created availabilities or error
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists and is active
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# For supervisors, we allow creating availability without requiring an active opening session
# This gives supervisors more flexibility than tutors
active_session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.status == 'active',
AvailabilityOpeningSession.is_active == True,
AvailabilityOpeningSession.start_time <= datetime.utcnow(),
AvailabilityOpeningSession.end_time >= datetime.utcnow()
).first()
# Set default limits if no active session
max_slots_limit = 50 # Default limit for supervisors
if active_session:
max_slots_limit = active_session.max_slots_per_tutor
# Check if supervisor has reached the maximum slots limit
existing_slots = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == supervisor_id,
).count()
if existing_slots >= max_slots_limit:
return custom_response(
success=False,
data=f"You have reached the maximum limit of {max_slots_limit} availability slots.",
status_code=400
)
created_availabilities = []
availabilities_data = payload.get('availabilities', [payload])
for availability_data in availabilities_data:
try:
# Validate time format
start_time = datetime.strptime(availability_data['start_time'], '%H:%M').time()
end_time = datetime.strptime(availability_data['end_time'], '%H:%M').time()
if end_time <= start_time:
return custom_response(
success=False,
data="End time must be after start time",
status_code=400
)
# Check for conflicts
conflict = self._check_supervisor_availability_conflict(
ctx, supervisor_id, availability_data['day_of_week'],
start_time, end_time, availability_data.get('is_recurring', True)
)
if conflict:
return custom_response(
success=False,
data=f"Availability conflict: {conflict}",
status_code=400
)
# Create availability
availability = SupervisorAvailability(
supervisor_id=supervisor_id,
day_of_week=availability_data['day_of_week'],
start_time=start_time,
end_time=end_time,
is_recurring=availability_data.get('is_recurring', True),
availability_type=availability_data.get('availability_type', 'teaching'),
location=availability_data.get('location'),
notes=availability_data.get('notes'),
is_approved=True, # Supervisors are auto-approved
approval_date=datetime.utcnow(),
approved_by=supervisor_id
)
if not availability_data.get('is_recurring', True):
availability.valid_from = availability_data.get('valid_from', date.today())
availability.valid_to = availability_data.get('valid_to', date.today() + timedelta(days=365))
ctx.session.add(availability)
created_availabilities.append(availability)
except ValueError as e:
return custom_response(
success=False,
data=f"Invalid time format: {str(e)}",
status_code=400
)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Supervisor availability created successfully.',
'availabilities': [self._supervisor_availability_to_dict(avail) for avail in created_availabilities]
},
status_code=201
)
def update_supervisor_availability(self, availability_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Update a supervisor availability slot
Args:
availability_id: ID of the availability slot
payload: Updated availability data
Returns:
Response with updated availability or error
"""
with DatabaseContextManager() as ctx:
availability = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Supervisor availability not found",
status_code=404
)
# Update fields
if 'day_of_week' in payload:
availability.day_of_week = payload['day_of_week']
if 'start_time' in payload:
availability.start_time = datetime.strptime(payload['start_time'], '%H:%M').time()
if 'end_time' in payload:
availability.end_time = datetime.strptime(payload['end_time'], '%H:%M').time()
if 'is_recurring' in payload:
availability.is_recurring = payload['is_recurring']
if 'availability_type' in payload:
availability.availability_type = payload['availability_type']
if 'location' in payload:
availability.location = payload['location']
if 'notes' in payload:
availability.notes = payload['notes']
ctx.session.commit()
return custom_response(
success=True,
data=self._supervisor_availability_to_dict(availability),
status_code=200
)
def delete_supervisor_availability(self, availability_id: str) -> Dict[str, Any]:
"""
Delete a supervisor availability slot
Args:
availability_id: ID of the availability slot
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
availability = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Supervisor availability not found",
status_code=404
)
ctx.session.delete(availability)
ctx.session.commit()
return custom_response(
success=True,
data="Supervisor availability deleted successfully",
status_code=200
)
def _check_supervisor_availability_conflict(self, ctx, supervisor_id: str, day_of_week: int,
start_time: time, end_time: time, is_recurring: bool) -> Optional[str]:
"""
Check for conflicts with existing supervisor availability
Args:
ctx: Database context
supervisor_id: ID of the supervisor
day_of_week: Day of the week (0-6)
start_time: Start time
end_time: End time
is_recurring: Whether the availability is recurring
Returns:
Conflict message if conflict exists, None otherwise
"""
# Check for overlapping availabilities on the same day
existing_availabilities = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == supervisor_id,
SupervisorAvailability.day_of_week == day_of_week,
SupervisorAvailability.is_cancelled == False
).all()
for existing in existing_availabilities:
if (start_time < existing.end_time and end_time > existing.start_time):
return f"Overlaps with existing availability on {self._get_day_name(day_of_week)} {existing.start_time}-{existing.end_time}"
return None
def _supervisor_availability_to_dict(self, availability: SupervisorAvailability) -> Dict[str, Any]:
"""
Convert supervisor availability to dictionary
Args:
availability: SupervisorAvailability object
Returns:
Dictionary representation of the availability
"""
# Get supervisor information
supervisor_info = None
if availability.supervisor:
supervisor_info = {
'id': availability.supervisor.id,
'first_name': availability.supervisor.first_name,
'last_name': availability.supervisor.last_name,
'email': availability.supervisor.email
}
return {
'id': availability.id,
'supervisor_id': availability.supervisor_id,
'day_of_week': availability.day_of_week,
'day_name': self._get_day_name(availability.day_of_week),
'start_time': availability.start_time.strftime('%H:%M'),
'end_time': availability.end_time.strftime('%H:%M'),
'is_recurring': availability.is_recurring,
'valid_from': availability.valid_from.isoformat() if availability.valid_from else None,
'valid_to': availability.valid_to.isoformat() if availability.valid_to else None,
'is_approved': availability.is_approved,
'availability_type': availability.availability_type,
'location': availability.location,
'notes': availability.notes,
'created_at': availability.created_at.isoformat() if availability.created_at else None,
'supervisor': supervisor_info,
'supervisor_name': f"{availability.supervisor.first_name} {availability.supervisor.last_name}" if availability.supervisor else 'Unknown Supervisor',
'user_type': 'supervisor'
}
def _get_day_name(self, day_of_week: int) -> str:
"""Get day name from day of week number"""
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
return days[day_of_week] if 0 <= day_of_week < 7 else 'Unknown'