mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
feat: Schema system for Basic Memory (#549)
Signed-off-by: phernandez <paul@basicmachines.co> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Paul Hernandez
@@ -0,0 +1,259 @@
|
||||
"""Schema validator for Basic Memory.
|
||||
|
||||
Validates a note's observations and relations against a resolved schema definition.
|
||||
The mapping rules ground schema fields in the existing Basic Memory note format:
|
||||
|
||||
Schema Declaration -> Grounded In
|
||||
-----------------------------------------------
|
||||
field: string -> observation [field] value
|
||||
field?(array): string -> multiple [field] observations
|
||||
field?: EntityType -> relation 'field [[Target]]'
|
||||
field?(array): EntityType -> multiple 'field' relations
|
||||
field?(enum): [values] -> observation [field] value where value is in set
|
||||
|
||||
Validation is soft by default (warn mode). Unmatched observations and relations
|
||||
are informational, not errors -- schemas are a subset, not a straitjacket.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field as dataclass_field
|
||||
|
||||
from basic_memory.schema.inference import ObservationData, RelationData
|
||||
from basic_memory.schema.parser import SchemaDefinition, SchemaField
|
||||
|
||||
|
||||
# --- Result Data Model ---
|
||||
|
||||
|
||||
@dataclass
|
||||
class FieldResult:
|
||||
"""Validation result for a single schema field."""
|
||||
|
||||
field: SchemaField
|
||||
status: str # "present" | "missing" | "enum_mismatch"
|
||||
values: list[str] = dataclass_field(default_factory=list) # Matched values
|
||||
message: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ValidationResult:
|
||||
"""Complete validation result for a note against a schema."""
|
||||
|
||||
note_identifier: str
|
||||
schema_entity: str
|
||||
passed: bool # True if no errors (warnings are OK)
|
||||
field_results: list[FieldResult] = dataclass_field(default_factory=list)
|
||||
unmatched_observations: dict[str, int] = dataclass_field(default_factory=dict) # cat -> count
|
||||
unmatched_relations: list[str] = dataclass_field(default_factory=list) # types not in schema
|
||||
warnings: list[str] = dataclass_field(default_factory=list)
|
||||
errors: list[str] = dataclass_field(default_factory=list)
|
||||
|
||||
|
||||
# --- Validation Logic ---
|
||||
|
||||
|
||||
def validate_note(
|
||||
note_identifier: str,
|
||||
schema: SchemaDefinition,
|
||||
observations: list[ObservationData],
|
||||
relations: list[RelationData],
|
||||
) -> ValidationResult:
|
||||
"""Validate a note against a schema definition.
|
||||
|
||||
Args:
|
||||
note_identifier: The note's title, permalink, or file path for reporting.
|
||||
schema: The resolved SchemaDefinition to validate against.
|
||||
observations: List of ObservationData from the note's observations.
|
||||
relations: List of RelationData from the note's relations.
|
||||
|
||||
Returns:
|
||||
A ValidationResult with per-field results, unmatched items, and warnings/errors.
|
||||
"""
|
||||
result = ValidationResult(
|
||||
note_identifier=note_identifier,
|
||||
schema_entity=schema.entity,
|
||||
passed=True,
|
||||
)
|
||||
|
||||
# Build lookup structures from the note's actual content
|
||||
obs_by_category = _group_observations(observations)
|
||||
rel_by_type = _group_relations(relations)
|
||||
|
||||
# Track which observation categories and relation types are matched by schema fields
|
||||
matched_categories: set[str] = set()
|
||||
matched_relation_types: set[str] = set()
|
||||
|
||||
# --- Validate each schema field ---
|
||||
for schema_field in schema.fields:
|
||||
field_result = _validate_field(schema_field, obs_by_category, rel_by_type)
|
||||
result.field_results.append(field_result)
|
||||
|
||||
# Track which categories/relation types this field consumed
|
||||
if schema_field.is_entity_ref:
|
||||
matched_relation_types.add(schema_field.name)
|
||||
else:
|
||||
matched_categories.add(schema_field.name)
|
||||
|
||||
# --- Generate warnings or errors based on validation mode ---
|
||||
# Trigger: field declared in schema but not found in note
|
||||
# Why: required missing = warning (or error in strict); optional missing = silent
|
||||
# Outcome: only required missing fields produce diagnostics
|
||||
if field_result.status == "missing" and schema_field.required:
|
||||
msg = _missing_field_message(schema_field)
|
||||
if schema.validation_mode == "strict":
|
||||
result.errors.append(msg)
|
||||
result.passed = False
|
||||
else:
|
||||
result.warnings.append(msg)
|
||||
|
||||
elif field_result.status == "enum_mismatch":
|
||||
msg = field_result.message or f"Field '{schema_field.name}' has invalid enum value"
|
||||
if schema.validation_mode == "strict":
|
||||
result.errors.append(msg)
|
||||
result.passed = False
|
||||
else:
|
||||
result.warnings.append(msg)
|
||||
|
||||
# --- Collect unmatched observations ---
|
||||
for category, values in obs_by_category.items():
|
||||
if category not in matched_categories:
|
||||
result.unmatched_observations[category] = len(values)
|
||||
|
||||
# --- Collect unmatched relations ---
|
||||
for rel_type in rel_by_type:
|
||||
if rel_type not in matched_relation_types:
|
||||
result.unmatched_relations.append(rel_type)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# --- Field Validation ---
|
||||
|
||||
|
||||
def _validate_field(
|
||||
schema_field: SchemaField,
|
||||
obs_by_category: dict[str, list[str]],
|
||||
rel_by_type: dict[str, list[str]],
|
||||
) -> FieldResult:
|
||||
"""Validate a single schema field against the note's data.
|
||||
|
||||
Entity ref fields map to relations; all other fields map to observations.
|
||||
"""
|
||||
# --- Entity reference fields map to relations ---
|
||||
if schema_field.is_entity_ref:
|
||||
return _validate_entity_ref_field(schema_field, rel_by_type)
|
||||
|
||||
# --- Enum fields require value membership check ---
|
||||
if schema_field.is_enum:
|
||||
return _validate_enum_field(schema_field, obs_by_category)
|
||||
|
||||
# --- Scalar and array fields map to observations ---
|
||||
return _validate_observation_field(schema_field, obs_by_category)
|
||||
|
||||
|
||||
def _validate_observation_field(
|
||||
schema_field: SchemaField,
|
||||
obs_by_category: dict[str, list[str]],
|
||||
) -> FieldResult:
|
||||
"""Validate a field that maps to observation categories."""
|
||||
values = obs_by_category.get(schema_field.name, [])
|
||||
|
||||
if not values:
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="missing",
|
||||
message=_missing_field_message(schema_field),
|
||||
)
|
||||
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="present",
|
||||
values=values,
|
||||
)
|
||||
|
||||
|
||||
def _validate_entity_ref_field(
|
||||
schema_field: SchemaField,
|
||||
rel_by_type: dict[str, list[str]],
|
||||
) -> FieldResult:
|
||||
"""Validate a field that maps to relations (entity references)."""
|
||||
targets = rel_by_type.get(schema_field.name, [])
|
||||
|
||||
if not targets:
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="missing",
|
||||
message=f"Missing relation: {schema_field.name} (no '{schema_field.name} [[...]]' "
|
||||
f"relation found)",
|
||||
)
|
||||
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="present",
|
||||
values=targets,
|
||||
)
|
||||
|
||||
|
||||
def _validate_enum_field(
|
||||
schema_field: SchemaField,
|
||||
obs_by_category: dict[str, list[str]],
|
||||
) -> FieldResult:
|
||||
"""Validate an enum field -- value must be in the allowed set."""
|
||||
values = obs_by_category.get(schema_field.name, [])
|
||||
|
||||
if not values:
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="missing",
|
||||
message=_missing_field_message(schema_field),
|
||||
)
|
||||
|
||||
# Check each value against the allowed enum values
|
||||
invalid_values = [v for v in values if v not in schema_field.enum_values]
|
||||
if invalid_values:
|
||||
allowed = ", ".join(schema_field.enum_values)
|
||||
invalid = ", ".join(invalid_values)
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="enum_mismatch",
|
||||
values=values,
|
||||
message=f"Field '{schema_field.name}' has invalid value(s): {invalid} "
|
||||
f"(allowed: {allowed})",
|
||||
)
|
||||
|
||||
return FieldResult(
|
||||
field=schema_field,
|
||||
status="present",
|
||||
values=values,
|
||||
)
|
||||
|
||||
|
||||
# --- Helper Functions ---
|
||||
|
||||
|
||||
def _group_observations(observations: list[ObservationData]) -> dict[str, list[str]]:
|
||||
"""Group observations by category."""
|
||||
result: dict[str, list[str]] = {}
|
||||
for obs in observations:
|
||||
result.setdefault(obs.category, []).append(obs.content)
|
||||
return result
|
||||
|
||||
|
||||
def _group_relations(relations: list[RelationData]) -> dict[str, list[str]]:
|
||||
"""Group relations by relation type."""
|
||||
result: dict[str, list[str]] = {}
|
||||
for rel in relations:
|
||||
result.setdefault(rel.relation_type, []).append(rel.target_name)
|
||||
return result
|
||||
|
||||
|
||||
def _missing_field_message(schema_field: SchemaField) -> str:
|
||||
"""Generate a human-readable message for a missing field."""
|
||||
kind = "required" if schema_field.required else "optional"
|
||||
|
||||
if schema_field.is_entity_ref:
|
||||
return (
|
||||
f"Missing {kind} field: {schema_field.name} "
|
||||
f"(no '{schema_field.name} [[...]]' relation found)"
|
||||
)
|
||||
|
||||
return f"Missing {kind} field: {schema_field.name} (expected [{schema_field.name}] observation)"
|
||||
Reference in New Issue
Block a user