70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from inbox.models import Inbox
|
|
from job.candidate.models import Manual_UPLOAD_CANDIDATE
|
|
from job.job_post.models import JobPosts
|
|
from role.models import EnumRoles, Roles
|
|
from search.serializers import (
|
|
serialize_search_candidate,
|
|
serialize_search_job,
|
|
serialize_search_manager,
|
|
)
|
|
from users.models import Users
|
|
from users.permissions import PermissionTag, has_permission
|
|
|
|
JOBS_CAP = 4
|
|
CANDIDATES_CAP = 4
|
|
MANAGERS_CAP = 3
|
|
|
|
|
|
class Search:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def fetch(self,q,limit,current_user):
|
|
query=(q or "").strip()
|
|
granted=current_user.get("permissions") or []
|
|
jobs=[]
|
|
candidates=[]
|
|
managers=[]
|
|
if query:
|
|
if has_permission(granted,PermissionTag.JOBS_VIEW):
|
|
jobs=await self._jobs(query,min(limit,JOBS_CAP))
|
|
if has_permission(granted,PermissionTag.CANDIDATES_VIEW):
|
|
candidates=await self._candidates(query,min(limit,CANDIDATES_CAP))
|
|
managers=await self._managers(query,min(limit,MANAGERS_CAP))
|
|
data={"jobs":jobs,"candidates":candidates,"managers":managers}
|
|
total=len(jobs)+len(candidates)+len(managers)
|
|
return data,total
|
|
|
|
async def _jobs(self,query,cap):
|
|
rows,_total=await JobPosts.fetch_job_posts(
|
|
self.session,search=query,top=cap,skip=0,active_only=False,include_deleted=False,
|
|
)
|
|
return [serialize_search_job(r) for r in rows]
|
|
|
|
async def _candidates(self,query,cap):
|
|
role=await Roles.get_role_by_name(self.session,EnumRoles.CANDIDATE.value)
|
|
if role is None:
|
|
return []
|
|
users=list(await Users.get_users(self.session,top=cap,search=query,role_id=role.id))
|
|
uids=[u.id for u in users]
|
|
inbox_hits=await Inbox.newest_cv_by_user_ids(self.session,uids)
|
|
manual_paths=await Manual_UPLOAD_CANDIDATE.file_paths_by_user_ids(self.session,uids)
|
|
rows=[]
|
|
for u in users:
|
|
key=str(u.id)
|
|
hit=inbox_hits.get(key) or {}
|
|
rows.append(serialize_search_candidate(
|
|
u.id,u.name,u.email,hit.get("inbox_id"),
|
|
hit.get("file_path") or manual_paths.get(key),
|
|
))
|
|
return rows
|
|
|
|
async def _managers(self,query,cap):
|
|
role=await Roles.get_role_by_name(self.session,EnumRoles.HIRING_MANAGER.value)
|
|
if role is None:
|
|
return []
|
|
users=await Users.get_users(self.session,top=cap,search=query,role_id=role.id)
|
|
return [serialize_search_manager(u) for u in users]
|