873 lines
35 KiB
Python
873 lines
35 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import TYPE_CHECKING, Optional
|
|
|
|
from sqlalchemy import DateTime, JSON, Index, String, and_, case, cast, func, or_, union_all
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import aliased
|
|
from sqlmodel import Field, Relationship, SQLModel, select
|
|
|
|
from job.job_post.enums import RequisitionStatus
|
|
|
|
if TYPE_CHECKING: # runtime import would be circular: users.models imports this module
|
|
from candidate_forms.models import Requisition
|
|
from users.models import Users
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class JobPosts(SQLModel, table=True):
|
|
__tablename__ = "job_posts"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
title: str = Field(index=True)
|
|
|
|
|
|
user: Optional["Users"] = Relationship(
|
|
back_populates="job_posts",
|
|
sa_relationship_kwargs={"lazy": "joined", "foreign_keys": "[JobPosts.created_by]"},
|
|
)
|
|
|
|
platform: str = Field(default="linkedin")
|
|
is_active: bool = Field(default=True)
|
|
is_deleted: bool = Field(default=False)
|
|
employment_type: str | None = Field(default=None)
|
|
location: str | None = Field(default=None)
|
|
experience_min: int | None = Field(default=None)
|
|
experience_max: int | None = Field(default=None)
|
|
requirements: list[str] = Field(default_factory=list, sa_type=JSON)
|
|
optional_skills: list[str] = Field(default_factory=list, sa_type=JSON)
|
|
salary: str = Field(default="Anonymous")
|
|
description: str | None = Field(default=None)
|
|
post_text: str
|
|
channel_id: str
|
|
buffer_post_id: str | None = Field(default=None)
|
|
buffer_external_link: str | None = Field(default=None)
|
|
buffer_sent_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
status: str = Field(default="draft")
|
|
buffer_error: str | None = Field(default=None)
|
|
|
|
requisition_status: str = Field(default="open", sa_column_kwargs={"server_default": "open"})
|
|
department: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
vacancies: int = Field(default=1, sa_column_kwargs={"server_default": "1"})
|
|
closed_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
# Who is working the req now (swappable). History lives in job_assignments
|
|
# with assignment_role=primary_recruiter; this column is the current pointer.
|
|
current_recruiter_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
# Who owns the requisition (stable). Optional. History lives in
|
|
# job_assignments with assignment_role=hiring_manager.
|
|
hiring_manager_id: uuid.UUID | None = Field(default=None, foreign_key="users.id", index=True)
|
|
# Annexure A employee requisition this job was opened from (optional 1:1).
|
|
# Distinct from requisition_status, which is the hiring lifecycle on this row.
|
|
# unique=True so two job posts cannot share one requisition; NULLs stay allowed.
|
|
requisition_id: uuid.UUID | None = Field(
|
|
default=None, foreign_key="requisitions.id", ondelete="SET NULL", unique=True, index=True,
|
|
)
|
|
requisition: Optional["Requisition"] = Relationship(
|
|
back_populates="job_post",
|
|
sa_relationship_kwargs={"lazy": "selectin"},
|
|
)
|
|
created_by: uuid.UUID = Field(foreign_key="users.id")
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id: str) -> uuid.UUID | None:
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def get_job_post_by_id(cls, session: AsyncSession, record_id: str):
|
|
uid = cls._as_uuid(record_id)
|
|
if uid is None:
|
|
return None
|
|
result = await session.execute(select(cls).where(cls.id == uid))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_by_requisition_id(cls, session: AsyncSession, requisition_id):
|
|
"""Live job post already opened from this Annexure A requisition, if any."""
|
|
uid = cls._as_uuid(requisition_id)
|
|
if uid is None:
|
|
return None
|
|
result = await session.execute(
|
|
select(cls).where(
|
|
cls.requisition_id == uid,
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_active_job_posts(cls, session: AsyncSession):
|
|
result = await session.execute(
|
|
select(cls).where(cls.is_active == True, cls.is_deleted == False) # noqa: E712
|
|
)
|
|
return result.scalars().all()
|
|
|
|
@classmethod
|
|
async def get_by_ids(cls, session: AsyncSession, ids: list[str], *, active_only: bool = True):
|
|
uids = []
|
|
for raw in ids or []:
|
|
uid = cls._as_uuid(raw)
|
|
if uid is not None:
|
|
uids.append(uid)
|
|
if not uids:
|
|
return []
|
|
statement = select(cls).where(cls.id.in_(uids))
|
|
if active_only:
|
|
statement = statement.where(cls.is_active == True, cls.is_deleted == False) # noqa: E712
|
|
result = await session.execute(statement)
|
|
rows = list(result.scalars().all())
|
|
by_id = {str(r.id): r for r in rows}
|
|
# Preserve request order so suggestion ranks stay stable.
|
|
return [by_id[str(u)] for u in uids if str(u) in by_id]
|
|
|
|
@classmethod
|
|
async def get_by_titles(cls, session: AsyncSession, titles: list[str], *, active_only: bool = False):
|
|
"""Match job posts whose title equals any of `titles` (trim + case-insensitive).
|
|
|
|
Used by sheet form-data: position_applied_for ↔ job_posts.title. Returns
|
|
non-deleted rows; inactive ones stay in the list so the UI can mark them
|
|
unavailable the same way inbox suggestions do.
|
|
"""
|
|
lowers = sorted({(t or "").strip().lower() for t in (titles or []) if (t or "").strip()})
|
|
if not lowers:
|
|
return []
|
|
statement = select(cls).where(
|
|
cls.is_deleted == False, # noqa: E712
|
|
func.lower(func.trim(cls.title)).in_(lowers),
|
|
)
|
|
if active_only:
|
|
statement = statement.where(cls.is_active == True) # noqa: E712
|
|
statement = statement.order_by(cls.created_at.desc())
|
|
result = await session.execute(statement)
|
|
return list(result.scalars().all())
|
|
|
|
@staticmethod
|
|
def title_ilike_pattern(applied_for: str) -> str | None:
|
|
"""ILIKE pattern so job_posts.title contains the form's Position Applied For."""
|
|
needle = (applied_for or "").strip()
|
|
if not needle:
|
|
return None
|
|
escaped = needle.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
|
return f"%{escaped}%"
|
|
|
|
@classmethod
|
|
async def ids_for_title_ilike(cls, session: AsyncSession, applied_for: str) -> list[uuid.UUID]:
|
|
"""All non-deleted job_posts.id whose title ILIKE-contains applied_for.
|
|
|
|
One form title can match many posts. Order is created_at DESC, id DESC
|
|
so the UI/ATS list is stable. Empty if blank or no row. Suggested,
|
|
not recruiter-assigned.
|
|
"""
|
|
pattern = cls.title_ilike_pattern(applied_for)
|
|
if not pattern:
|
|
return []
|
|
statement = (
|
|
select(cls.id)
|
|
.where(cls.is_deleted == False) # noqa: E712
|
|
.where(cls.title.ilike(pattern, escape="\\"))
|
|
.order_by(cls.created_at.desc(), cls.id.desc())
|
|
)
|
|
result = await session.execute(statement)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def ids_for_titles_ilike(
|
|
cls, session: AsyncSession, titles: list[str],
|
|
) -> dict[str, list[uuid.UUID]]:
|
|
"""Map stripped Position Applied For → every matching job_posts.id."""
|
|
out: dict[str, list[uuid.UUID]] = {}
|
|
seen: set[str] = set()
|
|
for raw in titles or []:
|
|
key = (raw or "").strip()
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
found = await cls.ids_for_title_ilike(session, key)
|
|
if found:
|
|
out[key] = found
|
|
return out
|
|
|
|
@classmethod
|
|
async def fetch_job_posts(
|
|
cls,
|
|
session: AsyncSession,
|
|
*,
|
|
search: str | None = None,
|
|
top: int | None = None,
|
|
skip: int = 0,
|
|
ids: list[str] | None = None,
|
|
active_only: bool = True,
|
|
include_deleted: bool = False,
|
|
department: str | None = None,
|
|
requisition_status: str | None = None,
|
|
employment_type: str | None = None,
|
|
hiring_manager_id: uuid.UUID | None = None,
|
|
restrict_ids: list | None = None,
|
|
):
|
|
if ids:
|
|
rows = await cls.get_by_ids(session, ids, active_only=active_only)
|
|
return rows, len(rows)
|
|
|
|
statement = select(cls)
|
|
if active_only:
|
|
statement = statement.where(cls.is_active == True, cls.is_deleted == False) # noqa: E712
|
|
elif not include_deleted:
|
|
statement = statement.where(cls.is_deleted == False) # noqa: E712
|
|
if restrict_ids is not None:
|
|
uids = []
|
|
for raw in restrict_ids:
|
|
uid = raw if isinstance(raw, uuid.UUID) else cls._as_uuid(raw)
|
|
if uid is not None:
|
|
uids.append(uid)
|
|
if not uids:
|
|
return [], 0
|
|
statement = statement.where(cls.id.in_(uids))
|
|
if search:
|
|
like = f"%{search.strip()}%"
|
|
statement = statement.where(
|
|
or_(cls.title.ilike(like), cls.location.ilike(like), cls.department.ilike(like))
|
|
)
|
|
if department:
|
|
statement = statement.where(cls.department == department)
|
|
if requisition_status:
|
|
statement = statement.where(cls.requisition_status == requisition_status)
|
|
if employment_type:
|
|
statement = statement.where(cls.employment_type == employment_type)
|
|
if hiring_manager_id is not None:
|
|
statement = statement.where(cls.hiring_manager_id == hiring_manager_id)
|
|
count_statement = select(func.count()).select_from(statement.subquery())
|
|
total = (await session.execute(count_statement)).scalar_one()
|
|
statement = statement.order_by(cls.created_at.desc())
|
|
if skip:
|
|
statement = statement.offset(skip)
|
|
if top is not None:
|
|
statement = statement.limit(top)
|
|
result = await session.execute(statement)
|
|
return list(result.scalars().all()), total
|
|
|
|
@classmethod
|
|
async def fetch_job_stats(
|
|
cls,
|
|
session: AsyncSession,
|
|
*,
|
|
job_post_id=None,
|
|
search: str | None = None,
|
|
ids: list[str] | None = None,
|
|
top: int | None = None,
|
|
skip: int = 0,
|
|
active_only: bool = False,
|
|
):
|
|
"""Per-job pipeline stage counts for every applicant assigned to the job.
|
|
|
|
Inbox, manual-upload / Add Candidate / CV-bank, and unpromoted sheet
|
|
rows. Duplicate emails (case-insensitive) count once per job — the
|
|
furthest pipeline stage is kept. Flagged is_duplicate rows are skipped.
|
|
Rows with no email still count, each as themselves. Jobs with zero
|
|
applicants still appear (LEFT JOIN).
|
|
"""
|
|
from g_sheet.models import FormData
|
|
from inbox.models import Inbox_Messages
|
|
from job.candidate.models import Manual_UPLOAD_CANDIDATE
|
|
from users.models import Users
|
|
|
|
job_uids = []
|
|
if job_post_id is not None:
|
|
uid = job_post_id if isinstance(job_post_id, uuid.UUID) else cls._as_uuid(job_post_id)
|
|
if uid is None:
|
|
return [], 0
|
|
job_uids = [uid]
|
|
elif ids:
|
|
for raw in ids:
|
|
uid = cls._as_uuid(raw)
|
|
if uid is not None:
|
|
job_uids.append(uid)
|
|
if not job_uids:
|
|
return [], 0
|
|
|
|
def dup_key(email_col, row_id):
|
|
# Same person = lower(trim(email)). No address -> unique per row
|
|
# so blank emails do not collapse into one applicant.
|
|
return func.coalesce(
|
|
func.nullif(func.lower(func.btrim(email_col)), ""),
|
|
func.concat("noid:", cast(row_id, String)),
|
|
)
|
|
|
|
inbox_q = (
|
|
select(
|
|
Inbox_Messages.assigned_job_post_id.label("job_post_id"),
|
|
dup_key(Inbox_Messages.message_from, Inbox_Messages.id).label("dup_key"),
|
|
cast(Inbox_Messages.application_status, String).label("stage"),
|
|
)
|
|
.where(Inbox_Messages.assigned_job_post_id.is_not(None))
|
|
.where(Inbox_Messages.is_duplicate == False) # noqa: E712
|
|
)
|
|
manual_stage = func.coalesce(
|
|
func.nullif(func.btrim(Manual_UPLOAD_CANDIDATE.status), ""),
|
|
"PENDING",
|
|
)
|
|
manual_email = func.coalesce(
|
|
func.nullif(func.btrim(Manual_UPLOAD_CANDIDATE.candidate_email), ""),
|
|
Users.email,
|
|
)
|
|
manual_q = (
|
|
select(
|
|
Manual_UPLOAD_CANDIDATE.job_post_id.label("job_post_id"),
|
|
dup_key(manual_email, Manual_UPLOAD_CANDIDATE.id).label("dup_key"),
|
|
manual_stage.label("stage"),
|
|
)
|
|
.select_from(Manual_UPLOAD_CANDIDATE)
|
|
.outerjoin(Users, Users.id == Manual_UPLOAD_CANDIDATE.user_id)
|
|
.where(Manual_UPLOAD_CANDIDATE.job_post_id.is_not(None))
|
|
)
|
|
# Unpromoted sheet applicants only — promoted rows already live on
|
|
# manual_upload_candidate (manual_upload_candidate_id set).
|
|
form_stage = case(
|
|
(FormData.processing_state == "rejected", "REJECTED"),
|
|
else_="PENDING",
|
|
)
|
|
form_q = (
|
|
select(
|
|
FormData.job_post_id.label("job_post_id"),
|
|
dup_key(FormData.candidate_email, FormData.id).label("dup_key"),
|
|
form_stage.label("stage"),
|
|
)
|
|
.where(FormData.job_post_id.is_not(None))
|
|
.where(FormData.manual_upload_candidate_id.is_(None))
|
|
.where(FormData.is_duplicate == False) # noqa: E712
|
|
)
|
|
if job_uids:
|
|
inbox_q = inbox_q.where(Inbox_Messages.assigned_job_post_id.in_(job_uids))
|
|
manual_q = manual_q.where(Manual_UPLOAD_CANDIDATE.job_post_id.in_(job_uids))
|
|
form_q = form_q.where(FormData.job_post_id.in_(job_uids))
|
|
|
|
apps = union_all(inbox_q, manual_q, form_q).subquery("applications")
|
|
stage_rank = case(
|
|
(apps.c.stage == "HIRED", 9),
|
|
(apps.c.stage == "APPROVED", 8),
|
|
(apps.c.stage == "OFFER", 7),
|
|
(apps.c.stage == "INTERVIEW", 6),
|
|
(apps.c.stage == "ASSESSMENT", 5),
|
|
(apps.c.stage.in_(["SCREENING", "PROCESS"]), 4),
|
|
(apps.c.stage == "PENDING", 3),
|
|
(apps.c.stage == "ONHOLD", 2),
|
|
(apps.c.stage.in_(["REJECTED", "CLOSED"]), 1),
|
|
else_=0,
|
|
)
|
|
unique_apps = (
|
|
select(apps.c.job_post_id, apps.c.dup_key, apps.c.stage)
|
|
.distinct(apps.c.job_post_id, apps.c.dup_key)
|
|
.order_by(apps.c.job_post_id, apps.c.dup_key, stage_rank.desc())
|
|
.subquery("unique_applicants")
|
|
)
|
|
stage = unique_apps.c.stage
|
|
|
|
def stage_count(*values):
|
|
return func.coalesce(func.sum(case((stage.in_(list(values)), 1), else_=0)), 0)
|
|
|
|
stats = (
|
|
select(
|
|
unique_apps.c.job_post_id,
|
|
func.count().label("total_applicants"),
|
|
stage_count("PENDING").label("shortlisting"),
|
|
stage_count("SCREENING", "PROCESS").label("screened"),
|
|
stage_count("ASSESSMENT").label("assessment"),
|
|
stage_count("INTERVIEW").label("interviewed"),
|
|
stage_count("OFFER").label("offered"),
|
|
stage_count("ONHOLD").label("on_hold"),
|
|
stage_count("REJECTED", "CLOSED").label("rejected"),
|
|
stage_count("APPROVED").label("approved"),
|
|
stage_count("HIRED").label("hired"),
|
|
)
|
|
.select_from(unique_apps)
|
|
.group_by(unique_apps.c.job_post_id)
|
|
.subquery("job_stage_stats")
|
|
)
|
|
|
|
# Alias so this join does not collide with the Users join inside
|
|
# the manual-upload subquery above.
|
|
Recruiter=aliased(Users)
|
|
statement = (
|
|
select(
|
|
cls.id.label("job_post_id"),
|
|
cls.title,
|
|
cls.department,
|
|
cls.location,
|
|
cls.requisition_status,
|
|
cls.current_recruiter_id,
|
|
cls.created_at,
|
|
Recruiter.name.label("recruiter_name"),
|
|
func.coalesce(stats.c.total_applicants, 0).label("total_applicants"),
|
|
func.coalesce(stats.c.shortlisting, 0).label("shortlisting"),
|
|
func.coalesce(stats.c.screened, 0).label("screened"),
|
|
func.coalesce(stats.c.assessment, 0).label("assessment"),
|
|
func.coalesce(stats.c.interviewed, 0).label("interviewed"),
|
|
func.coalesce(stats.c.offered, 0).label("offered"),
|
|
func.coalesce(stats.c.on_hold, 0).label("on_hold"),
|
|
func.coalesce(stats.c.rejected, 0).label("rejected"),
|
|
func.coalesce(stats.c.approved, 0).label("approved"),
|
|
func.coalesce(stats.c.hired, 0).label("hired"),
|
|
)
|
|
.select_from(cls)
|
|
.outerjoin(stats, stats.c.job_post_id == cls.id)
|
|
.outerjoin(Recruiter, Recruiter.id == cls.current_recruiter_id)
|
|
.where(cls.is_deleted == False) # noqa: E712
|
|
)
|
|
if active_only:
|
|
statement = statement.where(cls.is_active == True) # noqa: E712
|
|
if job_uids:
|
|
statement = statement.where(cls.id.in_(job_uids))
|
|
if search:
|
|
like = f"%{search.strip()}%"
|
|
statement = statement.where(
|
|
or_(cls.title.ilike(like), cls.location.ilike(like), cls.department.ilike(like))
|
|
)
|
|
|
|
count_statement = select(func.count()).select_from(statement.subquery())
|
|
total = (await session.execute(count_statement)).scalar_one()
|
|
statement = statement.order_by(cls.created_at.desc())
|
|
if job_post_id is None:
|
|
if skip:
|
|
statement = statement.offset(skip)
|
|
if top is not None:
|
|
statement = statement.limit(top)
|
|
result = await session.execute(statement)
|
|
return list(result.mappings().all()), int(total or 0)
|
|
|
|
@classmethod
|
|
async def list_departments(cls, session: AsyncSession, *, active_only: bool = False):
|
|
"""Distinct non-empty departments on non-deleted job posts.
|
|
|
|
The column default is "" — those rows are omitted so a dropdown never
|
|
offers a blank option. Closed requisitions still contribute unless
|
|
`active_only` is set: a past hiring department is a legitimate filter.
|
|
"""
|
|
statement = select(cls.department).where(
|
|
cls.is_deleted == False, # noqa: E712
|
|
cls.department != "",
|
|
)
|
|
if active_only:
|
|
statement = statement.where(cls.is_active == True) # noqa: E712
|
|
statement = statement.distinct().order_by(cls.department)
|
|
result = await session.execute(statement)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def ids_for_manager(cls, session: AsyncSession, user_id):
|
|
"""Job posts this user owns: assigned hiring_manager, or opened from
|
|
a requisition they created. The manager Candidates list and form
|
|
scope both follow this chain."""
|
|
uid = cls._as_uuid(user_id)
|
|
if uid is None:
|
|
return []
|
|
from candidate_forms.models import Requisition
|
|
|
|
assigned = await session.execute(
|
|
select(cls.id).where(
|
|
cls.hiring_manager_id == uid,
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
)
|
|
via_req = await session.execute(
|
|
select(cls.id)
|
|
.join(Requisition, cls.requisition_id == Requisition.id)
|
|
.where(
|
|
Requisition.created_by == uid,
|
|
Requisition.is_deleted == False, # noqa: E712
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
)
|
|
seen: set[uuid.UUID] = set()
|
|
out: list[uuid.UUID] = []
|
|
for row_id in list(assigned.scalars().all()) + list(via_req.scalars().all()):
|
|
if row_id not in seen:
|
|
seen.add(row_id)
|
|
out.append(row_id)
|
|
return out
|
|
|
|
@classmethod
|
|
async def ids_for_creator(cls, session: AsyncSession, user_id, created_by=False):
|
|
"""Jobs this recruiter should see on Candidates (when they lack
|
|
candidates.manage). created_by=True → created_by = session user only.
|
|
Otherwise: current_recruiter_id when set, else created_by."""
|
|
uid = cls._as_uuid(user_id)
|
|
if uid is None:
|
|
return []
|
|
if created_by:
|
|
result = await session.execute(
|
|
select(cls.id).where(
|
|
cls.created_by == uid,
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
)
|
|
return list(result.scalars().all())
|
|
result = await session.execute(
|
|
select(cls.id).where(
|
|
or_(
|
|
and_(cls.current_recruiter_id.is_not(None), cls.current_recruiter_id == uid),
|
|
and_(cls.current_recruiter_id.is_(None), cls.created_by == uid),
|
|
),
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def count_open_reqs_by_hiring_managers(cls, session: AsyncSession, user_ids):
|
|
"""Open requisitions per hiring manager, keyed by users.id."""
|
|
uids = [u for u in (user_ids or []) if u]
|
|
if not uids:
|
|
return {}
|
|
statement = (
|
|
select(cls.hiring_manager_id, func.count())
|
|
.where(
|
|
cls.hiring_manager_id.in_(uids),
|
|
cls.requisition_status == "open",
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
.group_by(cls.hiring_manager_id)
|
|
)
|
|
result = await session.execute(statement)
|
|
return {uid: int(n or 0) for uid, n in result.all()}
|
|
|
|
@classmethod
|
|
async def count_by_current_recruiter(
|
|
cls, session: AsyncSession, recruiter_id, *, status, department=None,
|
|
from_date=None, to_date=None,
|
|
):
|
|
"""Requisitions owned by current_recruiter_id in one requisition_status."""
|
|
uid = cls._as_uuid(recruiter_id)
|
|
if uid is None:
|
|
return 0
|
|
statement = select(func.count()).select_from(cls).where(
|
|
cls.current_recruiter_id == uid,
|
|
cls.requisition_status == status,
|
|
cls.is_deleted == False, # noqa: E712
|
|
)
|
|
if department:
|
|
statement = statement.where(cls.department == department)
|
|
if from_date is not None:
|
|
statement = statement.where(cls.closed_at >= from_date)
|
|
if to_date is not None:
|
|
statement = statement.where(cls.closed_at < to_date)
|
|
result = await session.execute(statement)
|
|
return int(result.scalar_one() or 0)
|
|
|
|
@classmethod
|
|
def _scoped(cls, statement, department=None, recruiter_id=None):
|
|
if department:
|
|
statement = statement.where(cls.department == department)
|
|
uid = cls._as_uuid(recruiter_id) if recruiter_id is not None else None
|
|
if uid is not None:
|
|
statement = statement.where(cls.current_recruiter_id == uid)
|
|
return statement
|
|
|
|
@classmethod
|
|
async def count_requisitions(
|
|
cls, session: AsyncSession, status=None, department=None, recruiter_id=None,
|
|
from_date=None, to_date=None, *, closed_in_window=False,
|
|
):
|
|
statement = select(func.count()).select_from(cls).where(cls.is_deleted == False) # noqa: E712
|
|
if status:
|
|
statement = statement.where(cls.requisition_status == status)
|
|
statement = cls._scoped(statement, department, recruiter_id)
|
|
if closed_in_window:
|
|
if from_date is not None:
|
|
statement = statement.where(cls.closed_at >= from_date)
|
|
if to_date is not None:
|
|
statement = statement.where(cls.closed_at < to_date)
|
|
result = await session.execute(statement)
|
|
return int(result.scalar_one() or 0)
|
|
|
|
@classmethod
|
|
async def list_open_reqs(cls, session: AsyncSession, department=None, recruiter_id=None):
|
|
"""Open, non-deleted requisitions — the zero-application fill for
|
|
analytics' per-job counts. Same scoping semantics as count_requisitions.
|
|
"""
|
|
statement = select(cls).where(
|
|
cls.is_deleted == False, # noqa: E712
|
|
cls.requisition_status == RequisitionStatus.OPEN.value,
|
|
)
|
|
statement = cls._scoped(statement, department, recruiter_id)
|
|
result = await session.execute(statement)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def count_open_snapshot(cls, session: AsyncSession, as_of, department=None, recruiter_id=None):
|
|
"""Jobs that existed and were still open at `as_of` (best-effort)."""
|
|
statement = select(func.count()).select_from(cls).where(
|
|
cls.is_deleted == False, # noqa: E712
|
|
cls.created_at < as_of,
|
|
or_(cls.closed_at.is_(None), cls.closed_at >= as_of),
|
|
cls.requisition_status == RequisitionStatus.OPEN.value,
|
|
)
|
|
statement = cls._scoped(statement, department, recruiter_id)
|
|
result = await session.execute(statement)
|
|
return int(result.scalar_one() or 0)
|
|
|
|
@classmethod
|
|
async def avg_time_to_fill(
|
|
cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
|
|
):
|
|
days = func.extract("epoch", cls.closed_at - cls.created_at) / 86400.0
|
|
statement = select(func.avg(days)).select_from(cls).where(
|
|
cls.is_deleted == False, # noqa: E712
|
|
cls.requisition_status.in_((
|
|
RequisitionStatus.CLOSED.value,
|
|
RequisitionStatus.COMPLETED.value,
|
|
)),
|
|
cls.closed_at.is_not(None),
|
|
)
|
|
if from_date is not None:
|
|
statement = statement.where(cls.closed_at >= from_date)
|
|
if to_date is not None:
|
|
statement = statement.where(cls.closed_at < to_date)
|
|
statement = cls._scoped(statement, department, recruiter_id)
|
|
result = await session.execute(statement)
|
|
value = result.scalar_one()
|
|
return float(value) if value is not None else None
|
|
|
|
@classmethod
|
|
async def insert_job_post(cls, session: AsyncSession, fields: dict):
|
|
row = cls(**fields)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
session.flush()
|
|
session.add(JobPostStatusHistory(
|
|
job_post_id=row.id,
|
|
from_status=None,
|
|
to_status=row.requisition_status or "open",
|
|
changed_by=row.created_by,
|
|
))
|
|
await session.commit()
|
|
return await cls.get_job_post_by_id(session, row.id)
|
|
|
|
@classmethod
|
|
async def mark_buffer_result(
|
|
cls,
|
|
session: AsyncSession,
|
|
record_id: str,
|
|
*,
|
|
buffer_post_id: str,
|
|
status: str,
|
|
external_link: str | None = None,
|
|
sent_at: datetime | None = None,
|
|
platform: str | None = None,
|
|
):
|
|
"""Record what Buffer reported.
|
|
`status` is the mapped Buffer PostStatus, not an assumption: a queued post lands
|
|
here as "scheduled" and only becomes "published" once Buffer says `sent`.
|
|
"""
|
|
row = await cls.get_job_post_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
row.status = status
|
|
row.buffer_post_id = buffer_post_id
|
|
row.buffer_external_link = external_link
|
|
row.buffer_sent_at = sent_at
|
|
if platform:
|
|
row.platform = platform
|
|
row.buffer_error = None
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def mark_failed(cls, session: AsyncSession, record_id: str, error: str):
|
|
row = await cls.get_job_post_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
row.status = "failed"
|
|
row.buffer_error = error
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def update_job_post(cls, session: AsyncSession, record_id: str, fields: dict):
|
|
row = await cls.get_job_post_by_id(session, record_id)
|
|
if not row or row.is_deleted:
|
|
return None
|
|
for key, value in fields.items():
|
|
setattr(row, key, value)
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
await session.commit()
|
|
return await cls.get_job_post_by_id(session, record_id)
|
|
|
|
@classmethod
|
|
async def soft_delete_job_post(cls, session: AsyncSession, record_id: str):
|
|
row = await cls.get_job_post_by_id(session, record_id)
|
|
if not row or row.is_deleted:
|
|
return None
|
|
row.is_deleted = True
|
|
row.is_active = False
|
|
row.requisition_id = None
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def set_requisition_status(
|
|
cls, session: AsyncSession, record_id: str, status: str, *, changed_by=None,
|
|
):
|
|
row = await cls.get_job_post_by_id(session, record_id)
|
|
if not row or row.is_deleted:
|
|
return None
|
|
previous = row.requisition_status
|
|
if previous == status:
|
|
return row
|
|
row.requisition_status = status
|
|
terminal = status in ("closed", "completed")
|
|
if terminal:
|
|
if previous not in ("closed", "completed") or row.closed_at is None:
|
|
row.closed_at = _now()
|
|
else:
|
|
row.closed_at = None
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
actor = cls._as_uuid(changed_by) if changed_by is not None else None
|
|
session.add(JobPostStatusHistory(
|
|
job_post_id=row.id,
|
|
from_status=previous,
|
|
to_status=status,
|
|
changed_by=actor,
|
|
))
|
|
await session.commit()
|
|
return await cls.get_job_post_by_id(session, record_id)
|
|
|
|
|
|
class JobPostStatusHistory(SQLModel, table=True):
|
|
"""Who changed job_posts.requisition_status, from what, to what, and when.
|
|
|
|
Distinct from job_assignments (ownership intervals). The Jobs History tab
|
|
merges both. Applied on prod by migrations/manual/016_job_post_status_history.sql.
|
|
"""
|
|
|
|
__tablename__ = "job_post_status_history"
|
|
__table_args__ = (
|
|
Index("ix_job_post_status_history_job_created", "job_post_id", "created_at"),
|
|
)
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
job_post_id: uuid.UUID = Field(foreign_key="job_posts.id", index=True)
|
|
from_status: str | None = Field(default=None)
|
|
to_status: str
|
|
changed_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
actor_kind: str = Field(default="user")
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
@classmethod
|
|
async def fetch_by_job(cls, session: AsyncSession, job_post_id):
|
|
uid = JobPosts._as_uuid(job_post_id)
|
|
if uid is None:
|
|
return []
|
|
result = await session.execute(
|
|
select(cls)
|
|
.where(cls.job_post_id == uid)
|
|
.order_by(cls.created_at.desc(), cls.id.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
class JobPostImages(SQLModel, table=True):
|
|
"""Cover image of a job post, stored as bytes IN the database.
|
|
|
|
Deliberately not on disk: production containers have ephemeral filesystems,
|
|
so a file-backed image dies on every redeploy. One row per post — the PK is
|
|
the job_posts FK, which makes re-upload a plain replace. Created in prod by
|
|
migrations/manual/009_job_post_images.sql (autogen is off there)."""
|
|
|
|
__tablename__ = "job_post_images"
|
|
|
|
job_post_id: uuid.UUID = Field(primary_key=True, foreign_key="job_posts.id")
|
|
content_type: str
|
|
file_name: str | None = Field(default=None)
|
|
data: bytes
|
|
uploaded_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
@classmethod
|
|
async def get(cls, session: AsyncSession, job_post_id: uuid.UUID):
|
|
result = await session.execute(select(cls).where(cls.job_post_id == job_post_id))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def upsert(cls, session: AsyncSession, job_post_id: uuid.UUID, *,
|
|
content_type: str, file_name: str | None, data: bytes,
|
|
uploaded_by: uuid.UUID | None):
|
|
row = await cls.get(session, job_post_id)
|
|
if row:
|
|
row.content_type = content_type
|
|
row.file_name = file_name
|
|
row.data = data
|
|
row.uploaded_by = uploaded_by
|
|
row.updated_at = _now()
|
|
else:
|
|
row = cls(
|
|
job_post_id=job_post_id, content_type=content_type,
|
|
file_name=file_name, data=data, uploaded_by=uploaded_by,
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
return row
|
|
|
|
|
|
class SocialPlatform(SQLModel, table=True):
|
|
"""Buffer publish aliases (fb → facebook). Spelling tolerance + UI list, not an allowlist."""
|
|
|
|
__tablename__ = "social_platforms"
|
|
|
|
id: int | None = Field(default=None, primary_key=True)
|
|
alias: str = Field(max_length=40, unique=True, index=True)
|
|
buffer_service: str
|
|
label: str
|
|
is_active: bool = Field(default=True)
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
@classmethod
|
|
async def get_by_alias(cls, session: AsyncSession, alias: str):
|
|
key = (alias or "").strip().lower()
|
|
if not key:
|
|
return None
|
|
result = await session.execute(select(cls).where(cls.alias == key))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def list_active(cls, session: AsyncSession):
|
|
result = await session.execute(
|
|
select(cls).where(cls.is_active == True).order_by(cls.id.asc()) # noqa: E712
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def list_aliases(cls, session: AsyncSession):
|
|
rows = await cls.list_active(session)
|
|
return [r.alias for r in rows]
|
|
|
|
@classmethod
|
|
async def alias_map(cls, session: AsyncSession) -> dict[str, str]:
|
|
"""alias → Buffer service name for normalize_platform / resolve_channel."""
|
|
rows = await cls.list_active(session)
|
|
return {r.alias: r.buffer_service for r in rows}
|
|
|
|
|
|
# Requisition must be registered before Users relationships trigger mapper
|
|
# configure — JobPosts.requisition_id FKs to app.requisitions.
|
|
import candidate_forms.models as _requisition_models # noqa: E402, F401
|
|
import users.models as _users_models # noqa: E402, F401 |