Fix lint errors

Author: Ista2000

.github/workflows/ci.yml

@@ -38,7 +38,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.9"
+          python-version: "3.10"
 
       - name: Install Poetry
         run: pip install poetry
@@ -52,7 +52,7 @@ jobs:
         run: |
           cd backend
           poetry run black --check src tests
-          poetry run mypy src
+          poetry run mypy src --install-types --non-interactive --ignore-missing-imports
 
       - name: Run tests
         run: |

backend/src/llm/llm.py

@@ -47,7 +47,7 @@ class LLM:
 
     def __init__(self) -> None:
         self.llms = [
-            ChatGroq(
+            ChatGroq(  # type: ignore[call-arg]
                 temperature=0,
                 groq_api_key=GROQ_API_KEY,
                 model_name=model_name,
@@ -64,9 +64,10 @@ def extract_job_from_page_data(
         for idx, llm in enumerate(self.llms):
             try:
                 chain = (
-                    PromptTemplate(template=EXTRACT_JOB_FROM_PAGE_DATA_TEMPLATE)
-                    | llm
-                    | PydanticOutputParser(pydantic_object=Job)
+                    PromptTemplate(
+                        template=EXTRACT_JOB_FROM_PAGE_DATA_TEMPLATE,
+                        input_variables=['page_data', 'source'],
+                    ) | llm | PydanticOutputParser(pydantic_object=Job)
                 )
                 response: Job = chain.invoke(
                     {
@@ -100,9 +101,10 @@ def extract_skills_from_resume(
         for llm in self.llms:
             try:
                 return (
-                    PromptTemplate(template=EXTRACT_KEYWORDS_FROM_RESUME_TEMPLATE)
-                    | llm
-                    | JsonOutputParser()
+                    PromptTemplate(
+                        template=EXTRACT_KEYWORDS_FROM_RESUME_TEMPLATE,
+                        input_variables=['resume_data', 'roles']
+                    ) | llm | JsonOutputParser()
                 ).invoke(
                     {"resume_data": resume_data, "roles": "\n".join(preferred_roles)}
                 )

backend/src/main.py

@@ -8,6 +8,7 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
+from sqlalchemy import Boolean, Column
 from sqlalchemy.orm import Session
 
 from .constants import EXPIRE_JOBS_AFTER_DAYS, STATIC_DIR_PATH
@@ -43,7 +44,7 @@ def mark_jobs_inactive(db: Session) -> None:
         )
 
         for job in old_jobs:
-            job.is_active = False
+            job.is_active = Column(Boolean, default=False)
         db.commit()
         print(f"{len(old_jobs)} jobs marked as inactive.")
     except Exception as e:
@@ -60,7 +61,7 @@ def expire_jobs(db: Session) -> None:
         for job in old_jobs:
             db.delete(job)
         db.commit()
-        job_collection.delete(ids=[job.url for job in old_jobs])
+        job_collection.delete(ids=[str(job.url) for job in old_jobs])
         print(f"{len(old_jobs)} jobs deleted.")
     except Exception as e:
         print(f"Error marking old jobs inactive: {e}")

backend/src/routers/auth.py

@@ -32,6 +32,7 @@
 )
 import sqlalchemy
 import sqlalchemy.exc
+from sqlalchemy import Boolean, Column, Integer, String
 from sqlalchemy.orm import Session
 
 from ..utils import get_normalized_locations_list_string, hash_password, verify_password
@@ -233,10 +234,10 @@ class Token(BaseModel):
 
 def authenticate_user(username: str, password: str, db: Session) -> Optional[User]:
     """Authenticate user with username and password"""
-    user: User = db.query(User).filter(User.email == username).first()
+    user = db.query(User).filter(User.email == username).first()
     if not user:
         return None
-    if not verify_password(password, user.hashed_password):
+    if not verify_password(password, str(user.hashed_password)):
         return None
     return user
 
@@ -260,10 +261,11 @@ def create_access_token(
     response_description="User data for myself",
 )
 def get_myself(email: user_dependency, db: db_dependency):
-    user: User = db.query(User).filter(User.email == email["email"]).first()
+    user = db.query(User).filter(User.email == email["email"]).first()
     if not user:
         raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED, detail="User not authenticated"
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="User not authenticated"
         )
     return user
 
@@ -276,7 +278,7 @@ def get_myself(email: user_dependency, db: db_dependency):
     response_description="User data for the the user with given email.",
 )
 def get_user(email: user_dependency, db: db_dependency, user_email: str):
-    user: User = db.query(User).filter(User.email == user_email).first()
+    user = db.query(User).filter(User.email == user_email).first()
     if not user:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
@@ -315,6 +317,11 @@ async def create_user(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail="Only PDF files are supported for resumes",
             )
+        if not resume.filename:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Resume filename not specified",
+            )
         try:
             # Save the file to the uploads directory
             resume_file_path = os.path.join(
@@ -355,7 +362,7 @@ async def create_user(
             is_admin=False,
             resume_url=(
                 os.path.join("static", user_obj.email, resume.filename)
-                if resume and not isinstance(resume, str)
+                if resume and not isinstance(resume, str) and resume.filename is not None
                 else None
             ),
             resume_text=resume_text,
@@ -468,7 +475,11 @@ async def update_user(
             detail=json.loads(e.json()),
         )
 
-    user: User = db.query(User).filter(User.email == email["email"]).first()
+    user = db.query(User).filter(User.email == email["email"]).first()
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="User not authenticated"
+        )
     resume_text = None
     resume_file_path = None
 
@@ -479,6 +490,11 @@ async def update_user(
                 detail="Only PDF files are supported for resumes",
             )
         try:
+            if not resume.filename:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Resume filename not specified",
+                )
             resume_file_path = os.path.join(
                 STATIC_DIR_PATH, user.email, resume.filename
             )
@@ -504,32 +520,33 @@ async def update_user(
         )
 
     try:
-        user.full_name = user_obj.full_name or user.full_name
-        user.experience_years = user_obj.experience_years or user.experience_years
+        user.full_name = Column(String, default=user_obj.full_name) if user_obj.full_name is not None else user.full_name
+        user.experience_years = Column(Integer, default=user_obj.experience_years) if user_obj.experience_years is not None else user.experience_years
         user.preferred_roles = (
-            json.dumps(user_obj.preferred_roles)
+            Column(String, default=json.dumps(user_obj.preferred_roles))
             if user_obj.preferred_roles
             else user.preferred_roles
         )
         user.preferred_locations = (
-            json.dumps(user_obj.preferred_locations)
+            Column(String, default=json.dumps(user_obj.preferred_locations))
             if user_obj.preferred_locations
             else user.preferred_locations
         )
         user.preferred_sources = (
-            json.dumps(user_obj.preferred_sources)
+            Column(String, default=json.dumps(user_obj.preferred_sources))
             if user_obj.preferred_sources
             else user.preferred_sources
         )
         user.receive_email_alerts = (
-            user_obj.receive_email_alerts
+            Column(Boolean, default=user_obj.receive_email_alerts)
             if user_obj.receive_email_alerts is not None
             else user.receive_email_alerts
         )
 
-        if resume and not isinstance(resume, str):
-            user.resume_url = os.path.join("static", user.email, resume.filename)
-            user.resume_text = resume_text
+        if resume and not isinstance(resume, str) and resume.filename:
+            user.resume_url = Column(String, default=os.path.join("static", user.email, resume.filename))
+            if resume_text:
+                user.resume_text = Column(String, default=resume_text)
 
         db.commit()
         db.refresh(user)
@@ -555,5 +572,5 @@ def login_for_access_token(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Incorrect username or password",
         )
-    access_token = create_access_token(email=user.email)
+    access_token = create_access_token(email=str(user.email))
     return {"access_token": access_token, "token_type": "bearer", "user": user}

backend/src/routers/job.py

@@ -85,9 +85,13 @@ def recommended_jobs(
     ),
 ):
     """Get all recommended jobs for the user based on their resume keywords"""
-    resume_text_json = (
-        db.query(User.resume_text).filter(User.email == user["email"]).first()[0]
-    )
+    resume_text = db.query(User.resume_text).filter(User.email == user["email"]).first()
+    if resume_text is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=f"User not authenticated",
+        )
+    resume_text_json = resume_text[0]
     if not resume_text_json:
         return {"NULL": db.query(Job).filter(Job.is_active == True).all()}
 

backend/src/scrapers/scraper_base.py

@@ -88,12 +88,17 @@ def parse_job_details(self, url: str, location: str):
             headers = {"User-Agent": "Mozilla/5.0"}
             response = requests.get(url, headers=headers)
             soup = BeautifulSoup(response.text, "html.parser")
-            page_data = soup.find("div", class_="top-card-layout__card").get_text(
-                strip=True, separator=" "
-            )
-            page_data += soup.find(
+            soup_find = soup.find("div", class_="top-card-layout__card")
+            page_data = ""
+            if soup_find:
+                page_data += soup_find.get_text(
+                    strip=True, separator=" "
+                )
+            soup_find = soup.find(
                 "div", class_="description__text description__text--rich"
-            ).get_text(strip=True, separator=" ")
+            )
+            if soup_find:
+                page_data += soup_find.get_text(strip=True, separator=" ")
             company = self.parse_job_company(soup)
             job_details = {
                 **self.infer_job_details(page_data, company, location),