211 lines
8.4 KiB
Python
211 lines
8.4 KiB
Python
"""FormData + SheetImportRun — spreadsheet mirror and background import runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import Column, DateTime, Index, delete, func, or_
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlmodel import Field, SQLModel, select
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class FormData(SQLModel, table=True):
|
|
"""One spreadsheet data row. raw_record keeps the full original header→value map."""
|
|
|
|
__tablename__ = "form_data"
|
|
__table_args__ = (
|
|
Index("ix_form_data_sheet_row_number", "sheet", "row_number", unique=True),
|
|
)
|
|
|
|
id: int | None = Field(default=None, primary_key=True)
|
|
sheet: str = Field(nullable=False, index=True)
|
|
name: str | None = Field(default=None, index=True)
|
|
degree: str | None = Field(default=None)
|
|
experience: str | None = Field(default=None)
|
|
age: int | None = Field(default=None)
|
|
age_raw: str | None = Field(default=None)
|
|
family_details: str | None = Field(default=None)
|
|
|
|
interview_date: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
interview_by: str | None = Field(default=None)
|
|
interview_time: str | None = Field(default=None)
|
|
interview_status: str | None = Field(default=None)
|
|
interview_notes: str | None = Field(default=None)
|
|
interview_result: str | None = Field(default=None)
|
|
|
|
second_interview_date: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
second_interview_by: str | None = Field(default=None)
|
|
second_interview_time: str | None = Field(default=None)
|
|
second_interview_status: str | None = Field(default=None)
|
|
second_interview_notes: str | None = Field(default=None)
|
|
second_interview_result: str | None = Field(default=None)
|
|
|
|
third_interview_date: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
third_interview_by: str | None = Field(default=None)
|
|
third_interview_time: str | None = Field(default=None)
|
|
third_interview_status: str | None = Field(default=None)
|
|
third_interview_notes: str | None = Field(default=None)
|
|
third_interview_result: str | None = Field(default=None)
|
|
|
|
fourth_interview_date: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
fourth_interview_by: str | None = Field(default=None)
|
|
fourth_interview_time: str | None = Field(default=None)
|
|
fourth_interview_status: str | None = Field(default=None)
|
|
fourth_interview_notes: str | None = Field(default=None)
|
|
fourth_interview_result: str | None = Field(default=None)
|
|
|
|
raw_record: dict | None = Field(default=None, sa_column=Column(JSONB))
|
|
row_number: int | None = Field(default=None)
|
|
imported_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=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
|
|
def _filters(cls, *, sheet=None, search=None):
|
|
filters = []
|
|
if sheet:
|
|
filters.append(cls.sheet == sheet)
|
|
if search:
|
|
pattern = f"%{search}%"
|
|
filters.append(or_(
|
|
cls.name.ilike(pattern),
|
|
cls.degree.ilike(pattern),
|
|
cls.experience.ilike(pattern),
|
|
cls.interview_by.ilike(pattern),
|
|
))
|
|
return filters
|
|
|
|
@classmethod
|
|
async def get_form_data_by_id(cls, session: AsyncSession, record_id):
|
|
try:
|
|
rid = int(record_id)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
result = await session.execute(select(cls).where(cls.id == rid))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def fetch_form_data(cls, session: AsyncSession, *, sheet=None, search=None, top=None, skip=None):
|
|
statement = select(cls).order_by(cls.sheet, cls.row_number)
|
|
for clause in cls._filters(sheet=sheet, search=search):
|
|
statement = statement.where(clause)
|
|
if skip:
|
|
statement = statement.offset(skip)
|
|
if top is not None:
|
|
statement = statement.limit(top)
|
|
result = await session.execute(statement)
|
|
return result.scalars().all()
|
|
|
|
@classmethod
|
|
async def count_form_data(cls, session: AsyncSession, *, sheet=None, search=None):
|
|
statement = select(func.count()).select_from(cls)
|
|
for clause in cls._filters(sheet=sheet, search=search):
|
|
statement = statement.where(clause)
|
|
result = await session.execute(statement)
|
|
return result.scalar_one()
|
|
|
|
@classmethod
|
|
async def get_sheet_names(cls, session: AsyncSession):
|
|
result = await session.execute(
|
|
select(cls.sheet).distinct().order_by(cls.sheet)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def delete_by_sheet(cls, session: AsyncSession, sheet: str, *, commit: bool = True):
|
|
count_result = await session.execute(
|
|
select(func.count()).select_from(cls).where(cls.sheet == sheet)
|
|
)
|
|
deleted = count_result.scalar_one()
|
|
await session.execute(delete(cls).where(cls.sheet == sheet))
|
|
if commit:
|
|
await session.commit()
|
|
return deleted
|
|
|
|
@classmethod
|
|
async def insert_form_data_bulk(cls, session: AsyncSession, records: list[dict], *, commit: bool = True):
|
|
rows = [cls(**fields) for fields in records]
|
|
session.add_all(rows)
|
|
if commit:
|
|
await session.commit()
|
|
return len(rows)
|
|
|
|
@classmethod
|
|
async def replace_sheet(cls, session: AsyncSession, sheet: str, records: list[dict]):
|
|
"""Delete + insert in one transaction so a mid-insert failure keeps prior rows."""
|
|
deleted = await cls.delete_by_sheet(session, sheet, commit=False)
|
|
inserted = await cls.insert_form_data_bulk(session, records, commit=False)
|
|
await session.commit()
|
|
return {"deleted": deleted, "inserted": inserted}
|
|
|
|
|
|
class SheetImportRun(SQLModel, table=True):
|
|
"""One Google Sheet → FormData import job (Taskiq). Survives tab close."""
|
|
|
|
__tablename__ = "sheet_import_runs"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
status: str = Field(default="queued", index=True) # queued|running|completed|failed
|
|
task_id: str | None = Field(default=None)
|
|
created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
tab: str | None = Field(default=None) # None = import all tabs
|
|
report: dict | None = Field(default=None, sa_column=Column(JSONB))
|
|
error: str | None = Field(default=None)
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id) -> uuid.UUID | None:
|
|
if record_id in (None, ""):
|
|
return None
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def get_by_id(cls, session: AsyncSession, record_id):
|
|
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(cls, session: AsyncSession):
|
|
result = await session.execute(
|
|
select(cls)
|
|
.where(cls.status.in_(("queued", "running")))
|
|
.order_by(cls.created_at.desc())
|
|
)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True):
|
|
row = cls(**fields)
|
|
session.add(row)
|
|
if commit:
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def update_run(cls, session: AsyncSession, record_id, fields: dict, *, commit: bool = True):
|
|
row = await cls.get_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
for key, value in fields.items():
|
|
setattr(row, key, value)
|
|
session.add(row)
|
|
if commit:
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|