feat: add feed inventory schema and purchase service
Implement FeedPurchased event handling: - Add migration for feed_inventory table - Create FeedInventoryProjection to track purchases - Create FeedService with purchase_feed method - Calculate price_per_kg_cents from bag details Purchases accumulate in inventory with: - purchased_kg, given_kg, balance_kg tracking - Last purchase price stored in cents - Timestamps for last purchase/given 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
88
src/animaltrack/projections/feed.py
Normal file
88
src/animaltrack/projections/feed.py
Normal file
@@ -0,0 +1,88 @@
|
||||
# ABOUTME: Projection for feed inventory tracking.
|
||||
# ABOUTME: Handles FeedPurchased and FeedGiven events.
|
||||
|
||||
from typing import Any
|
||||
|
||||
from animaltrack.events.types import FEED_PURCHASED
|
||||
from animaltrack.models.events import Event
|
||||
from animaltrack.projections.base import Projection
|
||||
|
||||
|
||||
class FeedInventoryProjection(Projection):
|
||||
"""Maintains feed inventory levels.
|
||||
|
||||
This projection handles feed purchase and given events, maintaining:
|
||||
- feed_inventory: Current inventory levels per feed type
|
||||
"""
|
||||
|
||||
def __init__(self, db: Any) -> None:
|
||||
"""Initialize the projection with a database connection.
|
||||
|
||||
Args:
|
||||
db: A fastlite database connection.
|
||||
"""
|
||||
super().__init__(db)
|
||||
|
||||
def get_event_types(self) -> list[str]:
|
||||
"""Return the event types this projection handles."""
|
||||
return [FEED_PURCHASED]
|
||||
|
||||
def apply(self, event: Event) -> None:
|
||||
"""Apply feed event to update inventory."""
|
||||
if event.type == FEED_PURCHASED:
|
||||
self._apply_feed_purchased(event)
|
||||
|
||||
def revert(self, event: Event) -> None:
|
||||
"""Revert feed event from inventory."""
|
||||
if event.type == FEED_PURCHASED:
|
||||
self._revert_feed_purchased(event)
|
||||
|
||||
def _apply_feed_purchased(self, event: Event) -> None:
|
||||
"""Apply feed purchase to inventory.
|
||||
|
||||
- Upsert feed_inventory row
|
||||
- Increment purchased_kg and balance_kg
|
||||
- Update last_purchase_price_per_kg_cents
|
||||
- Update last_purchase_at_utc
|
||||
"""
|
||||
feed_type_code = event.entity_refs.get("feed_type_code")
|
||||
total_kg = event.entity_refs.get("total_kg")
|
||||
price_per_kg_cents = event.entity_refs.get("price_per_kg_cents")
|
||||
ts_utc = event.ts_utc
|
||||
|
||||
# Upsert: create if not exists, update if exists
|
||||
self.db.execute(
|
||||
"""
|
||||
INSERT INTO feed_inventory
|
||||
(feed_type_code, purchased_kg, given_kg, balance_kg,
|
||||
last_purchase_price_per_kg_cents, last_purchase_at_utc, updated_at_utc)
|
||||
VALUES (?, ?, 0, ?, ?, ?, ?)
|
||||
ON CONFLICT(feed_type_code) DO UPDATE SET
|
||||
purchased_kg = purchased_kg + excluded.purchased_kg,
|
||||
balance_kg = balance_kg + excluded.purchased_kg,
|
||||
last_purchase_price_per_kg_cents = excluded.last_purchase_price_per_kg_cents,
|
||||
last_purchase_at_utc = excluded.last_purchase_at_utc,
|
||||
updated_at_utc = excluded.updated_at_utc
|
||||
""",
|
||||
(feed_type_code, total_kg, total_kg, price_per_kg_cents, ts_utc, ts_utc),
|
||||
)
|
||||
|
||||
def _revert_feed_purchased(self, event: Event) -> None:
|
||||
"""Revert feed purchase from inventory.
|
||||
|
||||
- Decrement purchased_kg and balance_kg
|
||||
"""
|
||||
feed_type_code = event.entity_refs.get("feed_type_code")
|
||||
total_kg = event.entity_refs.get("total_kg")
|
||||
ts_utc = event.ts_utc
|
||||
|
||||
self.db.execute(
|
||||
"""
|
||||
UPDATE feed_inventory
|
||||
SET purchased_kg = purchased_kg - ?,
|
||||
balance_kg = balance_kg - ?,
|
||||
updated_at_utc = ?
|
||||
WHERE feed_type_code = ?
|
||||
""",
|
||||
(total_kg, total_kg, ts_utc, feed_type_code),
|
||||
)
|
||||
110
src/animaltrack/services/feed.py
Normal file
110
src/animaltrack/services/feed.py
Normal file
@@ -0,0 +1,110 @@
|
||||
# ABOUTME: Service layer for feed operations.
|
||||
# ABOUTME: Coordinates event creation with projection updates for feed inventory.
|
||||
|
||||
from typing import Any
|
||||
|
||||
from animaltrack.db import transaction
|
||||
from animaltrack.events.payloads import FeedPurchasedPayload
|
||||
from animaltrack.events.processor import process_event
|
||||
from animaltrack.events.store import EventStore
|
||||
from animaltrack.events.types import FEED_PURCHASED
|
||||
from animaltrack.models.events import Event
|
||||
from animaltrack.projections import ProjectionRegistry
|
||||
from animaltrack.repositories.feed_types import FeedTypeRepository
|
||||
|
||||
|
||||
class FeedServiceError(Exception):
|
||||
"""Base exception for feed service errors."""
|
||||
|
||||
|
||||
class ValidationError(FeedServiceError):
|
||||
"""Raised when validation fails."""
|
||||
|
||||
|
||||
class FeedService:
|
||||
"""Service for feed-related operations.
|
||||
|
||||
Provides methods to purchase feed and record feed given events.
|
||||
All operations are atomic and maintain inventory consistency.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db: Any,
|
||||
event_store: EventStore,
|
||||
registry: ProjectionRegistry,
|
||||
) -> None:
|
||||
"""Initialize the service.
|
||||
|
||||
Args:
|
||||
db: A fastlite database connection.
|
||||
event_store: The event store for appending events.
|
||||
registry: Registry of projections to update.
|
||||
"""
|
||||
self.db = db
|
||||
self.event_store = event_store
|
||||
self.registry = registry
|
||||
self.feed_type_repo = FeedTypeRepository(db)
|
||||
|
||||
def purchase_feed(
|
||||
self,
|
||||
payload: FeedPurchasedPayload,
|
||||
ts_utc: int,
|
||||
actor: str,
|
||||
nonce: str | None = None,
|
||||
route: str | None = None,
|
||||
) -> Event:
|
||||
"""Record a feed purchase.
|
||||
|
||||
Creates a FeedPurchased event and updates inventory.
|
||||
Calculates total_kg and price_per_kg_cents from payload.
|
||||
|
||||
Args:
|
||||
payload: Validated purchase payload.
|
||||
ts_utc: Timestamp in milliseconds since epoch.
|
||||
actor: The user performing the purchase.
|
||||
nonce: Optional idempotency nonce.
|
||||
route: Required if nonce provided.
|
||||
|
||||
Returns:
|
||||
The created event.
|
||||
|
||||
Raises:
|
||||
ValidationError: If feed type doesn't exist or is inactive.
|
||||
"""
|
||||
# Validate feed type exists and is active
|
||||
feed_type = self.feed_type_repo.get(payload.feed_type_code)
|
||||
if feed_type is None:
|
||||
msg = f"Feed type '{payload.feed_type_code}' not found"
|
||||
raise ValidationError(msg)
|
||||
|
||||
if not feed_type.active:
|
||||
msg = f"Feed type '{payload.feed_type_code}' is inactive"
|
||||
raise ValidationError(msg)
|
||||
|
||||
# Calculate derived values
|
||||
total_kg = payload.bag_size_kg * payload.bags_count
|
||||
total_price_cents = payload.bag_price_cents * payload.bags_count
|
||||
price_per_kg_cents = total_price_cents // total_kg # Floor division
|
||||
|
||||
# Build entity_refs
|
||||
entity_refs = {
|
||||
"feed_type_code": payload.feed_type_code,
|
||||
"total_kg": total_kg,
|
||||
"price_per_kg_cents": price_per_kg_cents,
|
||||
}
|
||||
|
||||
with transaction(self.db):
|
||||
event = self.event_store.append_event(
|
||||
event_type=FEED_PURCHASED,
|
||||
ts_utc=ts_utc,
|
||||
actor=actor,
|
||||
entity_refs=entity_refs,
|
||||
payload=payload.model_dump(),
|
||||
nonce=nonce,
|
||||
route=route,
|
||||
)
|
||||
|
||||
process_event(event, self.registry)
|
||||
|
||||
return event
|
||||
Reference in New Issue
Block a user