Feature integrate with helios

app/Controllers/User/education_controller.py

@@ -4,7 +4,7 @@
 from sqlmodel import Session
 
 from Entities.UserDTOs.education_entity import CreateEducation, UpdateEducation, ReadEducation
-from Services.User.education_service import EducationService
+from Services.User.education_service import EducationService, get_education_service_with_publisher
 from Settings.logging_config import setup_logging
 from db import get_session
 
@@ -14,10 +14,9 @@
 
 
 @router.post("/", response_model=ReadEducation)
-def create_education(education_create: CreateEducation, session: Session = Depends(get_session)):
-    service = EducationService(session)
+def create_education(education_create: CreateEducation,  education_service: EducationService = Depends(get_education_service_with_publisher)):
     logger.info(f"Creating Education entry for profile_id={education_create.profile_id}")
-    education = service.create_education(education_create)
+    education = education_service.create_education(education_create)
     logger.info(f"Created Education with ID: {education.id}")
     return education
 
@@ -53,19 +52,17 @@ def list_educations(
 
 
 @router.put("/{education_id}", response_model=ReadEducation)
-def update_education(education_id: UUID, education_update: UpdateEducation, session: Session = Depends(get_session)):
-    service = EducationService(session)
+def update_education(education_id: UUID, education_update: UpdateEducation, education_service: EducationService = Depends(get_education_service_with_publisher)):
     logger.info(f"Updating Education ID: {education_id} with data: {education_update.dict(exclude_unset=True)}")
-    education = service.update_education(education_id, education_update)
+    education = education_service.update_education(education_id, education_update)
     logger.info(f"Updated Education ID: {education.id}")
     return education
 
 
 @router.delete("/{education_id}")
-def delete_education(education_id: UUID, session: Session = Depends(get_session)):
-    service = EducationService(session)
+def delete_education(education_id: UUID,  education_service: EducationService = Depends(get_education_service_with_publisher)):
     logger.info(f"Deleting Education ID: {education_id}")
-    message = service.delete_education(education_id)
+    message = education_service.delete_education(education_id)
     logger.info(message)
     return {"detail": message}
 

app/Controllers/User/leetcode_controller.py

@@ -14,14 +14,15 @@
     ReadLeetcodeTag,
 )
 
+from Services.User.leetcode_service import get_leetcode_service_with_publisher
+
 logger = setup_logging()
 
 router = APIRouter(prefix="/Dijkstra/v1/leetcode", tags=["Leetcode"])
 
 
 @router.post("/sync/{profile_id}/{lc_username}", response_model=ReadLeetcode)
-def sync_leetcode(profile_id: UUID, lc_username: str, session: Session = Depends(get_session)):
-    service = LeetCodeService(session)
+def sync_leetcode(profile_id: UUID, lc_username: str, service: LeetCodeService = Depends(get_leetcode_service_with_publisher)):
     logger.info(f"Syncing LeetCode data profile_id={profile_id} username={lc_username}")
     return service.create_or_update_from_api(profile_id, lc_username)
 
@@ -55,8 +56,7 @@ def get_leetcode_by_profile(profile_id: UUID, session: Session = Depends(get_ses
 
 
 @router.delete("/{leetcode_id}", status_code=204)
-def delete_leetcode(leetcode_id: UUID, session: Session = Depends(get_session)):
-    service = LeetCodeService(session)
+def delete_leetcode(leetcode_id: UUID, service: LeetCodeService = Depends(get_leetcode_service_with_publisher)):
     service.delete(leetcode_id)
     return Response(status_code=204)
 

app/Controllers/User/workexperience_controller.py

@@ -9,14 +9,15 @@
 from db import get_session
 from Schema.SQL.Enums.enums import EmploymentType, WorkLocationType, Domain
 
+from Services.User.workexperience_service import get_workexperience_service_with_publisher
+
 logger = setup_logging()
 
 router = APIRouter(prefix="/Dijkstra/v1/wp", tags=["Work Experiences"])
 
 
 @router.post("/", response_model=ReadWorkExperience)
-def create_work_experience(work_experience_create: CreateWorkExperience, session: Session = Depends(get_session)):
-    service = WorkExperienceService(session)
+def create_work_experience(work_experience_create: CreateWorkExperience, service: WorkExperienceService = Depends(get_workexperience_service_with_publisher)):
     logger.info(f"Creating Work Experience: {work_experience_create.title} at {work_experience_create.company_name}")
     work_experience = service.create_work_experience(work_experience_create)
     logger.info(f"Created Work Experience with ID: {work_experience.id}")
@@ -111,17 +112,15 @@ def autocomplete_work_experiences(
 
 @router.put("/{work_experience_id}", response_model=ReadWorkExperience)
 def update_work_experience(
-    work_experience_id: UUID, work_experience_update: UpdateWorkExperience, session: Session = Depends(get_session)
+    work_experience_id: UUID, work_experience_update: UpdateWorkExperience, service: WorkExperienceService = Depends(get_workexperience_service_with_publisher)
 ):
-    service = WorkExperienceService(session)
     logger.info(f"Updating Work Experience ID: {work_experience_id} with data: {work_experience_update.dict(exclude_unset=True)}")
     work_experience = service.update_work_experience(work_experience_id, work_experience_update)
     logger.info(f"Updated Work Experience ID: {work_experience.id}")
     return work_experience
 
 @router.delete("/{work_experience_id}")
-def delete_work_experience(work_experience_id: UUID, session: Session = Depends(get_session)):
-    service = WorkExperienceService(session)
+def delete_work_experience(work_experience_id: UUID, service: WorkExperienceService = Depends(get_workexperience_service_with_publisher)):
     logger.info(f"Deleting Work Experience ID: {work_experience_id}")
     message = service.delete_work_experience(work_experience_id)
     logger.info(f"Deleted Work Experience ID: {work_experience_id}")

app/Entities/UserDTOs/extended_entities.py

@@ -53,4 +53,3 @@ class ReadUserFull(ReadUser):
 
     class Config:
         from_attributes = True
-

app/Entities/UserDTOs/profile_entity_kafka_dto.py

@@ -0,0 +1,80 @@
+from pydantic import BaseModel
+from typing import Optional, List, Dict
+from Schema.SQL.Models.models import Profile
+from sqlalchemy import null
+
+from Schema.SQL.Enums.enums import SchoolType
+from Utils.utility_functions import calculate_months_served
+
+
+class EducationUpdateEventDTO(BaseModel):
+    """Education Update Event DTO"""
+    salary: int
+    time_served_months: int
+
+class CpgaUpdateEventDTO(BaseModel):
+    """CPGA Update Event DTO"""
+    cgpa: float
+
+class DsaMetricsUpdateEventDTO(BaseModel):
+    """Dsa Metrics Update Event DTO"""
+    contest_rating: int
+    global_rank: int
+
+class ProfileUpdateKafkaEvent(BaseModel):
+    """Kafka event payload for profile updates"""
+    user_id: str
+    work_experiences: Optional[List[EducationUpdateEventDTO]] = None
+    cgpa_metrics: Optional[CpgaUpdateEventDTO] = None
+    dsa_metrics: Optional[DsaMetricsUpdateEventDTO] = None
+
+def map_profile_to_kafka_event(profile: Profile) -> ProfileUpdateKafkaEvent:
+    """
+    Convert a Profile ORM object into a ProfileUpdateKafkaEvent DTO.
+    """
+
+    # ----------------------------
+    # Work Experiences Mapping
+    # ----------------------------
+    work_experience_dtos = []
+    for wx in profile.work_experience:
+        months_served = calculate_months_served(
+            wx.start_date_month,
+            wx.start_date_year,
+            wx.end_date_month,
+            wx.end_date_year
+        )
+        work_experience_dtos.append(
+            EducationUpdateEventDTO(
+                salary=(wx.yearly_salary_rupees or 2000000) / 100000, #TODO: This should be 0. Have kept it to a large number for dev testing
+                time_served_months=months_served,
+            )
+        )
+    # ----------------------------
+    # CGPA Mapping
+    # ----------------------------
+    edu = next(
+        (e for e in profile.education
+         if e.school_type in {SchoolType.UNIVERSITY, SchoolType.COLLEGE}),
+        None
+    )
+    #FIXME: Currently the CGPA is in the 4 scale while helios expects the 10 scale.
+    cgpa_dto = CpgaUpdateEventDTO(cgpa=edu.cgpa) if edu and edu.cgpa is not None else None
+
+    # ----------------------------
+    # DSA Metrics Mapping
+    # ----------------------------
+    leetcode_profile = profile.leetcode
+    dsa_dto = DsaMetricsUpdateEventDTO(global_rank=leetcode_profile.global_ranking, contest_rating=leetcode_profile.competition_rating) if leetcode_profile is not None and leetcode_profile.competition_rating is not None else None
+
+    # ----------------------------
+    # Construct final event DTO
+    # ----------------------------
+    event = ProfileUpdateKafkaEvent(
+        user_id=str(profile.user_rel.github_user_name),
+        work_experiences=work_experience_dtos,
+        cgpa_metrics=cgpa_dto,
+        dsa_metrics=dsa_dto,
+    )
+
+    return event

app/Entities/UserDTOs/workexperience_entity.py

@@ -154,6 +154,7 @@ class ReadWorkExperience(BaseModel):
     domain: Optional[List[Domain]]
     company_name: str
     company_logo: Optional[str]
+    yearly_salary_rupees: Optional[float]
     currently_working: bool
     location: Optional[ReadLocation]
     location_type: WorkLocationType

app/Schema/SQL/Models/models.py

@@ -184,7 +184,7 @@ class WorkExperience(UUIDBaseTable, table=True):
     tools_used: Optional[List[Tools]] = Field(
         sa_column=Column(ARRAY(SQLEnum(Tools, name="TOOLS")))
     )
-
+    yearly_salary_rupees: Optional[float] = None
     # Relationships
     profile_rel: Profile = Relationship(back_populates="work_experience")
     location_rel: Optional[Location] = Relationship(back_populates="work_experience")

app/Services/User/education_service.py

@@ -1,5 +1,6 @@
 from uuid import UUID
 from typing import List, Optional
+from fastapi.params import Depends
 from sqlmodel import Session
 
 from Repository.User.education_repository import EducationRepository
@@ -8,13 +9,16 @@
 from Entities.UserDTOs.location_entity import CreateLocation, ReadLocation
 from Schema.SQL.Models.models import Education, Profile, Location
 from Utils.Exceptions.user_exceptions import EducationNotFound, ProfileNotFound, LocationNotFound
+from Services.Kafka.producer_service import KafkaProducerService, get_kafka_producer
+from db import get_session
 
 
 class EducationService:
-    def __init__(self, session: Session):
+    def __init__(self, session: Session, kafka_producer: KafkaProducerService = None):
         self.repo = EducationRepository(session)
         self.location_repo = LocationRepository(session)
         self.session = session
+        self.kafka_producer = kafka_producer
 
     def _convert_to_read_dto(self, education: Education) -> ReadEducation:
         """Convert Education database model to ReadEducation DTO with populated location"""
@@ -73,6 +77,7 @@ def create_education(self, education_create: CreateEducation) -> ReadEducation:
 
         education = Education(**education_data)
         created_education = self.repo.create(education)
+        self.publish_profile_on_education_persist(created_education.profile_id)
         return self._convert_to_read_dto(created_education)
 
     def get_education(self, education_id: UUID) -> ReadEducation:
@@ -136,13 +141,16 @@ def update_education(self, education_id: UUID, education_update: UpdateEducation
             setattr(education, key, value)
 
         updated_education = self.repo.update(education)
+        self.publish_profile_on_education_persist(updated_education.profile_id)
         return self._convert_to_read_dto(updated_education)
 
     def delete_education(self, education_id: UUID) -> str:
         education = self.repo.get(education_id)
         if not education:
             raise EducationNotFound(education_id)
+        profile_id = education.profile_id
         self.repo.delete(education)
+        self.publish_profile_on_education_persist(profile_id)
         return f"Education {education_id} deleted successfully"
 
     def get_educations_by_profile_with_locations(self, profile_id: UUID) -> List[ReadEducation]:
@@ -158,4 +166,18 @@ def get_educations_by_github_username(self, github_username: str) -> List[ReadEd
 
         profile_service = ProfileService(self.session)
         profile_id = profile_service.get_profile_id_by_github_username(github_username)
-        return self.get_educations_by_profile(profile_id)
+        return self.get_educations_by_profile(profile_id)
+
+    def publish_profile_on_education_persist(self, profile_id: UUID):
+        """Publish profile update when education is created or updated"""
+        from Services.User.profile_service import ProfileService
+
+        profile_service = ProfileService(self.session, self.kafka_producer)
+        profile_service.publish_profile_update(profile_id)
+
+
+def get_education_service_with_publisher(
+    session: Session = Depends(get_session),
+    kafka: KafkaProducerService = Depends(get_kafka_producer),
+):
+    return EducationService(session=session, kafka_producer=kafka)

app/Services/User/leetcode_service.py

@@ -1,6 +1,7 @@
 from typing import Optional, List, Dict, Any
 from uuid import UUID
 import requests
+from fastapi import Depends
 from sqlmodel import Session
 
 from Settings.logging_config import setup_logging
@@ -27,16 +28,19 @@
     LeetcodeTagNotFound,
 )
 
+from Services.Kafka.producer_service import KafkaProducerService, get_kafka_producer
+from db import get_session
+
 logger = setup_logging()
 
 
 class LeetCodeService:
-    def __init__(self, session: Session):
+    def __init__(self, session: Session, kafka_producer: KafkaProducerService = None):
         self.session = session
         self.repo = LeetcodeRepository(session)
         self.badge_repo = LeetcodeBadgeRepository(session)
         self.tag_repo = LeetcodeTagRepository(session)
-
+        self.kafka_producer = kafka_producer
     # Basic read helpers (kept small to align with minimal service design)
     def get(self, leetcode_id: UUID) -> Optional[Leetcode]:
         return self.repo.get(leetcode_id)
@@ -187,7 +191,10 @@ def diff_count(name: str) -> Optional[int]:
             top_percentage=contest.get("topPercentage"),
             competition_badge=(contest.get("badge") or {}).get("name") if isinstance(contest.get("badge"), dict) else None,
         )
-        return self.repo.create(model)
+
+        model =  self.repo.create(model)
+        self.publish_profile_on_education_persist(model.profile_id)
+        return model
 
     def _fetch_api(self, username: str) -> Dict[str, Any]:
         try:
@@ -206,3 +213,15 @@ def _fetch_api(self, username: str) -> Dict[str, Any]:
         except Exception as e:  
             logger.exception("LeetCode API fetch failed")
             return {"error": str(e)}
+
+    def publish_profile_on_education_persist(self, profile_id: UUID):
+        """Publish profile update when education is created or updated"""
+        from Services.User.profile_service import ProfileService
+        profile_service = ProfileService(self.session, self.kafka_producer)
+        profile_service.publish_profile_update(profile_id)
+
+def get_leetcode_service_with_publisher(
+    session: Session = Depends(get_session),
+    kafka: KafkaProducerService = Depends(get_kafka_producer),
+):
+    return LeetCodeService(session=session, kafka_producer=kafka)