"""Database models for suggested chains.
Manages temporary chain suggestions before deployment.
Suggested chains are stored in the `suggested_chains` table and can be
deployed as real actions, edited, or discarded.
Table schema:
suggested_chains (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
site_id INTEGER NOT NULL,
chain_config TEXT NOT NULL, -- JSON string
source TEXT DEFAULT 'llm', -- 'llm' | 'template' | 'template_fallback'
confidence REAL DEFAULT 0.0,
category TEXT DEFAULT 'general',
status TEXT DEFAULT 'pending', -- 'pending' | 'deployed' | 'discarded'
deployed_action_id INTEGER, -- FK to actions table after deploy
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
import json
import logging
import secrets
from datetime import datetime
from typing import Any, Optional
logger = logging.getLogger("agentforms.suggested_chains")
# ─── Column-name allowlist helpers ──────────────────────────────────────
_SUGGESTED_CHAINS_ALLOWED_COLUMNS = {
"name",
"description",
"config",
"active",
"position",
"chain_config",
"status",
"source",
"confidence",
"category",
"deployed_action_id",
"created_at",
}
def _validate_columns(columns: list, allowed: set) -> list:
"""Validate column names against an allowlist. Returns list of 'col = ?' clauses."""
valid = []
for col in columns:
if col in allowed:
valid.append(f"{col} = ?")
else:
logger.warning(f"Blocked column in UPDATE: {col} (not in allowlist)")
return valid
from app.models import get_db
def create_suggestion(
user_id: int,
site_id: int,
chain_config: Any,
source: str = "template",
confidence: float = 0.5,
category: str = "general",
) -> str:
"""Create a new chain suggestion.
Args:
user_id: User ID
site_id: Site ID
chain_config: Chain config (dict or JSON string)
source: Source of suggestion ('llm', 'template', 'template_fallback')
confidence: Confidence score (0.0-1.0)
category: Category of suggestion
Returns:
Suggestion ID string (e.g., 'sc_abc123')
"""
if isinstance(chain_config, dict):
chain_config = json.dumps(chain_config)
# Generate suggestion ID
suggestion_id = f"sc_{secrets.token_hex(6)}"
conn = get_db()
conn.execute(
"""INSERT INTO suggested_chains
(id, user_id, site_id, chain_config, source, confidence, category, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
int(suggestion_id.replace("sc_", ""), 16),
user_id,
site_id,
chain_config,
source,
confidence,
category,
"pending",
),
)
conn.commit()
logger.info("Created suggestion %s for user %d site %d", suggestion_id, user_id, site_id)
return suggestion_id
def get_suggestion(suggestion_id: int) -> dict | None:
"""Get a suggestion by ID.
Returns suggestion dict or None.
"""
conn = get_db()
row = conn.execute(
"SELECT * FROM suggested_chains WHERE id = ?",
(suggestion_id,),
).fetchone()
if not row:
return None
result = dict(row)
# Parse chain_config from JSON string
if isinstance(result.get("chain_config"), str):
try:
result["chain_config"] = json.loads(result["chain_config"])
except (json.JSONDecodeError, TypeError):
pass
return result
def get_suggestions_for_site(site_id: int, status: str | None = None) -> list:
"""Get suggestions for a site.
Args:
site_id: Site ID
status: Filter by status ('pending', 'deployed', 'discarded') or None for all
"""
conn = get_db()
if status:
rows = conn.execute(
"SELECT * FROM suggested_chains WHERE site_id = ? AND status = ? ORDER BY created_at DESC",
(site_id, status),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM suggested_chains WHERE site_id = ? ORDER BY created_at DESC",
(site_id,),
).fetchall()
suggestions = []
for row in rows:
suggestion = dict(row)
# Parse chain_config from JSON string
if isinstance(suggestion.get("chain_config"), str):
try:
suggestion["chain_config"] = json.loads(suggestion["chain_config"])
except (json.JSONDecodeError, TypeError):
pass
suggestions.append(suggestion)
return suggestions
def update_suggestion(suggestion_id: int, chain_config: Any = None, status: str | None = None) -> bool:
"""Update a suggestion.
Args:
suggestion_id: Suggestion ID
chain_config: New chain config (dict or JSON string)
status: New status ('pending', 'deployed', 'discarded')
"""
conn = get_db()
updates = []
params = []
if chain_config is not None:
if isinstance(chain_config, dict):
chain_config = json.dumps(chain_config)
updates.append("chain_config = ?")
params.append(chain_config)
if status is not None:
updates.append("status = ?")
params.append(status)
if not updates:
return False
params.append(suggestion_id)
conn.execute(
f"UPDATE suggested_chains SET {', '.join(updates)} WHERE id = ?",
params,
)
conn.commit()
return True
def delete_suggestion(suggestion_id: int) -> bool:
"""Delete a suggestion.
Args:
suggestion_id: Suggestion ID
"""
conn = get_db()
result = conn.execute(
"DELETE FROM suggested_chains WHERE id = ?",
(suggestion_id,),
)
conn.commit()
return result.rowcount > 0
def deploy_suggestion(suggestion_id: int, deployed_action_id: int) -> bool:
"""Mark a suggestion as deployed and link to the action.
Args:
suggestion_id: Suggestion ID
deployed_action_id: ID of the created action
"""
return (
update_suggestion(
suggestion_id,
status="deployed",
)
and update_suggestion(
suggestion_id,
chain_config=None, # No config change
status=None, # No status change
)
or False
)
def cleanup_expired(before: datetime | None = None, max_age_hours: int = 24) -> int:
"""Delete expired suggestions.
Args:
before: Delete suggestions created before this time. Defaults to now - max_age_hours.
max_age_hours: Maximum age in hours (default 24).
Returns:
Number of deleted suggestions.
"""
if before is None:
from datetime import timedelta
before = datetime.now() - timedelta(hours=max_age_hours)
conn = get_db()
result = conn.execute(
"DELETE FROM suggested_chains WHERE status = 'pending' AND created_at < ?",
(before.isoformat(),),
)
conn.commit()
logger.info("Cleaned up %d expired suggestions", result.rowcount)
return result.rowcount