197 lines
6.9 KiB
Python
197 lines
6.9 KiB
Python
"""Custom-field metadata + validation.
|
|
|
|
The CustomFieldDefinition model in common/models.py stores schema rows; this
|
|
module is the cross-cutting validator everyone uses to coerce/check values
|
|
before persisting them on an entity's `custom_fields` JSONField.
|
|
|
|
See docs/cases/tier1/custom-fields.md and docs/cases/COORDINATION_DECISIONS.md.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
from rest_framework import serializers as drf_serializers
|
|
|
|
from common.object_registry import BUILT_IN
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def is_supported_target(target_model: str) -> bool:
|
|
return any(e.custom_field_target == target_model for e in BUILT_IN)
|
|
|
|
|
|
def validate_definition_options(field_type: str, options: Any) -> None:
|
|
"""Raise serializers.ValidationError for malformed dropdown options."""
|
|
if field_type != "dropdown":
|
|
if options not in (None, [], {}):
|
|
raise drf_serializers.ValidationError(
|
|
{"options": "options is only valid for dropdown fields"}
|
|
)
|
|
return
|
|
|
|
if not isinstance(options, list) or not options:
|
|
raise drf_serializers.ValidationError(
|
|
{
|
|
"options": "Dropdown fields require a non-empty list of {value, label} pairs"
|
|
}
|
|
)
|
|
|
|
seen: set[str] = set()
|
|
for idx, entry in enumerate(options):
|
|
if not isinstance(entry, dict):
|
|
raise drf_serializers.ValidationError(
|
|
{"options": f"Entry {idx} must be an object with value and label"}
|
|
)
|
|
value = entry.get("value")
|
|
label = entry.get("label")
|
|
if not isinstance(value, str) or not value:
|
|
raise drf_serializers.ValidationError(
|
|
{"options": f"Entry {idx} is missing a non-empty string value"}
|
|
)
|
|
if not isinstance(label, str) or not label:
|
|
raise drf_serializers.ValidationError(
|
|
{"options": f"Entry {idx} is missing a non-empty string label"}
|
|
)
|
|
if value in seen:
|
|
raise drf_serializers.ValidationError(
|
|
{"options": f"Duplicate dropdown value: {value!r}"}
|
|
)
|
|
seen.add(value)
|
|
|
|
|
|
def _coerce_value(field_type: str, raw: Any):
|
|
"""Coerce a raw input to the expected python/JSON type. Returns (value, error_or_None)."""
|
|
if raw is None or raw == "":
|
|
return None, None
|
|
|
|
if field_type in ("text", "textarea"):
|
|
return str(raw), None
|
|
|
|
if field_type == "number":
|
|
try:
|
|
if isinstance(raw, bool):
|
|
raise ValueError
|
|
if isinstance(raw, (int, float)):
|
|
return float(raw), None
|
|
return float(str(raw)), None
|
|
except (TypeError, ValueError):
|
|
return None, "must be a number"
|
|
|
|
if field_type == "checkbox":
|
|
if isinstance(raw, bool):
|
|
return raw, None
|
|
if isinstance(raw, str):
|
|
lowered = raw.strip().lower()
|
|
if lowered in ("true", "1", "yes", "on"):
|
|
return True, None
|
|
if lowered in ("false", "0", "no", "off"):
|
|
return False, None
|
|
return None, "must be a boolean"
|
|
|
|
if field_type == "date":
|
|
if isinstance(raw, date):
|
|
return raw.isoformat(), None
|
|
if isinstance(raw, str):
|
|
try:
|
|
return date.fromisoformat(raw).isoformat(), None
|
|
except ValueError:
|
|
return None, "must be a YYYY-MM-DD date"
|
|
return None, "must be a YYYY-MM-DD date"
|
|
|
|
if field_type == "dropdown":
|
|
return str(raw), None
|
|
|
|
return None, f"unsupported field_type {field_type!r}"
|
|
|
|
|
|
def validate_payload(
|
|
target_model: str,
|
|
value_dict: Any,
|
|
org,
|
|
*,
|
|
existing: dict | None = None,
|
|
) -> tuple[dict, dict]:
|
|
"""Validate and coerce a custom_fields payload for an entity.
|
|
|
|
Returns (cleaned_dict, errors_dict). Non-empty errors -> caller should 400.
|
|
Unknown keys are dropped silently and logged. `existing` is the value
|
|
currently stored on the entity, read from the database and never from the
|
|
request. Required fields already set on the entity are preserved across
|
|
PATCHes that omit them.
|
|
|
|
Values tied to soft-deleted (is_active=False) definitions are carried
|
|
forward on every save so admins can soft-delete a field without losing
|
|
history. Writes against an inactive key are rejected as unknown, and the
|
|
key arriving does not erase the stored value: the phone sends a record's
|
|
whole stored map back on save.
|
|
|
|
A value sent back exactly as stored is kept without being validated again,
|
|
so a record still holding a dropdown option an admin has since removed can
|
|
be saved. The stored copy is what is kept, so nothing the request carries
|
|
is written unchecked.
|
|
"""
|
|
from common.models import CustomFieldDefinition # avoid import cycle
|
|
|
|
if value_dict is None:
|
|
value_dict = {}
|
|
if not isinstance(value_dict, dict):
|
|
return {}, {"custom_fields": "must be an object"}
|
|
|
|
existing = existing or {}
|
|
|
|
all_definitions = list(
|
|
CustomFieldDefinition.objects.filter(org=org, target_model=target_model)
|
|
)
|
|
all_keys = {d.key for d in all_definitions}
|
|
active_by_key = {d.key: d for d in all_definitions if d.is_active}
|
|
|
|
cleaned: dict = {}
|
|
errors: dict = {}
|
|
|
|
# Carry forward existing recognized values the caller cannot change: the
|
|
# ones it left out, and soft-deleted ones whatever it sent. Values whose
|
|
# definition was hard-deleted are dropped. The schema is gone.
|
|
for key, value in existing.items():
|
|
if key in all_keys and (key not in value_dict or key not in active_by_key):
|
|
cleaned[key] = value
|
|
|
|
for key, raw in value_dict.items():
|
|
defn = active_by_key.get(key)
|
|
if defn is None:
|
|
logger.info(
|
|
"custom_fields: dropping unknown key %r on %s for org %s",
|
|
key,
|
|
target_model,
|
|
getattr(org, "id", org),
|
|
)
|
|
continue
|
|
if key in existing and raw == existing[key]:
|
|
cleaned[key] = existing[key]
|
|
continue
|
|
coerced, error = _coerce_value(defn.field_type, raw)
|
|
if error:
|
|
errors[key] = error
|
|
continue
|
|
if coerced is None or coerced == "":
|
|
cleaned.pop(key, None)
|
|
continue
|
|
if defn.field_type == "dropdown":
|
|
allowed = {opt.get("value") for opt in (defn.options or [])}
|
|
if coerced not in allowed:
|
|
errors[key] = f"must be one of {sorted(allowed)}"
|
|
continue
|
|
cleaned[key] = coerced
|
|
|
|
# Required fields: error if neither a new value nor an existing one is set.
|
|
for defn in active_by_key.values():
|
|
if not defn.is_required:
|
|
continue
|
|
present = cleaned.get(defn.key) not in (None, "")
|
|
if not present:
|
|
errors.setdefault(defn.key, "is required")
|
|
|
|
return cleaned, errors
|