import uuid from datetime import datetime, timezone from typing import TYPE_CHECKING, Optional from sqlalchemy import DateTime, JSON, Index, func, or_ from sqlalchemy.ext.asyncio import AsyncSession 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 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) # foreign_keys is required, not decoration: current_recruiter_id and # hiring_manager_id below are extra FKs into users.id, so the join is # ambiguous without it and every mapper fails to initialize. `user` is the # AUTHOR of the post. The recruiter and hiring-manager columns stay bare — # Users already carries five selectin relations that load on every # authenticated request. Same pairing as Notes.user / Notes.author. 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 is the hiring lifecycle (RequisitionStatus). Distinct from # `status`, which tracks Buffer publishing (draft/scheduled/published/failed). # server_default is load-bearing: this column arrives as an ALTER on a populated table. 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). Required at create. 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) 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_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()) @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, ): 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 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 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 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 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.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} import users.models as _users_models