First version

src/shared_models/__init__.py

@@ -0,0 +1,17 @@
+from .message import (
+	LINK_SCHEMES,
+	MAX_LEN,
+	Category,
+	Message,
+	MessageValidationError,
+	ValidationIssue,
+)
+
+__all__ = [
+	"LINK_SCHEMES",
+	"MAX_LEN",
+	"Category",
+	"Message",
+	"MessageValidationError",
+	"ValidationIssue",
+]

src/shared_models/message.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from enum import Enum
+from typing import Any, Mapping
+
+MAX_LEN = 255
+LINK_SCHEMES = ("http://", "https://")
+
+
+class Category(str, Enum):
+	ROYALROAD = "royal-road"
+	PODCASTS = "podcasts"
+	MIXTAPES = "mixtapes"
+	UTILITY = "utility"
+
+
+@dataclass(frozen=True)
+class ValidationIssue:
+	field: str
+	code: str # machine-readable: "required", "too_long", ...
+	message: str # human-readable
+
+	def to_dict(self) -> dict[str, str]:
+		return asdict(self)
+
+
+class MessageValidationError(ValueError):
+	"""Carries every problem found, not just the first."""
+
+	def __init__(self, issues: list[ValidationIssue]):
+		self.issues = issues
+		super().__init__("; ".join(f"{i.field}: {i.message}" for i in issues))
+
+	def to_list(self) -> list[dict[str, str]]:
+		return [i.to_dict() for i in self.issues]
+
+
+def _text(data, name, issues, *, required):
+	"""Return a stripped string, or None if absent/invalid (issue recorded)."""
+	if name not in data or data[name] is None:
+		if required:
+			issues.append(ValidationIssue(name, "required", f"{name} is required"))
+		return None
+	value = data[name]
+	if not isinstance(value, str):
+		issues.append(ValidationIssue(
+			name, "type", f"{name} must be a string, got {type(value).__name__}"))
+		return None
+	value = value.strip()
+	if required and not value:
+		issues.append(ValidationIssue(name, "empty", f"{name} must not be blank"))
+		return None
+	if len(value) > MAX_LEN:
+		issues.append(ValidationIssue(
+			name, "too_long",
+			f"{name} is {len(value)} characters, maximum is {MAX_LEN}"))
+		return None
+	return value
+
+
+@dataclass(frozen=True)
+class Message:
+	title: str
+	body: str
+	link: str | None = None
+	category: Category = Category.UTILITY
+	timestamp: int | None = None
+
+	_FIELDS = frozenset({"title", "body", "link", "category", "timestamp"})
+
+	@classmethod
+	def parse(cls, data: Any) -> "Message":
+		if not isinstance(data, Mapping):
+			raise MessageValidationError([ValidationIssue(
+				"message", "type",
+				f"message must be an object, got {type(data).__name__}")])
+
+		issues: list[ValidationIssue] = []
+
+		for key in sorted(map(str, data.keys() - cls._FIELDS)):
+			issues.append(ValidationIssue(key, "unknown_field", f"unexpected field {key!r}"))
+
+		title = _text(data, "title", issues, required=True)
+		body = _text(data, "body", issues, required=True)
+
+		link = _text(data, "link", issues, required=False) or None
+		if link and not link.lower().startswith(LINK_SCHEMES):
+			issues.append(ValidationIssue(
+				"link", "bad_scheme", "link must start with http:// or https://"))
+
+		category = Category.UTILITY
+		if data.get("category") is not None:
+			try:
+				category = Category(data["category"])
+			except (ValueError, TypeError):
+				allowed = ", ".join(c.value for c in Category)
+				issues.append(ValidationIssue(
+					"category", "invalid_choice",
+					f"category {data['category']!r} is not one of: {allowed}"))
+
+		timestamp = data.get("timestamp")
+		if timestamp is not None:
+			if isinstance(timestamp, bool) or not isinstance(timestamp, int):
+				issues.append(ValidationIssue(
+					"timestamp", "type", "timestamp must be an integer (unix seconds)"))
+			elif timestamp < 0:
+				issues.append(ValidationIssue(
+					"timestamp", "out_of_range", "timestamp must not be negative"))
+
+		if issues:
+			raise MessageValidationError(issues)
+		return cls(title, body, link, category, timestamp)
+
+	def to_dict(self) -> dict[str, Any]:
+		d = asdict(self)
+		d["category"] = self.category.value
+		return d