File size: 5,822 Bytes
f9474bf dc0fe62 ad9fe10 dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 f9474bf dc0fe62 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | import os
import json
import time
from typing import Optional, List
from backend.database import SessionLocal, Integration, Tool
from backend.composio_client import get_composio
DATA_DIR = os.environ.get("DATA_DIR", "/data")
CACHE_FILE = f"{DATA_DIR}/composio_cache.json"
CACHE_TTL = int(os.environ.get("CACHE_TTL", "86400"))
def _to_dict(obj):
if hasattr(obj, "model_dump"):
return obj.model_dump(mode="json")
if hasattr(obj, "__dict__"):
return {k: v for k, v in obj.__dict__.items() if not k.startswith("_")}
return {"slug": str(obj), "name": str(obj)}
class LocalDB:
def __init__(self, cache_file: str = CACHE_FILE):
self.cache_file = cache_file
os.makedirs(os.path.dirname(cache_file), exist_ok=True)
def save_to_json(self, toolkits: List[dict], tools: List[dict]):
data = {"toolkits": toolkits, "tools": tools, "cached_at": time.time(),
"toolkit_count": len(toolkits), "tool_count": len(tools)}
with open(self.cache_file, "w") as f:
json.dump(data, f, indent=2)
def load_from_json(self) -> Optional[dict]:
if os.path.exists(self.cache_file):
with open(self.cache_file) as f:
return json.load(f)
return None
def is_json_stale(self) -> bool:
data = self.load_from_json()
if not data:
return True
return (time.time() - data.get("cached_at", 0)) > CACHE_TTL
def seed_sqlite(self, toolkits: List[dict], tools: List[dict]):
db = SessionLocal()
try:
db.query(Tool).delete()
db.query(Integration).delete()
db.commit()
for tk in toolkits:
meta = tk.get("meta", {}) or {}
cats = (meta.get("categories") or []) if isinstance(meta, dict) else []
category = cats[0].get("name", "Other") if cats else "Other"
auth_type = "oauth2"
for s in (tk.get("auth_schemes") or []):
if "oauth" in str(s).lower():
auth_type = "oauth2"
break
auth_type = "api_key"
db.add(Integration(
id=tk.get("slug", tk.get("id", "")),
name=tk.get("name", tk.get("slug", "")),
description=tk.get("description", ""),
logo_url="",
category=category,
auth_type=auth_type,
tool_count=int(meta.get("tools_count", 0)) if isinstance(meta, dict) else 0,
trigger_count=0,
is_active=True
))
for t in tools:
toolkit = t.get("toolkit", {}) or {}
ts = toolkit.get("slug", "") if isinstance(toolkit, dict) else ""
tags = t.get("tags", ["General"]) or ["General"]
db.add(Tool(
id=t.get("slug", t.get("id", "")),
integration_id=ts,
name=t.get("name", ""),
description=t.get("description", ""),
input_schema=t.get("input_parameters", {"type": "object", "properties": {}}),
output_schema=t.get("output_parameters"),
category=tags[0] if tags else "General",
is_active=not t.get("is_deprecated", False)
))
db.commit()
self.save_to_json(toolkits, tools)
except Exception:
db.rollback()
raise
finally:
db.close()
async def sync_from_composio(api_key: str = None, force: bool = False) -> dict:
composio = get_composio()
local_db = LocalDB()
if composio is None:
data = local_db.load_from_json()
if data:
local_db.seed_sqlite(data.get("toolkits", []), data.get("tools", []))
return {"source": "local_cache", "toolkits": data.get("toolkit_count", 0), "tools": data.get("tool_count", 0)}
return {"source": "none", "toolkits": 0, "tools": 0}
if not force and not local_db.is_json_stale():
data = local_db.load_from_json()
if data and data.get("toolkits"):
local_db.seed_sqlite(data["toolkits"], data.get("tools", []))
return {"source": "local_cache", "toolkits": data["toolkit_count"], "tools": data["tool_count"]}
try:
raw_tk = composio.toolkits.get()
new_toolkits = [_to_dict(t) for t in raw_tk if hasattr(t, "slug") or (isinstance(t, dict) and t.get("slug"))]
new_tools = []
for tk in new_toolkits:
slug = tk.get("slug", "")
if not slug:
continue
try:
raw = composio.tools.get_raw_composio_tools(toolkits=[slug], limit=500)
new_tools.extend(_to_dict(r) for r in raw)
except Exception:
pass
local_db.seed_sqlite(new_toolkits, new_tools)
return {"source": "api", "toolkits": len(new_toolkits), "tools": len(new_tools)}
except Exception as e:
print(f"[Sync] SDK failed: {e}")
data = local_db.load_from_json()
if data:
local_db.seed_sqlite(data.get("toolkits", []), data.get("tools", []))
return {"source": "local_cache_fallback", "toolkits": data.get("toolkit_count", 0), "tools": data.get("tool_count", 0)}
return {"source": "fallback", "toolkits": 0, "tools": 0}
def seed_integrations():
from backend.database import init_db
init_db()
import asyncio
try:
result = asyncio.run(sync_from_composio())
print(f"[Seed] Result: {result}")
except Exception as e:
print(f"[Seed] Error: {e}")
if __name__ == "__main__":
seed_integrations()
|