LIMBFORGE AUTHORITATIVE SCHEMA AND API MAP ============================================================================ SERVER DATABASE CREATION / MIGRATION CONTEXT ---------------------------------------------------------------------------- 00018: ROOT = Path(__file__).resolve().parent 00019: STATIC = ROOT / "static" 00020: DATA = ROOT / "data" 00021: DATABASE = DATA / "limbforge.sqlite3" 00022: 00023: DATABASE_LOCK = threading.Lock() 00024: 00025: MEASUREMENTS = [ 00026: { 00027: "key": "cosmetic_envelope_length", 00028: "name": "Cosmetic envelope length", 00029: "instruction": ( 00030: "Measure between the specialist-agreed proximal reference " 00031: "and the intended distal end of the cosmetic cover." 00032: ), ... 00142: "focus": "oi-opening", 00143: "oi_only": True, 00144: }, 00145: ] 00146: 00147: MEASUREMENT_BY_KEY = { 00148: item["key"]: item 00149: for item in MEASUREMENTS 00150: } 00151: 00152: 00153: def utc_now() -> str: 00154: return datetime.now(timezone.utc).isoformat(timespec="seconds") 00155: 00156: 00157: def connect() -> sqlite3.Connection: 00158: connection = sqlite3.connect(DATABASE) 00159: connection.row_factory = sqlite3.Row 00160: connection.execute("PRAGMA foreign_keys = ON") 00161: return connection 00162: 00163: 00164: def initialise_database() -> None: 00165: DATA.mkdir(parents=True, exist_ok=True) 00166: 00167: with DATABASE_LOCK, connect() as database: 00168: database.executescript( 00169: """ 00170: CREATE TABLE IF NOT EXISTS projects ( 00171: id TEXT PRIMARY KEY, 00172: name TEXT NOT NULL, 00173: reference_id TEXT NOT NULL DEFAULT '', 00174: side TEXT NOT NULL, 00175: cover_class TEXT NOT NULL, 00176: osseointegration INTEGER NOT NULL DEFAULT 0, 00177: specialist_name TEXT NOT NULL DEFAULT '', ... 00179: exclusion_zones TEXT NOT NULL DEFAULT '', 00180: component_notes TEXT NOT NULL DEFAULT '', 00181: review_notes TEXT NOT NULL DEFAULT '', 00182: created_at TEXT NOT NULL, 00183: updated_at TEXT NOT NULL 00184: ); 00185: 00186: CREATE TABLE IF NOT EXISTS measurements ( 00187: project_id TEXT NOT NULL, 00188: measurement_key TEXT NOT NULL, 00189: value_mm REAL NOT NULL, 00190: notes TEXT NOT NULL DEFAULT '', 00191: status TEXT NOT NULL DEFAULT 'completed', 00192: updated_at TEXT NOT NULL, 00193: PRIMARY KEY (project_id, measurement_key), 00194: FOREIGN KEY (project_id) 00195: REFERENCES projects(id) 00196: ON DELETE CASCADE 00197: ); 00198: """ 00199: ) 00200: 00201: 00202: def row_to_project( 00203: row: sqlite3.Row, 00204: measurements: list[sqlite3.Row] | None = None, 00205: ) -> dict: 00206: project = dict(row) 00207: project["osseointegration"] = bool(project["osseointegration"]) 00208: 00209: if measurements is not None: 00210: project["measurements"] = { 00211: item["measurement_key"]: { 00212: "value_mm": item["value_mm"], 00213: "notes": item["notes"], 00214: "status": item["status"], 00215: "updated_at": item["updated_at"], 00216: } 00217: for item in measurements 00218: } 00219: 00220: return project 00221: 00222: 00223: def list_projects() -> list[dict]: 00224: with DATABASE_LOCK, connect() as database: 00225: rows = database.execute( 00226: """ 00227: SELECT * 00228: FROM projects 00229: ORDER BY updated_at DESC 00230: """ 00231: ).fetchall() 00232: 00233: return [row_to_project(row) for row in rows] 00234: 00235: 00236: def load_project(project_id: str) -> dict | None: 00237: with DATABASE_LOCK, connect() as database: 00238: project = database.execute( 00239: "SELECT * FROM projects WHERE id = ?", 00240: (project_id,), 00241: ).fetchone() 00242: 00243: if project is None: 00244: return None 00245: 00246: measurements = database.execute( 00247: """ 00248: SELECT * 00249: FROM measurements 00250: WHERE project_id = ? 00251: ORDER BY updated_at 00252: """, 00253: (project_id,), 00254: ).fetchall() 00255: 00256: return row_to_project(project, measurements) 00257: 00258: 00259: def create_project(payload: dict) -> dict: 00260: project_id = str(uuid.uuid4()) 00261: timestamp = utc_now() 00262: 00263: name = str(payload.get("name", "")).strip() ... 00274: raise ValueError("Cover class is required.") 00275: 00276: oi = bool(payload.get("osseointegration", False)) 00277: 00278: with DATABASE_LOCK, connect() as database: 00279: database.execute( 00280: """ 00281: INSERT INTO projects ( 00282: id, 00283: name, 00284: reference_id, 00285: side, 00286: cover_class, 00287: osseointegration, 00288: specialist_name, ... 00341: ) 00342: 00343: timestamp = utc_now() 00344: 00345: with DATABASE_LOCK, connect() as database: 00346: database.execute( 00347: """ 00348: UPDATE projects 00349: SET 00350: name = ?, 00351: reference_id = ?, 00352: side = ?, 00353: cover_class = ?, 00354: osseointegration = ?, 00355: specialist_name = ?, ... 00423: 00424: if definition is None: 00425: raise ValueError("Unknown measurement.") 00426: 00427: if definition["oi_only"] and not project["osseointegration"]: 00428: raise ValueError( 00429: "This measurement is available only for " 00430: "osseointegration projects." 00431: ) 00432: 00433: try: 00434: value = float(payload.get("value_mm")) 00435: except (TypeError, ValueError): 00436: raise ValueError("Enter a valid measurement in millimetres.") 00437: ... 00446: raise ValueError("Invalid measurement status.") 00447: 00448: timestamp = utc_now() 00449: 00450: with DATABASE_LOCK, connect() as database: 00451: database.execute( 00452: """ 00453: INSERT INTO measurements ( 00454: project_id, 00455: measurement_key, 00456: value_mm, 00457: notes, 00458: status, 00459: updated_at 00460: ) ... 00474: status, 00475: timestamp, 00476: ), 00477: ) 00478: 00479: database.execute( 00480: """ 00481: UPDATE projects 00482: SET updated_at = ? 00483: WHERE id = ? 00484: """, 00485: (timestamp, project_id), 00486: ) 00487: 00488: return load_project(project_id) ... 00499: 00500: if measurement_key not in MEASUREMENT_BY_KEY: 00501: raise ValueError("Unknown measurement.") 00502: 00503: with DATABASE_LOCK, connect() as database: 00504: database.execute( 00505: """ 00506: DELETE FROM measurements 00507: WHERE project_id = ? 00508: AND measurement_key = ? 00509: """, 00510: (project_id, measurement_key), 00511: ) 00512: 00513: database.execute( 00514: """ 00515: UPDATE projects 00516: SET updated_at = ? 00517: WHERE id = ? 00518: """, 00519: (utc_now(), project_id), 00520: ) 00521: 00522: return load_project(project_id) ... 00596: "service": "limbforge-shared-workspace", 00597: "database": str(DATABASE), 00598: } 00599: ) 00600: return 00601: 00602: if path == "/api/definitions": 00603: self.send_json(MEASUREMENTS) 00604: return 00605: 00606: if path == "/api/projects": 00607: self.send_json(list_projects()) 00608: return 00609: 00610: parts = [ 00611: unquote(part) 00612: for part in path.strip("/").split("/") 00613: if part 00614: ] 00615: 00616: if ( 00617: len(parts) == 3 00618: and parts[0] == "api" 00619: and parts[1] == "projects" 00620: ): 00621: project = load_project(parts[2]) 00622: 00623: if project is None: 00624: raise LookupError("Project not found.") 00625: 00626: self.send_json(project) ... 00631: except Exception as error: 00632: self.handle_api_error(error) 00633: 00634: def do_POST(self) -> None: 00635: path = urlparse(self.path).path 00636: 00637: try: 00638: if path == "/api/projects": 00639: self.send_json( 00640: create_project(self.read_json()), 00641: HTTPStatus.CREATED, 00642: ) 00643: return 00644: 00645: self.send_json( ... 00658: if part 00659: ] 00660: 00661: try: 00662: if ( 00663: len(parts) == 3 00664: and parts[0] == "api" 00665: and parts[1] == "projects" 00666: ): 00667: self.send_json( 00668: update_project( 00669: parts[2], 00670: self.read_json(), 00671: ) 00672: ) 00673: return 00674: 00675: if ( 00676: len(parts) == 5 00677: and parts[0] == "api" 00678: and parts[1] == "projects" 00679: and parts[3] == "measurements" 00680: ): 00681: self.send_json( 00682: save_measurement( 00683: parts[2], 00684: parts[4], 00685: self.read_json(), 00686: ) ... 00703: if part 00704: ] 00705: 00706: try: 00707: if ( 00708: len(parts) == 5 00709: and parts[0] == "api" 00710: and parts[1] == "projects" 00711: and parts[3] == "measurements" 00712: ): 00713: self.send_json( 00714: delete_measurement( 00715: parts[2], 00716: parts[4], 00717: ) 00718: ) SERVER PROJECT AND MEASUREMENT API CONTEXT ---------------------------------------------------------------------------- 00015: from pathlib import Path 00016: from urllib.parse import unquote, urlparse 00017: 00018: ROOT = Path(__file__).resolve().parent 00019: STATIC = ROOT / "static" 00020: DATA = ROOT / "data" 00021: DATABASE = DATA / "limbforge.sqlite3" 00022: 00023: DATABASE_LOCK = threading.Lock() 00024: 00025: MEASUREMENTS = [ 00026: { 00027: "key": "cosmetic_envelope_length", 00028: "name": "Cosmetic envelope length", 00029: "instruction": ( 00030: "Measure between the specialist-agreed proximal reference " 00031: "and the intended distal end of the cosmetic cover." 00032: ), 00033: "guide": "lf_guide_length.png", 00034: "focus": "full", 00035: "oi_only": False, ... 00137: "instruction": ( 00138: "Record the confirmed width of the cleaning, inspection and " 00139: "emergency-access opening." 00140: ), 00141: "guide": "lf_guide_oi.png", 00142: "focus": "oi-opening", 00143: "oi_only": True, 00144: }, 00145: ] 00146: 00147: MEASUREMENT_BY_KEY = { 00148: item["key"]: item 00149: for item in MEASUREMENTS 00150: } 00151: 00152: 00153: def utc_now() -> str: 00154: return datetime.now(timezone.utc).isoformat(timespec="seconds") 00155: 00156: 00157: def connect() -> sqlite3.Connection: 00158: connection = sqlite3.connect(DATABASE) 00159: connection.row_factory = sqlite3.Row ... 00166: 00167: with DATABASE_LOCK, connect() as database: 00168: database.executescript( 00169: """ 00170: CREATE TABLE IF NOT EXISTS projects ( 00171: id TEXT PRIMARY KEY, 00172: name TEXT NOT NULL, 00173: reference_id TEXT NOT NULL DEFAULT '', 00174: side TEXT NOT NULL, 00175: cover_class TEXT NOT NULL, 00176: osseointegration INTEGER NOT NULL DEFAULT 0, 00177: specialist_name TEXT NOT NULL DEFAULT '', 00178: clinic_name TEXT NOT NULL DEFAULT '', 00179: exclusion_zones TEXT NOT NULL DEFAULT '', 00180: component_notes TEXT NOT NULL DEFAULT '', 00181: review_notes TEXT NOT NULL DEFAULT '', 00182: created_at TEXT NOT NULL, 00183: updated_at TEXT NOT NULL 00184: ); 00185: 00186: CREATE TABLE IF NOT EXISTS measurements ( 00187: project_id TEXT NOT NULL, 00188: measurement_key TEXT NOT NULL, 00189: value_mm REAL NOT NULL, 00190: notes TEXT NOT NULL DEFAULT '', 00191: status TEXT NOT NULL DEFAULT 'completed', 00192: updated_at TEXT NOT NULL, 00193: PRIMARY KEY (project_id, measurement_key), 00194: FOREIGN KEY (project_id) 00195: REFERENCES projects(id) 00196: ON DELETE CASCADE 00197: ); 00198: """ 00199: ) 00200: 00201: 00202: def row_to_project( 00203: row: sqlite3.Row, 00204: measurements: list[sqlite3.Row] | None = None, 00205: ) -> dict: 00206: project = dict(row) 00207: project["osseointegration"] = bool(project["osseointegration"]) 00208: 00209: if measurements is not None: 00210: project["measurements"] = { 00211: item["measurement_key"]: { 00212: "value_mm": item["value_mm"], 00213: "notes": item["notes"], 00214: "status": item["status"], 00215: "updated_at": item["updated_at"], 00216: } 00217: for item in measurements 00218: } 00219: 00220: return project 00221: 00222: 00223: def list_projects() -> list[dict]: 00224: with DATABASE_LOCK, connect() as database: 00225: rows = database.execute( 00226: """ 00227: SELECT * ... 00236: def load_project(project_id: str) -> dict | None: 00237: with DATABASE_LOCK, connect() as database: 00238: project = database.execute( 00239: "SELECT * FROM projects WHERE id = ?", 00240: (project_id,), 00241: ).fetchone() 00242: 00243: if project is None: 00244: return None 00245: 00246: measurements = database.execute( 00247: """ 00248: SELECT * 00249: FROM measurements 00250: WHERE project_id = ? 00251: ORDER BY updated_at 00252: """, 00253: (project_id,), 00254: ).fetchall() 00255: 00256: return row_to_project(project, measurements) 00257: 00258: 00259: def create_project(payload: dict) -> dict: 00260: project_id = str(uuid.uuid4()) 00261: timestamp = utc_now() 00262: 00263: name = str(payload.get("name", "")).strip() 00264: side = str(payload.get("side", "")).strip() 00265: cover_class = str(payload.get("cover_class", "")).strip() 00266: 00267: if not name: 00268: raise ValueError("Project name is required.") 00269: 00270: if side not in {"Right", "Left", "Bilateral"}: 00271: raise ValueError("Select Right, Left or Bilateral.") 00272: 00273: if not cover_class: 00274: raise ValueError("Cover class is required.") 00275: 00276: oi = bool(payload.get("osseointegration", False)) 00277: 00278: with DATABASE_LOCK, connect() as database: 00279: database.execute( 00280: """ 00281: INSERT INTO projects ( 00282: id, 00283: name, 00284: reference_id, 00285: side, 00286: cover_class, 00287: osseointegration, 00288: specialist_name, 00289: clinic_name, 00290: exclusion_zones, 00291: component_notes, 00292: review_notes, 00293: created_at, 00294: updated_at 00295: ) 00296: VALUES (?, ?, ?, ?, ?, ?, '', '', '', '', '', ?, ?) 00297: """, 00298: ( 00299: project_id, 00300: name, 00301: str(payload.get("reference_id", "")).strip(), 00302: side, ... 00328: 00329: cover_class = str( 00330: payload.get("cover_class", existing["cover_class"]) 00331: ).strip() 00332: 00333: if not cover_class: 00334: raise ValueError("Cover class is required.") 00335: 00336: oi = bool( 00337: payload.get( 00338: "osseointegration", 00339: existing["osseointegration"], 00340: ) 00341: ) 00342: 00343: timestamp = utc_now() 00344: 00345: with DATABASE_LOCK, connect() as database: 00346: database.execute( 00347: """ 00348: UPDATE projects 00349: SET 00350: name = ?, 00351: reference_id = ?, 00352: side = ?, 00353: cover_class = ?, 00354: osseointegration = ?, 00355: specialist_name = ?, 00356: clinic_name = ?, 00357: exclusion_zones = ?, 00358: component_notes = ?, 00359: review_notes = ?, 00360: updated_at = ? 00361: WHERE id = ? 00362: """, 00363: ( 00364: name, 00365: str( 00366: payload.get( 00367: "reference_id", 00368: existing["reference_id"], 00369: ) ... 00378: ) 00379: ).strip(), 00380: str( 00381: payload.get( 00382: "clinic_name", 00383: existing["clinic_name"], 00384: ) 00385: ).strip(), 00386: str( 00387: payload.get( 00388: "exclusion_zones", 00389: existing["exclusion_zones"], 00390: ) 00391: ).strip(), 00392: str( 00393: payload.get( 00394: "component_notes", 00395: existing["component_notes"], 00396: ) 00397: ).strip(), 00398: str( 00399: payload.get( 00400: "review_notes", 00401: existing["review_notes"], 00402: ) 00403: ).strip(), 00404: timestamp, 00405: project_id, 00406: ), 00407: ) 00408: 00409: return load_project(project_id) 00410: 00411: 00412: def save_measurement( 00413: project_id: str, 00414: measurement_key: str, 00415: payload: dict, 00416: ) -> dict: 00417: project = load_project(project_id) 00418: 00419: if project is None: 00420: raise LookupError("Project not found.") 00421: 00422: definition = MEASUREMENT_BY_KEY.get(measurement_key) 00423: 00424: if definition is None: 00425: raise ValueError("Unknown measurement.") 00426: 00427: if definition["oi_only"] and not project["osseointegration"]: 00428: raise ValueError( 00429: "This measurement is available only for " 00430: "osseointegration projects." 00431: ) 00432: 00433: try: 00434: value = float(payload.get("value_mm")) 00435: except (TypeError, ValueError): 00436: raise ValueError("Enter a valid measurement in millimetres.") 00437: 00438: if value <= 0 or value > 2000: 00439: raise ValueError( 00440: "Measurement must be greater than zero and below 2000 mm." 00441: ) 00442: 00443: status = str(payload.get("status", "completed")) 00444: 00445: if status not in {"completed", "specialist_review"}: 00446: raise ValueError("Invalid measurement status.") 00447: 00448: timestamp = utc_now() 00449: 00450: with DATABASE_LOCK, connect() as database: 00451: database.execute( 00452: """ 00453: INSERT INTO measurements ( 00454: project_id, 00455: measurement_key, 00456: value_mm, 00457: notes, 00458: status, 00459: updated_at 00460: ) 00461: VALUES (?, ?, ?, ?, ?, ?) 00462: ON CONFLICT(project_id, measurement_key) 00463: DO UPDATE SET 00464: value_mm = excluded.value_mm, 00465: notes = excluded.notes, 00466: status = excluded.status, 00467: updated_at = excluded.updated_at 00468: """, 00469: ( 00470: project_id, 00471: measurement_key, 00472: value, 00473: str(payload.get("notes", "")).strip(), 00474: status, 00475: timestamp, 00476: ), 00477: ) 00478: 00479: database.execute( 00480: """ 00481: UPDATE projects 00482: SET updated_at = ? 00483: WHERE id = ? 00484: """, 00485: (timestamp, project_id), 00486: ) 00487: 00488: return load_project(project_id) 00489: 00490: 00491: def delete_measurement( 00492: project_id: str, 00493: measurement_key: str, 00494: ) -> dict: 00495: project = load_project(project_id) 00496: 00497: if project is None: 00498: raise LookupError("Project not found.") 00499: 00500: if measurement_key not in MEASUREMENT_BY_KEY: 00501: raise ValueError("Unknown measurement.") 00502: 00503: with DATABASE_LOCK, connect() as database: 00504: database.execute( 00505: """ 00506: DELETE FROM measurements 00507: WHERE project_id = ? 00508: AND measurement_key = ? 00509: """, 00510: (project_id, measurement_key), 00511: ) 00512: 00513: database.execute( 00514: """ 00515: UPDATE projects 00516: SET updated_at = ? 00517: WHERE id = ? 00518: """, 00519: (utc_now(), project_id), 00520: ) ... 00593: self.send_json( 00594: { 00595: "status": "ok", 00596: "service": "limbforge-shared-workspace", 00597: "database": str(DATABASE), 00598: } 00599: ) 00600: return 00601: 00602: if path == "/api/definitions": 00603: self.send_json(MEASUREMENTS) 00604: return 00605: 00606: if path == "/api/projects": 00607: self.send_json(list_projects()) 00608: return 00609: 00610: parts = [ 00611: unquote(part) 00612: for part in path.strip("/").split("/") 00613: if part 00614: ] 00615: 00616: if ( ... 00628: 00629: self.serve_static(path) 00630: 00631: except Exception as error: 00632: self.handle_api_error(error) 00633: 00634: def do_POST(self) -> None: 00635: path = urlparse(self.path).path 00636: 00637: try: 00638: if path == "/api/projects": 00639: self.send_json( 00640: create_project(self.read_json()), 00641: HTTPStatus.CREATED, 00642: ) 00643: return 00644: 00645: self.send_json( 00646: {"error": "Route not found."}, 00647: HTTPStatus.NOT_FOUND, 00648: ) ... 00669: parts[2], 00670: self.read_json(), 00671: ) 00672: ) 00673: return 00674: 00675: if ( 00676: len(parts) == 5 00677: and parts[0] == "api" 00678: and parts[1] == "projects" 00679: and parts[3] == "measurements" 00680: ): 00681: self.send_json( 00682: save_measurement( 00683: parts[2], 00684: parts[4], 00685: self.read_json(), 00686: ) 00687: ) 00688: return 00689: 00690: self.send_json( 00691: {"error": "Route not found."}, 00692: HTTPStatus.NOT_FOUND, ... 00701: unquote(part) 00702: for part in path.strip("/").split("/") 00703: if part 00704: ] 00705: 00706: try: 00707: if ( 00708: len(parts) == 5 00709: and parts[0] == "api" 00710: and parts[1] == "projects" 00711: and parts[3] == "measurements" 00712: ): 00713: self.send_json( 00714: delete_measurement( 00715: parts[2], 00716: parts[4], 00717: ) 00718: ) 00719: return 00720: 00721: self.send_json( 00722: {"error": "Route not found."}, 00723: HTTPStatus.NOT_FOUND, 00724: ) FRONTEND PROJECT FORM AND PAYLOAD CONTEXT ---------------------------------------------------------------------------- 00001: 00002: 00003:
00004: 00005: 00009: 00010: