from datetime import datetime
from sqlalchemy import func, and_, or_
from src.models import DatabaseContextManager
from src.models.models import (
CourseModule, Course, Tutor, Student, Enrollment, SubmissionStatus, tutor_course_association,
CourseResource, Assignment, NotificationPreference, AssignmentSubmission
)
from flask import current_app
from src.utils import (
custom_response,
send_email
)
import uuid
from typing import Dict
class CourseModuleManager:
def __init__(self):
self.table = CourseModule
def create_module(self, payload: Dict) -> Dict:
"""
Create a new course module with validation and notifications.
Args:
payload: {
"course_id": str,
"title": str,
"description": str (optional),
"sequence": int,
"start_date": str (ISO format, optional),
"end_date": str (ISO format, optional),
"objectives": str (optional),
"estimated_study_time": int (optional, in hours),
"is_assessment_module": bool (default False),
"notify_users": bool (default True) - whether to send notifications
}
Returns:
Response with created module or error
"""
with DatabaseContextManager() as ctx:
# Validate course exists
course = ctx.session.query(Course).filter(
Course.id == payload['course_id']
).first()
if not course:
return custom_response(
success=False,
data="Course not found",
status_code=404
)
# Validate sequence is unique for this course
existing_sequence = ctx.session.query(CourseModule).filter(
and_(
CourseModule.course_id == payload['course_id'],
CourseModule.sequence == payload['sequence']
)
).first()
if existing_sequence:
return custom_response(
success=False,
data=f"Module with sequence {payload['sequence']} already exists for this course",
status_code=400
)
# Validate dates if provided
start_date = None
end_date = None
if 'start_date' in payload:
try:
start_date = datetime.fromisoformat(payload['start_date'])
except ValueError:
return custom_response(
success=False,
data="Invalid start_date format. Use ISO format (YYYY-MM-DD)",
status_code=400
)
if 'end_date' in payload:
try:
end_date = datetime.fromisoformat(payload['end_date'])
except ValueError:
return custom_response(
success=False,
data="Invalid end_date format. Use ISO format (YYYY-MM-DD)",
status_code=400
)
if start_date and end_date and start_date > end_date:
return custom_response(
success=False,
data="start_date cannot be after end_date",
status_code=400
)
# Create the module
module = CourseModule(
id=str(uuid.uuid4()),
course_id=payload['course_id'],
title=payload['title'],
description=payload.get('description'),
sequence=payload['sequence'],
start_date=start_date,
end_date=end_date,
objectives=payload.get('objectives'),
estimated_study_time=payload.get('estimated_study_time', 0),
is_assessment_module=payload.get('is_assessment_module', False),
created_at=datetime.utcnow()
)
ctx.session.add(module)
ctx.session.commit()
# Send notifications if requested
if payload.get('notify_users', True):
self._send_module_creation_notifications(ctx, module, course)
return custom_response(
success=True,
data={
'message': 'Course module created successfully',
'module': {
'id': module.id,
'title': module.title,
'course_id': module.course_id,
'course_code': course.code,
'sequence': module.sequence,
'start_date': module.start_date.isoformat() if module.start_date else None,
'end_date': module.end_date.isoformat() if module.end_date else None,
'is_published': module.is_published
}
},
status_code=201
)
def _send_module_creation_notifications(self, ctx, module: CourseModule, course: Course) -> None:
"""
Send notifications about new module creation to tutors and enrolled students.
Args:
ctx: Database context
module: The newly created CourseModule
course: The parent Course
"""
# Get all tutors for this course
tutors = ctx.session.query(Tutor).join(
tutor_course_association,
Tutor.id == tutor_course_association.c.tutor_id
).filter(
tutor_course_association.c.course_id == module.course_id
).all()
# Get all enrolled students
students = ctx.session.query(Student).join(
Enrollment,
Student.id == Enrollment.student_id
).filter(
Enrollment.course_id == module.course_id,
Enrollment.status == 'active'
).all()
# Notification to tutors
for tutor in tutors:
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = f"New Module Added: {module.title} - {course.code}"
message = f"""
New Course Module Notification
Hello {tutor.first_name},
A new module has been added to your course {course.code} - {course.title}:
Module Title: {module.title}
Sequence: {module.sequence}
{f"
Start Date: {module.start_date.strftime('%B %d, %Y')}
" if module.start_date else ""}
{f"
End Date: {module.end_date.strftime('%B %d, %Y')}
" if module.end_date else ""}
Estimated Study Time: {module.estimated_study_time} hours
Assessment Module: {'Yes' if module.is_assessment_module else 'No'}
{f"
Objectives: {module.objectives}
" if module.objectives else ""}
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 module notification to tutor {tutor.id}: {str(e)}")
# Notification to students (only if module is published)
if module.is_published:
for student in students:
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == student.id
).first()
if student_prefs and student_prefs.receive_email:
subject = f"New Module Available: {module.title} - {course.code}"
message = f"""
New Course Module Available
Hello {student.first_name},
A new module is now available in your course {course.code} - {course.title}:
Module Title: {module.title}
{f"
Start Date: {module.start_date.strftime('%B %d, %Y')}
" if module.start_date else ""}
{f"
End Date: {module.end_date.strftime('%B %d, %Y')}
" if module.end_date else ""}
Estimated Study Time: {module.estimated_study_time} hours
Assessment Module: {'Yes' if module.is_assessment_module else 'No'}
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=student.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send module notification to student {student.id}: {str(e)}")
def get_module(self, module_id: str) -> Dict:
"""
Get a single course module with all its details.
Args:
module_id: ID of the module to retrieve
Returns:
Response with module details or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
# Get associated course
course = ctx.session.query(Course).filter(
Course.id == module.course_id
).first()
# Get resources for this module
resources = ctx.session.query(CourseResource).filter(
CourseResource.module_id == module_id
).all()
# Get assignments for this module
assignments = ctx.session.query(Assignment).filter(
Assignment.module_id == module_id
).all()
return custom_response(
success=True,
data={
'module': {
'id': module.id,
'title': module.title,
'description': module.description,
'course_id': module.course_id,
'course_code': course.code,
'course_title': course.title,
'sequence': module.sequence,
'start_date': module.start_date.isoformat() if module.start_date else None,
'end_date': module.end_date.isoformat() if module.end_date else None,
'is_published': module.is_published,
'objectives': module.objectives,
'estimated_study_time': module.estimated_study_time,
'is_assessment_module': module.is_assessment_module,
'created_at': module.created_at.isoformat(),
'resource_count': len(resources),
'assignment_count': len(assignments)
},
'resources': [{
'id': r.id,
'title': r.title,
'resource_type': r.resource_type.value,
'is_required': r.is_required,
'views': r.views,
'downloads': r.downloads
} for r in resources],
'assignments': [{
'id': a.id,
'title': a.title,
'assignment_type': a.assignment_type.value,
'due_date': a.due_date.isoformat() if a.due_date else None,
'is_published': a.is_published,
'total_points': a.total_points
} for a in assignments]
},
status_code=200
)
def get_course_modules(self, course_id: str, include_unpublished: bool = False) -> Dict:
"""
Get all modules for a course, optionally including unpublished ones.
Args:
course_id: ID of the course
include_unpublished: Whether to include unpublished modules (default False)
Returns:
Response with list of modules or error
"""
with DatabaseContextManager() as ctx:
# Verify course exists
course = ctx.session.query(Course).filter(
Course.id == course_id
).first()
if not course:
return custom_response(
success=False,
data="Course not found",
status_code=404
)
# Build query
query = ctx.session.query(CourseModule).filter(
CourseModule.course_id == course_id
)
if not include_unpublished:
query = query.filter(CourseModule.is_published == True)
modules = query.order_by(
CourseModule.sequence.asc()
).all()
return custom_response(
success=True,
data={
'course': {
'id': course.id,
'code': course.code,
'title': course.title,
'total_modules': len(modules)
},
'modules': [{
'id': m.id,
'title': m.title,
'sequence': m.sequence,
'start_date': m.start_date.isoformat() if m.start_date else None,
'end_date': m.end_date.isoformat() if m.end_date else None,
'is_published': m.is_published,
'estimated_study_time': m.estimated_study_time,
'is_assessment_module': m.is_assessment_module,
'resource_count': ctx.session.query(CourseResource).filter(
CourseResource.module_id == m.id
).count(),
'assignment_count': ctx.session.query(Assignment).filter(
Assignment.module_id == m.id
).count()
} for m in modules]
},
status_code=200
)
def update_module(self, module_id: str, payload: Dict) -> Dict:
"""
Update a course module with validation and notifications.
Args:
module_id: ID of the module to update
payload: {
"title": str (optional),
"description": str (optional),
"sequence": int (optional),
"start_date": str (ISO format, optional),
"end_date": str (ISO format, optional),
"objectives": str (optional),
"estimated_study_time": int (optional, in hours),
"is_assessment_module": bool (optional),
"is_published": bool (optional),
"notify_users": bool (default True) - whether to send notifications
}
Returns:
Response with updated module or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
# Track changes for notification
changes = {}
was_published = module.is_published
# Validate sequence if changed
if 'sequence' in payload and payload['sequence'] != module.sequence:
existing_sequence = ctx.session.query(CourseModule).filter(
CourseModule.course_id == module.course_id,
CourseModule.sequence == payload['sequence'],
CourseModule.id != module_id
).first()
if existing_sequence:
return custom_response(
success=False,
data=f"Module with sequence {payload['sequence']} already exists for this course",
status_code=400
)
changes['sequence'] = {
'old': module.sequence,
'new': payload['sequence']
}
module.sequence = payload['sequence']
# Validate dates if provided
if 'start_date' in payload:
try:
new_start_date = datetime.fromisoformat(payload['start_date'])
if module.start_date != new_start_date:
changes['start_date'] = {
'old': module.start_date.isoformat() if module.start_date else None,
'new': payload['start_date']
}
module.start_date = new_start_date
except ValueError:
return custom_response(
success=False,
data="Invalid start_date format. Use ISO format (YYYY-MM-DD)",
status_code=400
)
if 'end_date' in payload:
try:
new_end_date = datetime.fromisoformat(payload['end_date'])
if module.end_date != new_end_date:
changes['end_date'] = {
'old': module.end_date.isoformat() if module.end_date else None,
'new': payload['end_date']
}
module.end_date = new_end_date
except ValueError:
return custom_response(
success=False,
data="Invalid end_date format. Use ISO format (YYYY-MM-DD)",
status_code=400
)
if module.start_date and module.end_date and module.start_date > module.end_date:
return custom_response(
success=False,
data="start_date cannot be after end_date",
status_code=400
)
# Update other fields if provided
if 'title' in payload and payload['title'] != module.title:
changes['title'] = {
'old': module.title,
'new': payload['title']
}
module.title = payload['title']
if 'description' in payload and payload['description'] != module.description:
changes['description'] = {
'old': module.description,
'new': payload['description']
}
module.description = payload['description']
if 'objectives' in payload and payload['objectives'] != module.objectives:
changes['objectives'] = {
'old': module.objectives,
'new': payload['objectives']
}
module.objectives = payload['objectives']
if 'estimated_study_time' in payload and payload['estimated_study_time'] != module.estimated_study_time:
changes['estimated_study_time'] = {
'old': module.estimated_study_time,
'new': payload['estimated_study_time']
}
module.estimated_study_time = payload['estimated_study_time']
if 'is_assessment_module' in payload and payload['is_assessment_module'] != module.is_assessment_module:
changes['is_assessment_module'] = {
'old': module.is_assessment_module,
'new': payload['is_assessment_module']
}
module.is_assessment_module = payload['is_assessment_module']
if 'is_published' in payload and payload['is_published'] != module.is_published:
changes['is_published'] = {
'old': module.is_published,
'new': payload['is_published']
}
module.is_published = payload['is_published']
module.updated_at = datetime.utcnow()
ctx.session.commit()
# Send notifications if changes were made and requested
if changes and payload.get('notify_users', True):
course = ctx.session.query(Course).filter(
Course.id == module.course_id
).first()
self._send_module_update_notifications(ctx, module, course, changes, was_published)
return custom_response(
success=True,
data={
'message': 'Course module updated successfully',
'module_id': module_id,
'changes': changes
},
status_code=200
)
def _send_module_update_notifications(self, ctx, module: CourseModule, course: Course,
changes: Dict, was_published: bool) -> None:
"""
Send notifications about module updates to tutors and enrolled students.
Args:
ctx: Database context
module: The updated CourseModule
course: The parent Course
changes: Dictionary of changes made
was_published: Whether the module was published before the update
"""
# Get all tutors for this course
tutors = ctx.session.query(Tutor).join(
tutor_course_association,
Tutor.id == tutor_course_association.c.tutor_id
).filter(
tutor_course_association.c.course_id == module.course_id
).all()
# Get all enrolled students (only notify if module is now published or was already published)
students = []
if module.is_published or was_published:
students = ctx.session.query(Student).join(
Enrollment,
Student.id == Enrollment.student_id
).filter(
Enrollment.course_id == module.course_id,
Enrollment.status == 'active'
).all()
# Prepare changes list for email
changes_list = []
for field, values in changes.items():
changes_list.append(f"{field.replace('_', ' ').title()}: From '{values['old']}' to '{values['new']}'")
# Notification to tutors
for tutor in tutors:
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = f"Module Updated: {module.title} - {course.code}"
message = f"""
Course Module Update Notification
Hello {tutor.first_name},
The following module in your course {course.code} - {course.title} has been updated:
Module Title: {module.title}
Changes:
{"".join(changes_list) if changes_list else "- No specific changes listed
"}
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 module update notification to tutor {tutor.id}: {str(e)}")
# Notification to students (only if module is published)
if module.is_published:
for student in students:
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == student.id
).first()
if student_prefs and student_prefs.receive_email:
subject = f"Module Updated: {module.title} - {course.code}"
message = f"""
Course Module Update Notification
Hello {student.first_name},
The following module in your course {course.code} - {course.title} has been updated:
Module Title: {module.title}
Important Changes:
{"" if not changes_list else "".join([
f"- {field.replace('_', ' ').title()}: Changed
"
for field in changes.keys()
if field not in ['description', 'objectives']
])}
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=student.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send module update notification to student {student.id}: {str(e)}")
def delete_module(self, module_id: str, notify_users: bool = False) -> Dict:
"""
Delete a course module after validation.
Args:
module_id: ID of the module to delete
notify_users: Whether to send notifications about the deletion (default False)
Returns:
Response with deletion status or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
# Get course for notification purposes
course = ctx.session.query(Course).filter(
Course.id == module.course_id
).first()
# Check if module has resources or assignments
has_resources = ctx.session.query(CourseResource).filter(
CourseResource.module_id == module_id
).first()
has_assignments = ctx.session.query(Assignment).filter(
Assignment.module_id == module_id
).first()
if has_resources or has_assignments:
return custom_response(
success=False,
data="Cannot delete module with associated resources or assignments",
status_code=400
)
# Store module info for notification
module_info = {
'title': module.title,
'course_code': course.code,
'course_title': course.title
}
ctx.session.delete(module)
ctx.session.commit()
# Send notifications if requested
if notify_users:
self._send_module_deletion_notifications(ctx, module_info, course)
return custom_response(
success=True,
data={
'message': 'Course module deleted successfully',
'module_id': module_id
},
status_code=200
)
def _send_module_deletion_notifications(self, ctx, module_info: Dict, course: Course) -> None:
"""
Send notifications about module deletion to tutors and enrolled students.
Args:
ctx: Database context
module_info: Dictionary with module title and course info
course: The parent Course
"""
# Get all tutors for this course
tutors = ctx.session.query(Tutor).join(
tutor_course_association,
Tutor.id == tutor_course_association.c.tutor_id
).filter(
tutor_course_association.c.course_id == course.id
).all()
# Get all enrolled students (only if module was published)
students = ctx.session.query(Student).join(
Enrollment,
Student.id == Enrollment.student_id
).filter(
Enrollment.course_id == course.id,
Enrollment.status == 'active'
).all()
# Notification to tutors
for tutor in tutors:
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = f"Module Removed: {module_info['title']} - {course.code}"
message = f"""
Course Module Removal Notification
Hello {tutor.first_name},
The following module has been removed from your course {course.code} - {course.title}:
Module Title: {module_info['title']}
Course: {course.code} - {course.title}
If this was unexpected, please contact your supervisor.
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 module deletion notification to tutor {tutor.id}: {str(e)}")
# Notification to students
for student in students:
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == student.id
).first()
if student_prefs and student_prefs.receive_email:
subject = f"Module Removed: {module_info['title']} - {course.code}"
message = f"""
Course Module Removal Notification
Hello {student.first_name},
The following module has been removed from your course {course.code} - {course.title}:
Module Title: {module_info['title']}
If you have any questions about this change, please contact your tutor.
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=student.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send module deletion notification to student {student.id}: {str(e)}")
def publish_module(self, module_id: str, notify_users: bool = True) -> Dict:
"""
Publish a module to make it visible to students, with notifications.
Args:
module_id: ID of the module to publish
notify_users: Whether to send notifications (default True)
Returns:
Response with publication status or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
if module.is_published:
return custom_response(
success=False,
data="Module is already published",
status_code=400
)
module.is_published = True
module.published_at = datetime.utcnow()
ctx.session.commit()
# Send notifications if requested
if notify_users:
course = ctx.session.query(Course).filter(
Course.id == module.course_id
).first()
self._send_module_publication_notifications(ctx, module, course)
return custom_response(
success=True,
data={
'message': 'Course module published successfully',
'module_id': module_id,
'published_at': module.published_at.isoformat()
},
status_code=200
)
def _send_module_publication_notifications(self, ctx, module: CourseModule, course: Course) -> None:
"""
Send notifications about module publication to enrolled students.
Args:
ctx: Database context
module: The published CourseModule
course: The parent Course
"""
# Get all enrolled students
students = ctx.session.query(Student).join(
Enrollment,
Student.id == Enrollment.student_id
).filter(
Enrollment.course_id == module.course_id,
Enrollment.status == 'active'
).all()
# Notification to students
for student in students:
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == student.id
).first()
if student_prefs and student_prefs.receive_email:
subject = f"New Module Available: {module.title} - {course.code}"
message = f"""
New Course Module Available
Hello {student.first_name},
A new module is now available in your course {course.code} - {course.title}:
Module Title: {module.title}
{f"
Start Date: {module.start_date.strftime('%B %d, %Y')}
" if module.start_date else ""}
{f"
End Date: {module.end_date.strftime('%B %d, %Y')}
" if module.end_date else ""}
Estimated Study Time: {module.estimated_study_time} hours
Assessment Module: {'Yes' if module.is_assessment_module else 'No'}
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=student.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send module publication notification to student {student.id}: {str(e)}")
def get_module_progress(self, module_id: str, student_id: str = None) -> Dict:
"""
Get progress statistics for a module, optionally for a specific student.
Args:
module_id: ID of the module
student_id: Optional student ID to filter by
Returns:
Response with progress data or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
course = ctx.session.query(Course).filter(
Course.id == module.course_id
).first()
# Get all resources for this module
resources = ctx.session.query(CourseResource).filter(
CourseResource.module_id == module_id
).all()
# Get all assignments for this module
assignments = ctx.session.query(Assignment).filter(
Assignment.module_id == module_id
).all()
# Get student progress if student_id is provided
student_progress = None
if student_id:
# Verify student is enrolled in the course
enrollment = ctx.session.query(Enrollment).filter(
Enrollment.student_id == student_id,
Enrollment.course_id == module.course_id,
Enrollment.status == 'active'
).first()
if not enrollment:
return custom_response(
success=False,
data="Student is not enrolled in this course",
status_code=403
)
# Get resource views/downloads for this student
resource_stats = []
for resource in resources:
# In a real implementation, you would query a resource access log table
# Here we're just returning placeholder data
resource_stats.append({
'resource_id': resource.id,
'views': 1 if resource.id.endswith('0') else 0, # Placeholder logic
'downloads': 1 if resource.id.endswith('5') else 0, # Placeholder logic
'last_accessed': datetime.utcnow().isoformat() if resource.id.endswith('0') else None
})
# Get assignment submissions for this student
assignment_submissions = []
for assignment in assignments:
submission = ctx.session.query(AssignmentSubmission).filter(
AssignmentSubmission.assignment_id == assignment.id,
AssignmentSubmission.student_id == student_id
).first()
if submission:
assignment_submissions.append({
'assignment_id': assignment.id,
'status': submission.status.value,
'submitted_at': submission.submitted_at.isoformat() if submission.submitted_at else None,
'grade': submission.grade,
'is_late': submission.status == SubmissionStatus.late
})
else:
assignment_submissions.append({
'assignment_id': assignment.id,
'status': 'not_started',
'submitted_at': None,
'grade': None,
'is_late': False
})
# Calculate completion percentage
total_items = len(resources) + len(assignments)
completed_items = len([r for r in resource_stats if r['views'] > 0]) + \
len([a for a in assignment_submissions if a['status'] in ['submitted', 'graded']])
completion_percentage = round((completed_items / total_items) * 100, 1) if total_items > 0 else 0
student_progress = {
'student_id': student_id,
'resource_stats': resource_stats,
'assignment_submissions': assignment_submissions,
'completion_percentage': completion_percentage,
'started_at': datetime.utcnow().isoformat(), # Placeholder - would come from first access
'last_accessed': datetime.utcnow().isoformat() # Placeholder
}
return custom_response(
success=True,
data={
'module': {
'id': module.id,
'title': module.title,
'course_id': module.course_id,
'course_code': course.code,
'is_published': module.is_published
},
'stats': {
'total_resources': len(resources),
'total_assignments': len(assignments),
'published_resources': len([r for r in resources if r.is_published]),
'published_assignments': len([a for a in assignments if a.is_published]),
'average_completion': 75.5 # Placeholder - would calculate from student data
},
'student_progress': student_progress
},
status_code=200
)
def create_module_new(self, payload: Dict) -> Dict:
"""
Create a new course module with the new structure.
Args:
payload: {
"course_week_id": str,
"title": str,
"description": str (optional),
"module_type": str (default 'lecture'),
"estimated_duration_minutes": int (default 60),
"learning_objectives": str (optional),
"is_required": bool (default True),
"requester_id": str
}
Returns:
Response with created module or error
"""
with DatabaseContextManager() as ctx:
# Validate course week exists and get course_id through the level relationship
from src.models.models import CourseWeek, CourseLevel
week = ctx.session.query(CourseWeek).join(CourseLevel).filter(
CourseWeek.id == payload['course_week_id']
).first()
if not week:
return custom_response(
success=False,
data="Course week not found",
status_code=404
)
# Get the course_id from the level
level = ctx.session.query(CourseLevel).filter(
CourseLevel.id == week.course_level_id
).first()
if not level:
return custom_response(
success=False,
data="Course level not found",
status_code=404
)
# Create the module
module = CourseModule(
id=str(uuid.uuid4()),
course_id=level.course_id,
course_week_id=payload['course_week_id'],
title=payload['title'],
description=payload.get('description'),
module_type=payload.get('module_type', 'lecture'),
estimated_duration_minutes=payload.get('estimated_duration_minutes', 60),
learning_objectives=payload.get('learning_objectives'),
is_required=payload.get('is_required', True),
created_at=datetime.utcnow()
)
ctx.session.add(module)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Course module created successfully',
'module': {
'id': module.id,
'title': module.title,
'course_week_id': module.course_week_id,
'module_type': module.module_type,
'estimated_duration_minutes': module.estimated_duration_minutes,
'is_required': module.is_required,
'is_published': module.is_published
}
},
status_code=201
)
def update_module_new(self, module_id: str, payload: Dict) -> Dict:
"""
Update a course module with the new structure.
Args:
module_id: ID of the module to update
payload: Update data
Returns:
Response with updated module or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
# Update fields if provided
if 'title' in payload:
module.title = payload['title']
if 'description' in payload:
module.description = payload['description']
if 'module_type' in payload:
module.module_type = payload['module_type']
if 'estimated_duration_minutes' in payload:
module.estimated_duration_minutes = payload['estimated_duration_minutes']
if 'learning_objectives' in payload:
module.learning_objectives = payload['learning_objectives']
if 'is_required' in payload:
module.is_required = payload['is_required']
module.updated_at = datetime.utcnow()
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Course module updated successfully',
'module': {
'id': module.id,
'title': module.title,
'course_week_id': module.course_week_id,
'module_type': module.module_type,
'estimated_duration_minutes': module.estimated_duration_minutes,
'is_required': module.is_required,
'is_published': module.is_published
}
},
status_code=200
)
def delete_module_new(self, module_id: str, requester_id: str) -> Dict:
"""
Delete a course module.
Args:
module_id: ID of the module to delete
requester_id: ID of the user requesting deletion
Returns:
Response with deletion status or error
"""
with DatabaseContextManager() as ctx:
module = ctx.session.query(CourseModule).filter(
CourseModule.id == module_id
).first()
if not module:
return custom_response(
success=False,
data="Course module not found",
status_code=404
)
ctx.session.delete(module)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Course module deleted successfully',
'module_id': module_id
},
status_code=200
)
def get_modules_by_course(self, course_id: str, requester_id: str) -> Dict:
"""
Get all modules for a course with week information.
Args:
course_id: ID of the course
requester_id: ID of the user requesting
Returns:
Response with list of modules or error
"""
with DatabaseContextManager() as ctx:
# Verify course exists
course = ctx.session.query(Course).filter(
Course.id == course_id
).first()
if not course:
return custom_response(
success=False,
data="Course not found",
status_code=404
)
# Get all modules for this course with week information
from src.models.models import CourseWeek, CourseLevel
modules = ctx.session.query(CourseModule).join(
CourseWeek, CourseModule.course_week_id == CourseWeek.id
).join(
CourseLevel, CourseWeek.course_level_id == CourseLevel.id
).filter(
CourseModule.course_id == course_id
).all()
# Format the response
formatted_modules = []
for module in modules:
# Get week information
week = ctx.session.query(CourseWeek).filter(
CourseWeek.id == module.course_week_id
).first()
week_info = None
if week:
level = ctx.session.query(CourseLevel).filter(
CourseLevel.id == week.course_level_id
).first()
week_info = {
'id': week.id,
'week_title': week.week_title,
'week_number': week.week_number,
'level': {
'id': level.id if level else None,
'level_name': level.level_name if level else None
} if level else None
}
formatted_modules.append({
'id': module.id,
'title': module.title,
'description': module.description,
'module_type': module.module_type,
'estimated_duration_minutes': module.estimated_duration_minutes,
'learning_objectives': module.learning_objectives,
'is_required': module.is_required,
'is_published': module.is_published,
'course_week_id': module.course_week_id,
'week': week_info,
'created_at': module.created_at.isoformat(),
'updated_at': module.updated_at.isoformat()
})
return custom_response(
success=True,
data=formatted_modules,
status_code=200
)