150 lines
4.7 KiB
Python
150 lines
4.7 KiB
Python
"""
|
|
app/enrollment/storage.py
|
|
-------------------------
|
|
Flat-file persistence for enrollment submissions — one JSON document per
|
|
submission, in the directory named by config ENROLLMENT_DIR.
|
|
|
|
Why files and not a table
|
|
-------------------------
|
|
Enrollment happens BEFORE anything exists in the system: there is no contract,
|
|
no facility and no user account to key a row against, and the volume is a
|
|
handful of documents a year. A directory of readable JSON keeps this feature
|
|
completely outside the schema — no model, no migration, nothing to keep in sync
|
|
with the rest of the app. It can be backed up with `cp` and read with `cat`.
|
|
|
|
File naming
|
|
-----------
|
|
<YYYYmmdd-HHMMSS>-<8 hex>.json
|
|
|
|
Time-ordered so a plain directory listing sorts chronologically, with random
|
|
suffix so two submissions in the same second cannot collide. The stem is the
|
|
submission's id and is the ONLY thing the admin URLs accept — see _safe_id().
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import secrets
|
|
import tempfile
|
|
from datetime import datetime
|
|
|
|
from flask import current_app
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: Submission ids are generated by us and must round-trip through a URL and a
|
|
#: file path. Anything not matching is rejected before touching the filesystem,
|
|
#: so a crafted id can never escape the enrollment directory (path traversal).
|
|
_ID_RE = re.compile(r'^\d{8}-\d{6}-[0-9a-f]{8}$')
|
|
|
|
|
|
def enrollment_dir():
|
|
"""Absolute path of the submission directory, created on first use."""
|
|
path = current_app.config['ENROLLMENT_DIR']
|
|
os.makedirs(path, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def new_id(when=None):
|
|
"""Mint a time-ordered, collision-safe submission id."""
|
|
when = when or datetime.now()
|
|
return f'{when:%Y%m%d-%H%M%S}-{secrets.token_hex(4)}'
|
|
|
|
|
|
def _safe_id(submission_id):
|
|
"""Return the id if it is one of ours, else None.
|
|
|
|
Never interpolate an unvalidated id into a path — `../../etc/passwd` and
|
|
friends. Callers should 404 on None.
|
|
"""
|
|
if not submission_id or not _ID_RE.match(submission_id):
|
|
logger.warning('ENROLLMENT | rejected malformed id=%r', submission_id)
|
|
return None
|
|
return submission_id
|
|
|
|
|
|
def _path_for(submission_id):
|
|
sid = _safe_id(submission_id)
|
|
if sid is None:
|
|
return None
|
|
return os.path.join(enrollment_dir(), f'{sid}.json')
|
|
|
|
|
|
def save(record):
|
|
"""Write a submission atomically. Returns the id.
|
|
|
|
Written to a temp file in the same directory then os.replace()d, so a
|
|
crash mid-write can never leave a truncated JSON document that would break
|
|
the admin list for every other submission.
|
|
"""
|
|
sid = record['id']
|
|
path = _path_for(sid)
|
|
if path is None:
|
|
raise ValueError(f'refusing to save malformed id {sid!r}')
|
|
|
|
directory = os.path.dirname(path)
|
|
fd, tmp = tempfile.mkstemp(dir=directory, suffix='.tmp')
|
|
try:
|
|
with os.fdopen(fd, 'w', encoding='utf-8') as fh:
|
|
json.dump(record, fh, indent=2, ensure_ascii=False)
|
|
os.replace(tmp, path)
|
|
except Exception:
|
|
# Never leave the temp file behind on a failed write.
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
logger.info('ENROLLMENT | saved | id=%s project=%r',
|
|
sid, record.get('project_name'))
|
|
return sid
|
|
|
|
|
|
def load(submission_id):
|
|
"""Return one submission dict, or None if unknown/unreadable."""
|
|
path = _path_for(submission_id)
|
|
if path is None or not os.path.isfile(path):
|
|
return None
|
|
try:
|
|
with open(path, encoding='utf-8') as fh:
|
|
return json.load(fh)
|
|
except (OSError, ValueError):
|
|
logger.exception('ENROLLMENT | unreadable submission id=%s', submission_id)
|
|
return None
|
|
|
|
|
|
def load_all():
|
|
"""Return every submission, newest first.
|
|
|
|
A single corrupt file is skipped with a log line rather than breaking the
|
|
whole admin list.
|
|
"""
|
|
directory = enrollment_dir()
|
|
records = []
|
|
for name in sorted(os.listdir(directory), reverse=True):
|
|
if not name.endswith('.json'):
|
|
continue
|
|
rec = load(name[:-len('.json')])
|
|
if rec is not None:
|
|
records.append(rec)
|
|
return records
|
|
|
|
|
|
def update_office(submission_id, office, status):
|
|
"""Merge the office-use block + status into a stored submission.
|
|
|
|
Returns the updated record, or None if the id is unknown. Only these
|
|
fields are writable after submission — the customer's own answers are
|
|
immutable, so the file stays an accurate record of what they asked for.
|
|
"""
|
|
rec = load(submission_id)
|
|
if rec is None:
|
|
return None
|
|
rec.setdefault('office', {}).update(office)
|
|
rec['status'] = status
|
|
rec['updated_at'] = datetime.now().isoformat(timespec='seconds')
|
|
save(rec)
|
|
return rec
|