feat(seed): markable cohort for exam-marker (9P/Ph1)
Adds seed_cohort_9p_ph1.py — creates N student accounts (default 10) and enrols
them all into one class (Greenfield Year 9 Physics 9P/Ph1) so there is a real
cohort to mark. The canonical timetable seeds enrol one student per year-band,
leaving every class with <=1 student.
Uses the same paths as the canonical seeds (auth admin create user, profiles +
institute_memberships upsert, POST /database/timetable/classes/{id}/students as
school admin). Idempotent. Self-contained HTTP (runs inside the ccapi container).
Verified on dev .94: 10 created + enrolled, 0 errors; 9P/Ph1 roster = 11;
physics teacher sees all 11 under as-user RLS.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
CC Worker
parent
4b296cff74
commit
0ce654c6c6
218
run/initialization/seed_cohort_9p_ph1.py
Normal file
218
run/initialization/seed_cohort_9p_ph1.py
Normal file
@ -0,0 +1,218 @@
|
||||
"""
|
||||
seed_cohort_9p_ph1.py — Markable cohort for exam-marker testing.
|
||||
|
||||
Creates N student accounts and enrols them ALL into a single class (default the
|
||||
Greenfield Year 9 Physics class `9P/Ph1`), so there is a real cohort to mark.
|
||||
|
||||
Why: the canonical timetable seeds enrol "one student per year-group band"
|
||||
(seed_greenfield_timetable.py), so every class has <=1 student — too few for a
|
||||
results table / per-question stats. This seeder fills one class to a usable size.
|
||||
|
||||
Mechanics (identical paths to the canonical seeds — nothing bespoke server-side):
|
||||
- auth user: POST {SUPABASE_URL}/auth/v1/admin/users
|
||||
- profile: upsert public.profiles (school_id = institute)
|
||||
- membership: upsert public.institute_memberships (role 'student')
|
||||
- enrolment: POST {API_BASE_URL}/database/timetable/classes/{class_id}/students
|
||||
(as a school_admin; class_students upsert → idempotent)
|
||||
|
||||
Idempotent: re-running skips existing auth users and upserts everything else.
|
||||
|
||||
Env required: SUPABASE_URL, SERVICE_ROLE_KEY (API_BASE_URL defaults to api-dev)
|
||||
Optional env: COHORT_COUNT, COHORT_CLASS_CODE, COHORT_INSTITUTE_ID,
|
||||
SEED_STUDENT_PASSWORD, SEED_SCHOOL_ADMIN_PASSWORD
|
||||
|
||||
Run (dev):
|
||||
SUPABASE_URL=... SERVICE_ROLE_KEY=... API_BASE_URL=http://192.168.0.64:18000 \
|
||||
python3 -c "from run.initialization.seed_cohort_9p_ph1 import seed; seed()"
|
||||
"""
|
||||
import os
|
||||
import time
|
||||
import requests
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
# Greenfield Academy (the school actually populated on dev .94)
|
||||
GREENFIELD_ID = os.getenv("COHORT_INSTITUTE_ID", "a1b2c3d4-e5f6-7890-abcd-ef1234567890")
|
||||
GREENFIELD_DOMAIN = "greenfieldacademy.test"
|
||||
GREENFIELD_ADMIN_EMAIL = f"admin@{GREENFIELD_DOMAIN}"
|
||||
|
||||
CLASS_CODE = os.getenv("COHORT_CLASS_CODE", "9P/Ph1")
|
||||
COHORT_COUNT = int(os.getenv("COHORT_COUNT", "10"))
|
||||
|
||||
# Realistic-ish names so the results table doesn't read "Pupil 01..10".
|
||||
COHORT_NAMES = [
|
||||
("Amelia", "Clarke"), ("Noah", "Bennett"), ("Olivia", "Foster"), ("Leo", "Hughes"),
|
||||
("Ava", "Patel"), ("Jacob", "Reid"), ("Mia", "Turner"), ("Harry", "Ellis"),
|
||||
("Isla", "Morgan"), ("Oscar", "Khan"), ("Freya", "Walsh"), ("Theo", "Ndlovu"),
|
||||
]
|
||||
|
||||
DEFAULT_STUDENT_PASSWORD = "Student@Cc2025!"
|
||||
DEFAULT_SCHOOL_ADMIN_PASSWORD = "Admin@Cc2025!"
|
||||
|
||||
|
||||
def _ctx() -> Dict[str, str]:
|
||||
return {
|
||||
"supa_url": os.environ["SUPABASE_URL"].rstrip("/"),
|
||||
"service_key": os.environ["SERVICE_ROLE_KEY"],
|
||||
"api_base": os.environ.get("API_BASE_URL", "http://192.168.0.64:18000").rstrip("/"),
|
||||
"student_pw": os.getenv("SEED_STUDENT_PASSWORD", DEFAULT_STUDENT_PASSWORD),
|
||||
"admin_pw": os.getenv("SEED_SCHOOL_ADMIN_PASSWORD", DEFAULT_SCHOOL_ADMIN_PASSWORD),
|
||||
}
|
||||
|
||||
|
||||
def _sb_headers(ctx: Dict[str, str]) -> Dict[str, str]:
|
||||
return {
|
||||
"apikey": ctx["service_key"],
|
||||
"Authorization": f"Bearer {ctx['service_key']}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def _sign_in(ctx: Dict[str, str], email: str, password: str) -> str:
|
||||
r = requests.post(
|
||||
f"{ctx['supa_url']}/auth/v1/token?grant_type=password",
|
||||
headers={"apikey": ctx["service_key"], "Content-Type": "application/json"},
|
||||
json={"email": email, "password": password},
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()["access_token"]
|
||||
|
||||
|
||||
def _resolve_class_id(ctx: Dict[str, str]) -> Optional[str]:
|
||||
r = requests.get(
|
||||
f"{ctx['supa_url']}/rest/v1/classes",
|
||||
headers=_sb_headers(ctx),
|
||||
params={"class_code": f"eq.{CLASS_CODE}",
|
||||
"institute_id": f"eq.{GREENFIELD_ID}",
|
||||
"select": "id,name", "limit": "1"},
|
||||
)
|
||||
data = r.json() if r.ok else []
|
||||
return data[0]["id"] if data else None
|
||||
|
||||
|
||||
def _existing_auth_users(ctx: Dict[str, str]) -> Dict[str, str]:
|
||||
r = requests.get(
|
||||
f"{ctx['supa_url']}/auth/v1/admin/users",
|
||||
headers=_sb_headers(ctx), params={"per_page": 200},
|
||||
)
|
||||
r.raise_for_status()
|
||||
return {u["email"]: u["id"] for u in r.json().get("users", [])}
|
||||
|
||||
|
||||
def _create_auth_user(ctx: Dict[str, str], spec: Dict) -> Optional[str]:
|
||||
r = requests.post(
|
||||
f"{ctx['supa_url']}/auth/v1/admin/users",
|
||||
headers=_sb_headers(ctx),
|
||||
json={
|
||||
"email": spec["email"], "password": ctx["student_pw"], "email_confirm": True,
|
||||
"user_metadata": {
|
||||
"username": spec["username"], "full_name": spec["full_name"],
|
||||
"display_name": spec["display_name"], "user_type": "student",
|
||||
},
|
||||
},
|
||||
)
|
||||
if r.status_code in (200, 201):
|
||||
return r.json()["id"]
|
||||
return None
|
||||
|
||||
|
||||
def _upsert(ctx: Dict[str, str], table: str, row: Dict, on_conflict: str) -> bool:
|
||||
h = {**_sb_headers(ctx), "Prefer": "resolution=merge-duplicates,return=minimal"}
|
||||
r = requests.post(f"{ctx['supa_url']}/rest/v1/{table}",
|
||||
headers=h, json=row, params={"on_conflict": on_conflict})
|
||||
return r.ok
|
||||
|
||||
|
||||
def _cohort_specs() -> List[Dict]:
|
||||
specs = []
|
||||
for i in range(1, COHORT_COUNT + 1):
|
||||
first, last = COHORT_NAMES[(i - 1) % len(COHORT_NAMES)]
|
||||
prefix = f"cohort{i:02d}"
|
||||
specs.append({
|
||||
"email": f"{prefix}@{GREENFIELD_DOMAIN}",
|
||||
"username": f"{prefix}.{GREENFIELD_DOMAIN.replace('.', '_')}",
|
||||
"full_name": f"{first} {last}",
|
||||
"display_name": first,
|
||||
})
|
||||
return specs
|
||||
|
||||
|
||||
def seed(count: Optional[int] = None) -> Dict[str, Any]:
|
||||
global COHORT_COUNT
|
||||
if count is not None:
|
||||
COHORT_COUNT = count
|
||||
ctx = _ctx()
|
||||
results: Dict[str, Any] = {"class_code": CLASS_CODE, "requested": COHORT_COUNT,
|
||||
"created": 0, "reused": 0, "enrolled": 0, "errors": []}
|
||||
|
||||
print(f"COHORT SEED → {CLASS_CODE} @ {GREENFIELD_DOMAIN} (target {COHORT_COUNT} students)")
|
||||
|
||||
class_id = _resolve_class_id(ctx)
|
||||
if not class_id:
|
||||
results["errors"].append(f"class {CLASS_CODE} not found for institute {GREENFIELD_ID}")
|
||||
print(f" ✗ {results['errors'][-1]}")
|
||||
return results
|
||||
print(f" class_id = {class_id}")
|
||||
|
||||
existing = _existing_auth_users(ctx)
|
||||
specs = _cohort_specs()
|
||||
|
||||
# 1) accounts: auth user + profile + membership
|
||||
uids: Dict[str, str] = {}
|
||||
for spec in specs:
|
||||
email = spec["email"]
|
||||
uid = existing.get(email)
|
||||
if uid:
|
||||
results["reused"] += 1
|
||||
else:
|
||||
uid = _create_auth_user(ctx, spec)
|
||||
if not uid:
|
||||
results["errors"].append(f"create auth user {email}")
|
||||
print(f" ✗ create {email}")
|
||||
continue
|
||||
results["created"] += 1
|
||||
time.sleep(0.15)
|
||||
uids[email] = uid
|
||||
ok_p = _upsert(ctx, "profiles", {
|
||||
"id": uid, "email": email, "user_type": "student",
|
||||
"username": spec["username"], "full_name": spec["full_name"],
|
||||
"display_name": spec["display_name"], "school_id": GREENFIELD_ID,
|
||||
"neo4j_sync_status": "pending",
|
||||
}, on_conflict="id")
|
||||
ok_m = _upsert(ctx, "institute_memberships", {
|
||||
"profile_id": uid, "institute_id": GREENFIELD_ID, "role": "student", "metadata": {},
|
||||
}, on_conflict="profile_id,institute_id")
|
||||
if not (ok_p and ok_m):
|
||||
results["errors"].append(f"profile/membership {email} (p={ok_p} m={ok_m})")
|
||||
|
||||
# 2) enrol all into the class (via API, as school admin)
|
||||
admin_token = _sign_in(ctx, GREENFIELD_ADMIN_EMAIL, ctx["admin_pw"])
|
||||
for spec in specs:
|
||||
uid = uids.get(spec["email"])
|
||||
if not uid:
|
||||
continue
|
||||
r = requests.post(
|
||||
f"{ctx['api_base']}/database/timetable/classes/{class_id}/students",
|
||||
headers={"Authorization": f"Bearer {admin_token}", "Content-Type": "application/json"},
|
||||
json={"student_id": uid},
|
||||
)
|
||||
body = {}
|
||||
try:
|
||||
body = r.json()
|
||||
except Exception:
|
||||
pass
|
||||
if r.ok and (body.get("status") == "ok" or body.get("row") or body.get("id")):
|
||||
results["enrolled"] += 1
|
||||
print(f" ✓ {spec['email'].split('@')[0]} → {CLASS_CODE}")
|
||||
else:
|
||||
results["errors"].append(f"enrol {spec['email']}: {r.status_code} {str(body)[:120]}")
|
||||
print(f" ✗ enrol {spec['email']}: {r.status_code}")
|
||||
time.sleep(0.1)
|
||||
|
||||
print(f"\nDONE: created {results['created']}, reused {results['reused']}, "
|
||||
f"enrolled {results['enrolled']}/{COHORT_COUNT}, errors {len(results['errors'])}")
|
||||
return results
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import json
|
||||
print(json.dumps(seed(), indent=2, default=str))
|
||||
Loading…
x
Reference in New Issue
Block a user