From 7f3555ebaeea1c892ab83cba121820612803c8f8 Mon Sep 17 00:00:00 2001
From: Felix Blanke <info@fblanke.de>
Date: Tue, 4 Mar 2025 15:50:35 +0100
Subject: [PATCH] Retrieve schemas as needed
---
AKModel/utils.py | 30 +++++++++++++++++++++---------
1 file changed, 21 insertions(+), 9 deletions(-)
diff --git a/AKModel/utils.py b/AKModel/utils.py
index 3a0b3f0..1f9e821 100644
--- a/AKModel/utils.py
+++ b/AKModel/utils.py
@@ -4,24 +4,36 @@ from pathlib import Path
from jsonschema import Draft202012Validator
from jsonschema.protocols import Validator
from referencing import Registry, Resource
+import referencing.retrieval
from AKPlanning import settings
+def _construct_schema_path(uri: str | Path) -> Path:
+ schema_base_path = Path(settings.BASE_DIR).resolve()
+ uri_path = (schema_base_path / uri).resolve()
+ if not uri_path.is_relative_to(schema_base_path / "schemas"):
+ raise ValueError("Unallowed dictionary traversal")
+ return uri_path
+
+
+@referencing.retrieval.to_cached_resource()
+def retrieve_schema(uri: str) -> str:
+ # avoid dictionary traversals
+ uri_path = _construct_schema_path(uri)
+ with uri_path.open("r") as ff:
+ return ff.read()
+
+
def construct_schema_validator(schema: str | dict) -> Validator:
"""Construct a validator for a JSON schema.
In particular, all schemas from the 'schemas' directory
are loaded into the registry.
"""
- schema_base_path = Path(settings.BASE_DIR) / "schemas"
- resources = []
- for schema_path in schema_base_path.glob("**/*.schema.json"):
- with schema_path.open("r") as ff:
- res = Resource.from_contents(json.load(ff))
- resources.append((res.id(), res))
- registry = Registry().with_resources(resources)
+ registry = Registry(retrieve=retrieve_schema)
+
if isinstance(schema, str):
- with (schema_base_path / schema).open("r") as ff:
- schema = json.load(ff)
+ schema_uri = str(Path("schemas") / schema)
+ schema = registry.get_or_retrieve(schema_uri).value.contents
return Draft202012Validator(schema=schema, registry=registry)
--
GitLab