connections.py
Rama main · réplica de lectura del repositorio original.
"""Saved media connections for account workspaces and connection-bound jobs.
HTTP callers must check account ownership before resolving a client. Background
jobs resolve their saved connection directly.
"""
import json
import threading
import uuid
import database
from app.media_client import Connection, MediaClient
_clients = {}
_clients_lock = threading.Lock()
def initialize_schema(conn):
conn.execute('''CREATE TABLE IF NOT EXISTS media_connections (
id TEXT PRIMARY KEY, base_url TEXT NOT NULL, server_type TEXT NOT NULL,
username TEXT NOT NULL, password TEXT NOT NULL, user_id TEXT NOT NULL,
server_id TEXT NOT NULL, ai_settings TEXT NOT NULL DEFAULT '{}'
)''')
columns = {r['name'] for r in conn.execute('PRAGMA table_info(media_connections)')}
for name, definition in (
('owner_id', 'TEXT REFERENCES accounts(id)'), ('label', "TEXT NOT NULL DEFAULT ''"),
('webhook_secret', "TEXT NOT NULL DEFAULT ''"), ('api_key_hash', "TEXT NOT NULL DEFAULT ''"),
('api_key', "TEXT NOT NULL DEFAULT ''"),
('webhook_secret_updated_at', 'REAL'), ('webhook_setup_requested_at', 'REAL'),
('webhook_setup_acknowledged_at', 'REAL'), ('webhook_verified_at', 'REAL'),
('webhook_last_received_at', 'REAL')
):
if name not in columns:
conn.execute(f'ALTER TABLE media_connections ADD COLUMN {name} {definition}')
if 'connection_id' not in {r['name'] for r in conn.execute('PRAGMA table_info(schedules)')}:
conn.execute('ALTER TABLE schedules ADD COLUMN connection_id TEXT REFERENCES media_connections(id)')
conn.execute('''CREATE TABLE IF NOT EXISTS connection_presets (
id TEXT PRIMARY KEY, connection_id TEXT NOT NULL REFERENCES media_connections(id),
name TEXT NOT NULL, data TEXT NOT NULL,
UNIQUE(connection_id, name)
)''')
preset_columns = {r['name'] for r in conn.execute('PRAGMA table_info(connection_presets)')}
if 'tags_json' not in preset_columns:
conn.execute("ALTER TABLE connection_presets ADD COLUMN tags_json TEXT NOT NULL DEFAULT '[]'")
if 'is_favorite' not in preset_columns:
conn.execute("ALTER TABLE connection_presets ADD COLUMN is_favorite INTEGER NOT NULL DEFAULT 0")
conn.execute('''CREATE TABLE IF NOT EXISTS connection_recipes (
id TEXT PRIMARY KEY,
connection_id TEXT NOT NULL REFERENCES media_connections(id),
name TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
block_json TEXT NOT NULL,
tags_json TEXT NOT NULL DEFAULT '[]',
is_favorite INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)''')
conn.execute('CREATE INDEX IF NOT EXISTS idx_recipes_conn ON connection_recipes(connection_id)')
schedule_columns = {r['name'] for r in conn.execute('PRAGMA table_info(schedules)')}
if 'preset_id' not in schedule_columns:
conn.execute('ALTER TABLE schedules ADD COLUMN preset_id TEXT REFERENCES connection_presets(id)')
backfill_schedule_preset_ids(conn)
from app.ai_policy import migrate_provider_optin
migrate_provider_optin(conn)
def webhook_status(row):
"""Return the user-facing state for a connection's current webhook secret."""
if not row or not row.get('webhook_secret'):
return 'disabled'
changed = row.get('webhook_secret_updated_at') or 0
if (row.get('webhook_verified_at') or 0) >= changed and row.get('webhook_verified_at'):
return 'connected'
if (row.get('webhook_setup_acknowledged_at') or 0) >= changed:
return 'waiting_for_event'
if (row.get('webhook_setup_requested_at') or 0) >= changed:
return 'setup_requested'
return 'needs_setup'
def backfill_schedule_preset_ids(conn, connection_id=None):
"""Bind legacy name-based schedules to the matching preset in their connection."""
params = () if connection_id is None else (connection_id,)
where = ('WHERE s.preset_id IS NULL AND s.job_type IN (\'builder\', \'preset\')' if connection_id is None
else 'WHERE s.preset_id IS NULL AND s.job_type IN (\'builder\', \'preset\') AND s.connection_id=?')
rows = conn.execute(
f'SELECT s.id, s.connection_id, s.config_data FROM schedules s {where}', params
).fetchall()
for row in rows:
if not row['connection_id'] or not row['config_data']:
continue
try:
config_data = json.loads(row['config_data'])
except (TypeError, json.JSONDecodeError):
continue
preset_name = config_data.get('preset_name') if isinstance(config_data, dict) else None
if not preset_name:
continue
preset = conn.execute(
'SELECT id FROM connection_presets WHERE connection_id=? AND name=?',
(row['connection_id'], preset_name)
).fetchone()
if preset:
conn.execute('UPDATE schedules SET preset_id=? WHERE id=?', (preset['id'], row['id']))
def save_authenticated_connection(base_url, server_type, username, password, auth, ai_settings=None, *, owner_id=None, existing_id=None, label=""):
"""Only called after authentication. Existing connections survive active-user changes.
``ai_settings`` seeds a *new* connection only. An update never rewrites the stored
AI settings: saving media credentials used to rebuild that JSON from whatever the
connection form happened to hold, which erased a provider key the user configured
from the AI Hub. Provider changes go through /api/settings/ai instead.
"""
user_id = auth['User']['Id']
server_id = auth.get('ServerId') or ''
base_url = base_url.rstrip('/')
# Server identity is preferable to a URL alias; type is part of the namespace.
identity = f'{server_type}:{server_id or base_url}:{user_id}'
if owner_id:
identity = f'{owner_id}:{identity}'
connection_id = uuid.uuid5(uuid.NAMESPACE_URL, identity).hex
with database.get_db_connection() as conn:
if owner_id is None and conn.execute('SELECT 1 FROM accounts LIMIT 1').fetchone():
raise ValueError('Local accounts are enabled; configure connections while signed in.')
if existing_id:
existing = conn.execute('SELECT * FROM media_connections WHERE id=? AND owner_id=?', (existing_id, owner_id)).fetchone()
if not existing:
raise ValueError('Connection not found.')
if existing['user_id'] != user_id or existing['server_type'] != server_type or (existing['server_id'] and existing['server_id'] != server_id):
raise ValueError('This is a different server or media account. Add a new connection instead.')
connection_id = existing_id
elif owner_id:
existing = conn.execute("SELECT id FROM media_connections WHERE owner_id=? AND server_type=? AND user_id=? AND ((server_id=? AND server_id!='') OR (server_id='' AND base_url=?))",
(owner_id, server_type, user_id, server_id, base_url)).fetchone()
if existing:
connection_id = existing['id']
conn.execute('''INSERT INTO media_connections
(id, base_url, server_type, username, password, user_id, server_id, ai_settings)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET base_url=excluded.base_url,
username=excluded.username, password=excluded.password''',
(connection_id, base_url, server_type, username, password, user_id, server_id,
json.dumps(dict(ai_settings or {}))))
if owner_id:
conn.execute('UPDATE media_connections SET owner_id=?, label=? WHERE id=?', (owner_id, label, connection_id))
# Claim legacy presets once, never each time .env changes. Original tables
# stay intact for rollback. Jobs with a different user remain unassigned.
migrated = conn.execute("SELECT value FROM settings WHERE key='connection_migration_completed'").fetchone()
migrate_legacy = owner_id is None
if owner_id and not migrated:
owner = conn.execute('SELECT is_admin FROM accounts WHERE id=?', (owner_id,)).fetchone()
other_connections = conn.execute(
'SELECT 1 FROM media_connections WHERE owner_id=? AND id<>? LIMIT 1',
(owner_id, connection_id)
).fetchone()
migrate_legacy = bool(owner and owner['is_admin'] and not other_connections)
if not migrated and migrate_legacy:
for row in conn.execute('SELECT name, data FROM presets').fetchall():
conn.execute('INSERT OR IGNORE INTO connection_presets (id, connection_id, name, data) VALUES (?, ?, ?, ?)',
(uuid.uuid4().hex, connection_id, row['name'], row['data']))
conn.execute('UPDATE schedules SET connection_id=? WHERE connection_id IS NULL AND user_id=?',
(connection_id, user_id))
backfill_schedule_preset_ids(conn, connection_id)
conn.execute("INSERT INTO settings (key,value) VALUES ('connection_migration_completed', ?)", (connection_id,))
if owner_id is None:
conn.execute("INSERT OR REPLACE INTO settings (key,value) VALUES ('active_connection_id', ?)", (connection_id,))
conn.commit()
# Read back rather than trusting the argument: on an update the stored AI settings
# were deliberately left alone, so they are the only correct value here.
with database.get_db_connection() as conn:
stored = conn.execute('SELECT ai_settings FROM media_connections WHERE id=?', (connection_id,)).fetchone()
try:
saved_ai = json.loads(stored['ai_settings'] or '{}') if stored else {}
except (TypeError, ValueError):
saved_ai = {}
connection = Connection(connection_id, base_url, server_type, username, password, user_id, server_id, saved_ai)
media = MediaClient(connection, token=auth['AccessToken'], user_profile=auth.get('User'))
with _clients_lock:
_clients[connection_id] = media
return media
def active_connection_id():
with database.get_db_connection() as conn:
row = conn.execute("SELECT value FROM settings WHERE key='active_connection_id'").fetchone()
return row['value'] if row else None
def get_media_client(connection_id):
if not connection_id:
raise ValueError('No saved media connection. Configure the server in Settings.')
with database.get_db_connection() as conn:
row = conn.execute('SELECT * FROM media_connections WHERE id=?', (connection_id,)).fetchone()
if row is None:
raise ValueError('Saved media connection no longer exists.')
fields = ('id', 'base_url', 'server_type', 'username', 'password', 'user_id', 'server_id', 'ai_settings')
data = {key: row[key] for key in fields}
data['ai_settings'] = json.loads(data['ai_settings'])
connection = Connection(**data)
with _clients_lock:
media = _clients.get(connection_id)
if media is None or media.connection != connection:
media = MediaClient(connection)
_clients[connection_id] = media
return media
def active_media_client():
return get_media_client(active_connection_id())
def all_media_clients():
with database.get_db_connection() as conn:
ids = [row['id'] for row in conn.execute('SELECT id FROM media_connections')]
return [get_media_client(cid) for cid in ids]
def forget_media_client(connection_id):
"""Drop a removed connection from the process-local client registry."""
with _clients_lock:
_clients.pop(connection_id, None)
def delete_connection_data(conn, connection_id):
"""Delete local SQL data inside the caller's transaction; never contact a server."""
schedule_ids = [row['id'] for row in conn.execute(
'SELECT id FROM schedules WHERE connection_id=?', (connection_id,))]
conn.execute('DELETE FROM schedules WHERE connection_id=?', (connection_id,))
conn.execute('DELETE FROM connection_presets WHERE connection_id=?', (connection_id,))
conn.execute('DELETE FROM connection_recipes WHERE connection_id=?', (connection_id,))
# build_run_items cascade from build_runs.
conn.execute('DELETE FROM build_runs WHERE connection_id=?', (connection_id,))
conn.execute('UPDATE account_sessions SET connection_id=NULL WHERE connection_id=?', (connection_id,))
conn.execute("DELETE FROM settings WHERE key='active_connection_id' AND value=?", (connection_id,))
conn.execute('DELETE FROM media_connections WHERE id=?', (connection_id,))
return schedule_ids
def cleanup_deleted_connection(connection_id, schedule_ids):
"""Retire runtime jobs, credentials, caches and the AI index after SQL commits."""
import scheduler
from app import cache
from app.ai.enrichment_manager import stop_enrichment
from app.ai.vector_store import delete_connection_collection
stop_enrichment(connection_id)
for schedule_id in schedule_ids:
scheduler.scheduler_manager.remove_schedule(schedule_id)
scheduler._cancel_orphaned_jobs(schedule_id)
forget_media_client(connection_id)
cache.forget_connection(connection_id)
delete_connection_collection(connection_id)
def update_active_ai_setting(key, value):
media = active_media_client()
settings = dict(media.connection.ai_settings)
settings[key] = value
with database.get_db_connection() as conn:
conn.execute('UPDATE media_connections SET ai_settings=? WHERE id=?',
(json.dumps(settings), media.connection.id))
conn.commit()
def reload_connections():
"""Clear cached connection media clients to pick up reloaded database state."""
with _clients_lock:
_clients.clear()