utopia-ai
/
HR-ATS-Portal
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

talet pool api done

pull/7/head
ahmed.mujtaba 2026-08-10 21:01:51 +05:00
parent e21bb6f338
commit da3077c302
13 changed files with 181 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend/Dockerfile
View File

@ -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"]

2
backend/inbox/app.py
View File

@ -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))

BIN
backend/inbox/decoded_attachments/Mehdi Raza Content Writing Samples 001.docx Normal file
View File

Binary file not shown.

65
backend/inbox/models.py
View File

@ -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

9
backend/inbox/plugins.py
View File

@ -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:

7
backend/inbox/sync_tasks.py
View File

@ -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")

4
backend/inbox/views.py
View File

@ -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

38
backend/job/app.py
View File

@ -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))
@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))

31
backend/job/candidate/serializers.py
View File

@ -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": [],
}

52
backend/job/candidate/views.py
View File

@ -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))
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

2
backend/job/job_post/serializers.py
View File

@ -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,

12
backend/users/models.py
View File

@ -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)

12
docker-compose.yml
View File

@ -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