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

feat: add repositories and reference data seeder

Browse Source 
Implements Step 1.5: Creates ULID generation utility, repositories for
all reference types (species, locations, products, feed_types, users),
and an idempotent seeder that populates initial data. Updates CLI seed
command to run migrations and seed data.

Seed data:
- 3 users (ppetru, ines as admin; guest as recorder)
- 8 locations (Strip 1-4, Nursery 1-4)
- 3 species (duck, goose active; sheep inactive)
- 7 products (egg.duck, meat, offal, fat, bones, feathers, down)
- 3 feed types (starter, grower, layer - 20kg bags)

🤖 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-27 19:40:55 +00:00
parent 34b2421604
commit 4579229734
12 changed files with 1456 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/animaltrack/cli.py
View File

@@ -63,9 +63,28 @@ def main():
print(f"Created migration: {filepath}")
print(" Edit the file to add your schema changes")
elif args.command == "seed":
from animaltrack.config import Settings
from animaltrack.db import get_db
from animaltrack.migrations import run_migrations
from animaltrack.seeds import run_seeds
settings = Settings()
migrations_dir = "migrations"
print("Running migrations...")
success = run_migrations(
db_path=settings.db_path,
migrations_dir=migrations_dir,
verbose=False,
)
if not success:
print("Migration failed", file=sys.stderr)
sys.exit(1)
print("Loading seed data...")
# TODO: Implement seeder
print("Seeding not yet implemented")
db = get_db(settings.db_path)
run_seeds(db)
print("Seed data loaded successfully")
elif args.command == "serve":
print(f"Starting server on {args.host}:{args.port}...")
# TODO: Implement server

13
src/animaltrack/id_gen.py Normal file
View File

@@ -0,0 +1,13 @@
# ABOUTME: ULID generation utility for creating unique identifiers.
# ABOUTME: Provides a simple wrapper around the python-ulid library.
from ulid import ULID
def generate_id() -> str:
"""Generate a new ULID as a 26-character string.
Returns:
A 26-character uppercase alphanumeric ULID string.
"""
return str(ULID())

16
src/animaltrack/repositories/__init__.py Normal file
View File

@@ -0,0 +1,16 @@
# ABOUTME: Repositories package for AnimalTrack.
# ABOUTME: Exports all repository classes for reference data CRUD operations.
from animaltrack.repositories.feed_types import FeedTypeRepository
from animaltrack.repositories.locations import LocationRepository
from animaltrack.repositories.products import ProductRepository
from animaltrack.repositories.species import SpeciesRepository
from animaltrack.repositories.users import UserRepository
__all__ = [
"FeedTypeRepository",
"LocationRepository",
"ProductRepository",
"SpeciesRepository",
"UserRepository",
]

94
src/animaltrack/repositories/feed_types.py Normal file
View File

@@ -0,0 +1,94 @@
# ABOUTME: Repository for feed type reference data.
# ABOUTME: Provides CRUD operations for the feed_types table.
from typing import Any
from animaltrack.models.reference import FeedType
class FeedTypeRepository:
"""Repository for managing feed type data."""
def __init__(self, db: Any) -> None:
"""Initialize repository with database connection.
Args:
db: A fastlite database connection.
"""
self.db = db
def upsert(self, feed_type: FeedType) -> None:
"""Insert or update a feed type record.
Args:
feed_type: The feed type to upsert.
"""
self.db.execute(
"""
INSERT OR REPLACE INTO feed_types
(code, name, default_bag_size_kg, protein_pct, active, created_at_utc, updated_at_utc)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
feed_type.code,
feed_type.name,
feed_type.default_bag_size_kg,
feed_type.protein_pct,
int(feed_type.active),
feed_type.created_at_utc,
feed_type.updated_at_utc,
),
)
def get(self, code: str) -> FeedType | None:
"""Get a feed type by code.
Args:
code: The feed type code.
Returns:
The feed type if found, None otherwise.
"""
row = self.db.execute(
"""
SELECT code, name, default_bag_size_kg, protein_pct, active, created_at_utc, updated_at_utc
FROM feed_types WHERE code = ?
""",
(code,),
).fetchone()
if row is None:
return None
return FeedType(
code=row[0],
name=row[1],
default_bag_size_kg=row[2],
protein_pct=row[3],
active=bool(row[4]),
created_at_utc=row[5],
updated_at_utc=row[6],
)
def list_all(self) -> list[FeedType]:
"""Get all feed types.
Returns:
List of all feed types.
"""
rows = self.db.execute(
"""
SELECT code, name, default_bag_size_kg, protein_pct, active, created_at_utc, updated_at_utc
FROM feed_types
"""
).fetchall()
return [
FeedType(
code=row[0],
name=row[1],
default_bag_size_kg=row[2],
protein_pct=row[3],
active=bool(row[4]),
created_at_utc=row[5],
updated_at_utc=row[6],
)
for row in rows
]

104
src/animaltrack/repositories/locations.py Normal file
View File

@@ -0,0 +1,104 @@
# ABOUTME: Repository for location reference data.
# ABOUTME: Provides CRUD operations for the locations table.
from typing import Any
from animaltrack.models.reference import Location
class LocationRepository:
"""Repository for managing location data."""
def __init__(self, db: Any) -> None:
"""Initialize repository with database connection.
Args:
db: A fastlite database connection.
"""
self.db = db
def upsert(self, location: Location) -> None:
"""Insert or update a location record.
Args:
location: The location to upsert.
"""
self.db.execute(
"""
INSERT OR REPLACE INTO locations (id, name, active, created_at_utc, updated_at_utc)
VALUES (?, ?, ?, ?, ?)
""",
(
location.id,
location.name,
int(location.active),
location.created_at_utc,
location.updated_at_utc,
),
)
def get(self, location_id: str) -> Location | None:
"""Get a location by ID.
Args:
location_id: The location ID (ULID).
Returns:
The location if found, None otherwise.
"""
row = self.db.execute(
"SELECT id, name, active, created_at_utc, updated_at_utc FROM locations WHERE id = ?",
(location_id,),
).fetchone()
if row is None:
return None
return Location(
id=row[0],
name=row[1],
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
def get_by_name(self, name: str) -> Location | None:
"""Get a location by name.
Args:
name: The location name.
Returns:
The location if found, None otherwise.
"""
row = self.db.execute(
"SELECT id, name, active, created_at_utc, updated_at_utc FROM locations WHERE name = ?",
(name,),
).fetchone()
if row is None:
return None
return Location(
id=row[0],
name=row[1],
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
def list_all(self) -> list[Location]:
"""Get all locations.
Returns:
List of all locations.
"""
rows = self.db.execute(
"SELECT id, name, active, created_at_utc, updated_at_utc FROM locations"
).fetchall()
return [
Location(
id=row[0],
name=row[1],
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
for row in rows
]

97
src/animaltrack/repositories/products.py Normal file
View File

@@ -0,0 +1,97 @@
# ABOUTME: Repository for product reference data.
# ABOUTME: Provides CRUD operations for the products table.
from typing import Any
from animaltrack.models.reference import Product, ProductUnit
class ProductRepository:
"""Repository for managing product data."""
def __init__(self, db: Any) -> None:
"""Initialize repository with database connection.
Args:
db: A fastlite database connection.
"""
self.db = db
def upsert(self, product: Product) -> None:
"""Insert or update a product record.
Args:
product: The product to upsert.
"""
self.db.execute(
"""
INSERT OR REPLACE INTO products
(code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
product.code,
product.name,
product.unit.value,
int(product.collectable),
int(product.sellable),
int(product.active),
product.created_at_utc,
product.updated_at_utc,
),
)
def get(self, code: str) -> Product | None:
"""Get a product by code.
Args:
code: The product code.
Returns:
The product if found, None otherwise.
"""
row = self.db.execute(
"""
SELECT code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc
FROM products WHERE code = ?
""",
(code,),
).fetchone()
if row is None:
return None
return Product(
code=row[0],
name=row[1],
unit=ProductUnit(row[2]),
collectable=bool(row[3]),
sellable=bool(row[4]),
active=bool(row[5]),
created_at_utc=row[6],
updated_at_utc=row[7],
)
def list_all(self) -> list[Product]:
"""Get all products.
Returns:
List of all products.
"""
rows = self.db.execute(
"""
SELECT code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc
FROM products
"""
).fetchall()
return [
Product(
code=row[0],
name=row[1],
unit=ProductUnit(row[2]),
collectable=bool(row[3]),
sellable=bool(row[4]),
active=bool(row[5]),
created_at_utc=row[6],
updated_at_utc=row[7],
)
for row in rows
]

81
src/animaltrack/repositories/species.py Normal file
View File

@@ -0,0 +1,81 @@
# ABOUTME: Repository for species reference data.
# ABOUTME: Provides CRUD operations for the species table.
from typing import Any
from animaltrack.models.reference import Species
class SpeciesRepository:
"""Repository for managing species data."""
def __init__(self, db: Any) -> None:
"""Initialize repository with database connection.
Args:
db: A fastlite database connection.
"""
self.db = db
def upsert(self, species: Species) -> None:
"""Insert or update a species record.
Args:
species: The species to upsert.
"""
self.db.execute(
"""
INSERT OR REPLACE INTO species (code, name, active, created_at_utc, updated_at_utc)
VALUES (?, ?, ?, ?, ?)
""",
(
species.code,
species.name,
int(species.active),
species.created_at_utc,
species.updated_at_utc,
),
)
def get(self, code: str) -> Species | None:
"""Get a species by code.
Args:
code: The species code.
Returns:
The species if found, None otherwise.
"""
row = self.db.execute(
"SELECT code, name, active, created_at_utc, updated_at_utc FROM species WHERE code = ?",
(code,),
).fetchone()
if row is None:
return None
return Species(
code=row[0],
name=row[1],
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
def list_all(self) -> list[Species]:
"""Get all species.
Returns:
List of all species.
"""
rows = self.db.execute(
"SELECT code, name, active, created_at_utc, updated_at_utc FROM species"
).fetchall()
return [
Species(
code=row[0],
name=row[1],
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
for row in rows
]

81
src/animaltrack/repositories/users.py Normal file
View File

@@ -0,0 +1,81 @@
# ABOUTME: Repository for user reference data.
# ABOUTME: Provides CRUD operations for the users table.
from typing import Any
from animaltrack.models.reference import User, UserRole
class UserRepository:
"""Repository for managing user data."""
def __init__(self, db: Any) -> None:
"""Initialize repository with database connection.
Args:
db: A fastlite database connection.
"""
self.db = db
def upsert(self, user: User) -> None:
"""Insert or update a user record.
Args:
user: The user to upsert.
"""
self.db.execute(
"""
INSERT OR REPLACE INTO users (username, role, active, created_at_utc, updated_at_utc)
VALUES (?, ?, ?, ?, ?)
""",
(
user.username,
user.role.value,
int(user.active),
user.created_at_utc,
user.updated_at_utc,
),
)
def get(self, username: str) -> User | None:
"""Get a user by username.
Args:
username: The username.
Returns:
The user if found, None otherwise.
"""
row = self.db.execute(
"SELECT username, role, active, created_at_utc, updated_at_utc FROM users WHERE username = ?",
(username,),
).fetchone()
if row is None:
return None
return User(
username=row[0],
role=UserRole(row[1]),
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
def list_all(self) -> list[User]:
"""Get all users.
Returns:
List of all users.
"""
rows = self.db.execute(
"SELECT username, role, active, created_at_utc, updated_at_utc FROM users"
).fetchall()
return [
User(
username=row[0],
role=UserRole(row[1]),
active=bool(row[2]),
created_at_utc=row[3],
updated_at_utc=row[4],
)
for row in rows
]

255
src/animaltrack/seeds.py Normal file
View File

@@ -0,0 +1,255 @@
# ABOUTME: Idempotent seeder for reference data.
# ABOUTME: Seeds users, locations, species, products, and feed types.
import time
from typing import Any
from animaltrack.id_gen import generate_id
from animaltrack.models.reference import (
FeedType,
Location,
Product,
ProductUnit,
Species,
User,
UserRole,
)
from animaltrack.repositories import (
FeedTypeRepository,
LocationRepository,
ProductRepository,
SpeciesRepository,
UserRepository,
)
def run_seeds(db: Any) -> None:
"""Seed all reference data.
This function is idempotent - running it multiple times produces the same result.
Args:
db: A fastlite database connection with migrations applied.
"""
now_utc = int(time.time() * 1000)
_seed_users(db, now_utc)
_seed_locations(db, now_utc)
_seed_species(db, now_utc)
_seed_products(db, now_utc)
_seed_feed_types(db, now_utc)
def _seed_users(db: Any, now_utc: int) -> None:
"""Seed user data."""
repo = UserRepository(db)
users = [
User(
username="ppetru",
role=UserRole.ADMIN,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
User(
username="ines",
role=UserRole.ADMIN,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
User(
username="guest",
role=UserRole.RECORDER,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
]
for user in users:
repo.upsert(user)
def _seed_locations(db: Any, now_utc: int) -> None:
"""Seed location data.
Locations are created only if they don't exist (by name).
This preserves existing ULIDs on re-seeding.
"""
repo = LocationRepository(db)
location_names = [
"Strip 1",
"Strip 2",
"Strip 3",
"Strip 4",
"Nursery 1",
"Nursery 2",
"Nursery 3",
"Nursery 4",
]
for name in location_names:
existing = repo.get_by_name(name)
if existing is None:
location = Location(
id=generate_id(),
name=name,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
)
repo.upsert(location)
def _seed_species(db: Any, now_utc: int) -> None:
"""Seed species data."""
repo = SpeciesRepository(db)
species_list = [
Species(
code="duck",
name="Duck",
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Species(
code="goose",
name="Goose",
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Species(
code="sheep",
name="Sheep",
active=False,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
]
for species in species_list:
repo.upsert(species)
def _seed_products(db: Any, now_utc: int) -> None:
"""Seed product data."""
repo = ProductRepository(db)
products = [
Product(
code="egg.duck",
name="Duck Egg",
unit=ProductUnit.PIECE,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Product(
code="meat",
name="Meat",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Product(
code="offal",
name="Offal",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Product(
code="fat",
name="Fat",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Product(
code="bones",
name="Bones",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Product(
code="feathers",
name="Feathers",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
Product(
code="down",
name="Down",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
]
for product in products:
repo.upsert(product)
def _seed_feed_types(db: Any, now_utc: int) -> None:
"""Seed feed type data."""
repo = FeedTypeRepository(db)
feed_types = [
FeedType(
code="starter",
name="Starter Feed",
default_bag_size_kg=20,
protein_pct=None,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
FeedType(
code="grower",
name="Grower Feed",
default_bag_size_kg=20,
protein_pct=None,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
FeedType(
code="layer",
name="Layer Feed",
default_bag_size_kg=20,
protein_pct=None,
active=True,
created_at_utc=now_utc,
updated_at_utc=now_utc,
),
]
for feed_type in feed_types:
repo.upsert(feed_type)