15 KiB
15 KiB
HR-ATS Backend LLM Context Prompt
Copy everything below the line into any LLM session before asking it to write or edit backend code.
You are coding inside HR-ATS-Portal (backend/). Follow this house style exactly. Mirror the reference flow below line-for-line in shape. Do not invent layers, response shapes, flags, guards, or "defensive" checks the reference does not have. Write no more and no less than the reference does for the same job.
Reference flow (the canonical example)
POST /forms/requisition/create in backend/candidate_forms/. Every new endpoint copies this shape.
1. enums.py — enums + nested payload blocks
class EmploymentType(str,Enum):
PERMANENT = "permanent"
CONTRACT = "contract"
class Position(BaseModel):
department_id:uuid.UUID
department:Optional[str]
title:Optional[str]
type:Optional[EmploymentType]
period_from:Optional[date]=None
str, Enumclasses and the nestedBaseModelblocks a request body is built from live here.
2. app.py — request models inline + thin route
class RequisitionFormCreate(BaseModel):
form_type: str = "requisition"
position:Position
replacement_for:Optional[ReplacementFor]
initiated_by:Optional[str]
approved_by_hr:Optional[bool]
class RequisitionFormUpdate(BaseModel):
position:Optional[Position]=None
initiated_by:Optional[str]=None
approved_by_hr:Optional[bool]=None
@router.post("/forms/requisition/create")
async def create_requisition_form(
payload: RequisitionFormCreate,
current_user:dict=Depends(require_permission(PermissionTag.REQUISITIONS_CREATE)),
session:AsyncSession=Depends(get_session),
):
try:
service = RequisitionForm(session=session)
data = await service.create_form(payload.model_dump(exclude_unset=True), current_user)
return JSONResponse(content={"data": data, "status_code": 200})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
router = APIRouter(); mounted inmain.pyviaapp.include_router(...).- Paths are verb-in-path:
/<domain>/create,/<domain>/fetch,/<domain>/update,/<domain>/delete,/<domain>/search. No/api/v1, no REST-resource-only paths. - Methods:
postcreate,getfetch/search/count,patchupdate,deletedelete. - Param order:
payload→current_user→ query params →session. - Record id comes as a query param:
form_id:str=Query(...)(required) orQuery(None)(fetch one-or-all). - Body goes to the service as
payload.model_dump(exclude_unset=True). - Protection:
current_user:dict=Depends(require_permission(PermissionTag.X)). Several tags:require_permission(A, B, require_all=False). - The route body is only the try block above: build service, one
await,JSONResponse. Nothing else. - Envelope:
{"data": data, "status_code": 200}. List with count: add"total". A scalar goes insidedataas a dict ({"data":{"open":data},...}). - Request models stay inline in
app.py— Create has required fields without defaults, Update has every fieldOptional[...]=None.
3. views.py — service class
class RequisitionForm:
def __init__(self, session: AsyncSession):
self.session = session
async def create_form(self, payload, current_user):
payload["position"]["department_id"] = _as_uuid(payload["position"]["department_id"])
payload["created_by"] = _user_id(current_user)
row = await Requisition.insert_form(self.session, payload)
return serialize_requisition(row)
async def update_form(self, form_id, payload, current_user):
_user_id(current_user)
row = await Requisition.get_form_by_id(self.session, form_id)
if not row:
raise HTTPException(status_code=404, detail="Form not found")
if not payload:
raise HTTPException(status_code=400, detail="No fields to update")
updated = await Requisition.update_form(self.session, form_id, payload)
if not updated:
raise HTTPException(status_code=404, detail="Form not found")
return serialize_requisition(updated)
async def get_form_by_id(self, form_id, current_user):
created_by = None if is_admin(current_user) else _user_id(current_user)
if form_id:
row = await Requisition.get_form_by_id(self.session, record_id=form_id, created_by=created_by)
if not row:
raise HTTPException(status_code=404, detail="Form not found")
return serialize_requisition(row)
rows = await Requisition.get_form_by_id(self.session, created_by=created_by)
return [serialize_requisition(r) for r in rows]
- One class per resource,
__init__(self, session: AsyncSession)only. - Method parameters are untyped.
- A create method is: adjust the payload dict in place (coerce ids with plugin helpers, stamp
created_by) → call one model classmethod → return the serializer. That is the whole method. - Business errors:
raise HTTPException(status_code=..., detail="...")— 404 not found, 400 empty update, 401 auth, 422 invalid value. - Scope reads with
created_by = None if is_admin(current_user) else _user_id(current_user). - Always return serialized dicts / lists of dicts, never ORM rows.
4. plugins.py — shared helpers
def _as_uuid(value):
if value in (None, ""):
return None
try:
return uuid.UUID(str(value))
except (TypeError, ValueError):
return None
def _user_id(current_user):
if not current_user or not current_user.get("id"):
raise HTTPException(status_code=401, detail="Not authenticated")
uid = _as_uuid(current_user["id"])
if uid is None:
raise HTTPException(status_code=401, detail="Invalid user id")
return uid
- Every helper function and constant (
_as_uuid,_user_id,_aware, validators,FORM_TYPES, definitions payloads) lives in<domain>/plugins.pyand is imported by name intoviews.py. Never define helpers at the top ofviews.py,models.py,app.py, orserializers.py. - Helpers may raise
HTTPExceptionwhen they validate request data. - Before writing a helper, check the domain's
plugins.pyand reuse what exists.
5. models.py — SQLModel table + classmethod accessors
class Requisition(SQLModel, table=True):
__tablename__ = "requisitions"
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
department_id: Optional[uuid.UUID] = Field(default=None, foreign_key="departments.id")
department_ref: Optional["Department"] = Relationship(
back_populates="requisitions",
sa_relationship_kwargs={"uselist": False, "lazy": "selectin"},
)
position_title: Optional[str] = None
created_by: Optional[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))
is_deleted: bool = Field(default=False)
@classmethod
async def get_form_by_id(cls, session: AsyncSession, record_id=None, created_by=None):
qry = select(cls).where(cls.is_deleted == False) # noqa: E712
if created_by is not None:
qry = qry.where(cls.created_by == created_by)
if record_id not in (None, ""):
try:
uid = uuid.UUID(str(record_id))
except (TypeError, ValueError):
return None
qry = qry.where(cls.id == uid)
result = await session.execute(qry)
return result.scalars().first()
result = await session.execute(qry.order_by(cls.created_at.desc(),cls.id.desc()))
return list(result.scalars().all())
@classmethod
async def insert_form(cls, session: AsyncSession, fields: dict):
position = fields.get("position") if fields.get("position") else {}
row = cls(
department_id=position.get("department_id") if position.get("department_id") else None,
position_title=position.get("title") if position.get("title") else None,
employment_type=EmploymentType(position.get("type")) if position.get("type") else None,
jd_available=position.get("jd_available") if position.get("jd_available") is not None else None,
initiated_by=fields.get("initiated_by") if fields.get("initiated_by") else None,
created_by=fields.get("created_by") if fields.get("created_by") else None,
)
session.add(row)
await session.commit()
return await cls.get_form_by_id(session, row.id)
@classmethod
async def update_form(cls, session: AsyncSession, record_id, fields: dict):
row = await cls.get_form_by_id(session, record_id)
if not row:
return None
if "position" in fields:
position = fields.get("position") if fields.get("position") else {}
if "title" in position:
row.position_title = position.get("title") if position.get("title") else None
if "initiated_by" in fields:
row.initiated_by = fields.get("initiated_by") if fields.get("initiated_by") else None
row.updated_at = _now()
session.add(row)
await session.commit()
await session.refresh(row)
return row
@classmethod
async def soft_delete_form(cls, session: AsyncSession, record_id):
row = await cls.get_form_by_id(session, record_id)
if not row:
return None
row.is_deleted = True
row.updated_at = _now()
session.add(row)
await session.commit()
return row
- Standard columns on every table:
iduuid4 PK,created_byFKusers.id,created_at/updated_attz-aware via_now,is_deleted. - All DB access is
@classmethod async deftakingsessionfirst. No free functions, no repository class. - One accessor does one whole job. Insert maps the full (nested) payload dict to columns, adds, commits, and re-fetches in the same method. Never split a create into several helper calls or a second "create child" function when the mapping fits inline.
- One reader for one-or-many:
get_form_by_idreturns a row whenrecord_idis given, a list otherwise. Do not add a separatefetch_all. - Column mapping idiom:
x=src.get("k") if src.get("k") else None; booleans useis not None; enums wrapEnum(value). - Update idiom:
if "k" in fields:per field (nested blocks:if "block" in fields:then per-key), thenrow.updated_at = _now(), add, commit, refresh, return row. ReturnNonewhen missing — the view raises. - Soft delete only:
is_deleted = True(+updated_at). Reads always filtercls.is_deleted == False # noqa: E712. - Cross-domain relations: model name in quotes, import under
if TYPE_CHECKING:,back_populateson both sides, and a bottom-of-fileimport <other>.models as _<other>_models # noqa: E402, F401. - Commits happen inside write accessors; views never commit.
6. serializers.py — hand-built dicts
def _date(value):
return value.isoformat() if value else None
def serialize_requisition(row) -> dict:
return {
"id": str(row.id) if row.id else None,
"position": {
"department_id": str(row.department_id) if row.department_id else None,
"title": row.position_title,
"date": _date(row.date),
"type": _enum(row.employment_type),
},
"initiated_by": row.initiated_by,
"created_by": str(row.created_by) if row.created_by else None,
"created_at": row.created_at.isoformat() if row.created_at else None,
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
}
serialize_<resource>(row) -> dict. UUIDsstr(...) if ... else None, dates.isoformat(), enums via_enum.- The response shape mirrors the request shape (nested blocks back out as nested dicts), so the frontend sends and receives the same structure.
- No DB access, no Pydantic response models. Never include passwords.
7. Schema change → manual SQL migration
- New column / FK / index: add
backend/migrations/manual/<next_number>_<snake_name>.sqlwith a header comment explaining why, idempotent DDL (ADD COLUMN IF NOT EXISTS, guardedADD CONSTRAINT,CREATE INDEX IF NOT EXISTS), schemaapp.. Never edit an already-applied migration file.
Package layout (every domain)
backend/<domain>/
app.py # routes + inline request models
views.py # service class
models.py # SQLModel table + classmethod accessors
serializers.py # serialize_* dict builders
enums.py # str Enums + nested request BaseModel blocks
plugins.py # helpers + constants
permissions.py # auth domains only
- No
__init__.py. Run frombackend/; imports are top-level (candidate_forms.views,db_setup,users.permissions). - Config: module-level
load_dotenv()+os.getenv(...). Do not extenddb_setup.Settingsfor app secrets.
Layer duties
| Layer | Owns | Must NOT |
|---|---|---|
app.py |
Routes, inline request models, Depends, JSONResponse envelope |
Business rules, SQL, helper functions |
views.py |
Payload prep, business checks, HTTPException, call model, return serializer |
SQL, commits, helper definitions |
models.py |
Columns, relationships, queries, insert/update/soft-delete + commit | HTTPException, serializers |
serializers.py |
serialize_* → dict |
DB, Depends |
enums.py |
Enums, nested request blocks | Logic |
plugins.py |
Every helper and constant | Routes, DB writes |
Workflow when adding an endpoint
- Migration SQL if the schema changes.
- Columns + one classmethod accessor in
models.py. - Enum / nested block in
enums.pyif the body has one. - Helper in
plugins.pyonly if a view needs one that does not already exist. serialize_*inserializers.pyif the shape is new.- Service method in
views.py. - Inline request model + route in
app.pywith the exact try/except +JSONResponsewrapper andrequire_permission.
Auth (only when touching users/)
- Login / refresh: service returns the Users ORM row; the route mints tokens and calls
serialize_token(...)— never insideviews.py. Tokens at the root, user underdata. - PyJWT access + refresh with a
typeclaim;iat/expusedatetime.now(timezone.utc). - RBAC lives in
users/permissions.py(require_permission,is_admin,CurrentUser). Do not invent a second scheme.
Hard bans
- No extra flags, params, guards, try/excepts, logging, or validations the reference flow does not have.
- No splitting one create/update into multiple helper functions or extra model calls when one classmethod does it.
- No helpers defined outside
plugins.py; no duplicated helpers — import the existing one. - No repository / use-case / DTO layers; no Pydantic response models; no alternate envelopes; no
/api/v1. - No hard deletes.
- No drive-by refactors, renames, reformatting, or edits to unrelated domains.
- No
__init__.py. - Dependencies: pin in
backend/requirements.txtwith a# whycomment; secret names inbackend/.env.example.
Before finishing, put the new code next to the reference flow above and remove anything the reference would not have.