CAR_progress / database.py
puneethasshankar's picture
Upload 6 files
4624679 verified
Raw
History Blame Contribute Delete
28.5 kB
"""
database.py — Backend database layer for Job Intelligence Engine.
Architecture:
- Reads DATABASE_URL from the shared frontend/.env file.
- Uses a psycopg2 SimpleConnectionPool to manage connections efficiently.
- Provides typed query helpers for every table in the schema,
mirroring the Drizzle ORM schema in frontend/src/db/schema.ts.
Tables (in dependency order):
schools, programs, program_officer_assignments,
"user", placement_officers,
students, companies,
drives, placements,
officer_monthly_snapshot
"""
import os
import json
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2.pool import SimpleConnectionPool
from dotenv import load_dotenv
# ---------------------------------------------------------------------------
# Environment & Pool Setup
# ---------------------------------------------------------------------------
BASE_DIR = Path(__file__).resolve().parent.parent
FRONTEND_ENV_PATH = BASE_DIR / "frontend" / ".env"
if FRONTEND_ENV_PATH.exists():
print(f"[DB] Loading environment from {FRONTEND_ENV_PATH}")
load_dotenv(dotenv_path=FRONTEND_ENV_PATH)
else:
print(f"[DB] Warning: .env not found at {FRONTEND_ENV_PATH}. Falling back to system env.")
load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
db_pool: Optional[SimpleConnectionPool] = None
if DATABASE_URL:
try:
db_pool = SimpleConnectionPool(minconn=1, maxconn=10, dsn=DATABASE_URL)
print("[DB] Connection pool initialized successfully.")
except Exception as exc:
print(f"[DB] Failed to initialize connection pool: {exc}")
else:
print("[DB] DATABASE_URL is not set. Pool not initialized.")
# ---------------------------------------------------------------------------
# Connection Utilities
# ---------------------------------------------------------------------------
def get_db_connection():
"""Acquire a connection from the pool."""
if not db_pool:
raise RuntimeError("Database pool is not initialized.")
return db_pool.getconn()
def release_db_connection(conn):
"""Return a connection to the pool."""
if db_pool and conn:
db_pool.putconn(conn)
@contextmanager
def db_cursor(dict_cursor: bool = True):
"""
Context manager that acquires a connection + cursor, commits on exit,
and always releases the connection back to the pool.
Usage:
with db_cursor() as cur:
cur.execute("SELECT ...")
rows = cur.fetchall()
"""
conn = get_db_connection()
cursor_factory = RealDictCursor if dict_cursor else None
try:
cur = conn.cursor(cursor_factory=cursor_factory)
yield cur
conn.commit()
except Exception:
conn.rollback()
raise
finally:
cur.close()
release_db_connection(conn)
# ---------------------------------------------------------------------------
# Schema Reference
# (mirrors frontend/src/db/schema.ts — keep in sync)
# ---------------------------------------------------------------------------
# Enum values — mirrors pgEnum definitions in schema.ts
ROLE_ENUM = ("admin", "officer", "viewer")
DRIVE_TYPE_ENUM = ("full_time", "internship", "capstone")
PLACEMENT_TYPE_ENUM = ("full_time", "internship", "capstone", "higher_studies")
OFFER_STATUS_ENUM = ("offered", "accepted", "rejected")
# ---------------------------------------------------------------------------
# Table: schools
# Columns: school_id (PK), school_name
# ---------------------------------------------------------------------------
def get_all_schools() -> list[dict]:
with db_cursor() as cur:
cur.execute("SELECT school_id, school_name FROM schools ORDER BY school_name")
return cur.fetchall()
def get_school_by_id(school_id: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("SELECT school_id, school_name FROM schools WHERE school_id = %s", (school_id,))
return cur.fetchone()
def create_school(school_name: str) -> dict:
with db_cursor() as cur:
cur.execute("INSERT INTO schools (school_name) VALUES (%s) RETURNING *", (school_name,))
return cur.fetchone()
def update_school(school_id: int, school_name: str) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("UPDATE schools SET school_name = %s WHERE school_id = %s RETURNING *", (school_name, school_id))
return cur.fetchone()
def delete_school(school_id: int) -> bool:
with db_cursor() as cur:
cur.execute("DELETE FROM schools WHERE school_id = %s", (school_id,))
return cur.rowcount > 0
# ---------------------------------------------------------------------------
# Table: programs
# Columns: program_id (PK), school_id (FK), program_name,
# credit_weightage, total_eligible_students
# ---------------------------------------------------------------------------
def get_all_programs() -> list[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT p.program_id, p.school_id, s.school_name,
p.program_name, p.credit_weightage, p.total_eligible_students
FROM programs p
JOIN schools s ON s.school_id = p.school_id
ORDER BY s.school_name, p.program_name
""")
return cur.fetchall()
def get_programs_by_school(school_id: int) -> list[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT program_id, program_name, credit_weightage, total_eligible_students
FROM programs WHERE school_id = %s ORDER BY program_name
""", (school_id,))
return cur.fetchall()
def get_program_by_id(program_id: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT p.program_id, p.school_id, s.school_name,
p.program_name, p.credit_weightage, p.total_eligible_students
FROM programs p
JOIN schools s ON s.school_id = p.school_id
WHERE p.program_id = %s
""", (program_id,))
return cur.fetchone()
def create_program(school_id: int, program_name: str, credit_weightage: float, total_eligible_students: int) -> dict:
with db_cursor() as cur:
cur.execute("""
INSERT INTO programs (school_id, program_name, credit_weightage, total_eligible_students)
VALUES (%s, %s, %s, %s)
RETURNING *
""", (school_id, program_name, credit_weightage, total_eligible_students))
return cur.fetchone()
def update_program(program_id: int, school_id: int, program_name: str, credit_weightage: float, total_eligible_students: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
UPDATE programs
SET school_id = %s, program_name = %s, credit_weightage = %s, total_eligible_students = %s
WHERE program_id = %s
RETURNING *
""", (school_id, program_name, credit_weightage, total_eligible_students, program_id))
return cur.fetchone()
def delete_program(program_id: int) -> bool:
with db_cursor() as cur:
cur.execute("DELETE FROM programs WHERE program_id = %s", (program_id,))
return cur.rowcount > 0
# ---------------------------------------------------------------------------
# Table: program_officer_assignments
# Columns: program_id (FK, PK), officer_id (FK, PK) — composite PK
# ---------------------------------------------------------------------------
def get_officer_programs(officer_id: int) -> list[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT poa.program_id, poa.officer_id,
p.program_name, p.credit_weightage, p.total_eligible_students, s.school_name
FROM program_officer_assignments poa
JOIN programs p ON p.program_id = poa.program_id
JOIN schools s ON s.school_id = p.school_id
WHERE poa.officer_id = %s
""", (officer_id,))
return cur.fetchall()
def assign_officer_to_program(officer_id: int, program_id: int) -> None:
with db_cursor() as cur:
cur.execute("""
INSERT INTO program_officer_assignments (program_id, officer_id)
VALUES (%s, %s)
ON CONFLICT DO NOTHING
""", (program_id, officer_id))
def unassign_officer_from_program(officer_id: int, program_id: int) -> None:
with db_cursor() as cur:
cur.execute("""
DELETE FROM program_officer_assignments
WHERE program_id = %s AND officer_id = %s
""", (program_id, officer_id))
# ---------------------------------------------------------------------------
# Table: "user" (Better-Auth)
# Columns: id (PK, text), email (UNIQUE), role, createdAt, etc.
# ---------------------------------------------------------------------------
def get_user_by_email(email: str) -> Optional[dict]:
with db_cursor() as cur:
cur.execute('SELECT * FROM "user" WHERE email = %s', (email,))
return cur.fetchone()
def get_user_by_id(user_id: str) -> Optional[dict]:
with db_cursor() as cur:
cur.execute('SELECT id, email, role, "createdAt" FROM "user" WHERE id = %s', (user_id,))
return cur.fetchone()
def create_user(email: str, name: str, role: str) -> dict:
if role not in ROLE_ENUM:
raise ValueError(f"Invalid role '{role}'. Must be one of {ROLE_ENUM}")
import uuid
new_id = str(uuid.uuid4())
with db_cursor() as cur:
cur.execute("""
INSERT INTO "user" (id, name, email, "emailVerified", role, "createdAt", "updatedAt")
VALUES (%s, %s, %s, FALSE, %s, NOW(), NOW())
RETURNING id as user_id, email, role, "createdAt"
""", (new_id, name, email, role))
return cur.fetchone()
# ---------------------------------------------------------------------------
# Table: placement_officers
# Columns: officer_id (PK), user_id (FK), name, phone, school_id (FK)
# ---------------------------------------------------------------------------
def get_all_officers() -> list[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT po.officer_id, po.name, po.phone, po.email,
COALESCE(u.role, 'officer') as role,
COALESCE(
(SELECT json_agg(json_build_object('program_id', poa.program_id, 'program_name', p.program_name))
FROM program_officer_assignments poa
JOIN programs p ON p.program_id = poa.program_id
WHERE poa.officer_id = po.officer_id),
'[]'::json
) as assigned_programs
FROM placement_officers po
LEFT JOIN "user" u ON u.email = po.email
ORDER BY po.name
""")
return cur.fetchall()
def get_officer_by_id(officer_id: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT po.officer_id, po.name, po.phone,
u.email
FROM placement_officers po
JOIN "user" u ON u.id = po.user_id
WHERE po.officer_id = %s
""", (officer_id,))
return cur.fetchone()
def update_officer(officer_id: int, name: str, phone: Optional[str]) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
UPDATE placement_officers
SET name = %s, phone = %s
WHERE officer_id = %s
RETURNING *
""", (name, phone, officer_id))
return cur.fetchone()
def delete_officer(officer_id: int) -> bool:
with db_cursor() as cur:
cur.execute("DELETE FROM placement_officers WHERE officer_id = %s", (officer_id,))
return cur.rowcount > 0
def get_or_create_officer_by_email(email: str, name: str = "Officer") -> int:
"""
Finds the officer ID for a given email. If not found, auto-provisions:
1. A default school (if none exists)
2. A user record in 'users'
3. A 'placement_officers' record.
"""
with db_cursor() as cur:
# Check if officer already exists for this email
cur.execute("""
SELECT po.officer_id
FROM placement_officers po
JOIN "user" u ON u.id = po.user_id
WHERE u.email = %s
""", (email,))
row = cur.fetchone()
if row:
return row["officer_id"]
# 1. Ensure user exists
cur.execute('SELECT id as user_id FROM "user" WHERE email = %s', (email,))
user_row = cur.fetchone()
if user_row:
user_id = user_row["user_id"]
else:
import uuid
user_id = str(uuid.uuid4())
cur.execute("""
INSERT INTO "user" (id, name, email, "emailVerified", role, "createdAt", "updatedAt")
VALUES (%s, %s, %s, FALSE, 'officer', NOW(), NOW())
RETURNING id as user_id
""", (user_id, name, email))
user_id = cur.fetchone()["user_id"]
# 2. Create placement officer
cur.execute("""
INSERT INTO placement_officers (user_id, name)
VALUES (%s, %s)
RETURNING officer_id
""", (user_id, name))
return cur.fetchone()["officer_id"]
def create_student(name: str, usn: str, school_id: int, program_id: int, batch: str, is_active: bool = True) -> dict:
with db_cursor() as cur:
cur.execute("""
INSERT INTO students (name, usn, school_id, program_id, batch, is_active)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING *
""", (name, usn, school_id, program_id, batch, is_active))
return cur.fetchone()
def update_student(student_id: int, name: str, usn: str, school_id: int, program_id: int, batch: str, is_active: bool) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
UPDATE students
SET name = %s, usn = %s, school_id = %s, program_id = %s, batch = %s, is_active = %s
WHERE student_id = %s
RETURNING *
""", (name, usn, school_id, program_id, batch, is_active, student_id))
return cur.fetchone()
def delete_student(student_id: int) -> bool:
with db_cursor() as cur:
cur.execute("DELETE FROM students WHERE student_id = %s", (student_id,))
return cur.rowcount > 0
# ---------------------------------------------------------------------------
# Table: students
# Columns: student_id (PK), name, usn (UNIQUE), school_id (FK),
# program_id (FK), batch, is_active
# ---------------------------------------------------------------------------
def get_all_students(active_only: bool = True) -> list[dict]:
with db_cursor() as cur:
query = """
SELECT st.student_id, st.name, st.usn, st.batch, st.is_active,
st.school_id, sc.school_name,
st.program_id, pr.program_name, pr.credit_weightage,
COALESCE(
(SELECT json_agg(
json_build_object(
'placement_id', pl.placement_id,
'type', pl.placement_type,
'status', pl.offer_status,
'package', pl.package_lpa,
'month', pl.placement_month,
'year', pl.placement_year,
'company', c.company_name,
'drive_type', d.drive_type,
'is_self_placed', pl.is_self_placed,
'internship_stipend', pl.internship_stipend,
'officer_name', po.name,
'officer_id', pl.officer_id
)
) FROM placements pl
LEFT JOIN drives d ON d.drive_id = pl.drive_id
LEFT JOIN companies c ON c.company_id = d.company_id
LEFT JOIN placement_officers po ON po.officer_id = pl.officer_id
WHERE pl.student_id = st.student_id),
'[]'::json
) as placements
FROM students st
JOIN schools sc ON sc.school_id = st.school_id
JOIN programs pr ON pr.program_id = st.program_id
"""
if active_only:
query += " WHERE st.is_active = TRUE"
query += " ORDER BY st.name"
cur.execute(query)
return cur.fetchall()
def get_student_by_id(student_id: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT st.student_id, st.name, st.usn, st.batch, st.is_active,
st.school_id, sc.school_name,
st.program_id, pr.program_name, pr.credit_weightage
FROM students st
JOIN schools sc ON sc.school_id = st.school_id
JOIN programs pr ON pr.program_id = st.program_id
WHERE st.student_id = %s
""", (student_id,))
return cur.fetchone()
def get_student_by_usn(usn: str) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT st.student_id, st.name, st.usn, st.batch, st.is_active,
st.school_id, sc.school_name,
st.program_id, pr.program_name, pr.credit_weightage
FROM students st
JOIN schools sc ON sc.school_id = st.school_id
JOIN programs pr ON pr.program_id = st.program_id
WHERE st.usn = %s
""", (usn,))
return cur.fetchone()
def get_unplaced_students_for_officer(officer_id: int, month: int, year: int) -> int:
"""
Count active students under this officer's programs who have NOT
received an 'accepted' placement in the given month/year.
"""
with db_cursor() as cur:
cur.execute("""
SELECT COUNT(DISTINCT st.student_id)
FROM students st
JOIN program_officer_assignments poa ON poa.program_id = st.program_id
WHERE poa.officer_id = %s
AND st.is_active = TRUE
AND st.student_id NOT IN (
SELECT student_id FROM placements
WHERE offer_status = 'accepted'
AND placement_month = %s
AND placement_year = %s
)
""", (officer_id, month, year))
row = cur.fetchone()
return int(row["count"]) if row else 0
# ---------------------------------------------------------------------------
# Table: companies
# Columns: company_id (PK), company_name, industry
# ---------------------------------------------------------------------------
def get_all_companies() -> list[dict]:
with db_cursor() as cur:
cur.execute("SELECT company_id, company_name, industry, hr_details FROM companies ORDER BY company_name")
return cur.fetchall()
def get_company_by_id(company_id: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("SELECT company_id, company_name, industry, hr_details FROM companies WHERE company_id = %s", (company_id,))
return cur.fetchone()
def create_company(company_name: str, industry: Optional[str] = None, hr_details: Optional[dict] = None) -> dict:
with db_cursor() as cur:
cur.execute("""
INSERT INTO companies (company_name, industry, hr_details)
VALUES (%s, %s, %s)
RETURNING company_id, company_name, industry, hr_details
""", (company_name, industry, json.dumps(hr_details) if hr_details else None))
return cur.fetchone()
# ---------------------------------------------------------------------------
# Table: drives
# Columns: drive_id (PK), company_id (FK), drive_date, drive_type,
# min_package_lpa, max_package_lpa, is_rvce_drive
# ---------------------------------------------------------------------------
def get_all_drives(company_id: Optional[int] = None) -> list[dict]:
with db_cursor() as cur:
query = """
SELECT d.drive_id, d.company_id, c.company_name,
d.drive_date, d.drive_type,
d.min_package_lpa, d.max_package_lpa,
d.is_rvce_drive
FROM drives d
JOIN companies c ON c.company_id = d.company_id
"""
if company_id is not None:
cur.execute(query + " WHERE d.company_id = %s ORDER BY d.drive_date DESC", (company_id,))
else:
cur.execute(query + " ORDER BY d.drive_date DESC")
return cur.fetchall()
def get_drive_by_id(drive_id: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT d.drive_id, d.company_id, c.company_name,
d.drive_date, d.drive_type,
d.min_package_lpa, d.max_package_lpa,
d.is_rvce_drive
FROM drives d
JOIN companies c ON c.company_id = d.company_id
WHERE d.drive_id = %s
""", (drive_id,))
return cur.fetchone()
def create_drive(
company_id: int,
drive_type: str,
drive_date: Optional[datetime] = None,
min_package_lpa: Optional[float] = None,
max_package_lpa: Optional[float] = None,
is_rvce_drive: bool = False,
) -> dict:
if drive_type not in DRIVE_TYPE_ENUM:
raise ValueError(f"Invalid drive_type '{drive_type}'. Must be one of {DRIVE_TYPE_ENUM}")
with db_cursor() as cur:
cur.execute("""
INSERT INTO drives
(company_id, drive_date, drive_type, min_package_lpa, max_package_lpa, is_rvce_drive)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING drive_id, company_id, drive_date, drive_type,
min_package_lpa, max_package_lpa, is_rvce_drive
""", (company_id, drive_date, drive_type, min_package_lpa, max_package_lpa, is_rvce_drive))
return cur.fetchone()
# ---------------------------------------------------------------------------
# Table: placements
# Columns: placement_id (PK), student_id (FK), officer_id (FK), drive_id (FK),
# placement_type, package_lpa, offer_status, is_self_placed,
# placement_month, placement_year
# ---------------------------------------------------------------------------
def get_placements_by_officer(officer_id: int, is_admin: bool = False) -> list[dict]:
with db_cursor() as cur:
query = """
SELECT pl.placement_id, pl.offer_status, pl.placement_type,
pl.package_lpa, pl.is_self_placed,
pl.placement_month, pl.placement_year,
st.name AS student_name, st.usn,
pr.program_name,
co.company_name,
dr.drive_type,
po.name AS officer_name
FROM placements pl
JOIN students st ON st.student_id = pl.student_id
JOIN programs pr ON pr.program_id = st.program_id
LEFT JOIN drives dr ON dr.drive_id = pl.drive_id
LEFT JOIN companies co ON co.company_id = dr.company_id
LEFT JOIN placement_officers po ON po.officer_id = pl.officer_id
"""
if is_admin:
cur.execute(query + " ORDER BY pl.placement_year DESC, pl.placement_month DESC")
else:
cur.execute(query + " WHERE pl.officer_id = %s ORDER BY pl.placement_year DESC, pl.placement_month DESC", (officer_id,))
return cur.fetchall()
def update_placement(
placement_id: int,
package_lpa: Optional[float] = None,
internship_stipend: Optional[float] = None,
placement_type: str = "full_time",
offer_status: str = "accepted",
officer_id: Optional[int] = None
) -> Optional[dict]:
with db_cursor() as cur:
# Build dynamic query to only update provided fields if needed,
# but here we update all provided fields based on the API schema.
cur.execute("""
UPDATE placements
SET package_lpa = %s,
internship_stipend = %s,
placement_type = %s,
offer_status = %s,
officer_id = COALESCE(%s, officer_id)
WHERE placement_id = %s
RETURNING *
""", (package_lpa, internship_stipend, placement_type, offer_status, officer_id, placement_id))
return cur.fetchone()
def create_placement(
student_id: int,
officer_id: int,
placement_type: str,
offer_status: str,
placement_month: int,
placement_year: int,
drive_id: Optional[int] = None,
package_lpa: Optional[float] = None,
internship_stipend: Optional[float] = None,
is_self_placed: bool = False,
) -> dict:
if placement_type not in PLACEMENT_TYPE_ENUM:
raise ValueError(f"Invalid placement_type '{placement_type}'.")
if offer_status not in OFFER_STATUS_ENUM:
raise ValueError(f"Invalid offer_status '{offer_status}'.")
with db_cursor() as cur:
cur.execute("""
INSERT INTO placements
(student_id, officer_id, drive_id, placement_type,
package_lpa, internship_stipend, offer_status, is_self_placed,
placement_month, placement_year)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
RETURNING *
""", (
student_id, officer_id, drive_id, placement_type,
package_lpa, internship_stipend, offer_status, is_self_placed,
placement_month, placement_year,
))
return cur.fetchone()
def count_placements_for_officer(officer_id: int, month: int, year: int) -> int:
"""Count accepted placements for an officer in a given month/year."""
with db_cursor() as cur:
cur.execute("""
SELECT COUNT(*) FROM placements
WHERE officer_id = %s
AND offer_status = 'accepted'
AND placement_month = %s
AND placement_year = %s
""", (officer_id, month, year))
return int(cur.fetchone()["count"])
# ---------------------------------------------------------------------------
# Table: officer_monthly_snapshot
# Columns: snapshot_id (PK), officer_id (FK), month, year,
# starting_pool, target, placed, prism_credits, prism_score
# ---------------------------------------------------------------------------
def get_snapshot(officer_id: int, month: int, year: int) -> Optional[dict]:
with db_cursor() as cur:
cur.execute("""
SELECT * FROM officer_monthly_snapshot
WHERE officer_id = %s AND month = %s AND year = %s
""", (officer_id, month, year))
return cur.fetchone()
def upsert_snapshot(
officer_id: int,
month: int,
year: int,
starting_pool: int,
target: float,
placed: int,
prism_credits: float,
prism_score: float,
) -> dict:
"""Insert or update the officer's monthly snapshot."""
with db_cursor() as cur:
cur.execute("""
INSERT INTO officer_monthly_snapshot
(officer_id, month, year, starting_pool, target, placed, prism_credits, prism_score)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (officer_id, month, year)
DO UPDATE SET
starting_pool = EXCLUDED.starting_pool,
target = EXCLUDED.target,
placed = EXCLUDED.placed,
prism_credits = EXCLUDED.prism_credits,
prism_score = EXCLUDED.prism_score
RETURNING *
""", (officer_id, month, year, starting_pool, target, placed, prism_credits, prism_score))
return cur.fetchone()
def get_officer_history(officer_id: int, year: int) -> list[dict]:
"""Fetch all monthly snapshots for a given officer and year, ordered by month."""
with db_cursor() as cur:
cur.execute("""
SELECT month, year, starting_pool, target, placed, prism_credits, prism_score
FROM officer_monthly_snapshot
WHERE officer_id = %s AND year = %s
ORDER BY month
""", (officer_id, year))
return cur.fetchall()