from datetime import datetime, timezone from sqlalchemy import update,extract from sqlalchemy.orm import joinedload from sqlalchemy.sql import exists import random from core.config import get_db from core.security.exceptions import CoreDBError, GenericError from models import BursaryDiscountInfo, BursaryDiscountApplication, Participant, User, UserProfile, Course, AcademicYear from models.bursary_discount import BursaryCourseAllocation from schema.academic_year import AcademicYearResponseSchema from schema.bursary_discount import BursaryDiscountInfoResponse from schema.course import CourseResponseSchema class BursaryDiscountService: def create_info(self, request): with get_db() as db: payload = request.model_dump() try: instance = BursaryDiscountInfo(**payload) db.add(instance) db.commit() db.refresh(instance) return instance except Exception as e: db.rollback() raise CoreDBError(f"Could not create new bursary discount info: {e}") def list_all_info(self,academic_year_id): with get_db() as db: query = db.query(BursaryDiscountInfo).options(joinedload(BursaryDiscountInfo.academic_year)).filter( BursaryDiscountInfo.deleted_at.is_(None)) if academic_year_id: query=query.filter(BursaryDiscountInfo.academic_year_id==academic_year_id) return query.all() def list_bursary_info_academic_year(self): with get_db() as db: query = db.query(BursaryDiscountInfo).options(joinedload(BursaryDiscountInfo.academic_year)).filter( BursaryDiscountInfo.deleted_at.is_(None),BursaryDiscountInfo.is_enabled==True) bursary_info_academic_year=[] for row in query: bursary_info_academic_year.append(row.academic_year_id) records = db.query(AcademicYear).where(AcademicYear.deleted_at.is_(None),AcademicYear.id.in_(bursary_info_academic_year)).all() return [AcademicYearResponseSchema.model_validate(record).model_dump() for record in records] def bursary_discount_detail(self, bursary_discount_id: int): with get_db() as db: record = db.query(BursaryDiscountInfo).where(BursaryDiscountInfo.id == bursary_discount_id, BursaryDiscountInfo.deleted_at.is_(None)).first() if not record: raise GenericError(status_code=404, exc="Bursary discount not found") return BursaryDiscountInfoResponse.model_validate(record).model_dump() def update_bursary_discount(self, bursary_discount_id, request): payload = request.model_dump(exclude=['id']) with get_db() as db: result = (db.query(BursaryDiscountInfo).filter(BursaryDiscountInfo.id == bursary_discount_id, BursaryDiscountInfo.deleted_at.is_(None)).update(payload, synchronize_session="fetch")) db.commit() if result == 0: raise GenericError(status_code=422, exc="Could not update bursary discount") def toogle_enable(self,bursary_discount_id): with get_db() as db: db.execute( update(BursaryDiscountInfo) .where(BursaryDiscountInfo.id == bursary_discount_id) .values(is_enabled=~BursaryDiscountInfo.is_enabled) ) db.commit() def create_application(self, request, user): try: with get_db() as db: application_list=[] for participant_id in request.participant_id: application_number=f"{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(1000, 9999)}" participant = db.query(Participant).where(Participant.id == participant_id).first() application_list.append(BursaryDiscountApplication( application_number=application_number, parent_id=participant.parent_id, academic_year_id=request.academic_year_id, participant_id=participant_id, is_parent_user= user['user_type'] == 'parent', created_by=user['id'] )) db.bulk_save_objects(application_list) db.commit() except Exception as e: db.rollback() raise CoreDBError(f"Could not create new bursary discount application: {e}") def delete_bursary_discount(self, bursary_discount_id): with get_db() as db: bursary_discount = db.query(BursaryDiscountInfo).where(BursaryDiscountInfo.id == bursary_discount_id, BursaryDiscountInfo.deleted_at.is_(None)).first() if not bursary_discount: raise GenericError(status_code=404, exc="Bursary discount not found") bursary_discount.deleted_at = datetime.now(timezone.utc) db.commit() return {"message": "Bursary discount deleted successfully"} def list_application(self, academic_year_id, name, age,is_approved,application_number): with get_db() as db: query = db.query(BursaryDiscountApplication).options( joinedload(BursaryDiscountApplication.academic_year), joinedload(BursaryDiscountApplication.participant) ).filter( BursaryDiscountApplication.deleted_at.is_(None)) if is_approved is True: query = query.filter(BursaryDiscountApplication.is_approved == is_approved) if is_approved is False: query = query.filter(BursaryDiscountApplication.is_approved == False) if academic_year_id: query = query.filter(BursaryDiscountApplication.academic_year_id == academic_year_id) if application_number: query = query.filter(BursaryDiscountApplication.application_number == application_number) if name: participant = db.query(Participant).filter(Participant.first_name.icontains(name)).all() participant_id = [row.id for row in participant] query = query.filter(BursaryDiscountApplication.participant_id.in_(participant_id)) if age: current_year = datetime.now().year birth_year= current_year - age participant = db.query(Participant).filter(extract('year', Participant.dob) == birth_year).all() participant_id = [row.id for row in participant] query = query.filter(BursaryDiscountApplication.participant_id.in_(participant_id)) return query.all() def approve_application(self, id): with get_db() as db: db.query(BursaryDiscountApplication).filter(BursaryDiscountApplication.id == id).update( {"is_approved": True}) db.commit() def fetch_all_bursary_courses(self): with get_db() as db: records = db.query(Course).where(Course.is_bursary_product == True, Course.deleted_at.is_(None)).all() return [CourseResponseSchema.model_validate(record).model_dump() for record in records] def allocate_course(self, request): with get_db() as db: payload = request.model_dump(exclude=['is_fully_allocated']) instance_list = [BursaryCourseAllocation(course_id=row.course_id, discount_price=row.discount_price, bursary_discount_application_id=request.bursary_discount_application_id) for row in request.course] db.add_all(instance_list) db.query(BursaryDiscountApplication).filter( BursaryDiscountApplication.id == request.bursary_discount_application_id).update( {"is_fully_allocated": request.is_fully_allocated}) db.commit() def list_allocated_course(self, application_id): with get_db() as db: records = db.query(BursaryCourseAllocation).filter( BursaryCourseAllocation.bursary_discount_application_id == application_id).all() return [ dict( id=row.id, bursary_discount_application_id=row.bursary_discount_application_id, course_id=row.course_id, discount_price=row.discount_price, course=dict(id=row.course.id,name=row.course.name) ) for row in records ] def remove_course_allocation(self,allocation_id): with get_db() as db: db.query(BursaryCourseAllocation).filter(BursaryCourseAllocation.bursary_discount_application_id==allocation_id).delete() db.query(BursaryDiscountApplication).filter(BursaryDiscountApplication.id==allocation_id).update({"is_approved":False,"is_fully_allocated":False}) db.commit() def get_allocated_course_details(self, application_id: int): """ Get detailed information about allocated courses for a specific bursary application Args: application_id (int): ID of the bursary discount application Returns: dict: Details of allocated courses including application info and course details """ with get_db() as db: # Get the application details with related data application = db.query(BursaryDiscountApplication).options( joinedload(BursaryDiscountApplication.participant), joinedload(BursaryDiscountApplication.academic_year) ).filter( BursaryDiscountApplication.id == application_id, BursaryDiscountApplication.deleted_at.is_(None) ).first() if not application: raise GenericError(status_code=404, exc="Bursary application not found") # Get allocated courses allocated_courses = db.query(BursaryCourseAllocation).options( joinedload(BursaryCourseAllocation.course) ).filter( BursaryCourseAllocation.bursary_discount_application_id == application_id ).all() return { "application_details": { "id": application.id, "application_number": application.application_number, "participant_name": f"{application.participant.first_name} {application.participant.last_name}", "academic_year": application.academic_year.title, "is_approved": application.is_approved, "is_fully_allocated": application.is_fully_allocated }, "allocated_courses": [ { "allocation_id": course.id, "course_id": course.course.id, "course_name": course.course.name, "discount_price": course.discount_price, "academic_year": course.course.academic_year.title, # Assuming academic_year has a label field "term": course.course.term.name, # Assuming term has a name field "age_group": course.course.age_group.label # Assuming age_group has a label field } for course in allocated_courses ] }