jebus/enviPy-bayer
1
0
Fork 0
forked from enviPath/enviPy
Code Issues Pull Requests Actions Releases Activity

[Feature] Dynamic additional information rendering in frontend (#282)

Browse Source 
This implements a version of #274, relying on Pydantics built in JSON schema and JSON rendering.
Requires additional UI tagging in the ai model repo but will remove HTML tags.

Example scenario with filled information: 5882df9c-dae1-4d80-a40e-db4724271456/scenario/3a4d395a-6a6d-4154-8ce3-ced667fceec0

Reviewed-on: enviPath/enviPy#282
Co-authored-by: Tobias O <tobias.olenyi@envipath.com>
Co-committed-by: Tobias O <tobias.olenyi@envipath.com>
This commit is contained in:
Tobias O
2026-01-31 00:44:03 +13:00
committed by jebus
parent 9f63a9d4de
commit d80dfb5ee3
42 changed files with 3732 additions and 609 deletions
Download Patch File Download Diff File Expand all files Collapse all files

420
epapi/tests/v1/test_additional_information.py Normal file
View File

@ -0,0 +1,420 @@
"""
Tests for Additional Information API endpoints.
Tests CRUD operations on scenario additional information including the new PATCH endpoint.
"""
from django.test import TestCase, tag
import json
from uuid import uuid4
from epdb.logic import PackageManager, UserManager
from epdb.models import Scenario
@tag("api", "additional_information")
class AdditionalInformationAPITests(TestCase):
"""Test additional information API endpoints."""
@classmethod
def setUpTestData(cls):
"""Set up test data: user, package, and scenario."""
cls.user = UserManager.create_user(
"ai-test-user",
"ai-test@envipath.com",
"SuperSafe",
set_setting=False,
add_to_group=False,
is_active=True,
)
cls.other_user = UserManager.create_user(
"ai-other-user",
"ai-other@envipath.com",
"SuperSafe",
set_setting=False,
add_to_group=False,
is_active=True,
)
cls.package = PackageManager.create_package(
cls.user, "AI Test Package", "Test package for additional information"
)
# Package owned by other_user (no access for cls.user)
cls.other_package = PackageManager.create_package(
cls.other_user, "Other Package", "Package without access"
)
# Create a scenario for testing
cls.scenario = Scenario.objects.create(
package=cls.package,
name="Test Scenario",
description="Test scenario for additional information tests",
scenario_type="biodegradation",
scenario_date="2024-01-01",
additional_information={}, # Initialize with empty dict
)
cls.other_scenario = Scenario.objects.create(
package=cls.other_package,
name="Other Scenario",
description="Scenario in package without access",
scenario_type="biodegradation",
scenario_date="2024-01-01",
additional_information={},
)
def test_list_all_schemas(self):
"""Test GET /api/v1/information/schema/ returns all schemas."""
self.client.force_login(self.user)
response = self.client.get("/api/v1/information/schema/")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, dict)
# Should have multiple schemas
self.assertGreater(len(data), 0)
# Each schema should have RJSF format
for name, schema in data.items():
self.assertIn("schema", schema)
self.assertIn("uiSchema", schema)
self.assertIn("formData", schema)
self.assertIn("groups", schema)
def test_get_specific_schema(self):
"""Test GET /api/v1/information/schema/{model_name}/ returns specific schema."""
self.client.force_login(self.user)
# Assuming 'temperature' is a valid model
response = self.client.get("/api/v1/information/schema/temperature/")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIn("schema", data)
self.assertIn("uiSchema", data)
def test_get_nonexistent_schema_returns_404(self):
"""Test GET for non-existent schema returns 404."""
self.client.force_login(self.user)
response = self.client.get("/api/v1/information/schema/nonexistent/")
self.assertEqual(response.status_code, 404)
def test_list_scenario_information_empty(self):
"""Test GET /api/v1/scenario/{uuid}/information/ returns empty list initially."""
self.client.force_login(self.user)
response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIsInstance(data, list)
self.assertEqual(len(data), 0)
def test_create_additional_information(self):
"""Test POST creates additional information."""
self.client.force_login(self.user)
# Create temperature information (assuming temperature model exists)
payload = {"interval": {"start": 20, "end": 25}}
response = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["status"], "created")
self.assertIn("uuid", data)
self.assertIsNotNone(data["uuid"])
def test_create_with_invalid_data_returns_400(self):
"""Test POST with invalid data returns 400 with validation errors."""
self.client.force_login(self.user)
# Invalid data (missing required fields or wrong types)
payload = {"invalid_field": "value"}
response = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 400)
data = response.json()
# Should have validation error details in 'detail' field
self.assertIn("detail", data)
def test_validation_errors_are_user_friendly(self):
"""Test that validation errors are user-friendly and field-specific."""
self.client.force_login(self.user)
# Invalid data - wrong type (string instead of number in interval)
payload = {"interval": {"start": "not_a_number", "end": 25}}
response = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 400)
data = response.json()
# Parse the error response - Django Ninja wraps errors in 'detail'
error_str = data.get("detail") or data.get("error")
self.assertIsNotNone(error_str, "Response should contain error details")
# Parse the JSON error string
error_data = json.loads(error_str)
# Check structure
self.assertEqual(error_data.get("type"), "validation_error")
self.assertIn("field_errors", error_data)
self.assertIn("message", error_data)
# Ensure error messages are user-friendly (no Pydantic URLs or technical jargon)
error_str = json.dumps(error_data)
self.assertNotIn("pydantic", error_str.lower())
self.assertNotIn("https://errors.pydantic.dev", error_str)
self.assertNotIn("loc", error_str) # No technical field like 'loc'
# Check that error message is helpful
self.assertIn("Please", error_data["message"]) # User-friendly language
def test_patch_additional_information(self):
"""Test PATCH updates existing additional information."""
self.client.force_login(self.user)
# First create an item
create_payload = {"interval": {"start": 20, "end": 25}}
create_response = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(create_payload),
content_type="application/json",
)
item_uuid = create_response.json()["uuid"]
# Then update it with PATCH
update_payload = {"interval": {"start": 30, "end": 35}}
patch_response = self.client.patch(
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/",
data=json.dumps(update_payload),
content_type="application/json",
)
self.assertEqual(patch_response.status_code, 200)
data = patch_response.json()
self.assertEqual(data["status"], "updated")
self.assertEqual(data["uuid"], item_uuid) # UUID preserved
# Verify the data was updated
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
items = list_response.json()
self.assertEqual(len(items), 1)
updated_item = items[0]
self.assertEqual(updated_item["uuid"], item_uuid)
self.assertEqual(updated_item["data"]["interval"]["start"], 30)
self.assertEqual(updated_item["data"]["interval"]["end"], 35)
def test_patch_nonexistent_item_returns_404(self):
"""Test PATCH on non-existent item returns 404."""
self.client.force_login(self.user)
fake_uuid = str(uuid4())
payload = {"interval": {"start": 30, "end": 35}}
response = self.client.patch(
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{fake_uuid}/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 404)
def test_patch_with_invalid_data_returns_400(self):
"""Test PATCH with invalid data returns 400."""
self.client.force_login(self.user)
# First create an item
create_payload = {"interval": {"start": 20, "end": 25}}
create_response = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(create_payload),
content_type="application/json",
)
item_uuid = create_response.json()["uuid"]
# Try to update with invalid data
invalid_payload = {"invalid_field": "value"}
patch_response = self.client.patch(
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/",
data=json.dumps(invalid_payload),
content_type="application/json",
)
self.assertEqual(patch_response.status_code, 400)
def test_delete_additional_information(self):
"""Test DELETE removes additional information."""
self.client.force_login(self.user)
# Create an item
create_payload = {"interval": {"start": 20, "end": 25}}
create_response = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(create_payload),
content_type="application/json",
)
item_uuid = create_response.json()["uuid"]
# Delete it
delete_response = self.client.delete(
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/"
)
self.assertEqual(delete_response.status_code, 200)
data = delete_response.json()
self.assertEqual(data["status"], "deleted")
# Verify deletion
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
items = list_response.json()
self.assertEqual(len(items), 0)
def test_delete_nonexistent_item_returns_404(self):
"""Test DELETE on non-existent item returns 404."""
self.client.force_login(self.user)
fake_uuid = str(uuid4())
response = self.client.delete(
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{fake_uuid}/"
)
self.assertEqual(response.status_code, 404)
def test_multiple_items_crud(self):
"""Test creating, updating, and deleting multiple items."""
self.client.force_login(self.user)
# Create first item
item1_payload = {"interval": {"start": 20, "end": 25}}
response1 = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(item1_payload),
content_type="application/json",
)
item1_uuid = response1.json()["uuid"]
# Create second item (different type if available, or same type)
item2_payload = {"interval": {"start": 30, "end": 35}}
response2 = self.client.post(
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
data=json.dumps(item2_payload),
content_type="application/json",
)
item2_uuid = response2.json()["uuid"]
# Verify both exist
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
items = list_response.json()
self.assertEqual(len(items), 2)
# Update first item
update_payload = {"interval": {"start": 15, "end": 20}}
self.client.patch(
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item1_uuid}/",
data=json.dumps(update_payload),
content_type="application/json",
)
# Delete second item
self.client.delete(f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item2_uuid}/")
# Verify final state: one item with updated data
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
items = list_response.json()
self.assertEqual(len(items), 1)
self.assertEqual(items[0]["uuid"], item1_uuid)
self.assertEqual(items[0]["data"]["interval"]["start"], 15)
def test_unauthenticated_access_returns_401(self):
"""Test that unauthenticated requests return 401."""
# Don't log in
response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
self.assertEqual(response.status_code, 401)
def test_list_info_denied_without_permission(self):
"""User cannot list info for scenario in package they don't have access to"""
self.client.force_login(self.user)
response = self.client.get(f"/api/v1/scenario/{self.other_scenario.uuid}/information/")
self.assertEqual(response.status_code, 403)
def test_add_info_denied_without_permission(self):
"""User cannot add info to scenario in package they don't have access to"""
self.client.force_login(self.user)
payload = {"interval": {"start": 25, "end": 30}}
response = self.client.post(
f"/api/v1/scenario/{self.other_scenario.uuid}/information/temperature/",
json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 403)
def test_update_info_denied_without_permission(self):
"""User cannot update info in scenario they don't have access to"""
self.client.force_login(self.other_user)
# First create an item as other_user
create_payload = {"interval": {"start": 20, "end": 25}}
create_response = self.client.post(
f"/api/v1/scenario/{self.other_scenario.uuid}/information/temperature/",
data=json.dumps(create_payload),
content_type="application/json",
)
item_uuid = create_response.json()["uuid"]
# Try to update as user (who doesn't have access)
self.client.force_login(self.user)
update_payload = {"interval": {"start": 30, "end": 35}}
response = self.client.patch(
f"/api/v1/scenario/{self.other_scenario.uuid}/information/item/{item_uuid}/",
data=json.dumps(update_payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 403)
def test_delete_info_denied_without_permission(self):
"""User cannot delete info from scenario they don't have access to"""
self.client.force_login(self.other_user)
# First create an item as other_user
create_payload = {"interval": {"start": 20, "end": 25}}
create_response = self.client.post(
f"/api/v1/scenario/{self.other_scenario.uuid}/information/temperature/",
data=json.dumps(create_payload),
content_type="application/json",
)
item_uuid = create_response.json()["uuid"]
# Try to delete as user (who doesn't have access)
self.client.force_login(self.user)
response = self.client.delete(
f"/api/v1/scenario/{self.other_scenario.uuid}/information/item/{item_uuid}/"
)
self.assertEqual(response.status_code, 403)
def test_anonymous_user_denied(self):
"""Anonymous users cannot access private scenarios"""
# Ensure scenario package is not reviewed (private)
self.scenario.package.reviewed = False
self.scenario.package.save()
# Unauthenticated users get 401 from the auth layer
response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
self.assertEqual(response.status_code, 401)
def test_nonexistent_scenario_returns_404(self):
"""Test operations on non-existent scenario return 404."""
self.client.force_login(self.user)
fake_uuid = uuid4()
response = self.client.get(f"/api/v1/scenario/{fake_uuid}/information/")
self.assertEqual(response.status_code, 404)

325
epapi/tests/v1/test_scenario_creation.py Normal file
View File

@ -0,0 +1,325 @@
"""
Tests for Scenario Creation Endpoint Error Handling.
Tests comprehensive error handling for POST /api/v1/package/{uuid}/scenario/
including package not found, permission denied, validation errors, and database errors.
"""
from django.test import TestCase, tag
import json
from uuid import uuid4
from epdb.logic import PackageManager, UserManager
from epdb.models import Scenario
@tag("api", "scenario_creation")
class ScenarioCreationAPITests(TestCase):
"""Test scenario creation endpoint error handling."""
@classmethod
def setUpTestData(cls):
"""Set up test data: users and packages."""
cls.user = UserManager.create_user(
"scenario-test-user",
"scenario-test@envipath.com",
"SuperSafe",
set_setting=False,
add_to_group=False,
is_active=True,
)
cls.other_user = UserManager.create_user(
"other-user",
"other@envipath.com",
"SuperSafe",
set_setting=False,
add_to_group=False,
is_active=True,
)
cls.package = PackageManager.create_package(
cls.user, "Test Package", "Test package for scenario creation"
)
def test_create_scenario_package_not_found(self):
"""Test that non-existent package UUID returns 404."""
self.client.force_login(self.user)
fake_uuid = uuid4()
payload = {
"name": "Test Scenario",
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{fake_uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 404)
self.assertIn("Package not found", response.json()["detail"])
def test_create_scenario_insufficient_permissions(self):
"""Test that unauthorized access returns 403."""
self.client.force_login(self.other_user)
payload = {
"name": "Test Scenario",
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 403)
self.assertIn("permission", response.json()["detail"].lower())
def test_create_scenario_invalid_ai_type(self):
"""Test that unknown additional information type returns 400."""
self.client.force_login(self.user)
payload = {
"name": "Test Scenario",
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [
{"type": "invalid_type_that_does_not_exist", "data": {"some_field": "some_value"}}
],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 400)
response_data = response.json()
self.assertIn("Validation errors", response_data["detail"])
def test_create_scenario_validation_error(self):
"""Test that invalid additional information data returns 400."""
self.client.force_login(self.user)
# Use malformed data structure for an actual AI type
payload = {
"name": "Test Scenario",
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [
{
"type": "SomeValidType",
"data": None, # This should cause a validation error
}
],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
# Should return 400 for validation errors
self.assertIn(response.status_code, [400, 422])
def test_create_scenario_success(self):
"""Test that valid scenario creation returns 200."""
self.client.force_login(self.user)
payload = {
"name": "Test Scenario",
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["name"], "Test Scenario")
self.assertEqual(data["description"], "Test description")
# Verify scenario was actually created
scenario = Scenario.objects.get(name="Test Scenario")
self.assertEqual(scenario.package, self.package)
self.assertEqual(scenario.scenario_type, "biodegradation")
def test_create_scenario_with_additional_information(self):
"""Test creating scenario with valid additional information models."""
self.client.force_login(self.user)
# This test will succeed if the registry has valid models
# For now, test with empty additional_information
payload = {
"name": "Scenario with AI",
"description": "Test with additional information",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["name"], "Scenario with AI")
def test_create_scenario_auto_name(self):
"""Test that empty name triggers auto-generation."""
self.client.force_login(self.user)
payload = {
"name": "", # Empty name should be auto-generated
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
# Auto-generated name should follow pattern "Scenario N"
self.assertTrue(data["name"].startswith("Scenario "))
def test_create_scenario_xss_protection(self):
"""Test that XSS attempts are sanitized."""
self.client.force_login(self.user)
payload = {
"name": "<script>alert('xss')</script>Clean Name",
"description": "<img src=x onerror=alert('xss')>Description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
# XSS should be cleaned out
self.assertNotIn("<script>", data["name"])
self.assertNotIn("onerror", data["description"])
def test_create_scenario_missing_required_field(self):
"""Test that missing required fields returns validation error."""
self.client.force_login(self.user)
# Missing 'name' field entirely
payload = {
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
# Should return 422 for schema validation errors
self.assertEqual(response.status_code, 422)
def test_create_scenario_type_error_in_ai(self):
"""Test that TypeError in AI instantiation returns 400."""
self.client.force_login(self.user)
payload = {
"name": "Test Scenario",
"description": "Test description",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [
{
"type": "SomeType",
"data": "string instead of dict", # Wrong type
}
],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
# Should return 400 for validation errors
self.assertIn(response.status_code, [400, 422])
def test_create_scenario_default_values(self):
"""Test that default values are applied correctly."""
self.client.force_login(self.user)
# Minimal payload with only name
payload = {"name": "Minimal Scenario"}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["name"], "Minimal Scenario")
# Check defaults are applied
scenario = Scenario.objects.get(name="Minimal Scenario")
# Default description from model is "no description"
self.assertIn(scenario.description.lower(), ["", "no description"])
def test_create_scenario_unicode_characters(self):
"""Test that unicode characters are handled properly."""
self.client.force_login(self.user)
payload = {
"name": "Test Scenario 测试 🧪",
"description": "Description with émojis and spëcial çhars",
"scenario_date": "2024-01-01",
"scenario_type": "biodegradation",
"additional_information": [],
}
response = self.client.post(
f"/api/v1/package/{self.package.uuid}/scenario/",
data=json.dumps(payload),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIn("测试", data["name"])
self.assertIn("émojis", data["description"])

114
epapi/tests/v1/test_schema_generation.py Normal file
View File

@ -0,0 +1,114 @@
"""
Property-based tests for schema generation.
Tests that verify schema generation works correctly for all models,
regardless of their structure.
"""
import pytest
from typing import Type
from pydantic import BaseModel
from envipy_additional_information import registry, EnviPyModel
from epapi.utils.schema_transformers import build_rjsf_output
class TestSchemaGeneration:
"""Test that all models can generate valid RJSF schemas."""
@pytest.mark.parametrize("model_name,model_cls", list(registry.list_models().items()))
def test_all_models_generate_rjsf(self, model_name: str, model_cls: Type[BaseModel]):
"""Every model in the registry should generate valid RJSF format."""
# Skip non-EnviPyModel classes (parsers, etc.)
if not issubclass(model_cls, EnviPyModel):
pytest.skip(f"{model_name} is not an EnviPyModel")
# Should not raise exception
result = build_rjsf_output(model_cls)
# Verify structure
assert isinstance(result, dict), f"{model_name}: Result should be a dict"
assert "schema" in result, f"{model_name}: Missing 'schema' key"
assert "uiSchema" in result, f"{model_name}: Missing 'uiSchema' key"
assert "formData" in result, f"{model_name}: Missing 'formData' key"
assert "groups" in result, f"{model_name}: Missing 'groups' key"
# Verify types
assert isinstance(result["schema"], dict), f"{model_name}: schema should be dict"
assert isinstance(result["uiSchema"], dict), f"{model_name}: uiSchema should be dict"
assert isinstance(result["formData"], dict), f"{model_name}: formData should be dict"
assert isinstance(result["groups"], list), f"{model_name}: groups should be list"
# Verify schema has properties
assert "properties" in result["schema"], f"{model_name}: schema should have 'properties'"
assert isinstance(result["schema"]["properties"], dict), (
f"{model_name}: properties should be dict"
)
@pytest.mark.parametrize("model_name,model_cls", list(registry.list_models().items()))
def test_ui_schema_matches_schema_fields(self, model_name: str, model_cls: Type[BaseModel]):
"""uiSchema keys should match schema properties (or be nested for intervals)."""
if not issubclass(model_cls, EnviPyModel):
pytest.skip(f"{model_name} is not an EnviPyModel")
result = build_rjsf_output(model_cls)
schema_props = set(result["schema"]["properties"].keys())
ui_schema_keys = set(result["uiSchema"].keys())
# uiSchema should have entries for all top-level properties
# (intervals may have nested start/end, but the main field should be present)
assert ui_schema_keys.issubset(schema_props), (
f"{model_name}: uiSchema has keys not in schema: {ui_schema_keys - schema_props}"
)
@pytest.mark.parametrize("model_name,model_cls", list(registry.list_models().items()))
def test_groups_is_list_of_strings(self, model_name: str, model_cls: Type[BaseModel]):
"""Groups should be a list of strings."""
if not issubclass(model_cls, EnviPyModel):
pytest.skip(f"{model_name} is not an EnviPyModel")
result = build_rjsf_output(model_cls)
groups = result["groups"]
assert isinstance(groups, list), f"{model_name}: groups should be list"
assert all(isinstance(g, str) for g in groups), (
f"{model_name}: all groups should be strings, got {groups}"
)
assert len(groups) > 0, f"{model_name}: should have at least one group"
@pytest.mark.parametrize("model_name,model_cls", list(registry.list_models().items()))
def test_form_data_matches_schema(self, model_name: str, model_cls: Type[BaseModel]):
"""formData keys should match schema properties."""
if not issubclass(model_cls, EnviPyModel):
pytest.skip(f"{model_name} is not an EnviPyModel")
result = build_rjsf_output(model_cls)
schema_props = set(result["schema"]["properties"].keys())
form_data_keys = set(result["formData"].keys())
# formData should only contain keys that are in schema
assert form_data_keys.issubset(schema_props), (
f"{model_name}: formData has keys not in schema: {form_data_keys - schema_props}"
)
class TestWidgetTypes:
"""Test that widget types are valid."""
@pytest.mark.parametrize("model_name,model_cls", list(registry.list_models().items()))
def test_widget_types_are_valid(self, model_name: str, model_cls: Type[BaseModel]):
"""All widget types in uiSchema should be valid WidgetType values."""
from envipy_additional_information.ui_config import WidgetType
if not issubclass(model_cls, EnviPyModel):
pytest.skip(f"{model_name} is not an EnviPyModel")
result = build_rjsf_output(model_cls)
valid_widgets = {wt.value for wt in WidgetType}
for field_name, ui_config in result["uiSchema"].items():
widget = ui_config.get("ui:widget")
if widget:
assert widget in valid_widgets, (
f"{model_name}.{field_name}: Invalid widget '{widget}'. Valid: {valid_widgets}"
)

0
epapi/utils/__init__.py Normal file
View File

175
epapi/utils/schema_transformers.py Normal file
View File

@ -0,0 +1,175 @@
"""
Schema transformation utilities for converting Pydantic models to RJSF format.
This module provides functions to extract UI configuration from Pydantic models
and transform them into React JSON Schema Form (RJSF) compatible format.
"""
from typing import Type, Optional, Any
import jsonref
from pydantic import BaseModel
from envipy_additional_information.ui_config import UIConfig
from envipy_additional_information import registry
def extract_groups(model_cls: Type[BaseModel]) -> list[str]:
"""
Extract groups from registry-stored group information.
Args:
model_cls: The model class
Returns:
List of group names the model belongs to
"""
return registry.get_groups(model_cls)
def extract_ui_metadata(model_cls: Type[BaseModel]) -> dict[str, Any]:
"""
Extract model-level UI metadata from UI class.
Returns metadata attributes that are NOT UIConfig instances.
Common metadata includes: unit, description, title.
"""
metadata: dict[str, Any] = {}
if not hasattr(model_cls, "UI"):
return metadata
ui_class = getattr(model_cls, "UI")
# Iterate over all attributes in the UI class
for attr_name in dir(ui_class):
# Skip private attributes
if attr_name.startswith("_"):
continue
# Get the attribute value
try:
attr_value = getattr(ui_class, attr_name)
except AttributeError:
continue
# Skip callables but keep types/classes
if callable(attr_value) and not isinstance(attr_value, type):
continue
# Skip UIConfig instances (these are field-level configs, not metadata)
# This includes both UIConfig and IntervalConfig
if isinstance(attr_value, UIConfig):
continue
metadata[attr_name] = attr_value
return metadata
def extract_ui_config_from_model(model_cls: Type[BaseModel]) -> dict[str, Any]:
"""
Extract UI configuration from model's UI class.
Returns a dictionary mapping field names to their UI schema configurations.
Trusts the config classes to handle their own transformation logic.
"""
ui_configs: dict[str, Any] = {}
if not hasattr(model_cls, "UI"):
return ui_configs
ui_class = getattr(model_cls, "UI")
schema = model_cls.model_json_schema()
field_names = schema.get("properties", {}).keys()
# Extract config for each field
for field_name in field_names:
ui_config = getattr(ui_class, field_name)
if isinstance(ui_config, UIConfig):
ui_configs[field_name] = ui_config.to_ui_schema_field()
return ui_configs
def build_ui_schema(model_cls: Type[BaseModel]) -> dict:
"""Generate RJSF uiSchema from model's UI class."""
ui_schema = {}
# Extract field-level UI configs
field_configs = extract_ui_config_from_model(model_cls)
for field_name, config in field_configs.items():
ui_schema[field_name] = config
return ui_schema
def build_schema(model_cls: Type[BaseModel]) -> dict[str, Any]:
"""
Build JSON schema from Pydantic model, applying UI metadata.
Dereferences all $ref pointers to produce fully inlined schema.
This ensures the frontend receives schemas with enum values and nested
properties fully resolved, without needing client-side ref resolution.
Extracts model-level metadata from UI class (title, unit, etc.) and applies
it to the generated schema. This ensures UI metadata is the single source of truth.
"""
schema = model_cls.model_json_schema()
# Dereference $ref pointers (inlines $defs) using jsonref
# This ensures the frontend receives schemas with enum values and nested
# properties fully resolved, currently necessary for client-side rendering.
# FIXME: This is a hack to get the schema to work with alpine schema-form.js replace once we migrate to client-side framework.
schema = jsonref.replace_refs(schema, proxies=False)
# Remove $defs section as all refs are now inlined
if "$defs" in schema:
del schema["$defs"]
# Extract and apply UI metadata (title, unit, description, etc.)
ui_metadata = extract_ui_metadata(model_cls)
# Apply all metadata consistently as custom properties with x- prefix
# This ensures consistency and avoids conflicts with standard JSON Schema properties
for key, value in ui_metadata.items():
if value is not None:
schema[f"x-{key}"] = value
# Set standard title property from UI metadata for JSON Schema compliance
if "label" in ui_metadata:
schema["title"] = ui_metadata["label"]
return schema
def build_rjsf_output(model_cls: Type[BaseModel], initial_data: Optional[dict] = None) -> dict:
"""
Main function that returns complete RJSF format.
Trusts the config classes to handle their own transformation logic.
No special-case handling - if a config knows how to transform itself, it will.
Returns:
dict with keys: schema, uiSchema, formData, groups
"""
# Build schema with UI metadata applied
schema = build_schema(model_cls)
# Build UI schema - config classes handle their own transformation
ui_schema = build_ui_schema(model_cls)
# Extract groups from marker interfaces
groups = extract_groups(model_cls)
# Use provided initial_data or empty dict
form_data = initial_data if initial_data is not None else {}
return {
"schema": schema,
"uiSchema": ui_schema,
"formData": form_data,
"groups": groups,
}

50
epapi/v1/dal.py
View File

@ -1,12 +1,12 @@
from django.db.models import Model
from epdb.logic import PackageManager
from epdb.models import CompoundStructure, User, Package, Compound
from epdb.models import CompoundStructure, User, Package, Compound, Scenario
from uuid import UUID
from .errors import EPAPINotFoundError, EPAPIPermissionDeniedError
def get_compound_or_error(user, compound_uuid: UUID):
def get_compound_for_read(user, compound_uuid: UUID):
"""
Get compound by UUID with permission check.
"""
@ -23,7 +23,7 @@ def get_compound_or_error(user, compound_uuid: UUID):
return compound
def get_package_or_error(user, package_uuid: UUID):
def get_package_for_read(user, package_uuid: UUID):
"""
Get package by UUID with permission check.
"""
@ -41,14 +41,40 @@ def get_package_or_error(user, package_uuid: UUID):
return package
def get_user_packages_qs(user: User | None):
def get_scenario_for_read(user, scenario_uuid: UUID):
"""Get scenario by UUID with read permission check."""
try:
scenario = Scenario.objects.select_related("package").get(uuid=scenario_uuid)
except Scenario.DoesNotExist:
raise EPAPINotFoundError(f"Scenario with UUID {scenario_uuid} not found")
if not user or user.is_anonymous or not PackageManager.readable(user, scenario.package):
raise EPAPIPermissionDeniedError("Insufficient permissions to access this scenario.")
return scenario
def get_scenario_for_write(user, scenario_uuid: UUID):
"""Get scenario by UUID with write permission check."""
try:
scenario = Scenario.objects.select_related("package").get(uuid=scenario_uuid)
except Scenario.DoesNotExist:
raise EPAPINotFoundError(f"Scenario with UUID {scenario_uuid} not found")
if not user or user.is_anonymous or not PackageManager.writable(user, scenario.package):
raise EPAPIPermissionDeniedError("Insufficient permissions to modify this scenario.")
return scenario
def get_user_packages_for_read(user: User | None):
"""Get all packages readable by the user."""
if not user or user.is_anonymous:
return PackageManager.get_reviewed_packages()
return PackageManager.get_all_readable_packages(user, include_reviewed=True)
def get_user_entities_qs(model_class: Model, user: User | None):
def get_user_entities_for_read(model_class: Model, user: User | None):
"""Build queryset for reviewed package entities."""
if not user or user.is_anonymous:
@ -60,16 +86,14 @@ def get_user_entities_qs(model_class: Model, user: User | None):
return qs
def get_package_scoped_entities_qs(
model_class: Model, package_uuid: UUID, user: User | None = None
):
def get_package_entities_for_read(model_class: Model, package_uuid: UUID, user: User | None = None):
"""Build queryset for specific package entities."""
package = get_package_or_error(user, package_uuid)
package = get_package_for_read(user, package_uuid)
qs = model_class.objects.filter(package=package).select_related("package")
return qs
def get_user_structures_qs(user: User | None):
def get_user_structure_for_read(user: User | None):
"""Build queryset for structures accessible to the user (via compound->package)."""
if not user or user.is_anonymous:
@ -83,13 +107,13 @@ def get_user_structures_qs(user: User | None):
return qs
def get_package_compound_scoped_structure_qs(
def get_package_compound_structure_for_read(
package_uuid: UUID, compound_uuid: UUID, user: User | None = None
):
"""Build queryset for specific package compound structures."""
get_package_or_error(user, package_uuid)
compound = get_compound_or_error(user, compound_uuid)
get_package_for_read(user, package_uuid)
compound = get_compound_for_read(user, compound_uuid)
qs = CompoundStructure.objects.filter(compound=compound).select_related("compound__package")
return qs

237
epapi/v1/endpoints/additional_information.py Normal file
View File

@ -0,0 +1,237 @@
from ninja import Router, Body
from ninja.errors import HttpError
from uuid import UUID
from pydantic import ValidationError
from pydantic_core import ErrorDetails
from typing import Dict, Any
import logging
import json
from envipy_additional_information import registry
from envipy_additional_information.groups import GroupEnum
from epapi.utils.schema_transformers import build_rjsf_output
from ..dal import get_scenario_for_read, get_scenario_for_write
logger = logging.getLogger(__name__)
router = Router(tags=["Additional Information"])
@router.get("/information/schema/")
def list_all_schemas(request):
"""Return all schemas in RJSF format with lowercase class names as keys."""
result = {}
for name, cls in registry.list_models().items():
try:
result[name] = build_rjsf_output(cls)
except Exception as e:
logger.warning(f"Failed to generate schema for {name}: {e}")
continue
return result
@router.get("/information/schema/{model_name}/")
def get_model_schema(request, model_name: str):
"""Return RJSF schema for specific model."""
cls = registry.get_model(model_name.lower())
if not cls:
raise HttpError(404, f"Unknown model: {model_name}")
return build_rjsf_output(cls)
@router.get("/scenario/{uuid:scenario_uuid}/information/")
def list_scenario_info(request, scenario_uuid: UUID):
"""List all additional information for a scenario"""
scenario = get_scenario_for_read(request.user, scenario_uuid)
result = []
for ai in scenario.get_additional_information():
result.append(
{
"type": ai.__class__.__name__,
"uuid": getattr(ai, "uuid", None),
"data": ai.model_dump(mode="json"),
}
)
return result
def _format_validation_error(error: ErrorDetails) -> str:
"""Format a Pydantic validation error into a user-friendly message."""
msg = error["msg"] or "Invalid value"
error_type = error["type"] or ""
# Handle common validation types with friendly messages
if error_type == "enum":
expected = error["ctx"]["expected"] if error["ctx"] else ""
return f"Please select a valid option{': ' + expected if expected else ''}"
elif error_type == "missing":
return "This field is required"
elif error_type in ("string_type", "int_type", "float_type"):
type_name = error_type.replace("_type", "")
return f"Please enter a valid {type_name}"
elif error_type == "value_error":
# Use the message as-is for value errors
return msg
else:
# Default: use the message from Pydantic but clean it up
return msg.replace("Input should be ", "Please enter ").replace("Value error, ", "")
@router.post("/scenario/{uuid:scenario_uuid}/information/{model_name}/")
def add_scenario_info(
request, scenario_uuid: UUID, model_name: str, payload: Dict[str, Any] = Body(...)
):
"""Add new additional information to scenario"""
cls = registry.get_model(model_name.lower())
if not cls:
raise HttpError(404, f"Unknown model: {model_name}")
try:
instance = cls(**payload) # Pydantic validates
except ValidationError as e:
# Transform Pydantic validation errors into user-friendly format
field_errors: dict[str, list[str]] = {}
for error in e.errors():
# Get the field name from location tuple
loc = error.get("loc", ())
field = str(loc[-1]) if loc else "root"
# Format the error message
friendly_msg = _format_validation_error(error)
if field not in field_errors:
field_errors[field] = []
field_errors[field].append(friendly_msg)
# Return structured error for frontend parsing
error_response = {
"type": "validation_error",
"field_errors": field_errors,
"message": "Please correct the errors below",
}
raise HttpError(400, json.dumps(error_response))
scenario = get_scenario_for_write(request.user, scenario_uuid)
# Model method now returns the UUID
created_uuid = scenario.add_additional_information(instance)
return {"status": "created", "uuid": created_uuid}
@router.patch("/scenario/{uuid:scenario_uuid}/information/item/{uuid:ai_uuid}/")
def update_scenario_info(
request, scenario_uuid: UUID, ai_uuid: UUID, payload: Dict[str, Any] = Body(...)
):
"""Update existing additional information for a scenario"""
scenario = get_scenario_for_write(request.user, scenario_uuid)
ai_uuid_str = str(ai_uuid)
# Find item to determine type for validation
found_type = None
for type_name, items in scenario.additional_information.items():
if any(item.get("uuid") == ai_uuid_str for item in items):
found_type = type_name
break
if found_type is None:
raise HttpError(404, f"Additional information not found: {ai_uuid}")
# Get the model class for validation
cls = registry.get_model(found_type.lower())
if not cls:
raise HttpError(500, f"Unknown model type in data: {found_type}")
# Validate the payload against the model
try:
instance = cls(**payload)
except ValidationError as e:
# Transform Pydantic validation errors into user-friendly format
field_errors: dict[str, list[str]] = {}
for error in e.errors():
# Get the field name from location tuple
loc = error.get("loc", ())
field = str(loc[-1]) if loc else "root"
# Format the error message
friendly_msg = _format_validation_error(error)
if field not in field_errors:
field_errors[field] = []
field_errors[field].append(friendly_msg)
# Return structured error for frontend parsing
error_response = {
"type": "validation_error",
"field_errors": field_errors,
"message": "Please correct the errors below",
}
raise HttpError(400, json.dumps(error_response))
# Use model method for update
try:
scenario.update_additional_information(ai_uuid_str, instance)
except ValueError as e:
raise HttpError(404, str(e))
return {"status": "updated", "uuid": ai_uuid_str}
@router.delete("/scenario/{uuid:scenario_uuid}/information/item/{uuid:ai_uuid}/")
def delete_scenario_info(request, scenario_uuid: UUID, ai_uuid: UUID):
"""Delete additional information from scenario"""
scenario = get_scenario_for_write(request.user, scenario_uuid)
try:
scenario.remove_additional_information(str(ai_uuid))
except ValueError as e:
raise HttpError(404, str(e))
return {"status": "deleted"}
@router.get("/information/groups/")
def list_groups(request):
"""Return list of available group names."""
return {"groups": GroupEnum.values()}
@router.get("/information/groups/{group_name}/")
def get_group_models(request, group_name: str):
"""
Return models for a specific group organized by subcategory.
Args:
group_name: One of "sludge", "soil", or "sediment" (string)
Returns:
Dictionary with subcategories (exp, spike, comp, misc, or group name)
as keys and lists of model info as values
"""
# Convert string to enum (raises ValueError if invalid)
try:
group_enum = GroupEnum(group_name)
except ValueError:
valid = ", ".join(GroupEnum.values())
raise HttpError(400, f"Invalid group '{group_name}'. Valid: {valid}")
try:
group_data = registry.collect_group(group_enum)
except (ValueError, TypeError) as e:
raise HttpError(400, str(e))
result = {}
for subcategory, models in group_data.items():
result[subcategory] = [
{
"name": cls.__name__.lower(),
"class": cls.__name__,
"title": getattr(cls.UI, "title", cls.__name__)
if hasattr(cls, "UI")
else cls.__name__,
}
for cls in models
]
return result

6
epapi/v1/endpoints/compounds.py
View File

@ -6,7 +6,7 @@ from uuid import UUID
from epdb.models import Compound
from ..pagination import EnhancedPageNumberPagination
from ..schemas import CompoundOutSchema, ReviewStatusFilter
from ..dal import get_user_entities_qs, get_package_scoped_entities_qs
from ..dal import get_user_entities_for_read, get_package_entities_for_read
router = Router()
@ -21,7 +21,7 @@ def list_all_compounds(request):
"""
List all compounds from reviewed packages.
"""
return get_user_entities_qs(Compound, request.user).order_by("name").all()
return get_user_entities_for_read(Compound, request.user).order_by("name").all()
@router.get(
@ -38,4 +38,4 @@ def list_package_compounds(request, package_uuid: UUID):
List all compounds for a specific package.
"""
user = request.user
return get_package_scoped_entities_qs(Compound, package_uuid, user).order_by("name").all()
return get_package_entities_for_read(Compound, package_uuid, user).order_by("name").all()

6
epapi/v1/endpoints/models.py
View File

@ -6,7 +6,7 @@ from uuid import UUID
from epdb.models import EPModel
from ..pagination import EnhancedPageNumberPagination
from ..schemas import ModelOutSchema, ReviewStatusFilter
from ..dal import get_user_entities_qs, get_package_scoped_entities_qs
from ..dal import get_user_entities_for_read, get_package_entities_for_read
router = Router()
@ -21,7 +21,7 @@ def list_all_models(request):
"""
List all models from reviewed packages.
"""
return get_user_entities_qs(EPModel, request.user).order_by("name").all()
return get_user_entities_for_read(EPModel, request.user).order_by("name").all()
@router.get(
@ -38,4 +38,4 @@ def list_package_models(request, package_uuid: UUID):
List all models for a specific package.
"""
user = request.user
return get_package_scoped_entities_qs(EPModel, package_uuid, user).order_by("name").all()
return get_package_entities_for_read(EPModel, package_uuid, user).order_by("name").all()

4
epapi/v1/endpoints/packages.py
View File

@ -3,7 +3,7 @@ from ninja import Router
from ninja_extra.pagination import paginate
import logging
from ..dal import get_user_packages_qs
from ..dal import get_user_packages_for_read
from ..pagination import EnhancedPageNumberPagination
from ..schemas import PackageOutSchema, SelfReviewStatusFilter
@ -23,5 +23,5 @@ def list_all_packages(request):
"""
user = request.user
qs = get_user_packages_qs(user)
qs = get_user_packages_for_read(user)
return qs.order_by("name").all()

6
epapi/v1/endpoints/pathways.py
View File

@ -6,7 +6,7 @@ from uuid import UUID
from epdb.models import Pathway
from ..pagination import EnhancedPageNumberPagination
from ..schemas import PathwayOutSchema, ReviewStatusFilter
from ..dal import get_user_entities_qs, get_package_scoped_entities_qs
from ..dal import get_user_entities_for_read, get_package_entities_for_read
router = Router()
@ -22,7 +22,7 @@ def list_all_pathways(request):
List all pathways from reviewed packages.
"""
user = request.user
return get_user_entities_qs(Pathway, user).order_by("name").all()
return get_user_entities_for_read(Pathway, user).order_by("name").all()
@router.get(
@ -39,4 +39,4 @@ def list_package_pathways(request, package_uuid: UUID):
List all pathways for a specific package.
"""
user = request.user
return get_package_scoped_entities_qs(Pathway, package_uuid, user).order_by("name").all()
return get_package_entities_for_read(Pathway, package_uuid, user).order_by("name").all()

6
epapi/v1/endpoints/reactions.py
View File

@ -6,7 +6,7 @@ from uuid import UUID
from epdb.models import Reaction
from ..pagination import EnhancedPageNumberPagination
from ..schemas import ReactionOutSchema, ReviewStatusFilter
from ..dal import get_user_entities_qs, get_package_scoped_entities_qs
from ..dal import get_user_entities_for_read, get_package_entities_for_read
router = Router()
@ -22,7 +22,7 @@ def list_all_reactions(request):
List all reactions from reviewed packages.
"""
user = request.user
return get_user_entities_qs(Reaction, user).order_by("name").all()
return get_user_entities_for_read(Reaction, user).order_by("name").all()
@router.get(
@ -39,4 +39,4 @@ def list_package_reactions(request, package_uuid: UUID):
List all reactions for a specific package.
"""
user = request.user
return get_package_scoped_entities_qs(Reaction, package_uuid, user).order_by("name").all()
return get_package_entities_for_read(Reaction, package_uuid, user).order_by("name").all()

6
epapi/v1/endpoints/rules.py
View File

@ -6,7 +6,7 @@ from uuid import UUID
from epdb.models import Rule
from ..pagination import EnhancedPageNumberPagination
from ..schemas import ReviewStatusFilter, RuleOutSchema
from ..dal import get_user_entities_qs, get_package_scoped_entities_qs
from ..dal import get_user_entities_for_read, get_package_entities_for_read
router = Router()
@ -22,7 +22,7 @@ def list_all_rules(request):
List all rules from reviewed packages.
"""
user = request.user
return get_user_entities_qs(Rule, user).order_by("name").all()
return get_user_entities_for_read(Rule, user).order_by("name").all()
@router.get(
@ -39,4 +39,4 @@ def list_package_rules(request, package_uuid: UUID):
List all rules for a specific package.
"""
user = request.user
return get_package_scoped_entities_qs(Rule, package_uuid, user).order_by("name").all()
return get_package_entities_for_read(Rule, package_uuid, user).order_by("name").all()

98
epapi/v1/endpoints/scenarios.py
View File

@ -1,12 +1,22 @@
from django.conf import settings as s
from ninja import Router
from django.db import IntegrityError, OperationalError, DatabaseError
from ninja import Router, Body
from ninja.errors import HttpError
from ninja_extra.pagination import paginate
from uuid import UUID
from pydantic import ValidationError
import logging
import json
from epdb.models import Scenario
from epdb.logic import PackageManager
from epdb.views import _anonymous_or_real
from ..pagination import EnhancedPageNumberPagination
from ..schemas import ReviewStatusFilter, ScenarioOutSchema
from ..dal import get_user_entities_qs, get_package_scoped_entities_qs
from ..schemas import ReviewStatusFilter, ScenarioOutSchema, ScenarioCreateSchema
from ..dal import get_user_entities_for_read, get_package_entities_for_read
from envipy_additional_information import registry
logger = logging.getLogger(__name__)
router = Router()
@ -19,7 +29,7 @@ router = Router()
)
def list_all_scenarios(request):
user = request.user
return get_user_entities_qs(Scenario, user).order_by("name").all()
return get_user_entities_for_read(Scenario, user).order_by("name").all()
@router.get(
@ -33,4 +43,82 @@ def list_all_scenarios(request):
)
def list_package_scenarios(request, package_uuid: UUID):
user = request.user
return get_package_scoped_entities_qs(Scenario, package_uuid, user).order_by("name").all()
return get_package_entities_for_read(Scenario, package_uuid, user).order_by("name").all()
@router.post("/package/{uuid:package_uuid}/scenario/", response=ScenarioOutSchema)
def create_scenario(request, package_uuid: UUID, payload: ScenarioCreateSchema = Body(...)):
"""Create a new scenario with optional additional information."""
user = _anonymous_or_real(request)
try:
current_package = PackageManager.get_package_by_id(user, package_uuid)
except ValueError as e:
error_msg = str(e)
if "does not exist" in error_msg:
raise HttpError(404, f"Package not found: {package_uuid}")
elif "Insufficient permissions" in error_msg:
raise HttpError(403, "You do not have permission to access this package")
else:
logger.error(f"Unexpected ValueError from get_package_by_id: {error_msg}")
raise HttpError(400, "Invalid package request")
# Build additional information models from payload
additional_information_models = []
validation_errors = []
for ai_item in payload.additional_information:
# Get model class from registry
model_cls = registry.get_model(ai_item.type.lower())
if not model_cls:
validation_errors.append(f"Unknown additional information type: {ai_item.type}")
continue
try:
# Validate and create model instance
instance = model_cls(**ai_item.data)
additional_information_models.append(instance)
except ValidationError as e:
# Collect validation errors to return to user
error_messages = [err.get("msg", "Validation error") for err in e.errors()]
validation_errors.append(f"{ai_item.type}: {', '.join(error_messages)}")
except (TypeError, AttributeError, KeyError) as e:
logger.warning(f"Failed to instantiate {ai_item.type} model: {str(e)}")
validation_errors.append(f"{ai_item.type}: Invalid data structure - {str(e)}")
except Exception as e:
logger.error(f"Unexpected error instantiating {ai_item.type}: {str(e)}")
validation_errors.append(f"{ai_item.type}: Failed to process - please check your data")
# If there are validation errors, return them
if validation_errors:
raise HttpError(
400,
json.dumps(
{
"error": "Validation errors in additional information",
"details": validation_errors,
}
),
)
# Create scenario using the existing Scenario.create method
try:
new_scenario = Scenario.create(
package=current_package,
name=payload.name,
description=payload.description,
scenario_date=payload.scenario_date,
scenario_type=payload.scenario_type,
additional_information=additional_information_models,
)
except IntegrityError as e:
logger.error(f"Database integrity error creating scenario: {str(e)}")
raise HttpError(400, "Scenario creation failed - data constraint violation")
except OperationalError as e:
logger.error(f"Database operational error creating scenario: {str(e)}")
raise HttpError(503, "Database temporarily unavailable - please try again")
except (DatabaseError, AttributeError) as e:
logger.error(f"Error creating scenario: {str(e)}")
raise HttpError(500, "Failed to create scenario due to database error")
return new_scenario

8
epapi/v1/endpoints/structure.py
View File

@ -6,8 +6,8 @@ from uuid import UUID
from ..pagination import EnhancedPageNumberPagination
from ..schemas import CompoundStructureOutSchema, StructureReviewStatusFilter
from ..dal import (
get_user_structures_qs,
get_package_compound_scoped_structure_qs,
get_user_structure_for_read,
get_package_compound_structure_for_read,
)
router = Router()
@ -26,7 +26,7 @@ def list_all_structures(request):
List all structures from all packages.
"""
user = request.user
return get_user_structures_qs(user).order_by("name").all()
return get_user_structure_for_read(user).order_by("name").all()
@router.get(
@ -44,7 +44,7 @@ def list_package_structures(request, package_uuid: UUID, compound_uuid: UUID):
"""
user = request.user
return (
get_package_compound_scoped_structure_qs(package_uuid, compound_uuid, user)
get_package_compound_structure_for_read(package_uuid, compound_uuid, user)
.order_by("name")
.all()
)

14
epapi/v1/router.py
View File

@ -3,15 +3,16 @@ from ninja.security import SessionAuth
from .auth import BearerTokenAuth
from .endpoints import (
compounds,
models,
packages,
pathways,
reactions,
rules,
scenarios,
settings,
compounds,
rules,
reactions,
pathways,
models,
structure,
additional_information,
settings,
)
# Main router with authentication
@ -31,4 +32,5 @@ router.add_router("", reactions.router)
router.add_router("", pathways.router)
router.add_router("", models.router)
router.add_router("", structure.router)
router.add_router("", additional_information.router)
router.add_router("", settings.router)

19
epapi/v1/schemas.py
View File

@ -1,5 +1,5 @@
from ninja import FilterSchema, FilterLookup, Schema
from typing import Annotated, Optional
from typing import Annotated, Optional, List, Dict, Any
from uuid import UUID
@ -51,6 +51,23 @@ class ScenarioOutSchema(PackageEntityOutSchema):
pass
class AdditionalInformationItemSchema(Schema):
"""Schema for additional information item in scenario creation."""
type: str
data: Dict[str, Any]
class ScenarioCreateSchema(Schema):
"""Schema for creating a new scenario."""
name: str
description: str = ""
scenario_date: str = "No date"
scenario_type: str = "Not specified"
additional_information: List[AdditionalInformationItemSchema] = []
class CompoundOutSchema(PackageEntityOutSchema):
pass