my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	from datetime import UTC, datetime
#2	import io
#3	import json
#4	import gzip
#5	import zipfile
#6	from typing import Optional, List, Dict, Any
#7	from uuid import UUID
#8
#9	from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query, Form
#10	from fastapi.responses import StreamingResponse
#11	from pydantic import BaseModel
#12	from sqlalchemy.orm import Session, joinedload
#13	from sqlalchemy import and_
#14
#15	from app.database import get_db
#16	from app.models import (
#17	User, App, Memory, MemoryState, Category, memory_categories,
#18	MemoryStatusHistory, AccessControl
#19	)
#20	from app.utils.memory import get_memory_client
#21
#22	from uuid import uuid4
#23
#24	router = APIRouter(prefix="/api/v1/backup", tags=["backup"])
#25
#26	class ExportRequest(BaseModel):
#27	user_id: str
#28	app_id: Optional[UUID] = None
#29	from_date: Optional[int] = None
#30	to_date: Optional[int] = None
#31	include_vectors: bool = True
#32
#33	def _iso(dt: Optional[datetime]) -> Optional[str]:
#34	if isinstance(dt, datetime):
#35	try:
#36	return dt.astimezone(UTC).isoformat()
#37	except:
#38	return dt.replace(tzinfo=UTC).isoformat()
#39	return None
#40
#41	def _parse_iso(dt: Optional[str]) -> Optional[datetime]:
#42	if not dt:
#43	return None
#44	try:
#45	return datetime.fromisoformat(dt)
#46	except Exception:
#47	try:
#48	return datetime.fromisoformat(dt.replace("Z", "+00:00"))
#49	except Exception:
#50	return None
#51
#52	def _export_sqlite(db: Session, req: ExportRequest) -> Dict[str, Any]:
#53	user = db.query(User).filter(User.user_id == req.user_id).first()
#54	if not user:
#55	raise HTTPException(status_code=404, detail="User not found")
#56
#57	time_filters = []
#58	if req.from_date:
#59	time_filters.append(Memory.created_at >= datetime.fromtimestamp(req.from_date, tz=UTC))
#60	if req.to_date:
#61	time_filters.append(Memory.created_at <= datetime.fromtimestamp(req.to_date, tz=UTC))
#62
#63	mem_q = (
#64	db.query(Memory)
#65	.options(joinedload(Memory.categories), joinedload(Memory.app))
#66	.filter(
#67	Memory.user_id == user.id,
#68	*(time_filters or []),
#69	* ( [Memory.app_id == req.app_id] if req.app_id else [] ),
#70	)
#71	)
#72
#73	memories = mem_q.all()
#74	memory_ids = [m.id for m in memories]
#75
#76	app_ids = sorted({m.app_id for m in memories if m.app_id})
#77	apps = db.query(App).filter(App.id.in_(app_ids)).all() if app_ids else []
#78
#79	cats = sorted({c for m in memories for c in m.categories}, key = lambda c: str(c.id))
#80
#81	mc_rows = db.execute(
#82	memory_categories.select().where(memory_categories.c.memory_id.in_(memory_ids))
#83	).fetchall() if memory_ids else []
#84
#85	history = db.query(MemoryStatusHistory).filter(MemoryStatusHistory.memory_id.in_(memory_ids)).all() if memory_ids else []
#86
#87	acls = db.query(AccessControl).filter(
#88	AccessControl.subject_type == "app",
#89	AccessControl.subject_id.in_(app_ids) if app_ids else False
#90	).all() if app_ids else []
#91
#92	return {
#93	"user": {
#94	"id": str(user.id),
#95	"user_id": user.user_id,
#96	"name": user.name,
#97	"email": user.email,
#98	"metadata": user.metadata_,
#99	"created_at": _iso(user.created_at),
#100	"updated_at": _iso(user.updated_at)
#101	},
#102	"apps": [
#103	{
#104	"id": str(a.id),
#105	"owner_id": str(a.owner_id),
#106	"name": a.name,
#107	"description": a.description,
#108	"metadata": a.metadata_,
#109	"is_active": a.is_active,
#110	"created_at": _iso(a.created_at),
#111	"updated_at": _iso(a.updated_at),
#112	}
#113	for a in apps
#114	],
#115	"categories": [
#116	{
#117	"id": str(c.id),
#118	"name": c.name,
#119	"description": c.description,
#120	"created_at": _iso(c.created_at),
#121	"updated_at": _iso(c.updated_at),
#122	}
#123	for c in cats
#124	],
#125	"memories": [
#126	{
#127	"id": str(m.id),
#128	"user_id": str(m.user_id),
#129	"app_id": str(m.app_id) if m.app_id else None,
#130	"content": m.content,
#131	"metadata": m.metadata_,
#132	"state": m.state.value,
#133	"created_at": _iso(m.created_at),
#134	"updated_at": _iso(m.updated_at),
#135	"archived_at": _iso(m.archived_at),
#136	"deleted_at": _iso(m.deleted_at),
#137	"category_ids": [str(c.id) for c in m.categories], #TODO: figure out a way to add category names simply to this
#138	}
#139	for m in memories
#140	],
#141	"memory_categories": [
#142	{"memory_id": str(r.memory_id), "category_id": str(r.category_id)}
#143	for r in mc_rows
#144	],
#145	"status_history": [
#146	{
#147	"id": str(h.id),
#148	"memory_id": str(h.memory_id),
#149	"changed_by": str(h.changed_by),
#150	"old_state": h.old_state.value,
#151	"new_state": h.new_state.value,
#152	"changed_at": _iso(h.changed_at),
#153	}
#154	for h in history
#155	],
#156	"access_controls": [
#157	{
#158	"id": str(ac.id),
#159	"subject_type": ac.subject_type,
#160	"subject_id": str(ac.subject_id) if ac.subject_id else None,
#161	"object_type": ac.object_type,
#162	"object_id": str(ac.object_id) if ac.object_id else None,
#163	"effect": ac.effect,
#164	"created_at": _iso(ac.created_at),
#165	}
#166	for ac in acls
#167	],
#168	"export_meta": {
#169	"app_id_filter": str(req.app_id) if req.app_id else None,
#170	"from_date": req.from_date,
#171	"to_date": req.to_date,
#172	"version": "1",
#173	"generated_at": datetime.now(UTC).isoformat(),
#174	},
#175	}
#176
#177	def _export_logical_memories_gz(
#178	db: Session,
#179	*,
#180	user_id: str,
#181	app_id: Optional[UUID] = None,
#182	from_date: Optional[int] = None,
#183	to_date: Optional[int] = None
#184	) -> bytes:
#185	"""
#186	Export a provider-agnostic backup of memories so they can be restored to any vector DB
#187	by re-embedding content. One JSON object per line, gzip-compressed.
#188
#189	Schema (per line):
#190	{
#191	"id": "<uuid>",
#192	"content": "<text>",
#193	"metadata": {...},
#194	"created_at": "<iso8601 or null>",
#195	"updated_at": "<iso8601 or null>",
#196	"state": "active\|paused\|archived\|deleted",
#197	"app": "<app name or null>",
#198	"categories": ["catA", "catB", ...]
#199	}
#200	"""
#201
#202	user = db.query(User).filter(User.user_id == user_id).first()
#203	if not user:
#204	raise HTTPException(status_code=404, detail="User not found")
#205
#206	time_filters = []
#207	if from_date:
#208	time_filters.append(Memory.created_at >= datetime.fromtimestamp(from_date, tz=UTC))
#209	if to_date:
#210	time_filters.append(Memory.created_at <= datetime.fromtimestamp(to_date, tz=UTC))
#211
#212	q = (
#213	db.query(Memory)
#214	.options(joinedload(Memory.categories), joinedload(Memory.app))
#215	.filter(
#216	Memory.user_id == user.id,
#217	*(time_filters or []),
#218	)
#219	)
#220	if app_id:
#221	q = q.filter(Memory.app_id == app_id)
#222
#223	buf = io.BytesIO()
#224	with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
#225	for m in q.all():
#226	record = {
#227	"id": str(m.id),
#228	"content": m.content,
#229	"metadata": m.metadata_ or {},
#230	"created_at": _iso(m.created_at),
#231	"updated_at": _iso(m.updated_at),
#232	"state": m.state.value,
#233	"app": m.app.name if m.app else None,
#234	"categories": [c.name for c in m.categories],
#235	}
#236	gz.write((json.dumps(record) + "\n").encode("utf-8"))
#237	return buf.getvalue()
#238
#239	@router.post("/export")
#240	async def export_backup(req: ExportRequest, db: Session = Depends(get_db)):
#241	sqlite_payload = _export_sqlite(db=db, req=req)
#242	memories_blob = _export_logical_memories_gz(
#243	db=db,
#244	user_id=req.user_id,
#245	app_id=req.app_id,
#246	from_date=req.from_date,
#247	to_date=req.to_date,
#248
#249	)
#250
#251	#TODO: add vector store specific exports in future for speed
#252
#253	zip_buf = io.BytesIO()
#254	with zipfile.ZipFile(zip_buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
#255	zf.writestr("memories.json", json.dumps(sqlite_payload, indent=2))
#256	zf.writestr("memories.jsonl.gz", memories_blob)
#257
#258	zip_buf.seek(0)
#259	return StreamingResponse(
#260	zip_buf,
#261	media_type="application/zip",
#262	headers={"Content-Disposition": f'attachment; filename="memories_export_{req.user_id}.zip"'},
#263	)
#264
#265	@router.post("/import")
#266	async def import_backup(
#267	file: UploadFile = File(..., description="Zip with memories.json and memories.jsonl.gz"),
#268	user_id: str = Form(..., description="Import memories into this user_id"),
#269	mode: str = Query("overwrite"),
#270	db: Session = Depends(get_db)
#271	):
#272	if not file.filename.endswith(".zip"):
#273	raise HTTPException(status_code=400, detail="Expected a zip file.")
#274
#275	if mode not in {"skip", "overwrite"}:
#276	raise HTTPException(status_code=400, detail="Invalid mode. Must be 'skip' or 'overwrite'.")
#277
#278	user = db.query(User).filter(User.user_id == user_id).first()
#279	if not user:
#280	raise HTTPException(status_code=404, detail="User not found")
#281
#282	content = await file.read()
#283	try:
#284	with zipfile.ZipFile(io.BytesIO(content), "r") as zf:
#285	names = zf.namelist()
#286
#287	def find_member(filename: str) -> Optional[str]:
#288	for name in names:
#289	# Skip directory entries
#290	if name.endswith('/'):
#291	continue
#292	if name.rsplit('/', 1)[-1] == filename:
#293	return name
#294	return None
#295
#296	sqlite_member = find_member("memories.json")
#297	if not sqlite_member:
#298	raise HTTPException(status_code=400, detail="memories.json missing in zip")
#299
#300	memories_member = find_member("memories.jsonl.gz")
#301
#302	sqlite_data = json.loads(zf.read(sqlite_member))
#303	memories_blob = zf.read(memories_member) if memories_member else None
#304	except Exception:
#305	raise HTTPException(status_code=400, detail="Invalid zip file")
#306
#307	default_app = db.query(App).filter(App.owner_id == user.id, App.name == "openmemory").first()
#308	if not default_app:
#309	default_app = App(owner_id=user.id, name="openmemory", is_active=True, metadata_={})
#310	db.add(default_app)
#311	db.commit()
#312	db.refresh(default_app)
#313
#314	cat_id_map: Dict[str, UUID] = {}
#315	for c in sqlite_data.get("categories", []):
#316	cat = db.query(Category).filter(Category.name == c["name"]).first()
#317	if not cat:
#318	cat = Category(name=c["name"], description=c.get("description"))
#319	db.add(cat)
#320	db.commit()
#321	db.refresh(cat)
#322	cat_id_map[c["id"]] = cat.id
#323
#324	old_to_new_id: Dict[str, UUID] = {}
#325	for m in sqlite_data.get("memories", []):
#326	incoming_id = UUID(m["id"])
#327	existing = db.query(Memory).filter(Memory.id == incoming_id).first()
#328
#329	# Cross-user collision: always mint a new UUID and import as a new memory
#330	if existing and existing.user_id != user.id:
#331	target_id = uuid4()
#332	else:
#333	target_id = incoming_id
#334
#335	old_to_new_id[m["id"]] = target_id
#336
#337	# Same-user collision + skip mode: leave existing row untouched
#338	if existing and (existing.user_id == user.id) and mode == "skip":
#339	continue
#340
#341	# Same-user collision + overwrite mode: treat import as ground truth
#342	if existing and (existing.user_id == user.id) and mode == "overwrite":
#343	incoming_state = m.get("state", "active")
#344	existing.user_id = user.id
#345	existing.app_id = default_app.id
#346	existing.content = m.get("content") or ""
#347	existing.metadata_ = m.get("metadata") or {}
#348	try:
#349	existing.state = MemoryState(incoming_state)
#350	except Exception:
#351	existing.state = MemoryState.active
#352	# Update state-related timestamps from import (ground truth)
#353	existing.archived_at = _parse_iso(m.get("archived_at"))
#354	existing.deleted_at = _parse_iso(m.get("deleted_at"))
#355	existing.created_at = _parse_iso(m.get("created_at")) or existing.created_at
#356	existing.updated_at = _parse_iso(m.get("updated_at")) or existing.updated_at
#357	db.add(existing)
#358	db.commit()
#359	continue
#360
#361	new_mem = Memory(
#362	id=target_id,
#363	user_id=user.id,
#364	app_id=default_app.id,
#365	content=m.get("content") or "",
#366	metadata_=m.get("metadata") or {},
#367	state=MemoryState(m.get("state", "active")) if m.get("state") else MemoryState.active,
#368	created_at=_parse_iso(m.get("created_at")) or datetime.now(UTC),
#369	updated_at=_parse_iso(m.get("updated_at")) or datetime.now(UTC),
#370	archived_at=_parse_iso(m.get("archived_at")),
#371	deleted_at=_parse_iso(m.get("deleted_at")),
#372	)
#373	db.add(new_mem)
#374	db.commit()
#375
#376	for link in sqlite_data.get("memory_categories", []):
#377	mid = old_to_new_id.get(link["memory_id"])
#378	cid = cat_id_map.get(link["category_id"])
#379	if not (mid and cid):
#380	continue
#381	exists = db.execute(
#382	memory_categories.select().where(
#383	(memory_categories.c.memory_id == mid) & (memory_categories.c.category_id == cid)
#384	)
#385	).first()
#386
#387	if not exists:
#388	db.execute(memory_categories.insert().values(memory_id=mid, category_id=cid))
#389	db.commit()
#390
#391	for h in sqlite_data.get("status_history", []):
#392	hid = UUID(h["id"])
#393	mem_id = old_to_new_id.get(h["memory_id"], UUID(h["memory_id"]))
#394	exists = db.query(MemoryStatusHistory).filter(MemoryStatusHistory.id == hid).first()
#395	if exists and mode == "skip":
#396	continue
#397	rec = exists if exists else MemoryStatusHistory(id=hid)
#398	rec.memory_id = mem_id
#399	rec.changed_by = user.id
#400	try:
#401	rec.old_state = MemoryState(h.get("old_state", "active"))
#402	rec.new_state = MemoryState(h.get("new_state", "active"))
#403	except Exception:
#404	rec.old_state = MemoryState.active
#405	rec.new_state = MemoryState.active
#406	rec.changed_at = _parse_iso(h.get("changed_at")) or datetime.now(UTC)
#407	db.add(rec)
#408	db.commit()
#409
#410	memory_client = get_memory_client()
#411	vector_store = getattr(memory_client, "vector_store", None) if memory_client else None
#412
#413	if vector_store and memory_client and hasattr(memory_client, "embedding_model"):
#414	def iter_logical_records():
#415	if memories_blob:
#416	gz_buf = io.BytesIO(memories_blob)
#417	with gzip.GzipFile(fileobj=gz_buf, mode="rb") as gz:
#418	for raw in gz:
#419	yield json.loads(raw.decode("utf-8"))
#420	else:
#421	for m in sqlite_data.get("memories", []):
#422	yield {
#423	"id": m["id"],
#424	"content": m.get("content"),
#425	"metadata": m.get("metadata") or {},
#426	"created_at": m.get("created_at"),
#427	"updated_at": m.get("updated_at"),
#428	}
#429
#430	for rec in iter_logical_records():
#431	old_id = rec["id"]
#432	new_id = old_to_new_id.get(old_id, UUID(old_id))
#433	content = rec.get("content") or ""
#434	metadata = rec.get("metadata") or {}
#435	created_at = rec.get("created_at")
#436	updated_at = rec.get("updated_at")
#437
#438	if mode == "skip":
#439	try:
#440	get_fn = getattr(vector_store, "get", None)
#441	if callable(get_fn) and vector_store.get(str(new_id)):
#442	continue
#443	except Exception:
#444	pass
#445
#446	payload = dict(metadata)
#447	payload["data"] = content
#448	if created_at:
#449	payload["created_at"] = created_at
#450	if updated_at:
#451	payload["updated_at"] = updated_at
#452	payload["user_id"] = user_id
#453	payload.setdefault("source_app", "openmemory")
#454
#455	try:
#456	vec = memory_client.embedding_model.embed(content, "add")
#457	vector_store.insert(vectors=[vec], payloads=[payload], ids=[str(new_id)])
#458	except Exception as e:
#459	print(f"Vector upsert failed for memory {new_id}: {e}")
#460	continue
#461
#462	return {"message": f'Import completed into user "{user_id}"'}
#463
#464	return {"message": f'Import completed into user "{user_id}"'}
#465
#466
#467
#468
#469
#470
#471
#472
#473
#474
#475
#476
#477
#478
#479
#480
#481
#482
#483
#484
#485
#486
#487
#488
#489
#490
#491
#492
#493
#494
#495
#496
#497
#498
#499
#500

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public