diff --git a/backend/Dockerfile b/backend/Dockerfile index e462a2e..3a584e1 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -1,6 +1,7 @@ FROM python:3.12-slim WORKDIR /app +ENV PYTHONPATH=/app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt @@ -9,4 +10,4 @@ COPY . . # Runs the Taskiq worker against taskiq_management.broker_setup. # docker-compose overrides this command if needed. -CMD ["taskiq", "worker", "taskiq_management.broker_setup:broker", "inbox.tasks", "taskiq_management.tasks"] +CMD ["taskiq", "worker", "taskiq_management.broker_setup:broker", "inbox.tasks", "inbox.sync_tasks", "taskiq_management.tasks"] diff --git a/backend/inbox/app.py b/backend/inbox/app.py index 6d4321e..9f279c5 100644 --- a/backend/inbox/app.py +++ b/backend/inbox/app.py @@ -35,6 +35,8 @@ async def fetch_email( await service.enqueue_matching(list(service.pending_match_ids),force=False) account_setup=[] + if test_on: + return JSONResponse(content={"data":items_lst,"status_code":200}) if service.pending_confirmation_emails: account_setup=await service.send_account_setup(list(service.pending_confirmation_emails)) diff --git a/backend/inbox/decoded_attachments/Mehdi Raza Content Writing Samples 001.docx b/backend/inbox/decoded_attachments/Mehdi Raza Content Writing Samples 001.docx new file mode 100644 index 0000000..96dd43c Binary files /dev/null and b/backend/inbox/decoded_attachments/Mehdi Raza Content Writing Samples 001.docx differ diff --git a/backend/inbox/models.py b/backend/inbox/models.py index 97f9699..d63be73 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -12,6 +12,7 @@ from sqlalchemy import Column, DateTime, func, or_, update from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload from sqlmodel import Field, Relationship, SQLModel, select, true from users.models import Users @@ -42,9 +43,33 @@ class Inbox(SQLModel, table=True): created_at: datetime = Field(default_factory=datetime.now) updated_at: datetime = Field(default_factory=datetime.now) - # is_active: bool = Field(default=True) - # is_deleted: bool = Field(default=False) - # user: Users | None = Relationship(back_populates="inbox") + + user: Optional[Users] = Relationship( + back_populates="inbox", + sa_relationship_kwargs={"lazy": "joined"}, + ) + + @classmethod + async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None,limit:int=10,offset:int=0): + try: + qry = ( + select(cls) + .options(selectinload(cls.messages)) + .join(Users, cls.user_id == Users.id) + .join(Roles, Users.role_id == Roles.id) + .where(Roles.role_name == EnumRoles.CANDIDATE.value) + .limit(limit) + .offset(offset) + ) + if user_id: + qry = qry.where(cls.user_id == user_id) + result = await session.execute(qry) + rows = result.scalars().all() + if user_id and len(rows) == 1: + return rows[0] + return rows + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) class Inbox_Alerts(SQLModel, table=True): @@ -101,14 +126,6 @@ class Inbox_Messages(SQLModel, table=True): return body return email_data.get("bodyPreview") or "" - # @classmethod - # async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None): - # try: - # qryy=select(cls,Users).join(cls,cls.) - # if user_id - - # except Exception as e: - # raise HTTPException(status_code=500,detail=str(e)) @classmethod async def get_all_applications(cls,session:AsyncSession,message_id:uuid.UUID|int|None=None): @@ -219,11 +236,12 @@ class Inbox_Messages(SQLModel, table=True): if not cls._is_linkable_sender(address): return None try: - user=(await session.execute( - select(Users).where(func.lower(Users.email)==address) - )).scalars().first() + # id-only: avoid Users.job_posts selectin / role lazy loads under asyncio + user_id=(await session.execute( + select(Users.id).where(func.lower(Users.email)==address) + )).scalar_one_or_none() - if not user: + if user_id is None: role=await Roles.get_role_by_name(session,EnumRoles.CANDIDATE.value) user=Users( name=cls._sender_display_name(email_data,address), @@ -232,15 +250,18 @@ class Inbox_Messages(SQLModel, table=True): password=hash_password(DEFAULT_CANDIDATE_PASSWORD), ) session.add(user) + # autoflush=False: flush so users.id exists before inbox FK insert + # (Relationship helps ordering, but flush keeps this path explicit). + await session.flush() session.add(Inbox(user_id=user.id,message_id=email.id)) await session.commit() return address link=(await session.execute( - select(Inbox).where(Inbox.message_id==email.id,Inbox.user_id==user.id) - )).scalars().first() - if not link: - session.add(Inbox(user_id=user.id,message_id=email.id)) + select(Inbox.id).where(Inbox.message_id==email.id,Inbox.user_id==user_id) + )).scalar_one_or_none() + if link is None: + session.add(Inbox(user_id=user_id,message_id=email.id)) await session.commit() return None except IntegrityError: @@ -275,17 +296,21 @@ class Inbox_Messages(SQLModel, table=True): session.add(existing) await session.commit() await session.refresh(existing) - + if fields.get("attachment"): link_user=await cls._link_sender(session, email_data, existing) + # _link_sender may rollback (IntegrityError); that expires this row + await session.refresh(existing) return existing, link_user email = cls(**fields) session.add(email) await session.commit() + await session.refresh(email) if fields.get("attachment"): link_user=await cls._link_sender(session, email_data, email) + await session.refresh(email) return email, link_user @classmethod diff --git a/backend/inbox/plugins.py b/backend/inbox/plugins.py index 5176f49..9548c6d 100644 --- a/backend/inbox/plugins.py +++ b/backend/inbox/plugins.py @@ -36,14 +36,17 @@ async def request_email_confirmation(email): return response.status_code -async def fetch_read_status_delta(folder, since=None, limit=1000, max_pages=10, token=None): - """GET /sync/read-status -> the raw round dict.""" +async def fetch_read_status_delta(folder, since=None, limit=100, max_pages=10, token=None): + """GET /sync/read-status -> the raw round dict. + + Upstream Email API caps `limit` at 100; keep the default at that ceiling. + """ if not EMAIL_URL: raise RuntimeError("EMAIL_URL must be set") auth_token=token or EMAIL_API_TOKEN if not auth_token: raise RuntimeError("EMAIL_API_TOKEN must be set") - params={"folder":folder,"limit":limit,"max_pages":max_pages} + params={"folder":folder,"limit":min(int(limit or 100),100),"max_pages":max_pages} if since: params["since"]=since async with httpx.AsyncClient(timeout=15.0) as client: diff --git a/backend/inbox/sync_tasks.py b/backend/inbox/sync_tasks.py index 42dc6ac..42cc8e9 100644 --- a/backend/inbox/sync_tasks.py +++ b/backend/inbox/sync_tasks.py @@ -49,9 +49,14 @@ async def sync_read_status() -> dict: round_data=await fetch_read_status_delta( EMAIL_SYNC_FOLDER, since=since if rounds==1 else None, - limit=1000, + limit=100, max_pages=10, ) + except httpx.ConnectError as e: + # Email API down / unreachable from this process — soft-fail so the + # cron does not burn retries every minute. + logger.warning("sync_read_status unreachable: %s",e) + return {"error":"unreachable","detail":str(e)} except httpx.HTTPStatusError as e: if e.response.status_code==401: logger.warning("sync_read_status 401 — device-code sign-in required") diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 8c6092d..ea6a12d 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -51,7 +51,7 @@ class Email: except Exception as e: raise HTTPException(status_code=500,detail=str(e)) - async def get_email_by_id(self,message_id): + async def get_email_by_id(self,message_id,test_on=True): async with httpx.AsyncClient() as client: try: response=await client.get(f"{self.get_url}/emails/{message_id}", @@ -63,6 +63,8 @@ class Email: row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) if row.attachment and row.file_path and row.match_status is None: self.pending_match_ids.append(str(row.id)) + if test_on: + return data if new_user_email: self.pending_confirmation_emails.append(new_user_email) return data diff --git a/backend/job/app.py b/backend/job/app.py index 682bb75..3d95117 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -2,7 +2,7 @@ from fastapi import APIRouter,Depends,Query from fastapi.responses import JSONResponse from fastapi import HTTPException from db_setup import get_session -from job.candidate.views import FileRead +from job.candidate.views import FileRead,CandidateView from sqlalchemy.ext.asyncio import AsyncSession from users.permissions import PermissionTag, require_permission from job.job_post.views import JobPost,JobPostCreate @@ -11,6 +11,7 @@ from job.job_post.plugins import PlatformAlias from fastapi import UploadFile, File from dotenv import load_dotenv from datetime import datetime, time, timezone + load_dotenv() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -102,21 +103,20 @@ async def buffer_channels( except Exception as e: raise HTTPException(status_code=500,detail=str(e)) -# @router.get("/candidate/fetch") -# async def fetch_candidate( -# user_id:str=Query(None), -# current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), -# session: AsyncSession = Depends(get_session), -# ): -# try: -# service=CandidateView(session=session) -# if user_id: -# data=await service.get_candidate(user_id=user_id) -# else: -# data=await service.get_candidate() -# return JSONResponse(content={"data":data,"status_code":200}) -# except HTTPException: -# raise -# except Exception as e: -# raise HTTPException(status_code=500,detail=str(e)) - \ No newline at end of file +@router.get("/candidate/fetch") +async def fetch_candidate( + user_id:str=Query(None), + limit:int=Query(10), + offset:int=Query(0), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=CandidateView(session=session) + data=await service.get_candidate(user_id=user_id,limit=limit,offset=offset) + total=len(data) if isinstance(data,list) else 1 + return JSONResponse(content={"data":data,"total":total,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) diff --git a/backend/job/candidate/serializers.py b/backend/job/candidate/serializers.py index e69de29..9e6f6b7 100644 --- a/backend/job/candidate/serializers.py +++ b/backend/job/candidate/serializers.py @@ -0,0 +1,31 @@ +from inbox.models import Inbox +from typing import Any,List,Dict + + +def serialize_candidate_profile(link:Inbox|List[Inbox]|Dict[str,Any]|List[Dict[str,Any]]) -> Dict[str,Any]|List[Dict[str,Any]]: + if isinstance(link,list): + return [serialize_candidate_profile(item) for item in link] + if isinstance(link,dict): + return link + + user = link.user + message = link.messages + return { + "inbox_id": link.id, + "user_id": str(link.user_id) if link.user_id else None, + "name": user.name if user else None, + "email": user.email if user else None, + "is_active": user.is_active if user else None, + "message_id": str(link.message_id) if link.message_id else None, + "created_at": link.created_at.isoformat() if link.created_at else None, + "application_status": message.application_status if message else None, + "experience": message.experience if message else None, + "resume_text": message.resume_text if message else None, + "suggested_job_post_ids": list(message.suggested_job_post_ids or []) if message else [], + "match_summary": message.match_summary if message else None, + "match_reasoning": message.match_reasoning if message else None, + "match_status": message.match_status if message else None, + "match_error": message.match_error if message else None, + "matched_at": message.matched_at.isoformat() if message and message.matched_at else None, + "job_posts": [], + } diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 6b10aa3..6f430c7 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -5,7 +5,10 @@ from fastapi import HTTPException from pypdf import PdfReader from sqlalchemy import select from sqlmodel import true -from inbox.models import Inbox_Messages +from job.job_post.models import JobPosts +from job.job_post.serializers import serialize_job_post +from job.candidate.serializers import serialize_candidate_profile +from inbox.models import Inbox_Messages,Inbox from job.candidate.plugins import normalize_spaced_text class FileRead: @@ -68,12 +71,41 @@ class FileRead: # get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id) # get_file= -# class CandidateView: -# def __init__(self,session:AsyncSession): -# self.session=session - -# async def get_candidate(self,user_id=None): -# try: -# call_func=Inbox_Messages.get_candidate_profile(user_id=user_id) -# except Exception as e: -# raise HTTPException(status_code=500,detail=str(e)) \ No newline at end of file +class CandidateView: + def __init__(self,session:AsyncSession): + self.session=session + + async def get_candidate(self,user_id=None,limit=10,offset=0): + try: + rows=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=limit,offset=offset) + return await self.attach_job_posts(rows) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + async def get_job_post_by_id(self,record_id,data=None): + """Load full job_posts row and optionally append it onto a candidate payload.""" + try: + job_post_data=await JobPosts.get_job_post_by_id(session=self.session,record_id=record_id) + if not job_post_data: + return None + payload=serialize_job_post(job_post_data) + if isinstance(data,dict): + data.setdefault("job_posts",[]).append(payload) + return payload + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + async def attach_job_posts(self,data): + """Normalize list/single, serialize each record, attach full job_posts rows.""" + single=not isinstance(data,list) + records=[data] if single else list(data or []) + enriched=[] + for record in records: + payload=serialize_candidate_profile(record) + payload["job_posts"]=[] + for job_id in payload.get("suggested_job_post_ids") or []: + await self.get_job_post_by_id(record_id=job_id,data=payload) + enriched.append(payload) + return enriched[0] if single else enriched diff --git a/backend/job/job_post/serializers.py b/backend/job/job_post/serializers.py index 4589f3d..d58e3ad 100644 --- a/backend/job/job_post/serializers.py +++ b/backend/job/job_post/serializers.py @@ -13,6 +13,8 @@ def serialize_job_post(row) -> dict: "post_text": row.post_text, "channel_id": row.channel_id, "platform": row.platform, + "is_active": row.is_active, + "is_deleted": row.is_deleted, "buffer_post_id": row.buffer_post_id, "buffer_external_link": row.buffer_external_link, "buffer_sent_at": row.buffer_sent_at.isoformat() if row.buffer_sent_at else None, diff --git a/backend/users/models.py b/backend/users/models.py index cacf22d..bbfa0a1 100644 --- a/backend/users/models.py +++ b/backend/users/models.py @@ -1,5 +1,6 @@ import uuid from datetime import datetime +from typing import TYPE_CHECKING from sqlalchemy import func, or_ from sqlalchemy.ext.asyncio import AsyncSession @@ -9,6 +10,9 @@ from sqlmodel import Field, Relationship, SQLModel, select from role.models import Roles from job.job_post.models import JobPosts +if TYPE_CHECKING: # runtime import would cycle: inbox.models imports this module + from inbox.models import Inbox + class Users(SQLModel, table=True): __tablename__ = "users" @@ -16,7 +20,9 @@ class Users(SQLModel, table=True): name: str email: str = Field(unique=True) role_id: int | None = Field(nullable=True, foreign_key="roles.id") - role: Roles | None = Relationship(back_populates="users") + role: Roles | None = Relationship(back_populates="users", + sa_relationship_kwargs={"lazy": "selectin"} + ) # selectin, not joined: this is a one-to-many, so a joined load would repeat the # user row once per post. Without an explicit strategy the default is a lazy load, # which raises MissingGreenlet the moment anything touches it under asyncio. @@ -24,6 +30,10 @@ class Users(SQLModel, table=True): back_populates="user", sa_relationship_kwargs={"lazy": "selectin"}, ) + inbox: list["Inbox"] = Relationship( + back_populates="user", + sa_relationship_kwargs={"lazy": "selectin"}, + ) password: str created_at: datetime = Field(default_factory=datetime.now) updated_at: datetime = Field(default_factory=datetime.now) diff --git a/docker-compose.yml b/docker-compose.yml index be849d0..7f99556 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -18,6 +18,7 @@ services: build: context: ./backend container_name: hrms-taskiq-worker + working_dir: /app command: [ "taskiq", @@ -32,10 +33,14 @@ services: env_file: - ./backend/.env environment: + PYTHONPATH: /app REDIS_URL: redis://redis:6379/0 TASKIQ_QUEUE_NAME: inbox TASKIQ_WORKER_NAME: worker-01 + # .env uses localhost for the host-side API; containers must reach the host. DB_HOST: host.docker.internal + EMAIL_URL: http://host.docker.internal:5000 + BACKEND_URL: http://host.docker.internal:8000 extra_hosts: - "host.docker.internal:host-gateway" volumes: @@ -49,12 +54,19 @@ services: build: context: ./backend container_name: hrms-taskiq-scheduler + working_dir: /app command: ["taskiq", "scheduler", "taskiq_management.broker_setup:scheduler", "inbox.sync_tasks"] env_file: - ./backend/.env environment: + PYTHONPATH: /app REDIS_URL: redis://redis:6379/0 TASKIQ_QUEUE_NAME: inbox + DB_HOST: host.docker.internal + EMAIL_URL: http://host.docker.internal:5000 + BACKEND_URL: http://host.docker.internal:8000 + extra_hosts: + - "host.docker.internal:host-gateway" depends_on: redis: condition: service_healthy