Alo/animaltrack
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add event payload models with tag normalization

Browse Source 
Implements Step 2.3 of the plan:
- 17 Pydantic payload models for all event types per spec §6
- 6 enums: Sex, ReproStatus, LifeStage, AnimalStatus, Origin, Outcome
- Tag normalization per spec §20 (lowercase, spaces→hyphens, max 32 chars)
- ULID validation utilities
- 38 tests covering payloads, enums, and tag normalization

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Petru
2025-12-28 18:34:41 +00:00
parent 81eb0b8243
commit 80784ff636
5 changed files with 991 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
src/animaltrack/events/__init__.py
View File

@@ -1,7 +1,36 @@
# ABOUTME: Events package for event sourcing infrastructure.
# ABOUTME: Provides event types, store, and related exceptions.
# ABOUTME: Provides event types, store, payloads, and related exceptions.
from animaltrack.events.enums import (
AnimalStatus,
LifeStage,
Origin,
Outcome,
ReproStatus,
Sex,
)
from animaltrack.events.exceptions import ClockSkewError, DuplicateNonceError
from animaltrack.events.payloads import (
AnimalAttributesUpdatedPayload,
AnimalCohortCreatedPayload,
AnimalMergedPayload,
AnimalMovedPayload,
AnimalOutcomePayload,
AnimalPromotedPayload,
AnimalStatusCorrectedPayload,
AnimalTagEndedPayload,
AnimalTaggedPayload,
AttributeSet,
FeedGivenPayload,
FeedPurchasedPayload,
HatchRecordedPayload,
LocationArchivedPayload,
LocationCreatedPayload,
LocationRenamedPayload,
ProductCollectedPayload,
ProductSoldPayload,
YieldItem,
)
from animaltrack.events.store import EventStore
from animaltrack.events.types import (
ALL_EVENT_TYPES,
@@ -23,6 +52,7 @@ from animaltrack.events.types import (
PRODUCT_COLLECTED,
PRODUCT_SOLD,
)
from animaltrack.events.validation import normalize_tag, validate_ulid, validate_ulid_list
__all__ = [
# Event types
@@ -44,9 +74,44 @@ __all__ = [
"FEED_PURCHASED",
"FEED_GIVEN",
"ALL_EVENT_TYPES",
# Enums
"Sex",
"ReproStatus",
"LifeStage",
"AnimalStatus",
"Origin",
"Outcome",
# Exceptions
"ClockSkewError",
"DuplicateNonceError",
# Store
"EventStore",
# Validation
"normalize_tag",
"validate_ulid",
"validate_ulid_list",
# Payloads - Location
"LocationCreatedPayload",
"LocationRenamedPayload",
"LocationArchivedPayload",
# Payloads - Animal
"AnimalCohortCreatedPayload",
"AnimalPromotedPayload",
"AnimalMovedPayload",
"AnimalAttributesUpdatedPayload",
"AnimalTaggedPayload",
"AnimalTagEndedPayload",
"HatchRecordedPayload",
"AnimalOutcomePayload",
"AnimalMergedPayload",
"AnimalStatusCorrectedPayload",
# Payloads - Product
"ProductCollectedPayload",
"ProductSoldPayload",
# Payloads - Feed
"FeedPurchasedPayload",
"FeedGivenPayload",
# Payloads - Helper models
"YieldItem",
"AttributeSet",
]

59
src/animaltrack/events/enums.py Normal file
View File

@@ -0,0 +1,59 @@
# ABOUTME: Enums for event payloads and animal attributes.
# ABOUTME: Defines Sex, ReproStatus, LifeStage, AnimalStatus, Origin, and Outcome.
from enum import Enum
class Sex(str, Enum):
"""Biological sex of an animal."""
MALE = "male"
FEMALE = "female"
UNKNOWN = "unknown"
class ReproStatus(str, Enum):
"""Reproductive status of an animal."""
INTACT = "intact"
WETHER = "wether"
SPAYED = "spayed"
UNKNOWN = "unknown"
class LifeStage(str, Enum):
"""Life stage of an animal."""
HATCHLING = "hatchling"
JUVENILE = "juvenile"
SUBADULT = "subadult"
ADULT = "adult"
class AnimalStatus(str, Enum):
"""Current status of an animal."""
ALIVE = "alive"
DEAD = "dead"
HARVESTED = "harvested"
SOLD = "sold"
MERGED_INTO = "merged_into"
class Origin(str, Enum):
"""Origin of an animal (how it came to the farm)."""
HATCHED = "hatched"
PURCHASED = "purchased"
RESCUED = "rescued"
UNKNOWN = "unknown"
class Outcome(str, Enum):
"""Outcome type for animal lifecycle events."""
DEATH = "death"
HARVEST = "harvest"
SOLD = "sold"
PREDATOR_LOSS = "predator_loss"
UNKNOWN = "unknown"

370
src/animaltrack/events/payloads.py Normal file
View File

@@ -0,0 +1,370 @@
# ABOUTME: Pydantic models for all 17 event payload types.
# ABOUTME: Validates event data before storage per spec §6.
from typing import Literal
from pydantic import BaseModel, Field, field_validator
from animaltrack.events.enums import (
AnimalStatus,
LifeStage,
Origin,
Outcome,
ReproStatus,
Sex,
)
from animaltrack.events.validation import normalize_tag, validate_ulid
# =============================================================================
# Helper Models
# =============================================================================
class YieldItem(BaseModel):
"""A product yield from an animal outcome (harvest)."""
product_code: str
unit: Literal["piece", "kg"]
quantity: int = Field(..., ge=1)
weight_kg: float | None = None
notes: str | None = None
class AttributeSet(BaseModel):
"""Set of animal attributes that can be updated."""
sex: Sex | None = None
life_stage: LifeStage | None = None
repro_status: ReproStatus | None = None
# =============================================================================
# Location Event Payloads (3)
# =============================================================================
class LocationCreatedPayload(BaseModel):
"""Payload for LocationCreated event."""
location_id: str = Field(..., min_length=26, max_length=26)
name: str
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)
class LocationRenamedPayload(BaseModel):
"""Payload for LocationRenamed event."""
location_id: str = Field(..., min_length=26, max_length=26)
new_name: str
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)
class LocationArchivedPayload(BaseModel):
"""Payload for LocationArchived event."""
location_id: str = Field(..., min_length=26, max_length=26)
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)
# =============================================================================
# Animal Event Payloads (10)
# =============================================================================
class AnimalCohortCreatedPayload(BaseModel):
"""Payload for AnimalCohortCreated event."""
species: Literal["duck", "goose"]
count: int = Field(..., ge=1)
life_stage: LifeStage
sex: Sex = Sex.UNKNOWN
location_id: str = Field(..., min_length=26, max_length=26)
origin: Origin
notes: str | None = None
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)
class AnimalPromotedPayload(BaseModel):
"""Payload for AnimalPromoted event."""
animal_id: str = Field(..., min_length=26, max_length=26)
nickname: str | None = None
sex: Sex | None = None
repro_status: ReproStatus | None = None
distinguishing_traits: str | None = None
notes: str | None = None
@field_validator("animal_id")
@classmethod
def validate_animal_ulid(cls, v: str) -> str:
"""Validate animal_id is a valid ULID."""
return validate_ulid(v)
class AnimalMovedPayload(BaseModel):
"""Payload for AnimalMoved event."""
to_location_id: str = Field(..., min_length=26, max_length=26)
resolved_ids: list[str] = Field(..., min_length=1)
notes: str | None = None
@field_validator("to_location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate to_location_id is a valid ULID."""
return validate_ulid(v)
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
class AnimalAttributesUpdatedPayload(BaseModel):
"""Payload for AnimalAttributesUpdated event."""
resolved_ids: list[str] = Field(..., min_length=1)
set: AttributeSet
notes: str | None = None
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
class AnimalTaggedPayload(BaseModel):
"""Payload for AnimalTagged event."""
resolved_ids: list[str] = Field(..., min_length=1)
tag: str
notes: str | None = None
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
@field_validator("tag")
@classmethod
def normalize_tag_value(cls, v: str) -> str:
"""Normalize the tag per spec §20."""
return normalize_tag(v)
class AnimalTagEndedPayload(BaseModel):
"""Payload for AnimalTagEnded event."""
resolved_ids: list[str] = Field(..., min_length=1)
tag: str
notes: str | None = None
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
@field_validator("tag")
@classmethod
def normalize_tag_value(cls, v: str) -> str:
"""Normalize the tag per spec §20."""
return normalize_tag(v)
class HatchRecordedPayload(BaseModel):
"""Payload for HatchRecorded event."""
species: Literal["duck", "goose"]
location_id: str = Field(..., min_length=26, max_length=26)
assigned_brood_location_id: str | None = None
hatched_live: int = Field(..., ge=1)
notes: str | None = None
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)
@field_validator("assigned_brood_location_id")
@classmethod
def validate_brood_location_ulid(cls, v: str | None) -> str | None:
"""Validate assigned_brood_location_id is a valid ULID if provided."""
if v is not None:
return validate_ulid(v)
return v
class AnimalOutcomePayload(BaseModel):
"""Payload for AnimalOutcome event."""
outcome: Outcome
resolved_ids: list[str] = Field(..., min_length=1)
reason: str | None = None
yield_items: list[YieldItem] | None = None
notes: str | None = None
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
class AnimalMergedPayload(BaseModel):
"""Payload for AnimalMerged event."""
survivor_animal_id: str = Field(..., min_length=26, max_length=26)
merged_animal_ids: list[str] = Field(..., min_length=1)
notes: str | None = None
@field_validator("survivor_animal_id")
@classmethod
def validate_survivor_ulid(cls, v: str) -> str:
"""Validate survivor_animal_id is a valid ULID."""
return validate_ulid(v)
@field_validator("merged_animal_ids")
@classmethod
def validate_merged_ulids(cls, v: list[str]) -> list[str]:
"""Validate all merged_animal_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"merged_animal_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
class AnimalStatusCorrectedPayload(BaseModel):
"""Payload for AnimalStatusCorrected event (admin-only)."""
resolved_ids: list[str] = Field(..., min_length=1)
new_status: AnimalStatus
reason: str # Required for audit trail
notes: str | None = None
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
# =============================================================================
# Product Event Payloads (2)
# =============================================================================
class ProductCollectedPayload(BaseModel):
"""Payload for ProductCollected event."""
location_id: str = Field(..., min_length=26, max_length=26)
product_code: str
quantity: int = Field(..., ge=1)
resolved_ids: list[str] = Field(..., min_length=1)
notes: str | None = None
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)
@field_validator("resolved_ids")
@classmethod
def validate_resolved_ulids(cls, v: list[str]) -> list[str]:
"""Validate all resolved_ids are valid ULIDs."""
for i, ulid in enumerate(v):
if len(ulid) != 26:
msg = f"resolved_ids[{i}] must be 26 characters"
raise ValueError(msg)
return v
class ProductSoldPayload(BaseModel):
"""Payload for ProductSold event."""
product_code: str
quantity: int = Field(..., ge=1)
total_price_cents: int = Field(..., ge=0)
buyer: str | None = None
notes: str | None = None
# =============================================================================
# Feed Event Payloads (2)
# =============================================================================
class FeedPurchasedPayload(BaseModel):
"""Payload for FeedPurchased event."""
feed_type_code: str
bag_size_kg: int = Field(..., ge=1)
bags_count: int = Field(..., ge=1)
bag_price_cents: int = Field(..., ge=0)
vendor: str | None = None
notes: str | None = None
class FeedGivenPayload(BaseModel):
"""Payload for FeedGiven event."""
location_id: str = Field(..., min_length=26, max_length=26)
feed_type_code: str
amount_kg: int = Field(..., ge=1)
notes: str | None = None
@field_validator("location_id")
@classmethod
def validate_location_ulid(cls, v: str) -> str:
"""Validate location_id is a valid ULID."""
return validate_ulid(v)

76
src/animaltrack/events/validation.py Normal file
View File

@@ -0,0 +1,76 @@
# ABOUTME: Validation utilities for event payloads.
# ABOUTME: Includes tag normalization per spec §20 and ULID validators.
import re
def normalize_tag(tag: str) -> str:
"""Normalize a tag per spec §20.
Normalization rules:
- Convert to lowercase
- Trim leading/trailing whitespace
- Replace spaces with hyphens
- Keep only valid characters [a-z0-9:_-]
- Truncate to max 32 characters
- Raise ValueError if empty after normalization
Args:
tag: The raw tag string to normalize.
Returns:
The normalized tag string.
Raises:
ValueError: If the tag is empty after normalization.
"""
# Trim and lowercase
tag = tag.strip().lower()
# Spaces to hyphens
tag = tag.replace(" ", "-")
# Keep only valid chars: a-z, 0-9, colon, underscore, hyphen
tag = re.sub(r"[^a-z0-9:_-]", "", tag)
# Max 32 chars
tag = tag[:32]
# Check for empty
if not tag:
msg = "Tag is empty after normalization"
raise ValueError(msg)
return tag
def validate_ulid(value: str) -> str:
"""Validate that a value is a 26-character ULID.
Args:
value: The string to validate.
Returns:
The validated string.
Raises:
ValueError: If the value is not exactly 26 characters.
"""
if len(value) != 26:
msg = f"ULID must be exactly 26 characters, got {len(value)}"
raise ValueError(msg)
return value
def validate_ulid_list(values: list[str]) -> list[str]:
"""Validate that all values in a list are 26-character ULIDs.
Args:
values: The list of strings to validate.
Returns:
The validated list.
Raises:
ValueError: If any value is not exactly 26 characters.
"""
for i, value in enumerate(values):
if len(value) != 26:
msg = f"ULID at index {i} must be exactly 26 characters, got {len(value)}"
raise ValueError(msg)
return values