Files
agentic-codebase-memory-man…/snip/export/handlers.py

123 lines
3.9 KiB
Python

"""JSON import/export handlers for snippets."""
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from snip.db.database import Database, get_database
class ExportHandler:
def __init__(self, db_path: str | None = None):
self.db = get_database(db_path)
def export_all(self) -> dict[str, Any]:
"""Export all snippets."""
snippets = self.db.list_snippets(limit=10000)
for s in snippets:
if isinstance(s.get("tags"), str):
s["tags"] = json.loads(s["tags"])
return {
"version": "1.0",
"exported_at": datetime.utcnow().isoformat() + "Z",
"snippets": snippets,
}
def export_collection(self, collection_id: int) -> dict[str, Any]:
"""Export a specific collection."""
snippets = self.db.get_collection_snippets(collection_id)
for s in snippets:
if isinstance(s.get("tags"), str):
s["tags"] = json.loads(s["tags"])
return {
"version": "1.0",
"exported_at": datetime.utcnow().isoformat() + "Z",
"snippets": snippets,
}
def export_snippet(self, snippet_id: int) -> dict[str, Any] | None:
"""Export a single snippet."""
snippet = self.db.get_snippet(snippet_id)
if not snippet:
return None
if isinstance(snippet.get("tags"), str):
snippet["tags"] = json.loads(snippet["tags"])
return {
"version": "1.0",
"exported_at": datetime.utcnow().isoformat() + "Z",
"snippets": [snippet],
}
def write_export(self, file_path: str, data: dict[str, Any]):
"""Write export data to a file."""
Path(file_path).parent.mkdir(parents=True, exist_ok=True)
with open(file_path, "w") as f:
json.dump(data, f, indent=2)
def import_data(self, file_path: str, strategy: str = "skip") -> dict[str, Any]:
"""Import snippets from a JSON file."""
with open(file_path, "r") as f:
data = json.load(f)
snippets = data.get("snippets", [])
imported = 0
skipped = 0
replaced = 0
for snippet_data in snippets:
result = self.db.import_snippet(
snippet_data, strategy=strategy)
if result is None:
skipped += 1
elif result == -1:
replaced += 1
else:
imported += 1
return {
"imported": imported,
"skipped": skipped,
"replaced": replaced,
"total": len(snippets),
}
def generate_import_summary(self, results: dict[str, Any]) -> str:
"""Generate a human-readable import summary."""
lines = [
f"Import complete!",
f" Imported: {results.get('imported', 0)}",
f" Skipped: {results.get('skipped', 0)}",
f" Replaced: {results.get('replaced', 0)}",
]
return "\n".join(lines)
def export_snippets(snippets: list[dict[str, Any]], file_path: str):
"""Export snippets to a JSON file (standalone function)."""
export_data = {
"version": "1.0",
"exported_at": datetime.utcnow().isoformat() + "Z",
"snippets": snippets,
}
with open(file_path, "w") as f:
json.dump(export_data, f, indent=2)
def import_snippets(db: Database, file_path: str, strategy: str = "skip") -> tuple[int, int]:
"""Import snippets from a JSON file (standalone function)."""
with open(file_path, "r") as f:
data = json.load(f)
snippets = data.get("snippets", [])
imported = 0
skipped = 0
for snippet_data in snippets:
result = db.import_snippet(snippet_data, strategy=strategy)
if result is None:
skipped += 1
else:
imported += 1
return imported, skipped