my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	import logging
#2	from datetime import UTC, datetime
#3	from typing import List, Optional, Set
#4	from uuid import UUID
#5
#6	from app.database import get_db
#7	from app.models import (
#8	AccessControl,
#9	App,
#10	Category,
#11	Memory,
#12	MemoryAccessLog,
#13	MemoryState,
#14	MemoryStatusHistory,
#15	User,
#16	)
#17	from app.schemas import MemoryResponse
#18	from app.utils.memory import get_memory_client
#19	from app.utils.permissions import check_memory_access_permissions
#20	from fastapi import APIRouter, Depends, HTTPException, Query
#21	from fastapi_pagination import Page, Params
#22	from fastapi_pagination.ext.sqlalchemy import paginate as sqlalchemy_paginate
#23	from pydantic import BaseModel
#24	from sqlalchemy import func
#25	from sqlalchemy.orm import Session, joinedload
#26
#27	router = APIRouter(prefix="/api/v1/memories", tags=["memories"])
#28
#29
#30	def get_memory_or_404(db: Session, memory_id: UUID) -> Memory:
#31	memory = db.query(Memory).filter(Memory.id == memory_id).first()
#32	if not memory:
#33	raise HTTPException(status_code=404, detail="Memory not found")
#34	return memory
#35
#36
#37	def update_memory_state(db: Session, memory_id: UUID, new_state: MemoryState, user_id: UUID):
#38	memory = get_memory_or_404(db, memory_id)
#39	old_state = memory.state
#40
#41	# Update memory state
#42	memory.state = new_state
#43	if new_state == MemoryState.archived:
#44	memory.archived_at = datetime.now(UTC)
#45	elif new_state == MemoryState.deleted:
#46	memory.deleted_at = datetime.now(UTC)
#47
#48	# Record state change
#49	history = MemoryStatusHistory(
#50	memory_id=memory_id,
#51	changed_by=user_id,
#52	old_state=old_state,
#53	new_state=new_state
#54	)
#55	db.add(history)
#56	db.commit()
#57	return memory
#58
#59
#60	def get_accessible_memory_ids(db: Session, app_id: UUID) -> Set[UUID]:
#61	"""
#62	Get the set of memory IDs that the app has access to based on app-level ACL rules.
#63	Returns all memory IDs if no specific restrictions are found.
#64	"""
#65	# Get app-level access controls
#66	app_access = db.query(AccessControl).filter(
#67	AccessControl.subject_type == "app",
#68	AccessControl.subject_id == app_id,
#69	AccessControl.object_type == "memory"
#70	).all()
#71
#72	# If no app-level rules exist, return None to indicate all memories are accessible
#73	if not app_access:
#74	return None
#75
#76	# Initialize sets for allowed and denied memory IDs
#77	allowed_memory_ids = set()
#78	denied_memory_ids = set()
#79
#80	# Process app-level rules
#81	for rule in app_access:
#82	if rule.effect == "allow":
#83	if rule.object_id: # Specific memory access
#84	allowed_memory_ids.add(rule.object_id)
#85	else: # All memories access
#86	return None # All memories allowed
#87	elif rule.effect == "deny":
#88	if rule.object_id: # Specific memory denied
#89	denied_memory_ids.add(rule.object_id)
#90	else: # All memories denied
#91	return set() # No memories accessible
#92
#93	# Remove denied memories from allowed set
#94	if allowed_memory_ids:
#95	allowed_memory_ids -= denied_memory_ids
#96
#97	return allowed_memory_ids
#98
#99
#100	# List all memories with filtering
#101	@router.get("/", response_model=Page[MemoryResponse])
#102	async def list_memories(
#103	user_id: str,
#104	app_id: Optional[UUID] = None,
#105	from_date: Optional[int] = Query(
#106	None,
#107	description="Filter memories created after this date (timestamp)",
#108	examples=[1718505600]
#109	),
#110	to_date: Optional[int] = Query(
#111	None,
#112	description="Filter memories created before this date (timestamp)",
#113	examples=[1718505600]
#114	),
#115	categories: Optional[str] = None,
#116	params: Params = Depends(),
#117	search_query: Optional[str] = None,
#118	sort_column: Optional[str] = Query(None, description="Column to sort by (memory, categories, app_name, created_at)"),
#119	sort_direction: Optional[str] = Query(None, description="Sort direction (asc or desc)"),
#120	db: Session = Depends(get_db)
#121	):
#122	user = db.query(User).filter(User.user_id == user_id).first()
#123	if not user:
#124	raise HTTPException(status_code=404, detail="User not found")
#125
#126	# Build base query
#127	query = db.query(Memory).filter(
#128	Memory.user_id == user.id,
#129	Memory.state != MemoryState.deleted,
#130	Memory.state != MemoryState.archived,
#131	Memory.content.ilike(f"%{search_query}%") if search_query else True
#132	)
#133
#134	# Apply filters
#135	if app_id:
#136	query = query.filter(Memory.app_id == app_id)
#137
#138	if from_date:
#139	from_datetime = datetime.fromtimestamp(from_date, tz=UTC)
#140	query = query.filter(Memory.created_at >= from_datetime)
#141
#142	if to_date:
#143	to_datetime = datetime.fromtimestamp(to_date, tz=UTC)
#144	query = query.filter(Memory.created_at <= to_datetime)
#145
#146	# Add joins for app and categories after filtering
#147	query = query.outerjoin(App, Memory.app_id == App.id)
#148	query = query.outerjoin(Memory.categories)
#149
#150	# Apply category filter if provided
#151	if categories:
#152	category_list = [c.strip() for c in categories.split(",")]
#153	query = query.filter(Category.name.in_(category_list))
#154
#155	# Apply sorting if specified
#156	if sort_column:
#157	sort_field = getattr(Memory, sort_column, None)
#158	if sort_field:
#159	query = query.order_by(sort_field.desc()) if sort_direction == "desc" else query.order_by(sort_field.asc())
#160
#161	# Add eager loading for app and categories
#162	query = query.options(
#163	joinedload(Memory.app),
#164	joinedload(Memory.categories)
#165	).distinct(Memory.id)
#166
#167	# Get paginated results with transformer
#168	return sqlalchemy_paginate(
#169	query,
#170	params,
#171	transformer=lambda items: [
#172	MemoryResponse(
#173	id=memory.id,
#174	content=memory.content,
#175	created_at=memory.created_at,
#176	state=memory.state.value,
#177	app_id=memory.app_id,
#178	app_name=memory.app.name if memory.app else None,
#179	categories=[category.name for category in memory.categories],
#180	metadata_=memory.metadata_
#181	)
#182	for memory in items
#183	if check_memory_access_permissions(db, memory, app_id)
#184	]
#185	)
#186
#187
#188	# Get all categories
#189	@router.get("/categories")
#190	async def get_categories(
#191	user_id: str,
#192	db: Session = Depends(get_db)
#193	):
#194	user = db.query(User).filter(User.user_id == user_id).first()
#195	if not user:
#196	raise HTTPException(status_code=404, detail="User not found")
#197
#198	# Get unique categories associated with the user's memories
#199	# Get all memories
#200	memories = db.query(Memory).filter(Memory.user_id == user.id, Memory.state != MemoryState.deleted, Memory.state != MemoryState.archived).all()
#201	# Get all categories from memories
#202	categories = [category for memory in memories for category in memory.categories]
#203	# Get unique categories
#204	unique_categories = list(set(categories))
#205
#206	return {
#207	"categories": unique_categories,
#208	"total": len(unique_categories)
#209	}
#210
#211
#212	class CreateMemoryRequest(BaseModel):
#213	user_id: str
#214	text: str
#215	metadata: dict = {}
#216	infer: bool = True
#217	app: str = "openmemory"
#218
#219
#220	# Create new memory
#221	@router.post("/")
#222	async def create_memory(
#223	request: CreateMemoryRequest,
#224	db: Session = Depends(get_db)
#225	):
#226	user = db.query(User).filter(User.user_id == request.user_id).first()
#227	if not user:
#228	raise HTTPException(status_code=404, detail="User not found")
#229	# Get or create app
#230	app_obj = db.query(App).filter(App.name == request.app,
#231	App.owner_id == user.id).first()
#232	if not app_obj:
#233	app_obj = App(name=request.app, owner_id=user.id)
#234	db.add(app_obj)
#235	db.commit()
#236	db.refresh(app_obj)
#237
#238	# Check if app is active
#239	if not app_obj.is_active:
#240	raise HTTPException(status_code=403, detail=f"App {request.app} is currently paused on OpenMemory. Cannot create new memories.")
#241
#242	# Log what we're about to do
#243	logging.info(f"Creating memory for user_id: {request.user_id} with app: {request.app}")
#244
#245	# Try to get memory client safely
#246	try:
#247	memory_client = get_memory_client()
#248	if not memory_client:
#249	raise Exception("Memory client is not available")
#250	except Exception as client_error:
#251	logging.warning(f"Memory client unavailable: {client_error}. Creating memory in database only.")
#252	# Return a json response with the error
#253	return {
#254	"error": str(client_error)
#255	}
#256
#257	# Try to save to Qdrant via memory_client
#258	try:
#259	qdrant_response = memory_client.add(
#260	request.text,
#261	user_id=request.user_id, # Use string user_id to match search
#262	metadata={
#263	"source_app": "openmemory",
#264	"mcp_client": request.app,
#265	},
#266	infer=request.infer
#267	)
#268
#269	# Log the response for debugging
#270	logging.info(f"Qdrant response: {qdrant_response}")
#271
#272	# Process Qdrant response
#273	if isinstance(qdrant_response, dict) and 'results' in qdrant_response:
#274	created_memories = []
#275
#276	for result in qdrant_response['results']:
#277	if result['event'] == 'ADD':
#278	# Get the Qdrant-generated ID
#279	memory_id = UUID(result['id'])
#280
#281	# Check if memory already exists
#282	existing_memory = db.query(Memory).filter(Memory.id == memory_id).first()
#283
#284	if existing_memory:
#285	# Update existing memory
#286	existing_memory.state = MemoryState.active
#287	existing_memory.content = result['memory']
#288	memory = existing_memory
#289	else:
#290	# Create memory with the EXACT SAME ID from Qdrant
#291	memory = Memory(
#292	id=memory_id, # Use the same ID that Qdrant generated
#293	user_id=user.id,
#294	app_id=app_obj.id,
#295	content=result['memory'],
#296	metadata_=request.metadata,
#297	state=MemoryState.active
#298	)
#299	db.add(memory)
#300
#301	# Create history entry
#302	history = MemoryStatusHistory(
#303	memory_id=memory_id,
#304	changed_by=user.id,
#305	old_state=MemoryState.deleted if existing_memory else MemoryState.deleted,
#306	new_state=MemoryState.active
#307	)
#308	db.add(history)
#309
#310	created_memories.append(memory)
#311
#312	# Commit all changes at once
#313	if created_memories:
#314	db.commit()
#315	for memory in created_memories:
#316	db.refresh(memory)
#317
#318	# Return the first memory (for API compatibility)
#319	# but all memories are now saved to the database
#320	return created_memories[0]
#321	except Exception as qdrant_error:
#322	logging.warning(f"Qdrant operation failed: {qdrant_error}.")
#323	# Return a json response with the error
#324	return {
#325	"error": str(qdrant_error)
#326	}
#327
#328
#329
#330
#331	# Get memory by ID
#332	@router.get("/{memory_id}")
#333	async def get_memory(
#334	memory_id: UUID,
#335	db: Session = Depends(get_db)
#336	):
#337	memory = get_memory_or_404(db, memory_id)
#338	return {
#339	"id": memory.id,
#340	"text": memory.content,
#341	"created_at": int(memory.created_at.timestamp()),
#342	"state": memory.state.value,
#343	"app_id": memory.app_id,
#344	"app_name": memory.app.name if memory.app else None,
#345	"categories": [category.name for category in memory.categories],
#346	"metadata_": memory.metadata_
#347	}
#348
#349
#350	class DeleteMemoriesRequest(BaseModel):
#351	memory_ids: List[UUID]
#352	user_id: str
#353
#354	# Delete multiple memories
#355	@router.delete("/")
#356	async def delete_memories(
#357	request: DeleteMemoriesRequest,
#358	db: Session = Depends(get_db)
#359	):
#360	user = db.query(User).filter(User.user_id == request.user_id).first()
#361	if not user:
#362	raise HTTPException(status_code=404, detail="User not found")
#363
#364	# Get memory client to delete from vector store
#365	try:
#366	memory_client = get_memory_client()
#367	if not memory_client:
#368	raise HTTPException(
#369	status_code=503,
#370	detail="Memory client is not available"
#371	)
#372	except HTTPException:
#373	raise
#374	except Exception as client_error:
#375	logging.error(f"Memory client initialization failed: {client_error}")
#376	raise HTTPException(
#377	status_code=503,
#378	detail=f"Memory service unavailable: {str(client_error)}"
#379	)
#380
#381	# Delete from vector store then mark as deleted in database
#382	for memory_id in request.memory_ids:
#383	try:
#384	memory_client.delete(str(memory_id))
#385	except Exception as delete_error:
#386	logging.warning(f"Failed to delete memory {memory_id} from vector store: {delete_error}")
#387
#388	update_memory_state(db, memory_id, MemoryState.deleted, user.id)
#389
#390	return {"message": f"Successfully deleted {len(request.memory_ids)} memories"}
#391
#392
#393	# Archive memories
#394	@router.post("/actions/archive")
#395	async def archive_memories(
#396	memory_ids: List[UUID],
#397	user_id: UUID,
#398	db: Session = Depends(get_db)
#399	):
#400	for memory_id in memory_ids:
#401	update_memory_state(db, memory_id, MemoryState.archived, user_id)
#402	return {"message": f"Successfully archived {len(memory_ids)} memories"}
#403
#404
#405	class PauseMemoriesRequest(BaseModel):
#406	memory_ids: Optional[List[UUID]] = None
#407	category_ids: Optional[List[UUID]] = None
#408	app_id: Optional[UUID] = None
#409	all_for_app: bool = False
#410	global_pause: bool = False
#411	state: Optional[MemoryState] = None
#412	user_id: str
#413
#414	# Pause access to memories
#415	@router.post("/actions/pause")
#416	async def pause_memories(
#417	request: PauseMemoriesRequest,
#418	db: Session = Depends(get_db)
#419	):
#420
#421	global_pause = request.global_pause
#422	all_for_app = request.all_for_app
#423	app_id = request.app_id
#424	memory_ids = request.memory_ids
#425	category_ids = request.category_ids
#426	state = request.state or MemoryState.paused
#427
#428	user = db.query(User).filter(User.user_id == request.user_id).first()
#429	if not user:
#430	raise HTTPException(status_code=404, detail="User not found")
#431
#432	user_id = user.id
#433
#434	if global_pause:
#435	# Pause all memories
#436	memories = db.query(Memory).filter(
#437	Memory.state != MemoryState.deleted,
#438	Memory.state != MemoryState.archived
#439	).all()
#440	for memory in memories:
#441	update_memory_state(db, memory.id, state, user_id)
#442	return {"message": "Successfully paused all memories"}
#443
#444	if app_id:
#445	# Pause all memories for an app
#446	memories = db.query(Memory).filter(
#447	Memory.app_id == app_id,
#448	Memory.user_id == user.id,
#449	Memory.state != MemoryState.deleted,
#450	Memory.state != MemoryState.archived
#451	).all()
#452	for memory in memories:
#453	update_memory_state(db, memory.id, state, user_id)
#454	return {"message": f"Successfully paused all memories for app {app_id}"}
#455
#456	if all_for_app and memory_ids:
#457	# Pause all memories for an app
#458	memories = db.query(Memory).filter(
#459	Memory.user_id == user.id,
#460	Memory.state != MemoryState.deleted,
#461	Memory.id.in_(memory_ids)
#462	).all()
#463	for memory in memories:
#464	update_memory_state(db, memory.id, state, user_id)
#465	return {"message": "Successfully paused all memories"}
#466
#467	if memory_ids:
#468	# Pause specific memories
#469	for memory_id in memory_ids:
#470	update_memory_state(db, memory_id, state, user_id)
#471	return {"message": f"Successfully paused {len(memory_ids)} memories"}
#472
#473	if category_ids:
#474	# Pause memories by category
#475	memories = db.query(Memory).join(Memory.categories).filter(
#476	Category.id.in_(category_ids),
#477	Memory.state != MemoryState.deleted,
#478	Memory.state != MemoryState.archived
#479	).all()
#480	for memory in memories:
#481	update_memory_state(db, memory.id, state, user_id)
#482	return {"message": f"Successfully paused memories in {len(category_ids)} categories"}
#483
#484	raise HTTPException(status_code=400, detail="Invalid pause request parameters")
#485
#486
#487	# Get memory access logs
#488	@router.get("/{memory_id}/access-log")
#489	async def get_memory_access_log(
#490	memory_id: UUID,
#491	page: int = Query(1, ge=1),
#492	page_size: int = Query(10, ge=1, le=100),
#493	db: Session = Depends(get_db)
#494	):
#495	query = db.query(MemoryAccessLog).filter(MemoryAccessLog.memory_id == memory_id)
#496	total = query.count()
#497	logs = query.order_by(MemoryAccessLog.accessed_at.desc()).offset((page - 1) * page_size).limit(page_size).all()
#498
#499	# Get app name
#500	for log in logs:
#501	app = db.query(App).filter(App.id == log.app_id).first()
#502	log.app_name = app.name if app else None
#503
#504	return {
#505	"total": total,
#506	"page": page,
#507	"page_size": page_size,
#508	"logs": logs
#509	}
#510
#511
#512	class UpdateMemoryRequest(BaseModel):
#513	memory_content: str
#514	user_id: str
#515
#516	# Update a memory
#517	@router.put("/{memory_id}")
#518	async def update_memory(
#519	memory_id: UUID,
#520	request: UpdateMemoryRequest,
#521	db: Session = Depends(get_db)
#522	):
#523	user = db.query(User).filter(User.user_id == request.user_id).first()
#524	if not user:
#525	raise HTTPException(status_code=404, detail="User not found")
#526	memory = get_memory_or_404(db, memory_id)
#527	memory.content = request.memory_content
#528	db.commit()
#529	db.refresh(memory)
#530	return memory
#531
#532	class FilterMemoriesRequest(BaseModel):
#533	user_id: str
#534	page: int = 1
#535	size: int = 10
#536	search_query: Optional[str] = None
#537	app_ids: Optional[List[UUID]] = None
#538	category_ids: Optional[List[UUID]] = None
#539	sort_column: Optional[str] = None
#540	sort_direction: Optional[str] = None
#541	from_date: Optional[int] = None
#542	to_date: Optional[int] = None
#543	show_archived: Optional[bool] = False
#544
#545	@router.post("/filter", response_model=Page[MemoryResponse])
#546	async def filter_memories(
#547	request: FilterMemoriesRequest,
#548	db: Session = Depends(get_db)
#549	):
#550	user = db.query(User).filter(User.user_id == request.user_id).first()
#551	if not user:
#552	raise HTTPException(status_code=404, detail="User not found")
#553
#554	# Build base query
#555	query = db.query(Memory).filter(
#556	Memory.user_id == user.id,
#557	Memory.state != MemoryState.deleted,
#558	)
#559
#560	# Filter archived memories based on show_archived parameter
#561	if not request.show_archived:
#562	query = query.filter(Memory.state != MemoryState.archived)
#563
#564	# Apply search filter
#565	if request.search_query:
#566	query = query.filter(Memory.content.ilike(f"%{request.search_query}%"))
#567
#568	# Apply app filter
#569	if request.app_ids:
#570	query = query.filter(Memory.app_id.in_(request.app_ids))
#571
#572	# Add joins for app and categories
#573	query = query.outerjoin(App, Memory.app_id == App.id)
#574
#575	# Apply category filter
#576	if request.category_ids:
#577	query = query.join(Memory.categories).filter(Category.id.in_(request.category_ids))
#578	else:
#579	query = query.outerjoin(Memory.categories)
#580
#581	# Apply date filters
#582	if request.from_date:
#583	from_datetime = datetime.fromtimestamp(request.from_date, tz=UTC)
#584	query = query.filter(Memory.created_at >= from_datetime)
#585
#586	if request.to_date:
#587	to_datetime = datetime.fromtimestamp(request.to_date, tz=UTC)
#588	query = query.filter(Memory.created_at <= to_datetime)
#589
#590	# Apply sorting
#591	if request.sort_column and request.sort_direction:
#592	sort_direction = request.sort_direction.lower()
#593	if sort_direction not in ['asc', 'desc']:
#594	raise HTTPException(status_code=400, detail="Invalid sort direction")
#595
#596	sort_mapping = {
#597	'memory': Memory.content,
#598	'app_name': App.name,
#599	'created_at': Memory.created_at
#600	}
#601
#602	if request.sort_column not in sort_mapping:
#603	raise HTTPException(status_code=400, detail="Invalid sort column")
#604
#605	sort_field = sort_mapping[request.sort_column]
#606	if sort_direction == 'desc':
#607	query = query.order_by(sort_field.desc())
#608	else:
#609	query = query.order_by(sort_field.asc())
#610	else:
#611	# Default sorting
#612	query = query.order_by(Memory.created_at.desc())
#613
#614	# Add eager loading for categories and make the query distinct
#615	query = query.options(
#616	joinedload(Memory.categories)
#617	).distinct(Memory.id)
#618
#619	# Use fastapi-pagination's paginate function
#620	return sqlalchemy_paginate(
#621	query,
#622	Params(page=request.page, size=request.size),
#623	transformer=lambda items: [
#624	MemoryResponse(
#625	id=memory.id,
#626	content=memory.content,
#627	created_at=memory.created_at,
#628	state=memory.state.value,
#629	app_id=memory.app_id,
#630	app_name=memory.app.name if memory.app else None,
#631	categories=[category.name for category in memory.categories],
#632	metadata_=memory.metadata_
#633	)
#634	for memory in items
#635	]
#636	)
#637
#638
#639	@router.get("/{memory_id}/related", response_model=Page[MemoryResponse])
#640	async def get_related_memories(
#641	memory_id: UUID,
#642	user_id: str,
#643	params: Params = Depends(),
#644	db: Session = Depends(get_db)
#645	):
#646	# Validate user
#647	user = db.query(User).filter(User.user_id == user_id).first()
#648	if not user:
#649	raise HTTPException(status_code=404, detail="User not found")
#650
#651	# Get the source memory
#652	memory = get_memory_or_404(db, memory_id)
#653
#654	# Extract category IDs from the source memory
#655	category_ids = [category.id for category in memory.categories]
#656
#657	if not category_ids:
#658	return Page.create([], total=0, params=params)
#659
#660	# Build query for related memories
#661	query = db.query(Memory).distinct(Memory.id).filter(
#662	Memory.user_id == user.id,
#663	Memory.id != memory_id,
#664	Memory.state != MemoryState.deleted
#665	).join(Memory.categories).filter(
#666	Category.id.in_(category_ids)
#667	).options(
#668	joinedload(Memory.categories),
#669	joinedload(Memory.app)
#670	).order_by(
#671	func.count(Category.id).desc(),
#672	Memory.created_at.desc()
#673	).group_by(Memory.id)
#674
#675	# ⚡ Force page size to be 5
#676	params = Params(page=params.page, size=5)
#677
#678	return sqlalchemy_paginate(
#679	query,
#680	params,
#681	transformer=lambda items: [
#682	MemoryResponse(
#683	id=memory.id,
#684	content=memory.content,
#685	created_at=memory.created_at,
#686	state=memory.state.value,
#687	app_id=memory.app_id,
#688	app_name=memory.app.name if memory.app else None,
#689	categories=[category.name for category in memory.categories],
#690	metadata_=memory.metadata_
#691	)
#692	for memory in items
#693	]
#694	)

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public