|
|
@@ -0,0 +1,1142 @@
|
|
|
+"""Deterministic, cross-domain data quality profiling and SLA operations."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import copy
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import math
|
|
|
+import re
|
|
|
+import statistics
|
|
|
+import uuid
|
|
|
+from collections import Counter
|
|
|
+from collections.abc import Callable
|
|
|
+from datetime import datetime
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.common.timezone_utils import now_china
|
|
|
+
|
|
|
+MAX_RECORDS = 5_000
|
|
|
+MAX_FIELDS = 200
|
|
|
+MAX_CELLS = 200_000
|
|
|
+MAX_INLINE_CHARACTERS = 8_000_000
|
|
|
+MAX_SCALAR_CHARACTERS = 10_000
|
|
|
+MAX_ROLES = 30
|
|
|
+MAX_SAMPLE_LIMIT = 20
|
|
|
+ROLE_CHECKS = frozenset({"unique", "pattern", "distribution", "outlier"})
|
|
|
+TEMPLATE_KEYS = frozenset(
|
|
|
+ {"schema_version", "field_roles", "thresholds", "sample_limit"}
|
|
|
+)
|
|
|
+ROLE_KEYS = frozenset({"required", "checks"})
|
|
|
+THRESHOLD_KEYS = frozenset(
|
|
|
+ {
|
|
|
+ "completeness_min",
|
|
|
+ "uniqueness_min",
|
|
|
+ "pattern",
|
|
|
+ "volume_change_max_ratio",
|
|
|
+ "distribution_drift_max",
|
|
|
+ "freshness_max_seconds",
|
|
|
+ "quality_score_min",
|
|
|
+ }
|
|
|
+)
|
|
|
+SECRET_TOKENS = frozenset(
|
|
|
+ {
|
|
|
+ "apikey",
|
|
|
+ "authorization",
|
|
|
+ "connectionstring",
|
|
|
+ "credential",
|
|
|
+ "dsn",
|
|
|
+ "password",
|
|
|
+ "secret",
|
|
|
+ "token",
|
|
|
+ }
|
|
|
+)
|
|
|
+ROLE_PATTERN = re.compile(r"^[a-z][a-z0-9_]{0,62}$")
|
|
|
+FIELD_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_.]{0,199}$")
|
|
|
+CODE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{2,119}$")
|
|
|
+
|
|
|
+
|
|
|
+def _normalized_key(value: Any) -> str:
|
|
|
+ return re.sub(r"[^a-z0-9]", "", str(value).casefold())
|
|
|
+
|
|
|
+
|
|
|
+def _reject_secret_material(value: Any, path: str = "$") -> None:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ for key, item in value.items():
|
|
|
+ normalized = _normalized_key(key)
|
|
|
+ if any(token in normalized for token in SECRET_TOKENS):
|
|
|
+ raise ValueError(f"secret material is not allowed at {path}.{key}")
|
|
|
+ _reject_secret_material(item, f"{path}.{key}")
|
|
|
+ elif isinstance(value, list):
|
|
|
+ for index, item in enumerate(value):
|
|
|
+ _reject_secret_material(item, f"{path}[{index}]")
|
|
|
+
|
|
|
+
|
|
|
+def _closed_object(
|
|
|
+ value: Any,
|
|
|
+ allowed: frozenset[str] | set[str],
|
|
|
+ label: str,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ if not isinstance(value, dict):
|
|
|
+ raise ValueError(f"{label} must be an object")
|
|
|
+ _reject_secret_material(value)
|
|
|
+ unknown = sorted(set(value) - set(allowed))
|
|
|
+ if unknown:
|
|
|
+ raise ValueError(
|
|
|
+ f"{label} contains unsupported fields: {', '.join(unknown)}"
|
|
|
+ )
|
|
|
+ return copy.deepcopy(value)
|
|
|
+
|
|
|
+
|
|
|
+def _bounded_string(value: Any, label: str, maximum: int) -> str:
|
|
|
+ if not isinstance(value, str) or not value.strip():
|
|
|
+ raise ValueError(f"{label} is required")
|
|
|
+ normalized = value.strip()
|
|
|
+ if len(normalized) > maximum:
|
|
|
+ raise ValueError(f"{label} exceeds {maximum} characters")
|
|
|
+ return normalized
|
|
|
+
|
|
|
+
|
|
|
+def _uid(value: Any, label: str) -> str:
|
|
|
+ try:
|
|
|
+ return str(uuid.UUID(str(value)))
|
|
|
+ except (TypeError, ValueError, AttributeError) as error:
|
|
|
+ raise ValueError(f"{label} must be a UUID") from error
|
|
|
+
|
|
|
+
|
|
|
+def _ratio(value: Any, label: str, *, maximum: float = 1.0) -> float:
|
|
|
+ if isinstance(value, bool):
|
|
|
+ raise ValueError(f"{label} must be numeric")
|
|
|
+ try:
|
|
|
+ number = float(value)
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be numeric") from error
|
|
|
+ if not math.isfinite(number) or number < 0 or number > maximum:
|
|
|
+ raise ValueError(f"{label} must be between 0 and {maximum}")
|
|
|
+ return number
|
|
|
+
|
|
|
+
|
|
|
+def _positive_number(value: Any, label: str, *, maximum: float) -> float:
|
|
|
+ if isinstance(value, bool):
|
|
|
+ raise ValueError(f"{label} must be numeric")
|
|
|
+ try:
|
|
|
+ number = float(value)
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be numeric") from error
|
|
|
+ if not math.isfinite(number) or number <= 0 or number > maximum:
|
|
|
+ raise ValueError(f"{label} must be between 0 and {maximum}")
|
|
|
+ return number
|
|
|
+
|
|
|
+
|
|
|
+def _canonical(value: Any) -> str:
|
|
|
+ try:
|
|
|
+ return json.dumps(
|
|
|
+ value,
|
|
|
+ ensure_ascii=False,
|
|
|
+ sort_keys=True,
|
|
|
+ separators=(",", ":"),
|
|
|
+ )
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError("quality data must be JSON serializable") from error
|
|
|
+
|
|
|
+
|
|
|
+def _content_hash(value: Any) -> str:
|
|
|
+ return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def _parse_datetime(value: Any, label: str) -> datetime:
|
|
|
+ if isinstance(value, datetime):
|
|
|
+ parsed = value
|
|
|
+ else:
|
|
|
+ try:
|
|
|
+ parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be an ISO-8601 datetime") from error
|
|
|
+ if parsed.tzinfo is None:
|
|
|
+ raise ValueError(f"{label} must include a timezone")
|
|
|
+ return parsed
|
|
|
+
|
|
|
+
|
|
|
+def validate_quality_template(value: Any) -> dict[str, Any]:
|
|
|
+ """Validate a domain-neutral quality template with semantic field roles."""
|
|
|
+
|
|
|
+ template = _closed_object(value, TEMPLATE_KEYS, "quality template")
|
|
|
+ if template.get("schema_version") != "1.0":
|
|
|
+ raise ValueError("quality template schema_version must be 1.0")
|
|
|
+ raw_roles = template.get("field_roles")
|
|
|
+ if not isinstance(raw_roles, dict) or not 1 <= len(raw_roles) <= MAX_ROLES:
|
|
|
+ raise ValueError("field_roles must contain between 1 and 30 roles")
|
|
|
+ roles = {}
|
|
|
+ for raw_name, raw_role in raw_roles.items():
|
|
|
+ name = str(raw_name)
|
|
|
+ if not ROLE_PATTERN.fullmatch(name):
|
|
|
+ raise ValueError("quality field role name is invalid")
|
|
|
+ role = _closed_object(raw_role, ROLE_KEYS, f"field role {name}")
|
|
|
+ if not isinstance(role.get("required"), bool):
|
|
|
+ raise ValueError(f"field role {name} required must be a boolean")
|
|
|
+ checks = role.get("checks")
|
|
|
+ if not isinstance(checks, list):
|
|
|
+ raise ValueError(f"field role {name} checks must be an array")
|
|
|
+ normalized_checks = sorted({str(item) for item in checks})
|
|
|
+ if not set(normalized_checks) <= ROLE_CHECKS:
|
|
|
+ raise ValueError(f"field role {name} contains unsupported checks")
|
|
|
+ roles[name] = {
|
|
|
+ "required": role["required"],
|
|
|
+ "checks": normalized_checks,
|
|
|
+ }
|
|
|
+ thresholds = _closed_object(
|
|
|
+ template.get("thresholds"),
|
|
|
+ THRESHOLD_KEYS,
|
|
|
+ "quality thresholds",
|
|
|
+ )
|
|
|
+ required_thresholds = THRESHOLD_KEYS
|
|
|
+ missing = sorted(required_thresholds - set(thresholds))
|
|
|
+ if missing:
|
|
|
+ raise ValueError(
|
|
|
+ f"quality thresholds are missing fields: {', '.join(missing)}"
|
|
|
+ )
|
|
|
+ thresholds["completeness_min"] = _ratio(
|
|
|
+ thresholds["completeness_min"], "completeness_min"
|
|
|
+ )
|
|
|
+ thresholds["uniqueness_min"] = _ratio(
|
|
|
+ thresholds["uniqueness_min"], "uniqueness_min"
|
|
|
+ )
|
|
|
+ thresholds["volume_change_max_ratio"] = _ratio(
|
|
|
+ thresholds["volume_change_max_ratio"], "volume_change_max_ratio"
|
|
|
+ )
|
|
|
+ thresholds["distribution_drift_max"] = _ratio(
|
|
|
+ thresholds["distribution_drift_max"], "distribution_drift_max"
|
|
|
+ )
|
|
|
+ thresholds["freshness_max_seconds"] = _positive_number(
|
|
|
+ thresholds["freshness_max_seconds"],
|
|
|
+ "freshness_max_seconds",
|
|
|
+ maximum=31_536_000,
|
|
|
+ )
|
|
|
+ thresholds["quality_score_min"] = _ratio(
|
|
|
+ thresholds["quality_score_min"],
|
|
|
+ "quality_score_min",
|
|
|
+ maximum=100,
|
|
|
+ )
|
|
|
+ pattern = _bounded_string(thresholds["pattern"], "pattern", 120)
|
|
|
+ if (
|
|
|
+ not pattern.startswith("^")
|
|
|
+ or not pattern.endswith("$")
|
|
|
+ or "(?" in pattern
|
|
|
+ or re.search(r"\\[1-9]", pattern)
|
|
|
+ or re.search(r"\([^)]*[+*{][^)]*\)[+*{]", pattern)
|
|
|
+ or ".*.*" in pattern
|
|
|
+ ):
|
|
|
+ raise ValueError("pattern must be anchored and cannot use advanced groups")
|
|
|
+ try:
|
|
|
+ re.compile(pattern)
|
|
|
+ except re.error as error:
|
|
|
+ raise ValueError("pattern is invalid") from error
|
|
|
+ thresholds["pattern"] = pattern
|
|
|
+ sample_limit = template.get("sample_limit")
|
|
|
+ if (
|
|
|
+ isinstance(sample_limit, bool)
|
|
|
+ or not isinstance(sample_limit, int)
|
|
|
+ or not 1 <= sample_limit <= MAX_SAMPLE_LIMIT
|
|
|
+ ):
|
|
|
+ raise ValueError("sample_limit must be between 1 and 20")
|
|
|
+ return {
|
|
|
+ "schema_version": "1.0",
|
|
|
+ "field_roles": roles,
|
|
|
+ "thresholds": thresholds,
|
|
|
+ "sample_limit": sample_limit,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _scalar(value: Any) -> bool:
|
|
|
+ return value is None or isinstance(value, (str, int, float, bool))
|
|
|
+
|
|
|
+
|
|
|
+def _normalized_records(value: Any) -> list[dict[str, Any]]:
|
|
|
+ if not isinstance(value, list) or not 1 <= len(value) <= MAX_RECORDS:
|
|
|
+ raise ValueError("records must contain between 1 and 5000 rows")
|
|
|
+ if sum(len(item) for item in value if isinstance(item, dict)) > MAX_CELLS:
|
|
|
+ raise ValueError("quality records contain too many cells")
|
|
|
+ field_names = set()
|
|
|
+ records = []
|
|
|
+ inline_characters = 0
|
|
|
+ for raw in value:
|
|
|
+ if not isinstance(raw, dict):
|
|
|
+ raise ValueError("each quality record must be an object")
|
|
|
+ if len(raw) > MAX_FIELDS:
|
|
|
+ raise ValueError("a quality record contains too many fields")
|
|
|
+ record = {}
|
|
|
+ for raw_field, item in raw.items():
|
|
|
+ field = str(raw_field)
|
|
|
+ if not FIELD_PATTERN.fullmatch(field):
|
|
|
+ raise ValueError("quality record field name is invalid")
|
|
|
+ if not _scalar(item):
|
|
|
+ raise ValueError("quality record values must be scalar")
|
|
|
+ if isinstance(item, float) and not math.isfinite(item):
|
|
|
+ raise ValueError("quality record numeric values must be finite")
|
|
|
+ if (
|
|
|
+ isinstance(item, int)
|
|
|
+ and not isinstance(item, bool)
|
|
|
+ and item.bit_length() > 256
|
|
|
+ ):
|
|
|
+ raise ValueError("quality record integer value is too large")
|
|
|
+ if isinstance(item, str):
|
|
|
+ if len(item) > MAX_SCALAR_CHARACTERS:
|
|
|
+ raise ValueError("quality record scalar value is too large")
|
|
|
+ inline_characters += len(item)
|
|
|
+ if inline_characters > MAX_INLINE_CHARACTERS:
|
|
|
+ raise ValueError("quality records inline content is too large")
|
|
|
+ record[field] = item
|
|
|
+ field_names.add(field)
|
|
|
+ records.append(record)
|
|
|
+ if len(field_names) > MAX_FIELDS:
|
|
|
+ raise ValueError("quality records contain too many fields")
|
|
|
+ return records
|
|
|
+
|
|
|
+
|
|
|
+def _value_token(value: Any) -> str:
|
|
|
+ digest = hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()[:16]
|
|
|
+ return f"sha256:{digest}"
|
|
|
+
|
|
|
+
|
|
|
+def _type_name(value: Any) -> str:
|
|
|
+ if value is None:
|
|
|
+ return "null"
|
|
|
+ if isinstance(value, bool):
|
|
|
+ return "boolean"
|
|
|
+ if isinstance(value, int):
|
|
|
+ return "integer"
|
|
|
+ if isinstance(value, float):
|
|
|
+ return "number"
|
|
|
+ return "string"
|
|
|
+
|
|
|
+
|
|
|
+def _field_profile(
|
|
|
+ records: list[dict[str, Any]],
|
|
|
+ field: str,
|
|
|
+ *,
|
|
|
+ sample_limit: int,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ values = [item.get(field) for item in records]
|
|
|
+ populated = [item for item in values if item is not None and item != ""]
|
|
|
+ counts = Counter(_canonical(item) for item in populated)
|
|
|
+ originals = {}
|
|
|
+ for item in populated:
|
|
|
+ originals.setdefault(_canonical(item), item)
|
|
|
+ top_values = [
|
|
|
+ {"value_token": _value_token(originals[key]), "count": count}
|
|
|
+ for key, count in sorted(
|
|
|
+ counts.items(), key=lambda item: (-item[1], item[0])
|
|
|
+ )[:sample_limit]
|
|
|
+ ]
|
|
|
+ type_counts = Counter(_type_name(item) for item in values)
|
|
|
+ return {
|
|
|
+ "field_name": field,
|
|
|
+ "row_count": len(records),
|
|
|
+ "null_count": len(values) - len(populated),
|
|
|
+ "null_rate": round(
|
|
|
+ (len(values) - len(populated)) / len(records),
|
|
|
+ 6,
|
|
|
+ ),
|
|
|
+ "completeness_rate": round(len(populated) / len(records), 6),
|
|
|
+ "distinct_count": len(counts),
|
|
|
+ "uniqueness_rate": round(
|
|
|
+ len(counts) / len(populated), 6
|
|
|
+ )
|
|
|
+ if populated
|
|
|
+ else 0.0,
|
|
|
+ "type_counts": dict(sorted(type_counts.items())),
|
|
|
+ "top_values": top_values,
|
|
|
+ "sample_tokens": sorted({_value_token(item) for item in populated})[
|
|
|
+ :sample_limit
|
|
|
+ ],
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _distribution(profile: dict[str, Any]) -> dict[str, float]:
|
|
|
+ total = sum(item["count"] for item in profile["top_values"])
|
|
|
+ if total <= 0:
|
|
|
+ return {}
|
|
|
+ return {
|
|
|
+ item["value_token"]: item["count"] / total
|
|
|
+ for item in profile["top_values"]
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _distribution_distance(
|
|
|
+ current: dict[str, float],
|
|
|
+ previous: dict[str, float],
|
|
|
+) -> float:
|
|
|
+ keys = set(current) | set(previous)
|
|
|
+ return round(
|
|
|
+ 0.5
|
|
|
+ * sum(abs(current.get(key, 0.0) - previous.get(key, 0.0)) for key in keys),
|
|
|
+ 6,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _schema_signature(profiles: dict[str, dict[str, Any]]) -> dict[str, Any]:
|
|
|
+ return {
|
|
|
+ field: {
|
|
|
+ "types": sorted(
|
|
|
+ key for key in profile["type_counts"] if key != "null"
|
|
|
+ ),
|
|
|
+ }
|
|
|
+ for field, profile in sorted(profiles.items())
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _outlier_tokens(values: list[Any], sample_limit: int) -> list[str]:
|
|
|
+ numeric = [
|
|
|
+ float(item)
|
|
|
+ for item in values
|
|
|
+ if isinstance(item, (int, float)) and not isinstance(item, bool)
|
|
|
+ ]
|
|
|
+ if len(numeric) < 4:
|
|
|
+ return []
|
|
|
+ median = statistics.median(numeric)
|
|
|
+ deviations = [abs(item - median) for item in numeric]
|
|
|
+ mad = statistics.median(deviations)
|
|
|
+ if mad == 0:
|
|
|
+ outliers = [item for item in numeric if item != median]
|
|
|
+ else:
|
|
|
+ outliers = [
|
|
|
+ item
|
|
|
+ for item in numeric
|
|
|
+ if 0.6745 * abs(item - median) / mad > 3.5
|
|
|
+ ]
|
|
|
+ return sorted({_value_token(item) for item in outliers})[:sample_limit]
|
|
|
+
|
|
|
+
|
|
|
+class QualityOperationsService:
|
|
|
+ """Govern quality templates and persist deterministic operational evidence."""
|
|
|
+
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ publish_authorizer: Callable[[str], None],
|
|
|
+ uid_factory: Callable[[], str] = new_governance_uid,
|
|
|
+ now_factory: Callable[[], datetime] = now_china,
|
|
|
+ commit: Callable[[], None] = lambda: None,
|
|
|
+ rollback: Callable[[], None] = lambda: None,
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.publish_authorizer = publish_authorizer
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+ self.commit = commit
|
|
|
+ self.rollback = rollback
|
|
|
+
|
|
|
+ def create_template(
|
|
|
+ self,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ body = _closed_object(
|
|
|
+ payload,
|
|
|
+ {"code", "name", "owner_uid", "definition"},
|
|
|
+ "quality template request",
|
|
|
+ )
|
|
|
+ code = _bounded_string(body.get("code"), "code", 120).upper()
|
|
|
+ if not CODE_PATTERN.fullmatch(code):
|
|
|
+ raise ValueError("quality template code is invalid")
|
|
|
+ name = _bounded_string(body.get("name"), "name", 300)
|
|
|
+ owner_uid = _uid(body.get("owner_uid"), "owner_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ definition = validate_quality_template(body.get("definition"))
|
|
|
+ now = self.now_factory()
|
|
|
+ template_uid = self.uid_factory()
|
|
|
+ version_uid = self.uid_factory()
|
|
|
+ template = {
|
|
|
+ "uid": template_uid,
|
|
|
+ "code": code,
|
|
|
+ "name": name,
|
|
|
+ "owner_uid": owner_uid,
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": 1,
|
|
|
+ "active_version_uid": None,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ version = {
|
|
|
+ "uid": version_uid,
|
|
|
+ "template_uid": template_uid,
|
|
|
+ "version": 1,
|
|
|
+ "status": "draft",
|
|
|
+ "definition": definition,
|
|
|
+ "content_hash": _content_hash(definition),
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ "published_by": None,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.create_template(template, version)
|
|
|
+ self.commit()
|
|
|
+ return {**template, "latest_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def revise_template(
|
|
|
+ self,
|
|
|
+ template_uid: str,
|
|
|
+ definition: Any,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ uid = _uid(template_uid, "template_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ template = self.repository.get_template(uid, for_update=True)
|
|
|
+ if template is None:
|
|
|
+ raise LookupError("quality template was not found")
|
|
|
+ if int(template["current_version"]) != int(expected_version):
|
|
|
+ raise RuntimeError("quality template version conflict")
|
|
|
+ normalized = validate_quality_template(definition)
|
|
|
+ version_number = int(template["current_version"]) + 1
|
|
|
+ now = self.now_factory()
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "template_uid": uid,
|
|
|
+ "version": version_number,
|
|
|
+ "status": "draft",
|
|
|
+ "definition": normalized,
|
|
|
+ "content_hash": _content_hash(normalized),
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ "published_by": None,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ updated = {
|
|
|
+ **template,
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": version_number,
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.save_template_version(updated, version)
|
|
|
+ self.commit()
|
|
|
+ return {**updated, "latest_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def publish_template(
|
|
|
+ self,
|
|
|
+ template_uid: str,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ uid = _uid(template_uid, "template_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ self.publish_authorizer(actor)
|
|
|
+ template = self.repository.get_template(uid, for_update=True)
|
|
|
+ if template is None:
|
|
|
+ raise LookupError("quality template was not found")
|
|
|
+ if int(template["current_version"]) != int(expected_version):
|
|
|
+ raise RuntimeError("quality template version conflict")
|
|
|
+ version = next(
|
|
|
+ (
|
|
|
+ item
|
|
|
+ for item in self.repository.versions_for_template(uid)
|
|
|
+ if int(item["version"]) == int(expected_version)
|
|
|
+ ),
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ if version is None:
|
|
|
+ raise LookupError("quality template version was not found")
|
|
|
+ now = self.now_factory()
|
|
|
+ published_version = {
|
|
|
+ **version,
|
|
|
+ "status": "published",
|
|
|
+ "published_by": actor,
|
|
|
+ "published_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ published = {
|
|
|
+ **template,
|
|
|
+ "status": "published",
|
|
|
+ "active_version_uid": version["uid"],
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.publish_template(published, published_version)
|
|
|
+ self.commit()
|
|
|
+ return {**published, "active_version": published_version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_templates(self) -> list[dict[str, Any]]:
|
|
|
+ records = []
|
|
|
+ for template in self.repository.list_templates():
|
|
|
+ versions = self.repository.versions_for_template(template["uid"])
|
|
|
+ records.append(
|
|
|
+ {
|
|
|
+ **template,
|
|
|
+ "latest_version": versions[0] if versions else None,
|
|
|
+ "active_version": next(
|
|
|
+ (
|
|
|
+ item
|
|
|
+ for item in versions
|
|
|
+ if item["uid"] == template.get("active_version_uid")
|
|
|
+ ),
|
|
|
+ None,
|
|
|
+ ),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return records
|
|
|
+
|
|
|
+ def _binding(
|
|
|
+ self,
|
|
|
+ definition: dict[str, Any],
|
|
|
+ value: Any,
|
|
|
+ ) -> dict[str, str]:
|
|
|
+ if not isinstance(value, dict):
|
|
|
+ raise ValueError("field_bindings must be an object")
|
|
|
+ if set(value) != set(definition["field_roles"]):
|
|
|
+ raise ValueError("field_bindings must match template field roles")
|
|
|
+ result = {}
|
|
|
+ for role, field in value.items():
|
|
|
+ if not FIELD_PATTERN.fullmatch(str(field)):
|
|
|
+ raise ValueError("field binding is invalid")
|
|
|
+ result[str(role)] = str(field)
|
|
|
+ if len(set(result.values())) != len(result):
|
|
|
+ raise ValueError("field bindings must be unique")
|
|
|
+ return result
|
|
|
+
|
|
|
+ def execute(
|
|
|
+ self,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ body = _closed_object(
|
|
|
+ payload,
|
|
|
+ {
|
|
|
+ "template_uid",
|
|
|
+ "asset_uid",
|
|
|
+ "batch_key",
|
|
|
+ "field_bindings",
|
|
|
+ "source_observed_at",
|
|
|
+ "records",
|
|
|
+ },
|
|
|
+ "quality execution request",
|
|
|
+ )
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ template_uid = _uid(body.get("template_uid"), "template_uid")
|
|
|
+ asset_uid = _uid(body.get("asset_uid"), "asset_uid")
|
|
|
+ batch_key = _bounded_string(body.get("batch_key"), "batch_key", 160)
|
|
|
+ for existing in self.repository.list_runs(asset_uid):
|
|
|
+ if existing["batch_key"] == batch_key:
|
|
|
+ return existing
|
|
|
+ template = self.repository.get_template(template_uid)
|
|
|
+ if template is None or template.get("status") != "published":
|
|
|
+ raise RuntimeError("quality template must be published before execution")
|
|
|
+ version = self.repository.get_template_version(
|
|
|
+ template.get("active_version_uid")
|
|
|
+ )
|
|
|
+ if version is None or version.get("status") != "published":
|
|
|
+ raise RuntimeError("published quality template version was not found")
|
|
|
+ definition = validate_quality_template(version["definition"])
|
|
|
+ bindings = self._binding(definition, body.get("field_bindings"))
|
|
|
+ records = _normalized_records(body.get("records"))
|
|
|
+ bound_fields = set(bindings.values())
|
|
|
+ missing_fields = sorted(
|
|
|
+ field for field in bound_fields if all(field not in row for row in records)
|
|
|
+ )
|
|
|
+ if missing_fields:
|
|
|
+ raise ValueError(
|
|
|
+ f"bound fields are absent from records: {', '.join(missing_fields)}"
|
|
|
+ )
|
|
|
+ asset = self.repository.get_asset(asset_uid)
|
|
|
+ if asset is None:
|
|
|
+ raise LookupError("active metadata asset was not found")
|
|
|
+ source_observed_at = _parse_datetime(
|
|
|
+ body.get("source_observed_at"),
|
|
|
+ "source_observed_at",
|
|
|
+ )
|
|
|
+ now = self.now_factory()
|
|
|
+ if source_observed_at > now:
|
|
|
+ raise ValueError("source_observed_at cannot be in the future")
|
|
|
+ profiles = {
|
|
|
+ field: _field_profile(
|
|
|
+ records,
|
|
|
+ field,
|
|
|
+ sample_limit=definition["sample_limit"],
|
|
|
+ )
|
|
|
+ for field in sorted({key for row in records for key in row})
|
|
|
+ }
|
|
|
+ compiled_pattern = re.compile(definition["thresholds"]["pattern"])
|
|
|
+ for role, role_spec in definition["field_roles"].items():
|
|
|
+ if "pattern" not in role_spec["checks"]:
|
|
|
+ continue
|
|
|
+ field = bindings[role]
|
|
|
+ values = [
|
|
|
+ row.get(field)
|
|
|
+ for row in records
|
|
|
+ if row.get(field) not in (None, "")
|
|
|
+ ]
|
|
|
+ mismatches = [
|
|
|
+ item
|
|
|
+ for item in values
|
|
|
+ if not compiled_pattern.fullmatch(str(item))
|
|
|
+ ]
|
|
|
+ profiles[field]["pattern"] = {
|
|
|
+ "evaluated_count": len(values),
|
|
|
+ "mismatch_count": len(mismatches),
|
|
|
+ "match_rate": (
|
|
|
+ round((len(values) - len(mismatches)) / len(values), 6)
|
|
|
+ if values
|
|
|
+ else 0.0
|
|
|
+ ),
|
|
|
+ "mismatch_sample_tokens": sorted(
|
|
|
+ {_value_token(item) for item in mismatches}
|
|
|
+ )[: definition["sample_limit"]],
|
|
|
+ }
|
|
|
+ previous = self.repository.latest_run(asset_uid)
|
|
|
+ root_cause = self.repository.related_evidence(asset)
|
|
|
+ root_cause["responsibility"] = {
|
|
|
+ "owner_uid": template["owner_uid"],
|
|
|
+ "source": "quality_template",
|
|
|
+ }
|
|
|
+ run_uid = self.uid_factory()
|
|
|
+ findings = self._findings(
|
|
|
+ run_uid=run_uid,
|
|
|
+ template=template,
|
|
|
+ definition=definition,
|
|
|
+ bindings=bindings,
|
|
|
+ records=records,
|
|
|
+ profiles=profiles,
|
|
|
+ previous=previous,
|
|
|
+ source_observed_at=source_observed_at,
|
|
|
+ now=now,
|
|
|
+ root_cause=root_cause,
|
|
|
+ )
|
|
|
+ score = round(max(0.0, 100.0 - len(findings) * 12.0), 2)
|
|
|
+ comparison = self._comparison(
|
|
|
+ row_count=len(records),
|
|
|
+ score=score,
|
|
|
+ previous=previous,
|
|
|
+ )
|
|
|
+ run = {
|
|
|
+ "uid": run_uid,
|
|
|
+ "template_uid": template_uid,
|
|
|
+ "template_version_uid": version["uid"],
|
|
|
+ "template_hash": version["content_hash"],
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "source_uid": asset["source_uid"],
|
|
|
+ "business_domain_uid": asset.get("business_domain_uid"),
|
|
|
+ "batch_key": batch_key,
|
|
|
+ "status": "success",
|
|
|
+ "row_count": len(records),
|
|
|
+ "score": score,
|
|
|
+ "source_observed_at": source_observed_at.isoformat(),
|
|
|
+ "previous_run_uid": previous["uid"] if previous else None,
|
|
|
+ "comparison": comparison,
|
|
|
+ "profile": {
|
|
|
+ "fields": profiles,
|
|
|
+ "schema": _schema_signature(profiles),
|
|
|
+ },
|
|
|
+ "field_bindings": bindings,
|
|
|
+ "finding_count": len(findings),
|
|
|
+ "deterministic": True,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ events = self._sla_events(run, definition, findings)
|
|
|
+ metrics = [
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ **profile,
|
|
|
+ }
|
|
|
+ for profile in profiles.values()
|
|
|
+ ]
|
|
|
+ try:
|
|
|
+ self.repository.save_run(
|
|
|
+ run,
|
|
|
+ metrics,
|
|
|
+ findings,
|
|
|
+ events,
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return run
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def _finding(
|
|
|
+ self,
|
|
|
+ *,
|
|
|
+ run_uid: str,
|
|
|
+ template_uid: str,
|
|
|
+ asset_uid: str,
|
|
|
+ finding_type: str,
|
|
|
+ field_name: str | None,
|
|
|
+ severity: str,
|
|
|
+ actual: Any,
|
|
|
+ expected: Any,
|
|
|
+ root_cause: dict[str, Any],
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ recurrence_key = _content_hash(
|
|
|
+ {
|
|
|
+ "template_uid": template_uid,
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "finding_type": finding_type,
|
|
|
+ "field_name": field_name,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ occurrence = self.repository.recurrence_count(recurrence_key) + 1
|
|
|
+ return {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "finding_type": finding_type,
|
|
|
+ "field_name": field_name,
|
|
|
+ "severity": severity,
|
|
|
+ "status": "open",
|
|
|
+ "actual": copy.deepcopy(actual),
|
|
|
+ "expected": copy.deepcopy(expected),
|
|
|
+ "recurrence_key": recurrence_key,
|
|
|
+ "occurrence_number": occurrence,
|
|
|
+ "evidence": {
|
|
|
+ "deterministic": True,
|
|
|
+ "root_cause": copy.deepcopy(root_cause),
|
|
|
+ },
|
|
|
+ "created_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+
|
|
|
+ def _findings(
|
|
|
+ self,
|
|
|
+ *,
|
|
|
+ run_uid,
|
|
|
+ template,
|
|
|
+ definition,
|
|
|
+ bindings,
|
|
|
+ records,
|
|
|
+ profiles,
|
|
|
+ previous,
|
|
|
+ source_observed_at,
|
|
|
+ now,
|
|
|
+ root_cause,
|
|
|
+ ):
|
|
|
+ thresholds = definition["thresholds"]
|
|
|
+ findings = []
|
|
|
+
|
|
|
+ def add(kind, field, actual, expected, severity="error"):
|
|
|
+ findings.append(
|
|
|
+ self._finding(
|
|
|
+ run_uid=run_uid,
|
|
|
+ template_uid=template["uid"],
|
|
|
+ asset_uid=root_cause["asset"]["uid"],
|
|
|
+ finding_type=kind,
|
|
|
+ field_name=field,
|
|
|
+ severity=severity,
|
|
|
+ actual=actual,
|
|
|
+ expected=expected,
|
|
|
+ root_cause=root_cause,
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+ for role, role_spec in definition["field_roles"].items():
|
|
|
+ field = bindings[role]
|
|
|
+ profile = profiles[field]
|
|
|
+ if (
|
|
|
+ role_spec["required"]
|
|
|
+ and profile["completeness_rate"]
|
|
|
+ < thresholds["completeness_min"]
|
|
|
+ ):
|
|
|
+ add(
|
|
|
+ "completeness",
|
|
|
+ field,
|
|
|
+ profile["completeness_rate"],
|
|
|
+ {"minimum": thresholds["completeness_min"]},
|
|
|
+ )
|
|
|
+ if (
|
|
|
+ "unique" in role_spec["checks"]
|
|
|
+ and profile["uniqueness_rate"] < thresholds["uniqueness_min"]
|
|
|
+ ):
|
|
|
+ add(
|
|
|
+ "uniqueness",
|
|
|
+ field,
|
|
|
+ profile["uniqueness_rate"],
|
|
|
+ {"minimum": thresholds["uniqueness_min"]},
|
|
|
+ )
|
|
|
+ duplicate_count = sum(
|
|
|
+ count - 1
|
|
|
+ for count in Counter(
|
|
|
+ _canonical(row.get(field))
|
|
|
+ for row in records
|
|
|
+ if row.get(field) not in (None, "")
|
|
|
+ ).values()
|
|
|
+ if count > 1
|
|
|
+ )
|
|
|
+ if duplicate_count:
|
|
|
+ add(
|
|
|
+ "duplicate",
|
|
|
+ field,
|
|
|
+ {"duplicate_row_count": duplicate_count},
|
|
|
+ {"duplicate_row_count": 0},
|
|
|
+ )
|
|
|
+ if "pattern" in role_spec["checks"]:
|
|
|
+ pattern_profile = profile["pattern"]
|
|
|
+ if pattern_profile["mismatch_count"]:
|
|
|
+ add(
|
|
|
+ "pattern",
|
|
|
+ field,
|
|
|
+ {
|
|
|
+ "mismatch_count": pattern_profile["mismatch_count"],
|
|
|
+ "match_rate": pattern_profile["match_rate"],
|
|
|
+ "sample_tokens": pattern_profile[
|
|
|
+ "mismatch_sample_tokens"
|
|
|
+ ],
|
|
|
+ },
|
|
|
+ {"pattern": thresholds["pattern"]},
|
|
|
+ )
|
|
|
+ if "outlier" in role_spec["checks"]:
|
|
|
+ tokens = _outlier_tokens(
|
|
|
+ [row.get(field) for row in records],
|
|
|
+ definition["sample_limit"],
|
|
|
+ )
|
|
|
+ if tokens:
|
|
|
+ add(
|
|
|
+ "outlier",
|
|
|
+ field,
|
|
|
+ {"sample_tokens": tokens},
|
|
|
+ {"outlier_count": 0},
|
|
|
+ severity="warning",
|
|
|
+ )
|
|
|
+ freshness_seconds = max(0.0, (now - source_observed_at).total_seconds())
|
|
|
+ if freshness_seconds > thresholds["freshness_max_seconds"]:
|
|
|
+ add(
|
|
|
+ "freshness",
|
|
|
+ None,
|
|
|
+ {"seconds": round(freshness_seconds, 3)},
|
|
|
+ {"maximum_seconds": thresholds["freshness_max_seconds"]},
|
|
|
+ severity="critical",
|
|
|
+ )
|
|
|
+ if previous is not None:
|
|
|
+ previous_count = int(previous["row_count"])
|
|
|
+ ratio = (
|
|
|
+ abs(len(records) - previous_count) / previous_count
|
|
|
+ if previous_count
|
|
|
+ else 1.0
|
|
|
+ )
|
|
|
+ if ratio > thresholds["volume_change_max_ratio"]:
|
|
|
+ add(
|
|
|
+ "volume",
|
|
|
+ None,
|
|
|
+ {"row_count": len(records), "change_ratio": round(ratio, 6)},
|
|
|
+ {
|
|
|
+ "previous_row_count": previous_count,
|
|
|
+ "maximum_change_ratio": thresholds[
|
|
|
+ "volume_change_max_ratio"
|
|
|
+ ],
|
|
|
+ },
|
|
|
+ )
|
|
|
+ previous_profiles = previous.get("profile", {}).get("fields", {})
|
|
|
+ for role, role_spec in definition["field_roles"].items():
|
|
|
+ if "distribution" not in role_spec["checks"]:
|
|
|
+ continue
|
|
|
+ field = bindings[role]
|
|
|
+ previous_profile = previous_profiles.get(field)
|
|
|
+ if previous_profile is None:
|
|
|
+ continue
|
|
|
+ distance = _distribution_distance(
|
|
|
+ _distribution(profiles[field]),
|
|
|
+ _distribution(previous_profile),
|
|
|
+ )
|
|
|
+ if distance > thresholds["distribution_drift_max"]:
|
|
|
+ add(
|
|
|
+ "distribution",
|
|
|
+ field,
|
|
|
+ {"distance": distance},
|
|
|
+ {
|
|
|
+ "maximum_distance": thresholds[
|
|
|
+ "distribution_drift_max"
|
|
|
+ ]
|
|
|
+ },
|
|
|
+ severity="warning",
|
|
|
+ )
|
|
|
+ previous_schema = previous.get("profile", {}).get("schema", {})
|
|
|
+ current_schema = _schema_signature(profiles)
|
|
|
+ if current_schema != previous_schema:
|
|
|
+ add(
|
|
|
+ "schema",
|
|
|
+ None,
|
|
|
+ {
|
|
|
+ "current_hash": _content_hash(current_schema),
|
|
|
+ "changed": True,
|
|
|
+ },
|
|
|
+ {"previous_hash": _content_hash(previous_schema)},
|
|
|
+ )
|
|
|
+ return findings
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _comparison(*, row_count, score, previous):
|
|
|
+ if previous is None:
|
|
|
+ return {
|
|
|
+ "row_count_change_ratio": None,
|
|
|
+ "score_delta": None,
|
|
|
+ "period": "baseline",
|
|
|
+ }
|
|
|
+ previous_count = int(previous["row_count"])
|
|
|
+ change = (
|
|
|
+ (row_count - previous_count) / previous_count
|
|
|
+ if previous_count
|
|
|
+ else 1.0
|
|
|
+ )
|
|
|
+ return {
|
|
|
+ "row_count_change_ratio": round(change, 6),
|
|
|
+ "score_delta": round(score - float(previous["score"]), 2),
|
|
|
+ "period": "previous_run",
|
|
|
+ }
|
|
|
+
|
|
|
+ def _sla_events(self, run, definition, findings):
|
|
|
+ thresholds = definition["thresholds"]
|
|
|
+ freshness_finding = next(
|
|
|
+ (item for item in findings if item["finding_type"] == "freshness"),
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ checks = [
|
|
|
+ (
|
|
|
+ "freshness",
|
|
|
+ freshness_finding is None,
|
|
|
+ (
|
|
|
+ 0
|
|
|
+ if freshness_finding is None
|
|
|
+ else freshness_finding["actual"]["seconds"]
|
|
|
+ ),
|
|
|
+ thresholds["freshness_max_seconds"],
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ "quality_score",
|
|
|
+ run["score"] >= thresholds["quality_score_min"],
|
|
|
+ run["score"],
|
|
|
+ thresholds["quality_score_min"],
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ events = []
|
|
|
+ for sla_type, met, actual, threshold in checks:
|
|
|
+ previous = self.repository.latest_sla_event(
|
|
|
+ run["asset_uid"], sla_type
|
|
|
+ )
|
|
|
+ status = "met" if met else "violated"
|
|
|
+ if met and previous and previous["status"] == "violated":
|
|
|
+ status = "recovered"
|
|
|
+ recurrence = max(
|
|
|
+ (
|
|
|
+ item["occurrence_number"]
|
|
|
+ for item in findings
|
|
|
+ if (
|
|
|
+ sla_type == "quality_score"
|
|
|
+ or item["finding_type"] == "freshness"
|
|
|
+ )
|
|
|
+ ),
|
|
|
+ default=0,
|
|
|
+ )
|
|
|
+ events.append(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "run_uid": run["uid"],
|
|
|
+ "asset_uid": run["asset_uid"],
|
|
|
+ "sla_type": sla_type,
|
|
|
+ "status": status,
|
|
|
+ "severity": "critical" if not met else "info",
|
|
|
+ "actual": actual,
|
|
|
+ "threshold": threshold,
|
|
|
+ "owner_uid": (
|
|
|
+ self.repository.get_template(run["template_uid"])[
|
|
|
+ "owner_uid"
|
|
|
+ ]
|
|
|
+ ),
|
|
|
+ "escalation_level": (
|
|
|
+ min(3, max(1, recurrence)) if not met else 0
|
|
|
+ ),
|
|
|
+ "evidence": {
|
|
|
+ "run_uid": run["uid"],
|
|
|
+ "deterministic": True,
|
|
|
+ },
|
|
|
+ "created_at": run["created_at"],
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return events
|
|
|
+
|
|
|
+ def trend(self, asset_uid: str) -> dict[str, Any]:
|
|
|
+ uid = _uid(asset_uid, "asset_uid")
|
|
|
+ runs = sorted(
|
|
|
+ self.repository.list_runs(uid),
|
|
|
+ key=lambda item: (item["created_at"], item["uid"]),
|
|
|
+ )
|
|
|
+ points = [
|
|
|
+ {
|
|
|
+ "run_uid": item["uid"],
|
|
|
+ "created_at": item["created_at"],
|
|
|
+ "row_count": item["row_count"],
|
|
|
+ "score": item["score"],
|
|
|
+ "finding_count": item["finding_count"],
|
|
|
+ }
|
|
|
+ for item in runs
|
|
|
+ ]
|
|
|
+ latest = (
|
|
|
+ {
|
|
|
+ **runs[-1]["comparison"],
|
|
|
+ "run_uid": runs[-1]["uid"],
|
|
|
+ }
|
|
|
+ if runs
|
|
|
+ else None
|
|
|
+ )
|
|
|
+ return {
|
|
|
+ "asset_uid": uid,
|
|
|
+ "points": points,
|
|
|
+ "latest": latest,
|
|
|
+ "year_over_year": self._period_delta(runs, days=365),
|
|
|
+ "month_over_month": self._period_delta(runs, days=30),
|
|
|
+ }
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _period_delta(runs: list[dict[str, Any]], *, days: int):
|
|
|
+ if len(runs) < 2:
|
|
|
+ return None
|
|
|
+ latest = runs[-1]
|
|
|
+ latest_at = _parse_datetime(latest["created_at"], "created_at")
|
|
|
+ candidates = [
|
|
|
+ item
|
|
|
+ for item in runs[:-1]
|
|
|
+ if (
|
|
|
+ latest_at - _parse_datetime(item["created_at"], "created_at")
|
|
|
+ ).days
|
|
|
+ >= days
|
|
|
+ ]
|
|
|
+ if not candidates:
|
|
|
+ return None
|
|
|
+ baseline = candidates[-1]
|
|
|
+ return {
|
|
|
+ "baseline_run_uid": baseline["uid"],
|
|
|
+ "score_delta": round(
|
|
|
+ float(latest["score"]) - float(baseline["score"]), 2
|
|
|
+ ),
|
|
|
+ "row_count_delta": int(latest["row_count"])
|
|
|
+ - int(baseline["row_count"]),
|
|
|
+ }
|
|
|
+
|
|
|
+ def get_run(self, run_uid: str) -> dict[str, Any]:
|
|
|
+ run = self.repository.get_run(_uid(run_uid, "run_uid"))
|
|
|
+ if run is None:
|
|
|
+ raise LookupError("quality run was not found")
|
|
|
+ return {
|
|
|
+ **run,
|
|
|
+ "metrics": self.repository.list_metrics(run["uid"]),
|
|
|
+ "findings": self.repository.list_findings(run["uid"]),
|
|
|
+ "sla_events": self.repository.list_sla_events(run["uid"]),
|
|
|
+ }
|
|
|
+
|
|
|
+ def list_runs(self, asset_uid: str | None = None) -> list[dict[str, Any]]:
|
|
|
+ normalized = _uid(asset_uid, "asset_uid") if asset_uid else None
|
|
|
+ return self.repository.list_runs(normalized)
|