98 lines
3 KiB
Python
98 lines
3 KiB
Python
from typing import Any
|
|
|
|
from fastapi import (
|
|
Depends,
|
|
HTTPException,
|
|
)
|
|
from pydantic import (
|
|
BaseModel,
|
|
TypeAdapter,
|
|
ValidationError,
|
|
)
|
|
from starlette.responses import JSONResponse
|
|
|
|
from dump_things_service import (
|
|
HTTP_400_BAD_REQUEST,
|
|
HTTP_403_FORBIDDEN,
|
|
HTTP_422_UNPROCESSABLE_CONTENT,
|
|
Format,
|
|
)
|
|
from dump_things_service.abstract_config import (
|
|
check_collection,
|
|
get_config,
|
|
get_default_token_name,
|
|
)
|
|
from dump_things_service.api_key import api_key_header_scheme
|
|
from dump_things_service.converter import FormatConverter
|
|
from dump_things_service.instance_state import get_instance_state
|
|
from dump_things_service.utils import (
|
|
get_token_store,
|
|
join_default_token_permissions,
|
|
wrap_http_exception,
|
|
)
|
|
|
|
|
|
def validate_record(
|
|
collection: str,
|
|
data: BaseModel | str,
|
|
class_name: str,
|
|
model: Any,
|
|
input_format: Format,
|
|
api_key: str | None = Depends(api_key_header_scheme),
|
|
) -> JSONResponse:
|
|
|
|
instance_state = get_instance_state()
|
|
abstract_config = get_config()
|
|
|
|
if input_format == Format.json and isinstance(data, str):
|
|
raise HTTPException(
|
|
status_code=HTTP_400_BAD_REQUEST, detail='Invalid JSON data provided.'
|
|
)
|
|
|
|
if input_format == Format.ttl and not isinstance(data, str):
|
|
raise HTTPException(
|
|
status_code=HTTP_400_BAD_REQUEST, detail='Invalid ttl data provided.'
|
|
)
|
|
|
|
check_collection(abstract_config, collection)
|
|
|
|
token = (
|
|
get_default_token_name(abstract_config, collection)
|
|
if api_key is None
|
|
else api_key
|
|
)
|
|
|
|
store, token_permissions, user_id = get_token_store(
|
|
abstract_config,
|
|
instance_state,
|
|
collection,
|
|
token,
|
|
)
|
|
final_permissions = join_default_token_permissions(
|
|
abstract_config,
|
|
instance_state,
|
|
token_permissions,
|
|
collection,
|
|
)
|
|
if not final_permissions.incoming_write:
|
|
raise HTTPException(
|
|
status_code=HTTP_403_FORBIDDEN,
|
|
detail=f"Not authorized to validate records for collection '{collection}'.",
|
|
)
|
|
|
|
if input_format == Format.ttl:
|
|
with wrap_http_exception(ValueError, status_code=HTTP_422_UNPROCESSABLE_CONTENT, header='Conversion error'):
|
|
json_object = FormatConverter(
|
|
abstract_config.collections[collection].schema_location,
|
|
input_format=Format.ttl,
|
|
output_format=Format.json,
|
|
).convert(data, class_name)
|
|
with wrap_http_exception(ValidationError, status_code=HTTP_422_UNPROCESSABLE_CONTENT, header='Validation error'):
|
|
TypeAdapter(getattr(model, class_name)).validate_python(json_object)
|
|
else:
|
|
# Try to convert it into TTL to detect potential errors before storing
|
|
# the record
|
|
with wrap_http_exception(ValueError, status_code=HTTP_422_UNPROCESSABLE_CONTENT, header='Validation error'):
|
|
instance_state.validators[collection].validate(data)
|
|
|
|
return JSONResponse(True)
|