admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): simplify note write flow (#739)

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
Paul Hernandez
2026-04-15 17:56:27 -05:00
committed by GitHub
parent 8f2b25f0e0
commit 4e53bb83fd
17 changed files with 1296 additions and 1599 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/api/v2/routers/knowledge_router.py
+222 -441
View File
@@ -10,7 +10,7 @@ Key improvements:
- Simplified caching strategies
"""
from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends, Response, Path, Query
from fastapi import APIRouter, HTTPException, Response, Path
from loguru import logger
from basic_memory import telemetry
@@ -24,7 +24,6 @@ from basic_memory.deps import (
RelationRepositoryV2ExternalDep,
ProjectExternalIdPathDep,
TaskSchedulerDep,
FileServiceV2ExternalDep,
)
from basic_memory.schemas import DeleteEntitiesResponse
from basic_memory.schemas.base import Entity
@@ -75,34 +74,40 @@ async def get_graph(
Returns a flat node/edge structure optimized for rendering with graph libraries.
Only includes resolved relations (where to_id is not null).
"""
logger.info("API v2 request: get_graph")
with telemetry.operation(
"api.request.knowledge.get_graph",
entrypoint="api",
domain="knowledge",
action="get_graph",
):
logger.info("API v2 request: get_graph")
# Fetch all entities for this project
entities = await entity_repository.find_all(use_load_options=False)
nodes = [
GraphNode(
external_id=entity.external_id,
title=entity.title,
note_type=entity.note_type,
file_path=entity.file_path,
)
for entity in entities
]
# Fetch all entities for this project
entities = await entity_repository.find_all(use_load_options=False)
nodes = [
GraphNode(
external_id=entity.external_id,
title=entity.title,
note_type=entity.note_type,
file_path=entity.file_path,
)
for entity in entities
]
# Fetch all resolved relations (to_id is not null) with eager-loaded entities
relations = await relation_repository.find_all()
edges = [
GraphEdge(
from_id=relation.from_entity.external_id,
to_id=relation.to_entity.external_id,
relation_type=relation.relation_type,
)
for relation in relations
if relation.to_entity is not None
]
# Fetch all resolved relations (to_id is not null) with eager-loaded entities
relations = await relation_repository.find_all()
edges = [
GraphEdge(
from_id=relation.from_entity.external_id,
to_id=relation.to_entity.external_id,
relation_type=relation.relation_type,
)
for relation in relations
if relation.to_entity is not None
]
logger.info(f"API v2 response: graph with {len(nodes)} nodes and {len(edges)} edges")
return GraphResponse(nodes=nodes, edges=edges)
logger.info(f"API v2 response: graph with {len(nodes)} nodes and {len(edges)} edges")
return GraphResponse(nodes=nodes, edges=edges)
## Resolution endpoint
@@ -151,25 +156,13 @@ async def resolve_identifier(
):
logger.info(f"API v2 request: resolve_identifier for '{data.identifier}'")
with telemetry.scope(
"api.knowledge.resolve_entity.lookup_entity",
domain="knowledge",
action="resolve_entity",
phase="lookup_entity",
):
entity = await entity_repository.get_by_external_id(data.identifier)
entity = await entity_repository.get_by_external_id(data.identifier)
resolution_method = "external_id" if entity else "search"
if not entity:
with telemetry.scope(
"api.knowledge.resolve_entity.resolve_link",
domain="knowledge",
action="resolve_entity",
phase="resolve_link",
):
entity = await link_resolver.resolve_link(
data.identifier, source_path=data.source_path, strict=data.strict
)
entity = await link_resolver.resolve_link(
data.identifier, source_path=data.source_path, strict=data.strict
)
if entity:
if entity.permalink == data.identifier:
resolution_method = "permalink"
@@ -183,20 +176,14 @@ async def resolve_identifier(
if not entity:
raise HTTPException(status_code=404, detail=f"Entity not found: '{data.identifier}'")
with telemetry.scope(
"api.knowledge.resolve_entity.shape_response",
domain="knowledge",
action="resolve_entity",
phase="shape_response",
):
result = EntityResolveResponse(
external_id=entity.external_id,
entity_id=entity.id,
permalink=entity.permalink,
file_path=entity.file_path,
title=entity.title,
resolution_method=resolution_method,
)
result = EntityResolveResponse(
external_id=entity.external_id,
entity_id=entity.id,
permalink=entity.permalink,
file_path=entity.file_path,
title=entity.title,
resolution_method=resolution_method,
)
logger.debug(
f"API v2 response: resolved '{data.identifier}' to external_id={result.external_id} via {resolution_method}"
@@ -236,25 +223,13 @@ async def get_entity_by_id(
):
logger.info(f"API v2 request: get_entity_by_id entity_id={entity_id}")
with telemetry.scope(
"api.knowledge.get_entity.load_entity",
domain="knowledge",
action="get_entity",
phase="load_entity",
):
entity = await entity_repository.get_by_external_id(entity_id)
entity = await entity_repository.get_by_external_id(entity_id)
if not entity:
raise HTTPException(
status_code=404, detail=f"Entity with external_id '{entity_id}' not found"
)
with telemetry.scope(
"api.knowledge.get_entity.shape_response",
domain="knowledge",
action="get_entity",
phase="shape_response",
):
result = EntityResponseV2.model_validate(entity)
result = EntityResponseV2.model_validate(entity)
logger.info(f"API v2 response: external_id={entity_id}, title='{result.title}'")
return result
@@ -267,21 +242,15 @@ async def get_entity_by_id(
async def create_entity(
project_id: ProjectExternalIdPathDep,
data: Entity,
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
task_scheduler: TaskSchedulerDep,
file_service: FileServiceV2ExternalDep,
app_config: AppConfigDep,
fast: bool = Query(
True, description="If true, write quickly and defer indexing to background tasks."
),
) -> EntityResponseV2:
"""Create a new entity.
Args:
data: Entity data to create
fast: If True, defer indexing to background tasks
Returns:
Created entity with generated external_id (UUID) and file content
@@ -291,82 +260,26 @@ async def create_entity(
entrypoint="api",
domain="knowledge",
action="create_entity",
fast=fast,
):
logger.info(
"API v2 request", endpoint="create_entity", note_type=data.note_type, title=data.title
)
with telemetry.scope(
"api.knowledge.create_entity.write_entity",
domain="knowledge",
action="create_entity",
phase="write_entity",
fast=fast,
):
if fast:
entity = await entity_service.fast_write_entity(data)
written_content = None
search_content = None
else:
write_result = await entity_service.create_entity_with_content(data)
entity = write_result.entity
written_content = write_result.content
search_content = write_result.search_content
if fast:
with telemetry.scope(
"api.knowledge.create_entity.enqueue_reindex",
domain="knowledge",
action="create_entity",
phase="enqueue_reindex",
fast=fast,
):
task_scheduler.schedule(
"reindex_entity",
entity_id=entity.id,
project_id=project_id,
)
else:
with telemetry.scope(
"api.knowledge.create_entity.search_index",
domain="knowledge",
action="create_entity",
phase="search_index",
):
await search_service.index_entity(entity, content=search_content)
with telemetry.scope(
"api.knowledge.create_entity.vector_sync",
domain="knowledge",
action="create_entity",
phase="vector_sync",
):
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=entity.id,
project_id=project_id,
)
# Note writes are now internally consistent before the response returns. We only leave
# truly derived work, like semantic vectors, on the async scheduler.
write_result = await entity_service.create_entity_with_content(data)
entity = write_result.entity
await search_service.index_entity(entity, content=write_result.search_content)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=entity.id,
project_id=project_id,
)
result = EntityResponseV2.model_validate(entity)
if fast:
result = result.model_copy(update={"observations": [], "relations": []})
with telemetry.scope(
"api.knowledge.create_entity.read_content",
domain="knowledge",
action="create_entity",
phase="read_content",
source="file" if fast else "memory",
):
if fast:
content = await file_service.read_file_content(entity.file_path)
else:
# Non-fast writes already captured the markdown in memory. Reuse it here
# instead of re-reading the file; format_on_save is the one config that can
# still make the persisted file diverge because write_file only returns a checksum.
content = written_content
result = result.model_copy(update={"content": content})
# The write service already returns the canonical markdown accepted for this request.
result = result.model_copy(update={"content": write_result.content})
logger.info(
f"API v2 response: endpoint='create_entity' external_id={entity.external_id}, title={result.title}, permalink={result.permalink}, status_code=201"
@@ -381,18 +294,13 @@ async def create_entity(
async def update_entity_by_id(
data: Entity,
response: Response,
background_tasks: BackgroundTasks,
project_id: ProjectExternalIdPathDep,
entity_service: EntityServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
entity_repository: EntityRepositoryV2ExternalDep,
task_scheduler: TaskSchedulerDep,
file_service: FileServiceV2ExternalDep,
app_config: AppConfigDep,
entity_id: str = Path(..., description="Entity external ID (UUID)"),
fast: bool = Query(
True, description="If true, write quickly and defer indexing to background tasks."
),
) -> EntityResponseV2:
"""Update an entity by external ID.
@@ -401,7 +309,6 @@ async def update_entity_by_id(
Args:
entity_id: External ID (UUID string)
data: Updated entity data
fast: If True, defer indexing to background tasks
Returns:
Updated entity with file content
@@ -411,111 +318,43 @@ async def update_entity_by_id(
entrypoint="api",
domain="knowledge",
action="update_entity",
fast=fast,
):
logger.info(f"API v2 request: update_entity_by_id entity_id={entity_id}")
with telemetry.scope(
"api.knowledge.update_entity.load_entity",
domain="knowledge",
action="update_entity",
phase="load_entity",
):
existing = await entity_repository.get_by_external_id(entity_id)
existing = await entity_repository.get_by_external_id(entity_id)
created = existing is None
with telemetry.scope(
"api.knowledge.update_entity.write_entity",
domain="knowledge",
action="update_entity",
phase="write_entity",
fast=fast,
):
if fast:
entity = await entity_service.fast_write_entity(data, external_id=entity_id)
written_content = None
search_content = None
response.status_code = 200 if existing else 201
else:
if existing:
write_result = await entity_service.update_entity_with_content(existing, data)
entity = write_result.entity
written_content = write_result.content
search_content = write_result.search_content
response.status_code = 200
else:
write_result = await entity_service.create_entity_with_content(data)
entity = write_result.entity
written_content = write_result.content
search_content = write_result.search_content
if entity.external_id != entity_id:
entity = await entity_repository.update(
entity.id,
{"external_id": entity_id},
)
# external_id fixup only changes the DB row. The file content is unchanged,
# so the markdown captured during the write remains valid downstream.
if not entity:
raise HTTPException(
status_code=404,
detail=f"Entity with external_id '{entity_id}' not found",
)
response.status_code = 201
if fast:
with telemetry.scope(
"api.knowledge.update_entity.enqueue_reindex",
domain="knowledge",
action="update_entity",
phase="enqueue_reindex",
fast=fast,
):
task_scheduler.schedule(
"reindex_entity",
entity_id=entity.id,
project_id=project_id,
resolve_relations=created,
)
if existing:
write_result = await entity_service.update_entity_with_content(existing, data)
entity = write_result.entity
response.status_code = 200
else:
with telemetry.scope(
"api.knowledge.update_entity.search_index",
domain="knowledge",
action="update_entity",
phase="search_index",
):
await search_service.index_entity(entity, content=search_content)
with telemetry.scope(
"api.knowledge.update_entity.vector_sync",
domain="knowledge",
action="update_entity",
phase="vector_sync",
):
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=entity.id,
project_id=project_id,
write_result = await entity_service.create_entity_with_content(data)
entity = write_result.entity
if entity.external_id != entity_id:
entity = await entity_repository.update(
entity.id,
{"external_id": entity_id},
)
# external_id fixup only changes the DB row. The file content is unchanged,
# so the markdown captured during the write remains valid downstream.
if not entity:
raise HTTPException(
status_code=404,
detail=f"Entity with external_id '{entity_id}' not found",
)
response.status_code = 201
await search_service.index_entity(entity, content=write_result.search_content)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=entity.id,
project_id=project_id,
)
result = EntityResponseV2.model_validate(entity)
if fast:
result = result.model_copy(update={"observations": [], "relations": []})
with telemetry.scope(
"api.knowledge.update_entity.read_content",
domain="knowledge",
action="update_entity",
phase="read_content",
source="file" if fast else "memory",
):
if fast:
content = await file_service.read_file_content(entity.file_path)
else:
# Non-fast writes already captured the markdown in memory. Reuse it here
# instead of re-reading the file; format_on_save is the one config that can
# still make the persisted file diverge because write_file only returns a checksum.
content = written_content
result = result.model_copy(update={"content": content})
result = result.model_copy(update={"content": write_result.content})
logger.info(
f"API v2 response: external_id={entity_id}, created={created}, status_code={response.status_code}"
@@ -526,25 +365,19 @@ async def update_entity_by_id(
@router.patch("/entities/{entity_id}", response_model=EntityResponseV2)
async def edit_entity_by_id(
data: EditEntityRequest,
background_tasks: BackgroundTasks,
project_id: ProjectExternalIdPathDep,
entity_service: EntityServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
entity_repository: EntityRepositoryV2ExternalDep,
task_scheduler: TaskSchedulerDep,
file_service: FileServiceV2ExternalDep,
app_config: AppConfigDep,
entity_id: str = Path(..., description="Entity external ID (UUID)"),
fast: bool = Query(
True, description="If true, write quickly and defer indexing to background tasks."
),
) -> EntityResponseV2:
"""Edit an existing entity by external ID using operations like append, prepend, etc.
Args:
entity_id: External ID (UUID string)
data: Edit operation details
fast: If True, defer indexing to background tasks
Returns:
Updated entity with file content
@@ -557,110 +390,38 @@ async def edit_entity_by_id(
entrypoint="api",
domain="knowledge",
action="edit_entity",
fast=fast,
):
logger.info(
f"API v2 request: edit_entity_by_id entity_id={entity_id}, operation='{data.operation}'"
)
with telemetry.scope(
"api.knowledge.edit_entity.load_entity",
domain="knowledge",
action="edit_entity",
phase="load_entity",
):
entity = await entity_repository.get_by_external_id(entity_id)
entity = await entity_repository.get_by_external_id(entity_id)
if not entity: # pragma: no cover
raise HTTPException(
status_code=404, detail=f"Entity with external_id '{entity_id}' not found"
)
try:
with telemetry.scope(
"api.knowledge.edit_entity.write_entity",
domain="knowledge",
action="edit_entity",
phase="write_entity",
fast=fast,
):
if fast:
updated_entity = await entity_service.fast_edit_entity(
entity=entity,
operation=data.operation,
content=data.content,
section=data.section,
find_text=data.find_text,
expected_replacements=data.expected_replacements,
)
written_content = None
search_content = None
else:
identifier = entity.permalink or entity.file_path
write_result = await entity_service.edit_entity_with_content(
identifier=identifier,
operation=data.operation,
content=data.content,
section=data.section,
find_text=data.find_text,
expected_replacements=data.expected_replacements,
)
updated_entity = write_result.entity
written_content = write_result.content
search_content = write_result.search_content
if fast:
with telemetry.scope(
"api.knowledge.edit_entity.enqueue_reindex",
domain="knowledge",
action="edit_entity",
phase="enqueue_reindex",
fast=fast,
):
task_scheduler.schedule(
"reindex_entity",
entity_id=updated_entity.id,
project_id=project_id,
)
else:
with telemetry.scope(
"api.knowledge.edit_entity.search_index",
domain="knowledge",
action="edit_entity",
phase="search_index",
):
await search_service.index_entity(updated_entity, content=search_content)
with telemetry.scope(
"api.knowledge.edit_entity.vector_sync",
domain="knowledge",
action="edit_entity",
phase="vector_sync",
):
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=updated_entity.id,
project_id=project_id,
)
identifier = entity.permalink or entity.file_path
write_result = await entity_service.edit_entity_with_content(
identifier=identifier,
operation=data.operation,
content=data.content,
section=data.section,
find_text=data.find_text,
expected_replacements=data.expected_replacements,
)
updated_entity = write_result.entity
await search_service.index_entity(updated_entity, content=write_result.search_content)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=updated_entity.id,
project_id=project_id,
)
result = EntityResponseV2.model_validate(updated_entity)
if fast:
result = result.model_copy(update={"observations": [], "relations": []})
with telemetry.scope(
"api.knowledge.edit_entity.read_content",
domain="knowledge",
action="edit_entity",
phase="read_content",
source="file" if fast else "memory",
):
if fast:
content = await file_service.read_file_content(updated_entity.file_path)
else:
# Non-fast writes already captured the markdown in memory. Reuse it here
# instead of re-reading the file; format_on_save is the one config that can
# still make the persisted file diverge because write_file only returns a checksum.
content = written_content
result = result.model_copy(update={"content": content})
result = result.model_copy(update={"content": write_result.content})
logger.info(
f"API v2 response: external_id={entity_id}, operation='{data.operation}', status_code=200"
@@ -678,12 +439,10 @@ async def edit_entity_by_id(
@router.delete("/entities/{entity_id}", response_model=DeleteEntitiesResponse)
async def delete_entity_by_id(
background_tasks: BackgroundTasks,
project_id: ProjectExternalIdPathDep,
entity_service: EntityServiceV2ExternalDep,
entity_repository: EntityRepositoryV2ExternalDep,
entity_id: str = Path(..., description="Entity external ID (UUID)"),
search_service=Depends(lambda: None), # Optional for now
) -> DeleteEntitiesResponse:
"""Delete an entity by external ID.
@@ -695,23 +454,25 @@ async def delete_entity_by_id(
Note: Returns deleted=False if entity doesn't exist (idempotent)
"""
logger.info(f"API v2 request: delete_entity_by_id entity_id={entity_id}")
with telemetry.operation(
"api.request.knowledge.delete_entity",
entrypoint="api",
domain="knowledge",
action="delete_entity",
):
logger.info(f"API v2 request: delete_entity_by_id entity_id={entity_id}")
entity = await entity_repository.get_by_external_id(entity_id)
if entity is None:
logger.info(f"API v2 response: external_id={entity_id} not found, deleted=False")
return DeleteEntitiesResponse(deleted=False)
entity = await entity_repository.get_by_external_id(entity_id)
if entity is None:
logger.info(f"API v2 response: external_id={entity_id} not found, deleted=False")
return DeleteEntitiesResponse(deleted=False)
# Delete the entity using internal ID
deleted = await entity_service.delete_entity(entity.id)
# Delete the entity using internal ID
deleted = await entity_service.delete_entity(entity.id)
# Remove from search index if search service available
if search_service:
background_tasks.add_task(search_service.handle_delete, entity) # pragma: no cover
logger.info(f"API v2 response: external_id={entity_id}, deleted={deleted}")
logger.info(f"API v2 response: external_id={entity_id}, deleted={deleted}")
return DeleteEntitiesResponse(deleted=deleted)
return DeleteEntitiesResponse(deleted=deleted)
## Move endpoint
@@ -720,7 +481,6 @@ async def delete_entity_by_id(
@router.put("/entities/{entity_id}/move", response_model=EntityResponseV2)
async def move_entity(
data: MoveEntityRequestV2,
background_tasks: BackgroundTasks,
project_id: ProjectExternalIdPathDep,
entity_service: EntityServiceV2ExternalDep,
entity_repository: EntityRepositoryV2ExternalDep,
@@ -743,48 +503,58 @@ async def move_entity(
Returns:
Updated entity with new file path
"""
logger.info(
f"API v2 request: move_entity entity_id={entity_id}, destination='{data.destination_path}'"
)
try:
# First, get the entity by external_id to verify it exists
entity = await entity_repository.get_by_external_id(entity_id)
if not entity: # pragma: no cover
raise HTTPException(
status_code=404, detail=f"Entity with external_id '{entity_id}' not found"
)
# Move the entity using its current file path as identifier
moved_entity = await entity_service.move_entity(
identifier=entity.file_path, # Use file path for resolution
destination_path=data.destination_path,
project_config=project_config,
app_config=app_config,
with telemetry.operation(
"api.request.knowledge.move_entity",
entrypoint="api",
domain="knowledge",
action="move_entity",
):
logger.info(
f"API v2 request: move_entity entity_id={entity_id}, destination='{data.destination_path}'"
)
# Reindex at new location
reindexed_entity = await entity_service.link_resolver.resolve_link(data.destination_path)
if reindexed_entity:
await search_service.index_entity(reindexed_entity)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
try:
# First, get the entity by external_id to verify it exists
entity = await entity_repository.get_by_external_id(entity_id)
if not entity: # pragma: no cover
raise HTTPException(
status_code=404, detail=f"Entity with external_id '{entity_id}' not found"
)
# Move the entity using its current file path as identifier
moved_entity = await entity_service.move_entity(
identifier=entity.file_path, # Use file path for resolution
destination_path=data.destination_path,
project_config=project_config,
app_config=app_config,
entity_id=reindexed_entity.id,
project_id=project_id,
)
result = EntityResponseV2.model_validate(moved_entity)
# Reindex at new location
reindexed_entity = await entity_service.link_resolver.resolve_link(
data.destination_path
)
if reindexed_entity:
await search_service.index_entity(reindexed_entity)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=reindexed_entity.id,
project_id=project_id,
)
logger.info(f"API v2 response: moved external_id={entity_id} to '{data.destination_path}'")
result = EntityResponseV2.model_validate(moved_entity)
return result
logger.info(
f"API v2 response: moved external_id={entity_id} to '{data.destination_path}'"
)
except HTTPException: # pragma: no cover
raise # pragma: no cover
except Exception as e:
logger.error(f"Error moving entity: {e}")
raise HTTPException(status_code=400, detail=str(e))
return result
except HTTPException: # pragma: no cover
raise # pragma: no cover
except Exception as e:
logger.error(f"Error moving entity: {e}")
raise HTTPException(status_code=400, detail=str(e))
## Move directory endpoint
@@ -793,7 +563,6 @@ async def move_entity(
@router.post("/move-directory", response_model=DirectoryMoveResult)
async def move_directory(
data: MoveDirectoryRequestV2,
background_tasks: BackgroundTasks,
project_id: ProjectExternalIdPathDep,
entity_service: EntityServiceV2ExternalDep,
project_config: ProjectConfigV2ExternalDep,
@@ -814,40 +583,46 @@ async def move_directory(
Returns:
DirectoryMoveResult with counts and details of moved files
"""
logger.info(
f"API v2 request: move_directory source='{data.source_directory}', destination='{data.destination_directory}'"
)
try:
# Move the directory using the service
result = await entity_service.move_directory(
source_directory=data.source_directory,
destination_directory=data.destination_directory,
project_config=project_config,
app_config=app_config,
)
# Reindex moved entities
for file_path in result.moved_files:
entity = await entity_service.link_resolver.resolve_link(file_path)
if entity:
await search_service.index_entity(entity)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=entity.id,
project_id=project_id,
)
with telemetry.operation(
"api.request.knowledge.move_directory",
entrypoint="api",
domain="knowledge",
action="move_directory",
):
logger.info(
f"API v2 response: move_directory "
f"total={result.total_files}, success={result.successful_moves}, failed={result.failed_moves}"
f"API v2 request: move_directory source='{data.source_directory}', destination='{data.destination_directory}'"
)
return result
except Exception as e:
logger.error(f"Error moving directory: {e}")
raise HTTPException(status_code=400, detail=str(e))
try:
# Move the directory using the service
result = await entity_service.move_directory(
source_directory=data.source_directory,
destination_directory=data.destination_directory,
project_config=project_config,
app_config=app_config,
)
# Reindex moved entities
for file_path in result.moved_files:
entity = await entity_service.link_resolver.resolve_link(file_path)
if entity:
await search_service.index_entity(entity)
_schedule_vector_sync_if_enabled(
task_scheduler=task_scheduler,
app_config=app_config,
entity_id=entity.id,
project_id=project_id,
)
logger.info(
f"API v2 response: move_directory "
f"total={result.total_files}, success={result.successful_moves}, failed={result.failed_moves}"
)
return result
except Exception as e:
logger.error(f"Error moving directory: {e}")
raise HTTPException(status_code=400, detail=str(e))
## Delete directory endpoint
@@ -872,20 +647,26 @@ async def delete_directory(
Returns:
DirectoryDeleteResult with counts and details of deleted files
"""
logger.info(f"API v2 request: delete_directory directory='{data.directory}'")
with telemetry.operation(
"api.request.knowledge.delete_directory",
entrypoint="api",
domain="knowledge",
action="delete_directory",
):
logger.info(f"API v2 request: delete_directory directory='{data.directory}'")
try:
# Delete the directory using the service
result = await entity_service.delete_directory(
directory=data.directory,
)
try:
# Delete the directory using the service
result = await entity_service.delete_directory(
directory=data.directory,
)
logger.info(
f"API v2 response: delete_directory "
f"total={result.total_files}, success={result.successful_deletes}, failed={result.failed_deletes}"
)
return result
logger.info(
f"API v2 response: delete_directory "
f"total={result.total_files}, success={result.successful_deletes}, failed={result.failed_deletes}"
)
return result
except Exception as e:
logger.error(f"Error deleting directory: {e}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Error deleting directory: {e}")
raise HTTPException(status_code=400, detail=str(e))
src/basic_memory/cli/commands/doctor.py
+1 -1
View File
@@ -69,7 +69,7 @@ async def run_doctor() -> None:
content=f"# {api_note_title}\n\n- [note] API to file check",
entity_metadata={"tags": ["doctor"]},
)
api_result = await knowledge_client.create_entity(api_note.model_dump(), fast=False)
api_result = await knowledge_client.create_entity(api_note.model_dump())
api_file = project_path / api_result.file_path
if not api_file.exists():
src/basic_memory/deps/services.py
-25
View File
@@ -492,7 +492,6 @@ class LocalTaskScheduler:
async def get_task_scheduler(
entity_service: EntityServiceV2ExternalDep,
sync_service: SyncServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
project_config: ProjectConfigV2ExternalDep,
@@ -500,28 +499,6 @@ async def get_task_scheduler(
) -> TaskScheduler:
"""Create a scheduler that maps task specs to coroutines."""
scheduler: LocalTaskScheduler | None = None
async def _reindex_entity(
entity_id: int,
resolve_relations: bool = False,
**_: Any,
) -> None:
await entity_service.reindex_entity(entity_id)
# Trigger: caller requests relation resolution
# Why: resolve forward references created before the entity existed
# Outcome: updates unresolved relations pointing to this entity
if resolve_relations:
await sync_service.resolve_relations(entity_id=entity_id)
# Trigger: semantic search enabled in local config.
# Why: vector chunks are derived and should refresh after canonical reindex completes.
# Outcome: schedules out-of-band vector sync without extending write latency.
if app_config.semantic_search_enabled and scheduler is not None:
scheduler.schedule("sync_entity_vectors", entity_id=entity_id)
async def _resolve_relations(entity_id: int, **_: Any) -> None:
await sync_service.resolve_relations(entity_id=entity_id)
async def _sync_entity_vectors(entity_id: int, **_: Any) -> None:
await search_service.sync_entity_vectors(entity_id)
@@ -537,8 +514,6 @@ async def get_task_scheduler(
scheduler = LocalTaskScheduler(
{
"reindex_entity": _reindex_entity,
"resolve_relations": _resolve_relations,
"sync_entity_vectors": _sync_entity_vectors,
"sync_project": _sync_project,
"reindex_project": _reindex_project,
src/basic_memory/mcp/clients/knowledge.py
+1 -16
View File
@@ -44,9 +44,7 @@ class KnowledgeClient:
# --- Entity CRUD Operations ---
async def create_entity(
self, entity_data: dict[str, Any], *, fast: bool | None = None
) -> EntityResponse:
async def create_entity(self, entity_data: dict[str, Any]) -> EntityResponse:
"""Create a new entity.
Args:
@@ -58,18 +56,15 @@ class KnowledgeClient:
Raises:
ToolError: If the request fails
"""
params = {"fast": fast} if fast is not None else None
with telemetry.scope(
"mcp.client.knowledge.create_entity",
client_name="knowledge",
operation="create_entity",
fast=fast,
):
response = await call_post(
self.http_client,
f"{self._base_path}/entities",
json=entity_data,
params=params,
client_name="knowledge",
operation="create_entity",
path_template="/v2/projects/{project_id}/knowledge/entities",
@@ -80,8 +75,6 @@ class KnowledgeClient:
self,
entity_id: str,
entity_data: dict[str, Any],
*,
fast: bool | None = None,
) -> EntityResponse:
"""Update an existing entity (full replacement).
@@ -95,18 +88,15 @@ class KnowledgeClient:
Raises:
ToolError: If the request fails
"""
params = {"fast": fast} if fast is not None else None
with telemetry.scope(
"mcp.client.knowledge.update_entity",
client_name="knowledge",
operation="update_entity",
fast=fast,
):
response = await call_put(
self.http_client,
f"{self._base_path}/entities/{entity_id}",
json=entity_data,
params=params,
client_name="knowledge",
operation="update_entity",
path_template="/v2/projects/{project_id}/knowledge/entities/{entity_id}",
@@ -143,8 +133,6 @@ class KnowledgeClient:
self,
entity_id: str,
patch_data: dict[str, Any],
*,
fast: bool | None = None,
) -> EntityResponse:
"""Partially update an entity.
@@ -158,18 +146,15 @@ class KnowledgeClient:
Raises:
ToolError: If the request fails
"""
params = {"fast": fast} if fast is not None else None
with telemetry.scope(
"mcp.client.knowledge.patch_entity",
client_name="knowledge",
operation="patch_entity",
fast=fast,
):
response = await call_patch(
self.http_client,
f"{self._base_path}/entities/{entity_id}",
json=patch_data,
params=params,
client_name="knowledge",
operation="patch_entity",
path_template="/v2/projects/{project_id}/knowledge/entities/{entity_id}",
src/basic_memory/mcp/tools/edit_note.py
+2 -6
View File
@@ -374,9 +374,7 @@ async def edit_note(
directory=directory,
operation=operation,
)
result = await knowledge_client.create_entity(
entity.model_dump(), fast=False
)
result = await knowledge_client.create_entity(entity.model_dump())
file_created = True
else:
# find_replace/replace_section require existing content — re-raise
@@ -399,9 +397,7 @@ async def edit_note(
edit_data["expected_replacements"] = str(effective_replacements)
# Call the PATCH endpoint
result = await knowledge_client.patch_entity(
entity_id, edit_data, fast=False
)
result = await knowledge_client.patch_entity(entity_id, edit_data)
# --- Format response ---
# result is always set: either by create_entity (auto-create) or patch_entity (edit)
src/basic_memory/mcp/tools/write_note.py
+2 -2
View File
@@ -222,7 +222,7 @@ async def write_note(
logger.debug(f"Attempting to create entity permalink={entity.permalink}")
action = "Created" # Default to created
try:
result = await knowledge_client.create_entity(entity.model_dump(), fast=False)
result = await knowledge_client.create_entity(entity.model_dump())
action = "Created"
except Exception as e:
# If creation failed due to conflict (already exists), try to update
@@ -260,7 +260,7 @@ async def write_note(
) # pragma: no cover
entity_id = await knowledge_client.resolve_entity(entity.permalink)
result = await knowledge_client.update_entity(
entity_id, entity.model_dump(), fast=False
entity_id, entity.model_dump()
)
action = "Updated"
except Exception as update_error: # pragma: no cover
src/basic_memory/models/knowledge.py
+3 -3
View File
@@ -62,7 +62,7 @@ class Entity(Base):
)
# Core identity
id: Mapped[int] = mapped_column(Integer, primary_key=True)
id: Mapped[int] = mapped_column(Integer, primary_key=True) # pyright: ignore [reportIncompatibleVariableOverride]
# External UUID for API references - stable identifier that won't change
external_id: Mapped[str] = mapped_column(String, unique=True, default=lambda: str(uuid.uuid4()))
title: Mapped[str] = mapped_column(String)
@@ -229,7 +229,7 @@ class Observation(Base):
Index("ix_observation_category", "category"), # Add category index
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
id: Mapped[int] = mapped_column(Integer, primary_key=True) # pyright: ignore [reportIncompatibleVariableOverride]
project_id: Mapped[int] = mapped_column(Integer, ForeignKey("project.id"), index=True)
entity_id: Mapped[int] = mapped_column(Integer, ForeignKey("entity.id", ondelete="CASCADE"))
content: Mapped[str] = mapped_column(Text)
@@ -276,7 +276,7 @@ class Relation(Base):
Index("ix_relation_to_id", "to_id"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
id: Mapped[int] = mapped_column(Integer, primary_key=True) # pyright: ignore [reportIncompatibleVariableOverride]
project_id: Mapped[int] = mapped_column(Integer, ForeignKey("project.id"), index=True)
from_id: Mapped[int] = mapped_column(Integer, ForeignKey("entity.id", ondelete="CASCADE"))
to_id: Mapped[Optional[int]] = mapped_column(
src/basic_memory/services/entity_service.py
+572 -646
View File
File diff suppressed because it is too large Load Diff
tests/api/v2/test_knowledge_router.py
+15 -18
View File
@@ -375,10 +375,10 @@ async def test_update_entity_by_id(
@pytest.mark.asyncio
async def test_update_entity_by_id_fast_does_not_duplicate(
async def test_update_entity_by_id_does_not_duplicate(
client: AsyncClient, v2_project_url, entity_repository
):
"""Fast PUT updates the existing external_id without creating duplicates."""
"""PUT updates the existing external_id without creating duplicates."""
create_data = {
"title": "07 - Get Started",
"directory": "docs",
@@ -405,10 +405,10 @@ async def test_update_entity_by_id_fast_does_not_duplicate(
@pytest.mark.asyncio
async def test_put_entity_fast_returns_minimal_row(
async def test_put_entity_with_fast_param_returns_fully_indexed_row(
client: AsyncClient, v2_project_url, entity_repository
):
"""Fast PUT returns a minimal row and persists the external_id immediately."""
"""PUT ignores the legacy fast param and still returns a fully indexed row."""
external_id = str(uuid.uuid4())
update_data = {
"title": "FastPutEntity",
@@ -431,18 +431,19 @@ async def test_put_entity_fast_returns_minimal_row(
assert response.status_code == 201
created_entity = EntityResponseV2.model_validate(response.json())
assert created_entity.external_id == external_id
assert created_entity.observations == []
assert created_entity.relations == []
assert len(created_entity.observations) == 1
assert len(created_entity.relations) == 1
db_entity = await entity_repository.get_by_external_id(external_id)
assert db_entity is not None
@pytest.mark.asyncio
async def test_fast_create_schedules_reindex_task(
client: AsyncClient, v2_project_url, task_scheduler_spy
async def test_create_with_fast_param_does_not_schedule_reindex_task(
client: AsyncClient, v2_project_url, task_scheduler_spy, app_config
):
"""Fast create should enqueue a background reindex task."""
"""Legacy fast=true should not resurrect the removed reindex note-write path."""
app_config.semantic_search_enabled = False
start_count = len(task_scheduler_spy)
response = await client.post(
f"{v2_project_url}/knowledge/entities",
@@ -454,18 +455,14 @@ async def test_fast_create_schedules_reindex_task(
params={"fast": True},
)
assert response.status_code == 200
assert len(task_scheduler_spy) == start_count + 1
created_entity = EntityResponseV2.model_validate(response.json())
scheduled = task_scheduler_spy[-1]
assert scheduled["task_name"] == "reindex_entity"
assert scheduled["payload"]["entity_id"] == created_entity.id
assert len(task_scheduler_spy) == start_count
@pytest.mark.asyncio
async def test_non_fast_create_schedules_vector_sync_when_semantic_enabled(
async def test_create_schedules_vector_sync_when_semantic_enabled(
client: AsyncClient, v2_project_url, task_scheduler_spy, app_config
):
"""Non-fast create should schedule vector sync when semantic mode is enabled."""
"""Create should schedule vector sync when semantic mode is enabled."""
app_config.semantic_search_enabled = True
start_count = len(task_scheduler_spy)
@@ -488,10 +485,10 @@ async def test_non_fast_create_schedules_vector_sync_when_semantic_enabled(
@pytest.mark.asyncio
async def test_non_fast_create_skips_vector_sync_when_semantic_disabled(
async def test_create_skips_vector_sync_when_semantic_disabled(
client: AsyncClient, v2_project_url, task_scheduler_spy, app_config
):
"""Non-fast create should not schedule vector sync when semantic mode is disabled."""
"""Create should not schedule vector sync when semantic mode is disabled."""
app_config.semantic_search_enabled = False
start_count = len(task_scheduler_spy)
tests/api/v2/test_knowledge_router_telemetry.py
+9 -61
View File
@@ -9,7 +9,7 @@ from types import SimpleNamespace
from typing import Any, cast
import pytest
from fastapi import BackgroundTasks, Response
from fastapi import Response
from basic_memory.schemas.base import Entity
from basic_memory.schemas.request import EditEntityRequest
@@ -48,14 +48,12 @@ def _fake_entity(*, external_id: str = "entity-123", file_path: str = "notes/tes
)
def _assert_names_in_order(names: list[str], expected: list[str]) -> None:
cursor = 0
for expected_name in expected:
cursor = names.index(expected_name, cursor) + 1
def _assert_only_root_span(spans: list[tuple[str, dict]], expected_name: str) -> None:
assert [name for name, _ in spans] == [expected_name]
@pytest.mark.asyncio
async def test_create_entity_emits_root_and_nested_spans(monkeypatch) -> None:
async def test_create_entity_emits_only_root_span(monkeypatch) -> None:
spans, fake_span = _capture_spans()
monkeypatch.setattr(knowledge_router_module.telemetry, "span", fake_span)
@@ -81,10 +79,6 @@ async def test_create_entity_emits_root_and_nested_spans(monkeypatch) -> None:
def schedule(self, *args, **kwargs):
return None
class FakeFileService:
async def read_file_content(self, path):
raise AssertionError("non-fast create should not re-read file content")
result = await knowledge_router_module.create_entity(
project_id=123,
data=Entity(
@@ -94,30 +88,18 @@ async def test_create_entity_emits_root_and_nested_spans(monkeypatch) -> None:
content_type="text/markdown",
content="telemetry content",
),
background_tasks=BackgroundTasks(),
entity_service=cast(Any, FakeEntityService()),
search_service=cast(Any, FakeSearchService()),
task_scheduler=FakeTaskScheduler(),
file_service=cast(Any, FakeFileService()),
app_config=cast(Any, SimpleNamespace(semantic_search_enabled=False)),
fast=False,
)
assert result.content == response_content
_assert_names_in_order(
[name for name, _ in spans],
[
"api.request.knowledge.create_entity",
"api.knowledge.create_entity.write_entity",
"api.knowledge.create_entity.search_index",
"api.knowledge.create_entity.vector_sync",
"api.knowledge.create_entity.read_content",
],
)
_assert_only_root_span(spans, "api.request.knowledge.create_entity")
@pytest.mark.asyncio
async def test_update_entity_emits_root_and_nested_spans(monkeypatch) -> None:
async def test_update_entity_emits_only_root_span(monkeypatch) -> None:
spans, fake_span = _capture_spans()
monkeypatch.setattr(knowledge_router_module.telemetry, "span", fake_span)
@@ -145,10 +127,6 @@ async def test_update_entity_emits_root_and_nested_spans(monkeypatch) -> None:
def schedule(self, *args, **kwargs):
return None
class FakeFileService:
async def read_file_content(self, path):
raise AssertionError("non-fast update should not re-read file content")
response = Response()
result = await knowledge_router_module.update_entity_by_id(
data=Entity(
@@ -159,34 +137,21 @@ async def test_update_entity_emits_root_and_nested_spans(monkeypatch) -> None:
content="updated telemetry content",
),
response=response,
background_tasks=BackgroundTasks(),
project_id=123,
entity_service=cast(Any, FakeEntityService()),
search_service=cast(Any, FakeSearchService()),
entity_repository=cast(Any, FakeEntityRepository()),
task_scheduler=FakeTaskScheduler(),
file_service=cast(Any, FakeFileService()),
app_config=cast(Any, SimpleNamespace(semantic_search_enabled=False)),
entity_id=entity.external_id,
fast=False,
)
assert result.content == response_content
_assert_names_in_order(
[name for name, _ in spans],
[
"api.request.knowledge.update_entity",
"api.knowledge.update_entity.load_entity",
"api.knowledge.update_entity.write_entity",
"api.knowledge.update_entity.search_index",
"api.knowledge.update_entity.vector_sync",
"api.knowledge.update_entity.read_content",
],
)
_assert_only_root_span(spans, "api.request.knowledge.update_entity")
@pytest.mark.asyncio
async def test_edit_entity_emits_root_and_nested_spans(monkeypatch) -> None:
async def test_edit_entity_emits_only_root_span(monkeypatch) -> None:
spans, fake_span = _capture_spans()
monkeypatch.setattr(knowledge_router_module.telemetry, "span", fake_span)
@@ -214,33 +179,16 @@ async def test_edit_entity_emits_root_and_nested_spans(monkeypatch) -> None:
def schedule(self, *args, **kwargs):
return None
class FakeFileService:
async def read_file_content(self, path):
raise AssertionError("non-fast edit should not re-read file content")
result = await knowledge_router_module.edit_entity_by_id(
data=EditEntityRequest(operation="append", content="edited telemetry content"),
background_tasks=BackgroundTasks(),
project_id=123,
entity_service=cast(Any, FakeEntityService()),
search_service=cast(Any, FakeSearchService()),
entity_repository=cast(Any, FakeEntityRepository()),
task_scheduler=FakeTaskScheduler(),
file_service=cast(Any, FakeFileService()),
app_config=cast(Any, SimpleNamespace(semantic_search_enabled=False)),
entity_id=entity.external_id,
fast=False,
)
assert result.content == response_content
_assert_names_in_order(
[name for name, _ in spans],
[
"api.request.knowledge.edit_entity",
"api.knowledge.edit_entity.load_entity",
"api.knowledge.edit_entity.write_entity",
"api.knowledge.edit_entity.search_index",
"api.knowledge.edit_entity.vector_sync",
"api.knowledge.edit_entity.read_content",
],
)
_assert_only_root_span(spans, "api.request.knowledge.edit_entity")
tests/mcp/clients/test_clients.py
+61
View File
@@ -44,6 +44,7 @@ class TestKnowledgeClient:
async def mock_call_post(client, url, **kwargs):
assert "/v2/projects/proj-123/knowledge/entities" in url
assert kwargs.get("params") is None
return mock_response
monkeypatch.setattr(knowledge_mod, "call_post", mock_call_post)
@@ -53,6 +54,66 @@ class TestKnowledgeClient:
result = await client.create_entity({"title": "Test"})
assert result.title == "Test"
@pytest.mark.asyncio
async def test_update_entity(self, monkeypatch):
"""Test update_entity calls correct endpoint without fast query params."""
from basic_memory.mcp.clients import knowledge as knowledge_mod
mock_response = MagicMock()
mock_response.json.return_value = {
"permalink": "test",
"title": "Test",
"file_path": "test.md",
"note_type": "note",
"content_type": "text/markdown",
"observations": [],
"relations": [],
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
}
async def mock_call_put(client, url, **kwargs):
assert "/v2/projects/proj-123/knowledge/entities/entity-123" in url
assert kwargs.get("params") is None
return mock_response
monkeypatch.setattr(knowledge_mod, "call_put", mock_call_put)
mock_http = MagicMock()
client = KnowledgeClient(mock_http, "proj-123")
result = await client.update_entity("entity-123", {"title": "Test"})
assert result.title == "Test"
@pytest.mark.asyncio
async def test_patch_entity(self, monkeypatch):
"""Test patch_entity calls correct endpoint without fast query params."""
from basic_memory.mcp.clients import knowledge as knowledge_mod
mock_response = MagicMock()
mock_response.json.return_value = {
"permalink": "test",
"title": "Test",
"file_path": "test.md",
"note_type": "note",
"content_type": "text/markdown",
"observations": [],
"relations": [],
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
}
async def mock_call_patch(client, url, **kwargs):
assert "/v2/projects/proj-123/knowledge/entities/entity-123" in url
assert kwargs.get("params") is None
return mock_response
monkeypatch.setattr(knowledge_mod, "call_patch", mock_call_patch)
mock_http = MagicMock()
client = KnowledgeClient(mock_http, "proj-123")
result = await client.patch_entity("entity-123", {"operation": "append"})
assert result.title == "Test"
@pytest.mark.asyncio
async def test_resolve_entity(self, monkeypatch):
"""Test resolve_entity returns external_id."""
tests/services/test_entity_service.py
-149
View File
@@ -514,69 +514,6 @@ async def test_update_note_entity_content(entity_service: EntityService, file_se
assert metadata.get("status") == "draft"
@pytest.mark.asyncio
async def test_fast_write_and_reindex_entity(
entity_repository: EntityRepository,
observation_repository,
relation_repository,
entity_parser: EntityParser,
file_service: FileService,
link_resolver,
search_service: SearchService,
app_config: BasicMemoryConfig,
):
"""Fast write should defer observations/relations until reindex."""
service = EntityService(
entity_repository=entity_repository,
observation_repository=observation_repository,
relation_repository=relation_repository,
entity_parser=entity_parser,
file_service=file_service,
link_resolver=link_resolver,
search_service=search_service,
app_config=app_config,
)
schema = EntitySchema(
title="Reindex Target",
directory="test",
note_type="note",
content=dedent("""
# Reindex Target
- [note] Deferred observation
- relates_to [[Other Entity]]
""").strip(),
)
external_id = str(uuid.uuid4())
fast_entity = await service.fast_write_entity(schema, external_id=external_id)
assert fast_entity.external_id == external_id
assert len(fast_entity.observations) == 0
assert len(fast_entity.relations) == 0
await service.reindex_entity(fast_entity.id)
reindexed = await entity_repository.get_by_external_id(external_id)
assert reindexed is not None
assert len(reindexed.observations) == 1
assert len(reindexed.relations) == 1
@pytest.mark.asyncio
async def test_fast_write_entity_generates_external_id(entity_service: EntityService):
"""Fast write should generate an external_id when one is not provided."""
title = f"Fast Write {uuid.uuid4()}"
schema = EntitySchema(
title=title,
directory="test",
note_type="note",
)
fast_entity = await entity_service.fast_write_entity(schema)
assert fast_entity.external_id
@pytest.mark.asyncio
async def test_create_or_update_new(entity_service: EntityService, file_service: FileService):
"""Should create a new entity."""
@@ -2499,92 +2436,6 @@ async def test_update_preserves_created_by(entity_service: EntityService):
assert updated.last_updated_by == editor_id # updated
@pytest.mark.asyncio
async def test_fast_write_entity_sets_user_tracking(entity_service: EntityService):
"""fast_write_entity sets created_by and last_updated_by on create."""
user_id = str(uuid.uuid4())
entity_service.get_user_id = lambda: user_id
schema = EntitySchema(
title="Fast Write Tracked",
directory="test",
entity_type="note",
)
entity = await entity_service.fast_write_entity(schema, external_id=str(uuid.uuid4()))
assert entity.created_by == user_id
assert entity.last_updated_by == user_id
@pytest.mark.asyncio
async def test_fast_write_entity_update_preserves_created_by(entity_service: EntityService):
"""fast_write_entity update path preserves created_by, sets last_updated_by."""
creator_id = str(uuid.uuid4())
editor_id = str(uuid.uuid4())
external_id = str(uuid.uuid4())
# Create
entity_service.get_user_id = lambda: creator_id
schema = EntitySchema(
title="Fast Write Update",
directory="test",
entity_type="note",
)
entity = await entity_service.fast_write_entity(schema, external_id=external_id)
assert entity.created_by == creator_id
# Update (same external_id triggers update path)
entity_service.get_user_id = lambda: editor_id
update_schema = EntitySchema(
title="Fast Write Update",
directory="test",
entity_type="note",
content="Updated",
)
updated = await entity_service.fast_write_entity(update_schema, external_id=external_id)
assert updated.created_by == creator_id # preserved
assert updated.last_updated_by == editor_id # updated
@pytest.mark.asyncio
async def test_fast_edit_entity_sets_last_updated_by(entity_service: EntityService):
"""fast_edit_entity sets last_updated_by on edit."""
creator_id = str(uuid.uuid4())
editor_id = str(uuid.uuid4())
# Create entity first
entity_service.get_user_id = lambda: creator_id
schema = EntitySchema(
title="Fast Edit Tracked",
directory="test",
entity_type="note",
content="Original content",
)
entity = await entity_service.fast_write_entity(schema, external_id=str(uuid.uuid4()))
# Edit as different user
entity_service.get_user_id = lambda: editor_id
edited = await entity_service.fast_edit_entity(
entity=entity,
operation="append",
content="\nAppended content",
)
assert edited.created_by == creator_id # preserved
assert edited.last_updated_by == editor_id # updated
@pytest.mark.asyncio
async def test_fast_write_entity_null_user_id(entity_service: EntityService):
"""fast_write_entity with default get_user_id (None) leaves tracking fields null."""
schema = EntitySchema(
title="No User Tracking",
directory="test",
entity_type="note",
)
entity = await entity_service.fast_write_entity(schema, external_id=str(uuid.uuid4()))
assert entity.created_by is None
assert entity.last_updated_by is None
# --- Concurrent Delete Resilience ---
tests/services/test_entity_service_prepare.py
+334
View File
@@ -0,0 +1,334 @@
"""Parity tests for prepare-first entity write semantics."""
from __future__ import annotations
import pytest
from basic_memory.file_utils import ParseError, parse_frontmatter, remove_frontmatter
from basic_memory.schemas import Entity as EntitySchema
from basic_memory.services.exceptions import EntityAlreadyExistsError
@pytest.mark.asyncio
async def test_prepare_create_entity_content_matches_create_entity_with_content(
entity_service,
) -> None:
schema = EntitySchema(
title="Prepared Create",
directory="notes",
note_type="note",
content="---\nstatus: draft\npermalink: prepared/create\n---\nCreate body",
)
prepared = await entity_service.prepare_create_entity_content(schema)
result = await entity_service.create_entity_with_content(schema)
assert prepared.file_path.as_posix() == result.entity.file_path
assert prepared.markdown_content == result.content
assert prepared.search_content == result.search_content
assert prepared.entity_fields["title"] == result.entity.title
assert prepared.entity_fields["note_type"] == result.entity.note_type
assert prepared.entity_fields["permalink"] == result.entity.permalink
assert prepared.entity_fields["entity_metadata"] == result.entity.entity_metadata
@pytest.mark.asyncio
async def test_prepare_create_entity_content_can_skip_storage_existence_check(
entity_service,
) -> None:
async def fail_if_called(*args, **kwargs):
raise AssertionError("file_service.exists should not be called")
entity_service.file_service.exists = fail_if_called
prepared = await entity_service.prepare_create_entity_content(
EntitySchema(
title="Prepared Create No HEAD",
directory="notes",
note_type="note",
content="Create body",
),
check_storage_exists=False,
)
assert prepared.file_path.as_posix() == "notes/Prepared Create No HEAD.md"
assert prepared.entity_fields["title"] == "Prepared Create No HEAD"
@pytest.mark.asyncio
async def test_prepare_update_entity_content_matches_update_entity_with_content(
entity_service,
file_service,
) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Prepared Update",
directory="notes",
note_type="note",
content="---\nstatus: draft\nowner: alice\n---\nOriginal body",
)
)
existing_content = await file_service.read_file_content(created.file_path)
update_schema = EntitySchema(
title="Prepared Update",
directory="notes",
note_type="note",
content="---\nstatus: published\nreviewed_by: bob\n---\nUpdated body",
)
prepared = await entity_service.prepare_update_entity_content(
created,
update_schema,
existing_content,
)
result = await entity_service.update_entity_with_content(created, update_schema)
prepared_frontmatter = parse_frontmatter(prepared.markdown_content)
assert prepared.markdown_content == result.content
assert prepared.search_content == result.search_content
assert prepared.entity_fields["title"] == result.entity.title
assert prepared.entity_fields["note_type"] == result.entity.note_type
assert prepared.entity_fields["permalink"] == result.entity.permalink
assert prepared_frontmatter["owner"] == "alice"
assert prepared_frontmatter["status"] == "published"
assert prepared_frontmatter["reviewed_by"] == "bob"
@pytest.mark.asyncio
async def test_prepare_update_entity_content_can_change_file_path(
entity_service,
file_service,
) -> None:
"""Full replacements should carry title/directory renames through prepare state."""
created = await entity_service.create_entity(
EntitySchema(
title="Original Name",
directory="notes",
note_type="note",
content="Original body",
)
)
existing_content = await file_service.read_file_content(created.file_path)
update_schema = EntitySchema(
title="Renamed Note",
directory="journal",
note_type="note",
content="Renamed body",
)
prepared = await entity_service.prepare_update_entity_content(
created,
update_schema,
existing_content,
)
result = await entity_service.update_entity_with_content(created, update_schema)
assert prepared.file_path.as_posix() == "journal/Renamed Note.md"
assert result.entity.file_path == "journal/Renamed Note.md"
assert result.content == prepared.markdown_content
assert prepared.entity_fields["permalink"] != created.permalink
assert prepared.entity_fields["permalink"] == result.entity.permalink
assert not await file_service.exists("notes/Original Name.md")
assert await file_service.exists("journal/Renamed Note.md")
@pytest.mark.asyncio
async def test_prepare_update_entity_content_preserves_permalink_when_move_updates_disabled(
entity_service,
file_service,
) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Stable Permalink",
directory="notes",
note_type="note",
content="Original body",
)
)
entity_service.app_config.update_permalinks_on_move = False
existing_content = await file_service.read_file_content(created.file_path)
update_schema = EntitySchema(
title="Renamed Stable Permalink",
directory="journal",
note_type="note",
content="Renamed body",
)
prepared = await entity_service.prepare_update_entity_content(
created,
update_schema,
existing_content,
)
result = await entity_service.update_entity_with_content(created, update_schema)
prepared_frontmatter = parse_frontmatter(prepared.markdown_content)
assert prepared.file_path.as_posix() == "journal/Renamed Stable Permalink.md"
assert result.entity.file_path == "journal/Renamed Stable Permalink.md"
assert prepared.entity_fields["permalink"] == created.permalink
assert result.entity.permalink == created.permalink
assert prepared_frontmatter["permalink"] == created.permalink
assert not await file_service.exists("notes/Stable Permalink.md")
assert await file_service.exists("journal/Renamed Stable Permalink.md")
@pytest.mark.asyncio
async def test_update_entity_with_content_rejects_rename_conflicts_before_writing(
entity_service,
file_service,
) -> None:
source = await entity_service.create_entity(
EntitySchema(
title="Source Note",
directory="notes",
note_type="note",
content="Source body",
)
)
target = await entity_service.create_entity(
EntitySchema(
title="Target Note",
directory="notes",
note_type="note",
content="Target body",
)
)
source_content = await file_service.read_file_content(source.file_path)
target_content = await file_service.read_file_content(target.file_path)
with pytest.raises(
EntityAlreadyExistsError,
match="file already exists at destination path: notes/Target Note.md",
):
await entity_service.update_entity_with_content(
source,
EntitySchema(
title="Target Note",
directory="notes",
note_type="note",
content="Overwritten body",
),
)
assert await file_service.read_file_content(source.file_path) == source_content
assert await file_service.read_file_content(target.file_path) == target_content
assert await file_service.exists(source.file_path)
assert await file_service.exists(target.file_path)
@pytest.mark.asyncio
async def test_prepare_edit_entity_content_matches_edit_entity_with_content(
entity_service,
file_service,
) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Prepared Edit",
directory="notes",
note_type="note",
content="Before edit",
)
)
current_content = await file_service.read_file_content(created.file_path)
prepared = await entity_service.prepare_edit_entity_content(
created,
current_content,
operation="find_replace",
content="After edit",
find_text="Before edit",
)
result = await entity_service.edit_entity_with_content(
identifier=created.permalink,
operation="find_replace",
content="After edit",
find_text="Before edit",
)
assert prepared.markdown_content == result.content
assert prepared.search_content == result.search_content
assert prepared.entity_fields["title"] == result.entity.title
assert prepared.entity_fields["note_type"] == result.entity.note_type
assert prepared.entity_fields["permalink"] == result.entity.permalink
@pytest.mark.asyncio
async def test_prepare_edit_entity_content_prepend_preserves_valid_frontmatter(
entity_service,
file_service,
) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Prepared Prepend Frontmatter",
directory="notes",
note_type="note",
content="---\nstatus: draft\ntags:\n - one\n---\nOriginal body",
)
)
current_content = await file_service.read_file_content(created.file_path)
prepared = await entity_service.prepare_edit_entity_content(
created,
current_content,
operation="prepend",
content="Prepended line",
)
assert parse_frontmatter(prepared.markdown_content) == {
"title": "Prepared Prepend Frontmatter",
"type": "note",
"status": "draft",
"tags": ["one"],
"permalink": created.permalink,
}
assert remove_frontmatter(prepared.markdown_content) == "Prepended line\nOriginal body"
@pytest.mark.asyncio
async def test_prepare_edit_entity_content_prepend_fails_for_malformed_frontmatter(
entity_service,
) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Prepared Prepend Parse Error",
directory="notes",
note_type="note",
content="Original body",
)
)
malformed_content = "---\nstatus: [draft\n---\nOriginal body"
with pytest.raises(ParseError, match="Invalid YAML in frontmatter"):
await entity_service.prepare_edit_entity_content(
created,
malformed_content,
operation="prepend",
content="Prepended line",
)
@pytest.mark.asyncio
async def test_prepare_edit_entity_content_prepend_without_frontmatter_uses_simple_prepend(
entity_service,
) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Prepared Prepend Simple",
directory="notes",
note_type="note",
content="Original body",
)
)
prepared = await entity_service.prepare_edit_entity_content(
created,
"Original body",
operation="prepend",
content="Prepended line",
)
assert prepared.markdown_content == "Prepended line\nOriginal body"
tests/services/test_entity_service_telemetry.py
+13 -70
View File
@@ -1,4 +1,4 @@
"""Telemetry coverage for entity service write/edit/reindex paths."""
"""Telemetry coverage for the lower-level file spans used by entity service."""
from __future__ import annotations
@@ -9,7 +9,7 @@ import pytest
from basic_memory.schemas import Entity as EntitySchema
entity_service_module = importlib.import_module("basic_memory.services.entity_service")
telemetry_module = importlib.import_module("basic_memory.telemetry")
def _capture_spans():
@@ -23,16 +23,14 @@ def _capture_spans():
return spans, fake_span
def _assert_names_in_order(names: list[str], expected: list[str]) -> None:
cursor = 0
for expected_name in expected:
cursor = names.index(expected_name, cursor) + 1
def _span_names(spans: list[tuple[str, dict]]) -> list[str]:
return [name for name, _ in spans]
@pytest.mark.asyncio
async def test_create_entity_emits_expected_phase_spans(entity_service, monkeypatch) -> None:
async def test_create_entity_emits_file_write_span(entity_service, monkeypatch) -> None:
spans, fake_span = _capture_spans()
monkeypatch.setattr(entity_service_module.telemetry, "span", fake_span)
monkeypatch.setattr(telemetry_module, "span", fake_span)
schema = EntitySchema(
title="Telemetry Create",
@@ -45,22 +43,11 @@ async def test_create_entity_emits_expected_phase_spans(entity_service, monkeypa
entity = await entity_service.create_entity(schema)
assert entity.title == "Telemetry Create"
span_names = [name for name, _ in spans]
_assert_names_in_order(
span_names,
[
"entity_service.create.resolve_permalink",
"entity_service.create.write_file",
"file_service.write",
"entity_service.create.parse_markdown",
"entity_service.create.upsert_entity",
"entity_service.create.update_checksum",
],
)
assert "file_service.write" in _span_names(spans)
@pytest.mark.asyncio
async def test_edit_entity_emits_expected_phase_spans(entity_service, monkeypatch) -> None:
async def test_edit_entity_emits_file_read_and_write_spans(entity_service, monkeypatch) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Telemetry Edit",
@@ -72,7 +59,7 @@ async def test_edit_entity_emits_expected_phase_spans(entity_service, monkeypatc
)
spans, fake_span = _capture_spans()
monkeypatch.setattr(entity_service_module.telemetry, "span", fake_span)
monkeypatch.setattr(telemetry_module, "span", fake_span)
updated = await entity_service.edit_entity(
created.file_path,
@@ -81,51 +68,7 @@ async def test_edit_entity_emits_expected_phase_spans(entity_service, monkeypatc
)
assert updated.id == created.id
span_names = [name for name, _ in spans]
_assert_names_in_order(
span_names,
[
"entity_service.edit.resolve_entity",
"entity_service.edit.read_file",
"file_service.read",
"entity_service.edit.apply_operation",
"entity_service.edit.write_file",
"file_service.write",
"entity_service.edit.parse_markdown",
"entity_service.edit.upsert_entity",
"entity_service.edit.update_checksum",
],
)
@pytest.mark.asyncio
async def test_reindex_entity_emits_expected_phase_spans(entity_service, monkeypatch) -> None:
created = await entity_service.create_entity(
EntitySchema(
title="Telemetry Reindex",
directory="notes",
note_type="note",
content_type="text/markdown",
content="Reindex telemetry content",
)
)
spans, fake_span = _capture_spans()
monkeypatch.setattr(entity_service_module.telemetry, "span", fake_span)
await entity_service.reindex_entity(created.id)
span_names = [name for name, _ in spans]
_assert_names_in_order(
span_names,
[
"entity_service.reindex.load_entity",
"entity_service.reindex.read_file",
"file_service.read_content",
"entity_service.reindex.parse_markdown",
"entity_service.reindex.upsert_entity",
"entity_service.reindex.update_checksum",
],
)
if entity_service.search_service is not None:
assert "entity_service.reindex.search_index" in span_names
span_names = _span_names(spans)
assert "file_service.read" in span_names
assert "file_service.write" in span_names
assert span_names.index("file_service.read") < span_names.index("file_service.write")
tests/services/test_entity_service_write_result.py
+34
View File
@@ -1,10 +1,17 @@
"""Tests for EntityWriteResult content variants."""
import sys
import pytest
from basic_memory.file_utils import remove_frontmatter
from basic_memory.schemas import Entity as EntitySchema
skip_on_windows = pytest.mark.skipif(
sys.platform == "win32",
reason="formatter command uses POSIX shell redirection",
)
@pytest.mark.asyncio
async def test_create_entity_with_content_returns_full_and_search_content(
@@ -106,3 +113,30 @@ async def test_edit_entity_with_content_returns_full_and_search_content(
assert result.content == file_content
assert result.search_content == remove_frontmatter(file_content)
assert result.search_content == "Edited body content"
@skip_on_windows
@pytest.mark.asyncio
async def test_create_entity_with_content_returns_persisted_content_after_format_on_save(
entity_service, file_service, app_config
) -> None:
app_config.format_on_save = True
app_config.formatter_command = "sh -c 'echo modified > {file}'"
file_service.app_config = app_config
result = await entity_service.create_entity_with_content(
EntitySchema(
title="FormattedWriteResult",
directory="notes",
note_type="note",
content="Original body content",
)
)
file_path = file_service.get_entity_path(result.entity)
file_content, _ = await file_service.read_file(file_path)
assert file_content == "modified\n"
assert result.content == file_content
assert result.search_content == remove_frontmatter(file_content)
assert result.search_content == "modified"
tests/services/test_task_scheduler_semantic.py
+27 -71
View File
@@ -1,4 +1,4 @@
"""Task scheduler semantic indexing tests."""
"""Task scheduler tests for derived async work."""
import asyncio
from pathlib import Path
@@ -10,14 +10,6 @@ from basic_memory.config import BasicMemoryConfig, ProjectConfig
from basic_memory.deps.services import get_task_scheduler
class StubEntityService:
def __init__(self) -> None:
self.reindexed: list[int] = []
async def reindex_entity(self, entity_id: int) -> None:
self.reindexed.append(entity_id)
class StubSyncService:
def __init__(self) -> None:
self.resolved: list[int] = []
@@ -42,70 +34,9 @@ class StubSearchService:
self.reindexed_project = True
@pytest.mark.asyncio
async def test_reindex_entity_task_chains_vector_sync_when_semantic_enabled(tmp_path):
"""Reindex task should enqueue vector sync when semantic mode is enabled."""
entity_service = StubEntityService()
sync_service = StubSyncService()
search_service = StubSearchService()
app_config = BasicMemoryConfig(
env="test",
projects={"test-project": str(tmp_path)},
default_project="test-project",
semantic_search_enabled=True,
)
project_config = ProjectConfig(name="test-project", home=tmp_path)
scheduler = await get_task_scheduler(
entity_service=cast(Any, entity_service),
sync_service=cast(Any, sync_service),
search_service=cast(Any, search_service),
project_config=project_config,
app_config=app_config,
)
# Enable background tasks for this test — uses stubs, no real DB race risk
cast(Any, scheduler)._test_mode = False
scheduler.schedule("reindex_entity", entity_id=42)
await asyncio.sleep(0.05)
assert entity_service.reindexed == [42]
assert search_service.vector_synced == [42]
@pytest.mark.asyncio
async def test_reindex_entity_task_skips_vector_sync_when_semantic_disabled(tmp_path):
"""Reindex task should not enqueue vector sync when semantic mode is disabled."""
entity_service = StubEntityService()
sync_service = StubSyncService()
search_service = StubSearchService()
app_config = BasicMemoryConfig(
env="test",
projects={"test-project": str(tmp_path)},
default_project="test-project",
semantic_search_enabled=False,
)
project_config = ProjectConfig(name="test-project", home=tmp_path)
scheduler = await get_task_scheduler(
entity_service=cast(Any, entity_service),
sync_service=cast(Any, sync_service),
search_service=cast(Any, search_service),
project_config=project_config,
app_config=app_config,
)
# Enable background tasks for this test — uses stubs, no real DB race risk
cast(Any, scheduler)._test_mode = False
scheduler.schedule("reindex_entity", entity_id=42)
await asyncio.sleep(0.05)
assert entity_service.reindexed == [42]
assert search_service.vector_synced == []
@pytest.mark.asyncio
async def test_sync_entity_vectors_task_maps_to_search_service(tmp_path):
"""Explicit sync_entity_vectors task should call SearchService sync method."""
entity_service = StubEntityService()
sync_service = StubSyncService()
search_service = StubSearchService()
app_config = BasicMemoryConfig(
@@ -117,7 +48,6 @@ async def test_sync_entity_vectors_task_maps_to_search_service(tmp_path):
project_config = ProjectConfig(name="test-project", home=tmp_path)
scheduler = await get_task_scheduler(
entity_service=cast(Any, entity_service),
sync_service=cast(Any, sync_service),
search_service=cast(Any, search_service),
project_config=project_config,
@@ -129,3 +59,29 @@ async def test_sync_entity_vectors_task_maps_to_search_service(tmp_path):
await asyncio.sleep(0.05)
assert search_service.vector_synced == [7]
@pytest.mark.asyncio
async def test_sync_project_task_maps_to_sync_service(tmp_path):
"""Explicit sync_project task should call SyncService sync method."""
sync_service = StubSyncService()
search_service = StubSearchService()
app_config = BasicMemoryConfig(
env="test",
projects={"test-project": str(tmp_path)},
default_project="test-project",
semantic_search_enabled=True,
)
project_config = ProjectConfig(name="test-project", home=tmp_path)
scheduler = await get_task_scheduler(
sync_service=cast(Any, sync_service),
search_service=cast(Any, search_service),
project_config=project_config,
app_config=app_config,
)
cast(Any, scheduler)._test_mode = False
scheduler.schedule("sync_project", force_full=True)
await asyncio.sleep(0.05)
assert sync_service.synced == [(str(tmp_path), "test-project", True)]
tests/services/test_upsert_entity_optimization.py
-90
View File
@@ -9,8 +9,6 @@ Verifies that:
from __future__ import annotations
import importlib
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
@@ -27,9 +25,6 @@ from basic_memory.markdown.schemas import (
from basic_memory.schemas import Entity as EntitySchema
from basic_memory.services.entity_service import EntityService
entity_service_module = importlib.import_module("basic_memory.services.entity_service")
# --- Helpers ---
@@ -48,17 +43,6 @@ def _make_markdown(
)
def _capture_spans():
spans: list[tuple[str, dict]] = []
@contextmanager
def fake_span(name: str, **attrs):
spans.append((name, attrs))
yield
return spans, fake_span
# --- Optimization 1: No redundant get_by_file_path in update_entity_relations ---
@@ -167,80 +151,6 @@ async def test_create_or_update_entity_uses_lightweight_exact_resolution(
]
# --- Telemetry sub-spans ---
@pytest.mark.asyncio
async def test_upsert_update_emits_sub_spans(entity_service: EntityService, monkeypatch):
"""upsert_entity_from_markdown (update path) should emit sub-spans for each DB phase."""
entity = await entity_service.create_entity(
EntitySchema(
title="Span Test",
directory="notes",
note_type="note",
content="# Span Test\n\n## Observations\n- [fact] original",
)
)
spans, fake_span = _capture_spans()
monkeypatch.setattr(entity_service_module.telemetry, "span", fake_span)
markdown = _make_markdown(
title="Span Test",
observations=[MarkdownObservation(content="updated", category="fact")],
)
await entity_service.upsert_entity_from_markdown(Path(entity.file_path), markdown, is_new=False)
span_names = [name for name, _ in spans]
# update_entity_and_observations sub-spans
assert "upsert.update.fetch_entity" in span_names
assert "upsert.update.delete_observations" in span_names
assert "upsert.update.insert_observations" in span_names
assert "upsert.update.save_entity" in span_names
# update_entity_relations sub-spans
assert "upsert.relations.delete_existing" in span_names
assert "upsert.relations.reload_entity" in span_names
@pytest.mark.asyncio
async def test_upsert_with_relations_emits_resolve_and_insert_spans(
entity_service: EntityService, monkeypatch
):
"""When relations exist, resolve_links and insert_relations spans should be emitted."""
# Create two entities so the relation can resolve
await entity_service.create_entity(
EntitySchema(
title="Target Entity",
directory="notes",
note_type="note",
content="# Target Entity",
)
)
source = await entity_service.create_entity(
EntitySchema(
title="Source Entity",
directory="notes",
note_type="note",
content="# Source Entity",
)
)
spans, fake_span = _capture_spans()
monkeypatch.setattr(entity_service_module.telemetry, "span", fake_span)
markdown = _make_markdown(
title="Source Entity",
relations=[MarkdownRelation(type="links_to", target="Target Entity")],
)
await entity_service.upsert_entity_from_markdown(Path(source.file_path), markdown, is_new=False)
span_names = [name for name, _ in spans]
assert "upsert.relations.resolve_links" in span_names
assert "upsert.relations.insert_relations" in span_names
@pytest.mark.asyncio
async def test_upsert_with_relations_uses_lightweight_exact_resolution(
entity_service: EntityService, monkeypatch