feat(S5): pet rule CRUD query functions

tests/unit/test_pet_rules_queries.py

@@ -0,0 +1,36 @@
+import json
+import pytest
+from vigilar.storage.queries import (
+    count_pet_rules, delete_pet_rule, get_all_enabled_rules,
+    get_pet_rules, insert_pet_rule, update_pet_rule,
+)
+
+def test_insert_and_get(test_db):
+    rid = insert_pet_rule(test_db, "pet1", "Outdoor timer",
+        json.dumps([{"type": "in_zone_longer_than", "zone": "EXTERIOR", "minutes": 45}]),
+        "push_notify", "{pet_name} outside", 30, 0)
+    assert rid > 0
+    rules = get_pet_rules(test_db, "pet1")
+    assert len(rules) == 1
+    assert rules[0]["name"] == "Outdoor timer"
+
+def test_get_all_enabled(test_db):
+    insert_pet_rule(test_db, "pet1", "A", json.dumps([]), "push_notify", "", 30, 0)
+    insert_pet_rule(test_db, "pet2", "B", json.dumps([]), "log_event", "", 60, 0)
+    assert len(get_all_enabled_rules(test_db)) == 2
+
+def test_update(test_db):
+    rid = insert_pet_rule(test_db, "pet1", "Test", json.dumps([]), "push_notify", "", 30, 0)
+    update_pet_rule(test_db, rid, name="Updated", cooldown_minutes=60)
+    rules = get_pet_rules(test_db, "pet1")
+    assert rules[0]["name"] == "Updated"
+
+def test_delete(test_db):
+    rid = insert_pet_rule(test_db, "pet1", "Test", json.dumps([]), "push_notify", "", 30, 0)
+    delete_pet_rule(test_db, rid)
+    assert len(get_pet_rules(test_db, "pet1")) == 0
+
+def test_count(test_db):
+    for i in range(5):
+        insert_pet_rule(test_db, "pet1", f"Rule {i}", json.dumps([]), "push_notify", "", 30, i)
+    assert count_pet_rules(test_db, "pet1") == 5

vigilar/storage/queries.py

@@ -546,3 +546,52 @@ def update_package_status(engine: Engine, package_id: int, status: str, **kwargs
     with engine.begin() as conn:
         conn.execute(package_events.update().where(package_events.c.id == package_id)
             .values(status=status, **kwargs))
+
+
+# --- Pet Rules ---
+
+def insert_pet_rule(engine, pet_id, name, conditions, action, action_message, cooldown_minutes, priority) -> int:
+    from vigilar.storage.schema import pet_rules
+    with engine.begin() as conn:
+        result = conn.execute(pet_rules.insert().values(
+            pet_id=pet_id, name=name, enabled=1, conditions=conditions, action=action,
+            action_message=action_message, cooldown_minutes=cooldown_minutes,
+            priority=priority, created_at=time.time()))
+        return result.inserted_primary_key[0]
+
+
+def get_pet_rules(engine, pet_id) -> list[dict]:
+    from vigilar.storage.schema import pet_rules
+    with engine.connect() as conn:
+        return [dict(r) for r in conn.execute(
+            select(pet_rules).where(pet_rules.c.pet_id == pet_id)
+            .order_by(pet_rules.c.priority, pet_rules.c.id)).mappings().all()]
+
+
+def get_all_enabled_rules(engine) -> list[dict]:
+    from vigilar.storage.schema import pet_rules
+    with engine.connect() as conn:
+        return [dict(r) for r in conn.execute(
+            select(pet_rules).where(pet_rules.c.enabled == 1)
+            .order_by(pet_rules.c.priority, pet_rules.c.id)).mappings().all()]
+
+
+def update_pet_rule(engine, rule_id, **updates) -> None:
+    from vigilar.storage.schema import pet_rules
+    with engine.begin() as conn:
+        conn.execute(pet_rules.update().where(pet_rules.c.id == rule_id).values(**updates))
+
+
+def delete_pet_rule(engine, rule_id) -> None:
+    from vigilar.storage.schema import pet_rules
+    with engine.begin() as conn:
+        conn.execute(pet_rules.delete().where(pet_rules.c.id == rule_id))
+
+
+def count_pet_rules(engine, pet_id) -> int:
+    from sqlalchemy import func
+    from vigilar.storage.schema import pet_rules
+    with engine.connect() as conn:
+        return conn.execute(
+            select(func.count()).select_from(pet_rules).where(pet_rules.c.pet_id == pet_id)
+        ).scalar() or 0