forked from enviPath/enviPy
Tobias O
d80dfb5ee3
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>
125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
from django.conf import settings as s
|
|
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, 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()
|
|
|
|
|
|
@router.get("/scenarios/", response=EnhancedPageNumberPagination.Output[ScenarioOutSchema])
|
|
@paginate(
|
|
EnhancedPageNumberPagination,
|
|
page_size=s.API_PAGINATION_DEFAULT_PAGE_SIZE,
|
|
filter_schema=ReviewStatusFilter,
|
|
)
|
|
def list_all_scenarios(request):
|
|
user = request.user
|
|
return get_user_entities_for_read(Scenario, user).order_by("name").all()
|
|
|
|
|
|
@router.get(
|
|
"/package/{uuid:package_uuid}/scenario/",
|
|
response=EnhancedPageNumberPagination.Output[ScenarioOutSchema],
|
|
)
|
|
@paginate(
|
|
EnhancedPageNumberPagination,
|
|
page_size=s.API_PAGINATION_DEFAULT_PAGE_SIZE,
|
|
filter_schema=ReviewStatusFilter,
|
|
)
|
|
def list_package_scenarios(request, package_uuid: UUID):
|
|
user = request.user
|
|
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
|