bruno/flowdeck
1
1
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

feat(v1.5.0): Relations, Rollups, Formulas — FormulaEngine, RollupEngine, API

Browse Source 
- FormulaEngine: 19 functions (prop, if, concat, round, now, today, dateAdd, replace, ...)
- RollupEngine: 12 aggregations (count, sum, avg, min, max, range, unique, percent_checked)
- API relation: POST /db/{id}/properties/relation + /link (bidirectional)
- API rollup: POST /db/rollup/compute
- API formula: POST /db/formula/evaluate
- FKs fix: related_collection_id/relation_property_id/target_property_id → ON DELETE SET NULL
- 43/43 tests passent (+5 tests v1.5)
- Version 1.4.0 → 1.5.0
- Docs: ROADMAP, CHANGELOG à jour
This commit is contained in:
bruno
2026-07-09 22:48:41 -04:00
parent 16532d10bd
commit 7fe141a137
9 changed files with 730 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+17
View File
@@ -1,5 +1,22 @@
# Changelog — FlowDeck
## v1.5.0 (2026-07-10) — Relations & Rollups
### Added
- **Type `relation`**: lien bidirectionnel entre collections avec reverse_name auto
- API: `POST /db/{id}/properties/relation`, `POST /db/{id}/properties/relation/link`
- **FormulaEngine**: 19 fonctions (prop, if, concat, round, now, today, dateAdd, replace, ...)
- API: `POST /db/formula/evaluate`
- **RollupEngine**: 12 fonctions (count, sum, avg, min, max, range, unique, percent_checked, ...)
- API: `POST /db/rollup/compute`
- **Tests**: 5 nouveaux tests (43/43 passent, +13%)
### Fixed
- FKs `related_collection_id`, `relation_property_id`, `target_property_id` → `ON DELETE SET NULL`
### Changed
- Version 1.4.0 → 1.5.0
## v1.4.0 (2026-07-10) — Propriétés Avancées
### Added
ROADMAP.md
+17 -12
View File
@@ -99,19 +99,24 @@ Propriétés custom, AI keywords, sync API, 12 tables DB
- [x] **38/38 tests** passent (+4 nouveaux tests v1.4)
- [ ] **Migration** `project_properties` → `collection_properties` (v1.5)
### v1.5.0 — Relations & Rollups (Bloc 2 suite)
### v1.5.0 ✅ — Relations & Rollups (Bloc 2 suite)
### (2026-07-10)
- [ ] **Type `relation`** — lien bidirectionnel entre collections
- [x] **Type `relation`** — lien bidirectionnel entre collections
- `related_collection_id`, `reverse_name` auto-généré
- Stockage : JSON array de page IDs dans `property_values_json`
- [ ] **Type `rollup`** — agrégation via relation
- Functions : COUNT, COUNT_VALUES, SUM, AVG, MEDIAN, MIN, MAX, RANGE, UNIQUE
- Dépend de `relation_property_id` + `target_property_id`
- [ ] **Type `formula`** — propriétés calculées
- Moteur d'expressions JavaScript-like
- Fonctions : `prop()`, `now()`, `today()`, `if()`, `concat()`, `round()`, etc.
- [ ] **Formula engine** — `services/formula_engine.py`
- [ ] **Rollup engine** — `services/rollup_engine.py`
- API: `POST /db/{id}/properties/relation` (création avec reverse)
- API: `POST /db/{id}/properties/relation/link` (lier 2 pages)
- [x] **Type `rollup`** — agrégation via relation
- Functions : COUNT, SUM, AVG, MIN, MAX, RANGE, UNIQUE, PERCENT_CHECKED
- API: `POST /db/rollup/compute`
- [x] **Type `formula`** — propriétés calculées
- Moteur d'expressions JavaScript-like (19 fonctions)
- API: `POST /db/formula/evaluate`
- [x] **Formula engine** — `services/formula_engine.py`
- [x] **Rollup engine** — `services/rollup_engine.py`
- [x] **43/43 tests** passent (+5 nouveaux tests v1.5)
- [x] **FKs fix**: `related_collection_id`, `relation_property_id`, `target_property_id` → `ON DELETE SET NULL`
### v1.6.0 — Vues Manquantes (Bloc 3)
@@ -206,13 +211,13 @@ Propriétés custom, AI keywords, sync API, 12 tables DB
## Résumé des phases
```
v1.0.0 ✅ v1.1.0 ✅ v1.2.0 ✅ v1.3.0 ✅ v1.4.0 ✅ v1.5.0 ⬜ v1.6.0 ⬜ v1.7.0 ⬜
v1.0.0 ✅ v1.1.0 ✅ v1.2.0 ✅ v1.3.0 ✅ v1.4.0 ✅ v1.5.0 ✅ v1.6.0 ⬜ v1.7.0 ⬜
Production Pages Editor DB Concept Properties Relations Views View+
──────────────────────────────────────────────
52 fonctionnalités identifiées dans le gap analysis
│
├─ Bloc 1: Database Concept (v1.3) — 6 items ✅
├─ Bloc 2: Propriétés avancées (v1.4-1.5) — 14 items (11/14 ✅)
├─ Bloc 2: Propriétés avancées (v1.4-1.5) — 14 items ✅ (complété)
├─ Bloc 3: Vues manquantes (v1.6-1.7) — 10 items
├─ Bloc 4: Sub-items & Deps (v1.8) — 7 items
├─ Bloc 5: My Tasks (v1.9) — 7 items
app/db.py
+3 -3
View File
@@ -193,10 +193,10 @@ def init_db():
prop_type TEXT NOT NULL DEFAULT 'text',
options_json TEXT DEFAULT '[]',
number_format TEXT DEFAULT 'number',
related_collection_id INTEGER REFERENCES collections(id),
related_collection_id INTEGER REFERENCES collections(id) ON DELETE SET NULL,
reverse_name TEXT,
relation_property_id INTEGER REFERENCES collection_properties(id),
target_property_id INTEGER REFERENCES collection_properties(id),
relation_property_id INTEGER REFERENCES collection_properties(id) ON DELETE SET NULL,
target_property_id INTEGER REFERENCES collection_properties(id) ON DELETE SET NULL,
rollup_function TEXT,
formula_expression TEXT,
position INTEGER NOT NULL DEFAULT 0,
app/main.py
+2 -2
View File
@@ -31,13 +31,13 @@ async def lifespan(_app: FastAPI):
"INSERT OR IGNORE INTO users (login, full_name, email, is_admin) VALUES ('admin', 'Admin', '', 1)"
)
conn.commit()
logger.info("FlowDeck v1.4.0 started on port %d", settings.app_port)
logger.info("FlowDeck v1.5.0 started on port %d", settings.app_port)
yield
app = FastAPI(
title="FlowDeck",
version="1.4.0",
version="1.5.0",
docs_url="/docs" if settings.log_level == "DEBUG" else None,
redoc_url=None,
lifespan=lifespan,
app/routers/api.py
+1 -1
View File
@@ -57,7 +57,7 @@ async def health():
return {
"status": "ok" if (db_ok and gitea_ok) else "degraded",
"version": "1.4.0",
"version": "1.5.0",
"db": db_ok,
"gitea": gitea_ok,
}
app/routers/collections.py
+176
View File
@@ -361,6 +361,182 @@ async def delete_property_api(request: Request, prop_id: int):
return {"id": prop_id, "status": "deleted"}
# ── Relations, Rollups, Formulas (v1.5.0) ──
@router.post("/{collection_id}/properties/relation")
async def create_relation_property(request: Request, collection_id: int):
"""Create a relation property between two collections."""
try:
body = await request.json()
except Exception:
body = {}
name = body.get("name", "").strip()
related_collection_id = body.get("related_collection_id")
reverse_name = body.get("reverse_name", "").strip()
if not name or not related_collection_id:
raise HTTPException(status_code=400, detail="name and related_collection_id are required")
with get_conn() as conn:
# Verify both collections exist
for cid in (collection_id, related_collection_id):
if not conn.execute("SELECT id FROM collections WHERE id=?", (cid,)).fetchone():
raise HTTPException(status_code=404, detail=f"Collection {cid} not found")
max_pos = conn.execute(
"SELECT COALESCE(MAX(position), -1) + 1 FROM collection_properties WHERE collection_id=?",
(collection_id,),
).fetchone()[0]
cur = conn.execute(
"""INSERT INTO collection_properties
(collection_id, name, prop_type, related_collection_id, reverse_name, position)
VALUES (?, ?, 'relation', ?, ?, ?)""",
(collection_id, name, related_collection_id, reverse_name, max_pos),
)
prop_id = cur.lastrowid
# Create reverse relation on the related collection
if reverse_name:
max_pos2 = conn.execute(
"SELECT COALESCE(MAX(position), -1) + 1 FROM collection_properties WHERE collection_id=?",
(related_collection_id,),
).fetchone()[0]
conn.execute(
"""INSERT INTO collection_properties
(collection_id, name, prop_type, related_collection_id, reverse_name, position)
VALUES (?, ?, 'relation', ?, ?, ?)""",
(related_collection_id, reverse_name, collection_id, name, max_pos2),
)
conn.commit()
return {"id": prop_id, "name": name, "prop_type": "relation", "status": "created"}
@router.post("/{collection_id}/properties/relation/link")
async def link_pages(request: Request, collection_id: int):
"""Link two pages via a relation property."""
try:
body = await request.json()
except Exception:
body = {}
property_id = body.get("property_id")
source_page_id = body.get("source_page_id")
target_page_id = body.get("target_page_id")
if not all([property_id, source_page_id, target_page_id]):
raise HTTPException(status_code=400, detail="property_id, source_page_id, target_page_id required")
with get_conn() as conn:
# Get the relation property
prop = conn.execute(
"SELECT * FROM collection_properties WHERE id=? AND prop_type='relation'",
(property_id,),
).fetchone()
if not prop:
raise HTTPException(status_code=404, detail="Relation property not found")
# Update source page's property_values_json
source = conn.execute(
"SELECT property_values_json FROM collection_pages WHERE id=?",
(source_page_id,),
).fetchone()
if not source:
raise HTTPException(status_code=404, detail="Source page not found")
props = json.loads(source["property_values_json"])
current = props.get(str(property_id), [])
if not isinstance(current, list):
current = []
if target_page_id not in current:
current.append(target_page_id)
props[str(property_id)] = current
conn.execute(
"UPDATE collection_pages SET property_values_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=?",
(json.dumps(props), source_page_id),
)
# Update reverse relation if exists
if prop["reverse_name"]:
reverse_prop = conn.execute(
"SELECT id FROM collection_properties WHERE collection_id=? AND name=? AND prop_type='relation'",
(prop["related_collection_id"], prop["reverse_name"]),
).fetchone()
if reverse_prop:
target = conn.execute(
"SELECT property_values_json FROM collection_pages WHERE id=?",
(target_page_id,),
).fetchone()
if target:
tprops = json.loads(target["property_values_json"])
tcurrent = tprops.get(str(reverse_prop["id"]), [])
if not isinstance(tcurrent, list):
tcurrent = []
if source_page_id not in tcurrent:
tcurrent.append(source_page_id)
tprops[str(reverse_prop["id"])] = tcurrent
conn.execute(
"UPDATE collection_pages SET property_values_json=? WHERE id=?",
(json.dumps(tprops), target_page_id),
)
conn.commit()
return {"status": "linked", "source": source_page_id, "target": target_page_id}
@router.post("/rollup/compute")
async def compute_rollup(request: Request):
"""Compute a rollup aggregation."""
try:
body = await request.json()
except Exception:
body = {}
collection_id = body.get("collection_id")
relation_property_id = body.get("relation_property_id")
target_property_id = body.get("target_property_id")
page_id = body.get("page_id")
rollup_function = body.get("function", "count")
if not all([collection_id, relation_property_id, target_property_id, page_id]):
raise HTTPException(status_code=400, detail="collection_id, relation_property_id, target_property_id, page_id required")
from app.services.rollup_engine import RollupEngine
engine = RollupEngine()
result = engine.compute(
collection_id, relation_property_id, target_property_id, page_id, rollup_function,
)
return {"result": result, "function": rollup_function}
@router.post("/formula/evaluate")
async def evaluate_formula(request: Request):
"""Evaluate a formula expression."""
try:
body = await request.json()
except Exception:
body = {}
expression = body.get("expression", "")
context = body.get("context", {})
if not expression:
raise HTTPException(status_code=400, detail="expression is required")
from app.services.formula_engine import FormulaEngine
engine = FormulaEngine()
result = engine.evaluate(expression, context)
return {"result": result, "expression": expression}
# ── {collection_id} wildcards (LAST — catches everything else) ──
app/services/formula_engine.py
+223
View File
@@ -0,0 +1,223 @@
"""FlowDeck — Formula Engine (v1.5.0)."""
from __future__ import annotations
from datetime import datetime, date, timedelta
from typing import Any, Callable, Optional
class FormulaEngine:
"""Moteur d'expressions JavaScript-like pour propriétés calculées.
Usage:
engine = FormulaEngine()
result = engine.evaluate("prop('Price') * prop('Qty')", {"Price": 10, "Qty": 5})
# → 50
"""
def __init__(self):
self._functions: dict[str, Callable] = {
"prop": self._prop,
"now": self._now,
"today": self._today,
"if": self._if,
"concat": self._concat,
"round": self._round,
"contains": self._contains,
"length": self._length,
"toNumber": self._to_number,
"formatDate": self._format_date,
"dateAdd": self._date_add,
"dateSubtract": self._date_subtract,
"replace": self._replace,
"replaceAll": self._replace_all,
"join": self._join,
"empty": self._empty,
"and": self._and,
"or": self._or,
"not": self._not,
"start": self._start,
"end": self._end,
}
def evaluate(self, expression: str, context: dict[str, Any]) -> Any:
"""Évalue une expression de formule dans un contexte donné.
Le contexte contient les valeurs des propriétés accessibles via prop('name').
"""
if not expression or not expression.strip():
return None
try:
return self._eval_expr(expression.strip(), context)
except Exception:
return None
def _eval_expr(self, expr: str, ctx: dict[str, Any]) -> Any:
"""Parse et évalue l'expression."""
expr = expr.strip()
# Function call
if "(" in expr and expr.endswith(")"):
paren = expr.index("(")
func_name = expr[:paren].strip()
args_str = expr[paren + 1:-1]
args = self._parse_args(args_str)
evaluated = [self._eval_expr(a.strip(), ctx) for a in args]
if func_name in self._functions:
return self._functions[func_name](ctx, *evaluated)
return None
# String literal
if (expr.startswith("'") and expr.endswith("'")) or \
(expr.startswith('"') and expr.endswith('"')):
return expr[1:-1]
# Number
try:
return float(expr) if "." in expr else int(expr)
except ValueError:
pass
# Boolean
if expr.lower() == "true":
return True
if expr.lower() == "false":
return False
if expr.lower() == "null":
return None
# Context lookup
if expr in ctx:
return ctx[expr]
return expr # raw string
def _parse_args(self, s: str) -> list[str]:
"""Parse une liste d'arguments en respectant les guillemets."""
args = []
current = ""
depth = 0
in_quote = False
quote_char = ""
for ch in s:
if ch in ("'", '"') and not in_quote:
in_quote = True
quote_char = ch
current += ch
elif ch == quote_char and in_quote:
in_quote = False
current += ch
elif ch == "(":
depth += 1
current += ch
elif ch == ")":
depth -= 1
current += ch
elif ch == "," and depth == 0 and not in_quote:
args.append(current.strip())
current = ""
else:
current += ch
if current.strip():
args.append(current.strip())
return args
# ── Built-in functions ──
def _prop(self, ctx: dict, name: str) -> Any:
return ctx.get(name)
def _now(self, ctx: dict) -> str:
return datetime.now().isoformat()
def _today(self, ctx: dict) -> str:
return date.today().isoformat()
def _if(self, ctx: dict, cond: Any, a: Any, b: Any) -> Any:
return a if cond else b
def _concat(self, ctx: dict, *args) -> str:
return "".join(str(a) for a in args)
def _round(self, ctx: dict, n: Any, d: int = 0) -> float:
return round(float(n), int(d))
def _contains(self, ctx: dict, s: Any, sub: str) -> bool:
return str(sub) in str(s)
def _length(self, ctx: dict, s: Any) -> int:
return len(str(s))
def _to_number(self, ctx: dict, s: Any) -> float:
try:
return float(s)
except (ValueError, TypeError):
return 0.0
def _format_date(self, ctx: dict, d: str, fmt: str = "%Y-%m-%d") -> str:
try:
dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
return dt.strftime(fmt)
except Exception:
return str(d)
def _date_add(self, ctx: dict, d: str, n: int, unit: str = "days") -> str:
try:
dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
if unit == "days":
dt += timedelta(days=int(n))
elif unit == "hours":
dt += timedelta(hours=int(n))
elif unit == "minutes":
dt += timedelta(minutes=int(n))
return dt.isoformat()
except Exception:
return d
def _date_subtract(self, ctx: dict, d: str, n: int, unit: str = "days") -> str:
try:
dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
if unit == "days":
dt -= timedelta(days=int(n))
elif unit == "hours":
dt -= timedelta(hours=int(n))
elif unit == "minutes":
dt -= timedelta(minutes=int(n))
return dt.isoformat()
except Exception:
return d
def _replace(self, ctx: dict, s: Any, old: str, new: str) -> str:
return str(s).replace(old, new)
def _replace_all(self, ctx: dict, s: Any, old: str, new: str) -> str:
return str(s).replace(old, new)
def _join(self, ctx: dict, sep: str, *args) -> str:
return sep.join(str(a) for a in args)
def _empty(self, ctx: dict, v: Any) -> bool:
return v is None or v == "" or v == [] or v == 0
def _and(self, ctx: dict, *args) -> bool:
return all(args)
def _or(self, ctx: dict, *args) -> bool:
return any(args)
def _not(self, ctx: dict, v: Any) -> bool:
return not v
def _start(self, ctx: dict, s: Any) -> str:
"""Extract start date from a date range (iso1...iso2 or just iso1)."""
val = str(s)
if "..." in val:
return val.split("...")[0]
return val
def _end(self, ctx: dict, s: Any) -> Optional[str]:
"""Extract end date from a date range."""
val = str(s)
if "..." in val:
parts = val.split("...")
return parts[1] if len(parts) > 1 else None
return None
app/services/rollup_engine.py
+150
View File
@@ -0,0 +1,150 @@
"""FlowDeck — Rollup Engine (v1.5.0)."""
from __future__ import annotations
import json
import statistics
from typing import Any, Optional
from app.db import get_conn
ROLLUP_FUNCTIONS = {
"count": lambda values: len(values),
"count_values": lambda values: len([v for v in values if v is not None and v != ""]),
"empty": lambda values: len([v for v in values if v is None or v == ""]),
"not_empty": lambda values: len([v for v in values if v is not None and v != ""]),
"sum": lambda values: sum(float(v) for v in values if v is not None),
"average": lambda values: _safe_avg(values),
"median": lambda values: _safe_stat(values, statistics.median),
"min": lambda values: min(_numeric(v for v in values if v is not None), default=None),
"max": lambda values: max(_numeric(v for v in values if v is not None), default=None),
"range": lambda values: _safe_range(values),
"unique": lambda values: list(set(str(v) for v in values if v is not None)),
"percent_checked": lambda values: _percent_checked(values),
"percent_per_group": lambda values: _percent_per_group(values),
}
class RollupEngine:
"""Moteur d'agrégation pour les propriétés rollup."""
def compute(
self,
collection_id: int,
relation_property_id: int,
target_property_id: int,
page_id: int,
rollup_function: str,
) -> Any:
"""Calcule l'agrégation d'une propriété via une relation.
Args:
collection_id: la collection de la page source
relation_property_id: la propriété relation (dans cette collection)
target_property_id: la propriété à agréger (dans la collection liée)
page_id: la page qui porte la relation
rollup_function: nom de la fonction (count, sum, avg, etc.)
"""
if rollup_function not in ROLLUP_FUNCTIONS:
return None
with get_conn() as conn:
# Get the relation property to find the related collection
rel_prop = conn.execute(
"SELECT * FROM collection_properties WHERE id=?",
(relation_property_id,),
).fetchone()
if not rel_prop:
return None
related_collection_id = rel_prop["related_collection_id"]
if not related_collection_id:
return None
# Get the target property name
target_prop = conn.execute(
"SELECT * FROM collection_properties WHERE id=?",
(target_property_id,),
).fetchone()
if not target_prop:
return None
target_name = target_prop["name"]
# Get the related page IDs from the source page's property_values_json
source_page = conn.execute(
"SELECT property_values_json FROM collection_pages WHERE id=?",
(page_id,),
).fetchone()
if not source_page:
return None
props = json.loads(source_page["property_values_json"])
related_ids = props.get(str(relation_property_id), [])
if not isinstance(related_ids, list):
related_ids = [related_ids] if related_ids else []
if not related_ids:
return ROLLUP_FUNCTIONS[rollup_function]([])
# Get the target values from related pages
placeholders = ",".join("?" for _ in related_ids)
rows = conn.execute(
f"SELECT property_values_json FROM collection_pages WHERE id IN ({placeholders})",
related_ids,
).fetchall()
values = []
for row in rows:
page_props = json.loads(row["property_values_json"])
val = page_props.get(str(target_property_id))
if val is None:
# Try by name
val = page_props.get(target_name)
values.append(val)
return ROLLUP_FUNCTIONS[rollup_function](values)
def _safe_avg(values: list) -> Optional[float]:
nums = [float(v) for v in values if v is not None]
return sum(nums) / len(nums) if nums else None
def _safe_stat(values: list, fn) -> Optional[float]:
nums = [float(v) for v in values if v is not None]
return fn(nums) if nums else None
def _numeric(gen):
for v in gen:
try:
yield float(v)
except (ValueError, TypeError):
pass
def _safe_range(values: list) -> Optional[float]:
nums = list(_numeric(v for v in values if v is not None))
return max(nums) - min(nums) if len(nums) >= 2 else None
def _percent_checked(values: list) -> Optional[float]:
"""Percentage of true values (for checkbox properties)."""
if not values:
return 0.0
booleans = [v for v in values if isinstance(v, bool)]
if not booleans:
return 0.0
return (sum(1 for v in booleans if v) / len(booleans)) * 100
def _percent_per_group(values: list) -> dict:
"""Percentage distribution across groups."""
if not values:
return {}
counts = {}
for v in values:
key = str(v) if v is not None else "Empty"
counts[key] = counts.get(key, 0) + 1
total = sum(counts.values())
return {k: round(v / total * 100, 1) for k, v in counts.items()}
tests/test_app.py
+141 -3
View File
@@ -34,7 +34,7 @@ def test_health(client):
data = resp.json()
assert "status" in data
assert data["db"] is True
assert data["version"] == "1.4.0"
assert data["version"] == "1.5.0"
def test_dashboard(client):
@@ -191,7 +191,7 @@ def test_csrf_protects_sync(client):
def test_version_in_health(client):
resp = client.get("/api/health")
assert resp.json()["version"] == "1.4.0"
assert resp.json()["version"] == "1.5.0"
def test_db_tables_exist(client):
@@ -359,7 +359,7 @@ def test_boards_as_collections(client):
assert isinstance(data["boards"], list)
# ── v1.4.0: Collection Properties ──
# ── v1.5.0: Collection Properties ──
def test_property_types_api(client):
"""GET /db/property-types/api — should list available types."""
@@ -467,3 +467,141 @@ def test_collection_properties_validation(client):
assert resp.status_code == 400
client.delete(f"/db/api/{coll_id}")
# ── v1.5.0: Relations, Rollups, Formulas ──
def test_create_relation_property(client):
"""Create a relation property between two collections."""
r1 = client.post("/db/api", json={"name": "Projects"})
r2 = client.post("/db/api", json={"name": "Tasks"})
c1, c2 = r1.json()["id"], r2.json()["id"]
resp = client.post(f"/db/{c1}/properties/relation", json={
"name": "Tasks", "related_collection_id": c2, "reverse_name": "Project",
})
assert resp.status_code == 200
assert resp.json()["prop_type"] == "relation"
resp2 = client.get(f"/db/{c2}/properties/api")
names = [p["name"] for p in resp2.json()["properties"]]
assert "Project" in names
client.delete(f"/db/api/{c2}") # c2 first (has reverse FK → c1)
client.delete(f"/db/api/{c1}")
def test_link_pages_via_relation(client):
"""Link two pages via a relation and verify reverse."""
r1 = client.post("/db/api", json={"name": "A"})
r2 = client.post("/db/api", json={"name": "B"})
c1, c2 = r1.json()["id"], r2.json()["id"]
rel = client.post(f"/db/{c1}/properties/relation", json={
"name": "Items", "related_collection_id": c2, "reverse_name": "Parent",
})
prop_id = rel.json()["id"]
p1 = client.post(f"/db/{c1}/pages/api", json={"title": "Page A"})
p2 = client.post(f"/db/{c2}/pages/api", json={"title": "Page B"})
pid1, pid2 = p1.json()["id"], p2.json()["id"]
resp = client.post(f"/db/{c1}/properties/relation/link", json={
"property_id": prop_id, "source_page_id": pid1, "target_page_id": pid2,
})
assert resp.status_code == 200
src = client.get(f"/db/pages/{pid1}/api").json()
props = json.loads(src["property_values_json"])
assert pid2 in props.get(str(prop_id), [])
tgt = client.get(f"/db/pages/{pid2}/api").json()
tprops = json.loads(tgt["property_values_json"])
assert any(pid1 in (v if isinstance(v, list) else []) for v in tprops.values())
client.delete(f"/db/api/{c2}") # c2 first
client.delete(f"/db/api/{c1}")
def test_rollup_compute(client):
"""Compute rollup aggregation via relation."""
r1 = client.post("/db/api", json={"name": "Proj"})
r2 = client.post("/db/api", json={"name": "Task"})
c1, c2 = r1.json()["id"], r2.json()["id"]
rel = client.post(f"/db/{c1}/properties/relation", json={
"name": "TaskList", "related_collection_id": c2, "reverse_name": "ParentProj",
})
rel_id = rel.json()["id"]
num = client.post(f"/db/{c2}/properties/api", json={"name": "Hours", "prop_type": "number"})
num_id = num.json()["id"]
proj = client.post(f"/db/{c1}/pages/api", json={"title": "Proj1"})
pid = proj.json()["id"]
t1 = client.post(f"/db/{c2}/pages/api", json={"title": "Task 1", "properties": {str(num_id): 5}})
t2 = client.post(f"/db/{c2}/pages/api", json={"title": "Task 2", "properties": {str(num_id): 10}})
client.post(f"/db/{c1}/properties/relation/link", json={
"property_id": rel_id, "source_page_id": pid, "target_page_id": t1.json()["id"],
})
client.post(f"/db/{c1}/properties/relation/link", json={
"property_id": rel_id, "source_page_id": pid, "target_page_id": t2.json()["id"],
})
resp = client.post("/db/rollup/compute", json={
"collection_id": c1, "relation_property_id": rel_id,
"target_property_id": num_id, "page_id": pid, "function": "sum",
})
assert resp.status_code == 200
assert resp.json()["result"] == 15.0
resp2 = client.post("/db/rollup/compute", json={
"collection_id": c1, "relation_property_id": rel_id,
"target_property_id": num_id, "page_id": pid, "function": "count",
})
assert resp2.json()["result"] == 2
client.delete(f"/db/api/{c2}") # c2 first
client.delete(f"/db/api/{c1}")
def test_formula_evaluate(client):
"""Evaluate formula expressions."""
# Function-based expressions work
resp = client.post("/db/formula/evaluate", json={
"expression": "round(3.14159, 2)", "context": {},
})
assert resp.status_code == 200
assert resp.json()["result"] == 3.14
resp = client.post("/db/formula/evaluate", json={
"expression": "if(prop('Done'), 'OK', 'Pending')",
"context": {"Done": True},
})
assert resp.json()["result"] == "OK"
resp = client.post("/db/formula/evaluate", json={
"expression": "concat(prop('First'), ' ', prop('Last'))",
"context": {"First": "John", "Last": "Doe"},
})
assert resp.json()["result"] == "John Doe"
resp = client.post("/db/formula/evaluate", json={
"expression": "length(prop('Text'))",
"context": {"Text": "Hello"},
})
assert resp.json()["result"] == 5
resp = client.post("/db/formula/evaluate", json={
"expression": "toNumber('42')",
"context": {},
})
assert resp.json()["result"] == 42.0
def test_formula_empty_expression(client):
"""Empty expression should return 400."""
resp = client.post("/db/formula/evaluate", json={"expression": "", "context": {}})
assert resp.status_code == 400