update_entity

src/basic_memory/api/routers/knowledge_router.py

@@ -21,6 +21,7 @@ from basic_memory.schemas import (
     DeleteObservationsRequest,
     DeleteRelationsRequest,
     DeleteEntitiesRequest,
+    UpdateEntityRequest,
 )
 from basic_memory.schemas.base import PathId
 from basic_memory.services.exceptions import EntityNotFoundError
@@ -49,6 +50,31 @@ async def create_entities(
     )
 
 
+@router.put("/entities/{path_id:path}", response_model=EntityResponse)
+async def update_entity(
+    path_id: PathId,
+    data: UpdateEntityRequest,
+    background_tasks: BackgroundTasks,
+    knowledge_service: KnowledgeServiceDep,
+    search_service = Depends(get_search_service)
+) -> EntityResponse:
+    """Update an existing entity and reindex it."""
+    try:
+        # Convert request to dict, excluding None values
+        update_data = data.model_dump(exclude_none=True)
+        
+        # Update the entity
+        updated_entity = await knowledge_service.update_entity(path_id, update_data)
+        
+        # Reindex since content changed
+        await search_service.index_entity(updated_entity, background_tasks=background_tasks)
+        
+        return EntityResponse.model_validate(updated_entity)
+        
+    except EntityNotFoundError:
+        raise HTTPException(status_code=404, detail=f"Entity with {path_id} not found")
+
+
 @router.post("/relations", response_model=EntityListResponse)
 async def create_relations(
     data: CreateRelationsRequest,
@@ -100,7 +126,6 @@ async def get_entity(path_id: PathId, entity_service: EntityServiceDep) -> Entit
         raise HTTPException(status_code=404, detail=f"Entity with {path_id} not found")
 
 
-
 @router.post("/nodes", response_model=EntityListResponse)
 async def open_nodes(data: OpenNodesRequest, entity_service: EntityServiceDep) -> EntityListResponse:
     """Open specific nodes by their names."""

src/basic_memory/schemas/__init__.py

@@ -27,7 +27,7 @@ from basic_memory.schemas.request import (
     CreateEntityRequest,
     SearchNodesRequest,
     OpenNodesRequest,
-    CreateRelationsRequest,
+    CreateRelationsRequest, UpdateEntityRequest,
 )
 
 # Response models
@@ -61,6 +61,7 @@ __all__ = [
     "SearchNodesRequest",
     "OpenNodesRequest",
     "CreateRelationsRequest",
+    "UpdateEntityRequest",
     # Responses
     "SQLAlchemyModel",
     "ObservationResponse",

src/basic_memory/schemas/base.py

@@ -251,6 +251,7 @@ class Entity(BaseModel):
     name: str
     entity_type: EntityType
     entity_metadata: Optional[Dict] = Field(default=None, description="Optional metadata")
+    content: Optional[str] = None
     description: Optional[str] = None
     observations: List[Observation] = []
 

src/basic_memory/schemas/request.py

@@ -5,7 +5,7 @@ from annotated_types import MaxLen, MinLen
 
 from pydantic import BaseModel, StringConstraints
 
-from basic_memory.schemas.base import Observation, Entity, Relation, PathId, ObservationCategory
+from basic_memory.schemas.base import Observation, Entity, Relation, PathId, ObservationCategory, EntityType
 
 
 class ObservationCreate(BaseModel):
@@ -88,7 +88,15 @@ class CreateRelationsRequest(BaseModel):
     relations: List[Relation]
 
 
-## document
+## update
+
+class UpdateEntityRequest(BaseModel):
+    """Request to update an existing entity."""
+    name: Optional[str] = None
+    entity_type: Optional[EntityType] = None
+    description: Optional[str] = None
+    content: Optional[str] = None
+    entity_metadata: Optional[Dict[str, Any]] = None
 
 
 DocumentPathId = Annotated[

src/basic_memory/services/knowledge/entity_operations.py

@@ -1,6 +1,7 @@
 """Entity operations for knowledge service."""
 
-from typing import Sequence, List
+from datetime import datetime, UTC
+from typing import Sequence, List, Dict, Any, Optional
 
 from loguru import logger
 
@@ -63,6 +64,72 @@ class EntityOperations:
             
         return created
 
+    async def update_entity(
+            self,
+            path_id: str,
+            content: Optional[str] = None,
+            metadata: Optional[Dict[str, Any]] = None,
+            **update_fields: Any
+    ) -> EntityModel:
+        """Update an entity's content and metadata.
+
+        Args:
+            path_id: Entity's path ID
+            content: Optional new content
+            metadata: Optional metadata updates
+            **update_fields: Additional entity fields to update
+
+        Returns:
+            Updated entity
+
+        Raises:
+            EntityNotFoundError: If entity doesn't exist
+        """
+        logger.debug(f"Updating entity with path_id: {path_id}")
+
+        # Get existing entity
+        entity = await self.entity_service.get_by_path_id(path_id)
+        if not entity:
+            raise EntityNotFoundError(f"Entity not found: {path_id}")
+
+        try:
+            # Build update data
+            update_data = {}
+
+            # Add any direct field updates
+            if update_fields:
+                update_data.update(update_fields)
+
+            # Handle metadata update
+            if metadata is not None:
+                # Update existing metadata
+                new_metadata = dict(entity.entity_metadata or {})
+                new_metadata.update(metadata)
+                update_data["entity_metadata"] = new_metadata
+
+            # Update entity in database if we have changes
+            if update_data:
+                entity = await self.entity_service.update_entity(
+                    entity.path_id, update_data
+                )
+
+            # Always write file if we have any updates
+            if update_data or content is not None:
+                _, checksum = await self.file_operations.write_entity_file(
+                    entity=entity,
+                    content=content
+                )
+                # Update checksum in DB
+                entity = await self.entity_service.update_entity(
+                    entity.path_id, {"checksum": checksum}
+                )
+
+            return entity
+
+        except Exception as e:
+            logger.error(f"Failed to update entity: {e}")
+            raise
+
     async def delete_entity(self, path_id: str) -> bool:
         """Delete entity and its file."""
         logger.debug(f"Deleting entity: {path_id}")

src/basic_memory/services/knowledge/knowledge_service.py

@@ -1,11 +1,11 @@
 """Main knowledge service implementation."""
 
 from pathlib import Path
-from typing import List, Sequence, Tuple
+from typing import List, Sequence, Tuple, Dict, Any, Optional
 
-from loguru import logger
 
 from basic_memory.markdown.knowledge_writer import KnowledgeWriter
+from basic_memory.markdown.note_writer import NoteWriter
 from basic_memory.models import Entity as EntityModel
 from basic_memory.schemas import Entity as EntitySchema
 from basic_memory.schemas import Relation as RelationSchema
@@ -19,7 +19,6 @@ from .file_operations import FileOperations
 from .entity_operations import EntityOperations
 from .relation_operations import RelationOperations
 from .observation_operations import ObservationOperations
-from ...markdown.note_writer import NoteWriter
 
 
 class KnowledgeService:
@@ -31,7 +30,7 @@ class KnowledgeService:
     - Entity CRUD operations
     - Relations between entities
     - Observations about entities
-    
+
     Acts as the main coordinator for all knowledge operations, ensuring
     consistency between database and filesystem.
     """
@@ -46,39 +45,37 @@ class KnowledgeService:
         note_writer: NoteWriter,
         base_path: Path,
     ):
-        
         self.base_path = base_path
-        
+
         # Initialize operations in dependency order
         self.file_ops = FileOperations(
             entity_service=entity_service,
             file_service=file_service,
             knowledge_writer=knowledge_writer,
             note_writer=note_writer,
-            base_path=base_path
+            base_path=base_path,
         )
 
         self.entity_ops = EntityOperations(
-            entity_service=entity_service,
-            file_operations=self.file_ops
+            entity_service=entity_service, file_operations=self.file_ops
         )
 
         self.relation_ops = RelationOperations(
             relation_service=relation_service,
             entity_service=entity_service,
-            file_operations=self.file_ops
+            file_operations=self.file_ops,
         )
 
         self.observation_ops = ObservationOperations(
             observation_service=observation_service,
             entity_service=entity_service,
-            file_operations=self.file_ops
+            file_operations=self.file_ops,
         )
 
     # Entity operations
     async def get_entity_by_path_id(self, path_id: str) -> EntityModel:
         return await self.entity_ops.get_by_path_id(path_id)
-    
+
     async def create_entity(self, entity: EntitySchema) -> EntityModel:
         """Create a new entity."""
         return await self.entity_ops.create_entity(entity)
@@ -87,6 +84,28 @@ class KnowledgeService:
         """Create multiple entities."""
         return await self.entity_ops.create_entities(entities)
 
+    async def update_entity(
+        self,
+        path_id: str,
+        content: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **update_fields: Any,
+    ) -> EntityModel:
+        """Update an entity's content and metadata.
+
+        Args:
+            path_id: Entity's path ID
+            content: Optional new content
+            metadata: Optional metadata updates
+            **update_fields: Additional entity fields to update
+
+        Returns:
+            Updated entity
+        """
+        return await self.entity_ops.update_entity(
+            path_id=path_id, content=content, metadata=metadata, **update_fields
+        )
+
     async def delete_entity(self, path_id: str) -> bool:
         """Delete an entity and its file."""
         return await self.entity_ops.delete_entity(path_id)
@@ -114,19 +133,12 @@ class KnowledgeService:
 
     # Observation operations
     async def add_observations(
-        self, 
-        path_id: str, 
-        observations: List[ObservationCreate], 
-        context: str | None = None
+        self, path_id: str, observations: List[ObservationCreate], context: str | None = None
     ) -> EntityModel:
         """Add observations to an entity."""
         return await self.observation_ops.add_observations(path_id, observations, context)
 
-    async def delete_observations(
-        self,
-        path_id: str, 
-        observations: List[str]
-    ) -> EntityModel:
+    async def delete_observations(self, path_id: str, observations: List[str]) -> EntityModel:
         """Delete observations from an entity."""
         return await self.observation_ops.delete_observations(path_id, observations)
 
@@ -138,4 +150,4 @@ class KnowledgeService:
     async def write_entity_file(self, entity: EntityModel) -> Path:
         """Write entity to filesystem."""
         path, _ = await self.file_ops.write_entity_file(entity)
-        return path
+        return path

tests/services/test_knowledge_service.py

@@ -1,26 +1,27 @@
 """Tests for KnowledgeService."""
 
 from pathlib import Path
-from typing import List
 
 import pytest
 import yaml
-from sqlalchemy.exc import IntegrityError
 
 from basic_memory.models import Entity as EntityModel
 from basic_memory.models.knowledge import EntityType
 from basic_memory.schemas import Entity as EntitySchema, Relation as RelationSchema
-from basic_memory.schemas.base import ObservationCategory
-from basic_memory.schemas.request import ObservationCreate
 from basic_memory.services import EntityService
-from basic_memory.services.exceptions import EntityNotFoundError, FileOperationError
 from basic_memory.services.knowledge import KnowledgeService
 
 
 @pytest.mark.asyncio
 async def test_get_entity_path(knowledge_service: KnowledgeService):
     """Should generate correct filesystem path for entity."""
-    entity = EntityModel(id=1, path_id="test-entity", name="test-entity", entity_type=EntityType.KNOWLEDGE, description="Test entity")
+    entity = EntityModel(
+        id=1,
+        path_id="test-entity",
+        name="test-entity",
+        entity_type=EntityType.KNOWLEDGE,
+        description="Test entity",
+    )
     path = knowledge_service.get_entity_path(entity)
     assert path == Path(knowledge_service.base_path / "test-entity.md")
 
@@ -29,7 +30,9 @@ async def test_get_entity_path(knowledge_service: KnowledgeService):
 async def test_create_entity(knowledge_service: KnowledgeService):
     """Should create entity in DB and write file correctly."""
     # Setup
-    entity_schema = EntitySchema(name="test-entity", entity_type=EntityType.KNOWLEDGE, description="Test entity")
+    entity_schema = EntitySchema(
+        name="test-entity", entity_type=EntityType.KNOWLEDGE, description="Test entity"
+    )
 
     # Execute
     created = await knowledge_service.create_entity(entity_schema)
@@ -41,7 +44,7 @@ async def test_create_entity(knowledge_service: KnowledgeService):
     assert created.checksum is not None
     assert created.path_id == "test_entity"
     assert created.file_path == "test_entity.md"
-    
+
     # Verify file was written
     file_path = knowledge_service.get_entity_path(created)
     assert await knowledge_service.file_exists(file_path)
@@ -57,12 +60,13 @@ async def test_create_entity(knowledge_service: KnowledgeService):
     assert "modified" in metadata
 
 
-
 @pytest.mark.asyncio
 async def test_create_multiple_entities(knowledge_service: KnowledgeService):
     """Should create multiple entities successfully."""
     entities = [
-        EntitySchema(name=f"entity-{i}", entity_type=EntityType.KNOWLEDGE, description=f"Test entity {i}")
+        EntitySchema(
+            name=f"entity-{i}", entity_type=EntityType.KNOWLEDGE, description=f"Test entity {i}"
+        )
         for i in range(3)
     ]
 
@@ -113,195 +117,125 @@ async def test_create_relations(knowledge_service: KnowledgeService, entity_serv
 
 
 @pytest.mark.asyncio
-async def test_add_observations_observation(knowledge_service: KnowledgeService):
-    """Should add observations and update entity file."""
+async def test_update_knowledge_entity_description(knowledge_service: KnowledgeService):
+    """Should update knowledge entity description and write to file."""
     # Create test entity
     entity = await knowledge_service.create_entity(
-        EntitySchema(name="test", entity_type=EntityType.KNOWLEDGE, description="Test entity")
+        EntitySchema(
+            name="test",
+            entity_type=EntityType.KNOWLEDGE,
+            description="Test entity",
+            entity_metadata={"status": "draft"},
+        )
     )
 
-    # Add observations
-    observations = [
-        ObservationCreate(content="Test observation 1", category=ObservationCategory.TECH),
-        ObservationCreate(content="Test observation 2", category=ObservationCategory.DESIGN),
-    ]
-    context = "Test context"
-    updated_entity = await knowledge_service.add_observations(
-        entity.path_id, observations, context
+    # Update description
+    updated = await knowledge_service.update_entity(
+        entity.path_id, description="Updated description"
     )
 
-    # Verify observations in DB
-    assert len(updated_entity.observations) == 2
-    assert updated_entity.observations[0].content == "Test observation 1"
-    assert updated_entity.observations[0].category == "tech"
-    assert updated_entity.observations[0].context == context
-    assert updated_entity.observations[1].content == "Test observation 2"
-    assert updated_entity.observations[1].category == "design"
-    assert updated_entity.observations[1].context == context
-    
-    # Verify file was updated
-    file_path = knowledge_service.get_entity_path(updated_entity)
+    # Verify file has new description but preserved metadata
+    file_path = knowledge_service.get_entity_path(updated)
     content, _ = await knowledge_service.read_file(file_path)
-    
-    for obs in observations:
-        expected_line = f"- [{obs.category.value}] {obs.content} ({context})"
-        assert expected_line in content
 
-    # Also verify the Observations section header exists
-    assert "## Observations" in content
+    assert "Updated description" in content
+
+    # Verify metadata was preserved
+    _, frontmatter, _ = content.split("---", 2)
+    metadata = yaml.safe_load(frontmatter)
+    assert metadata["status"] == "draft"
 
 
 @pytest.mark.asyncio
-async def test_delete_entity(knowledge_service: KnowledgeService):
-    """Should delete entity and its file."""
+async def test_update_note_entity_content(knowledge_service: KnowledgeService):
+    """Should update note content directly."""
     # Create test entity
     entity = await knowledge_service.create_entity(
-        EntitySchema(name="test", entity_type=EntityType.KNOWLEDGE, description="Test entity")
-    )
-    file_path = knowledge_service.get_entity_path(entity)
-
-    # Verify file exists
-    assert await knowledge_service.file_exists(file_path)
-
-    # Delete entity
-    success = await knowledge_service.delete_entity(entity.path_id)
-    assert success
-
-    # Verify file was deleted
-    assert not await knowledge_service.file_exists(file_path)
-
-    # Verify entity was deleted from DB
-    with pytest.raises(EntityNotFoundError):
-        await knowledge_service.get_entity_by_path_id(entity.path_id)
-
-
-@pytest.mark.asyncio
-async def test_delete_multiple_entities(knowledge_service: KnowledgeService):
-    """Should delete multiple entities and their files."""
-    # Create test entities
-    entities = []
-    for i in range(3):
-        entity = await knowledge_service.create_entity(
-            EntitySchema(name=f"test-{i}", entity_type=EntityType.KNOWLEDGE, description=f"Test entity {i}")
+        EntitySchema(
+            name="test",
+            entity_type=EntityType.NOTE,
+            description="Test note",
+            entity_metadata={"status": "draft"},
         )
-        entities.append(entity)
+    )
 
-    # Delete entities
-    success = await knowledge_service.delete_entities([e.path_id for e in entities])
-    assert success
+    # Update content
+    new_content = "# Updated Content\n\nThis is new content."
+    updated = await knowledge_service.update_entity(entity.path_id, content=new_content)
 
-    # Verify files were deleted
-    for entity in entities:
-        file_path = knowledge_service.get_entity_path(entity)
-        assert not await knowledge_service.file_exists(file_path)
-        with pytest.raises(EntityNotFoundError):
-            await knowledge_service.get_entity_by_path_id(entity.path_id)
+    # Verify file has new content but preserved metadata
+    file_path = knowledge_service.get_entity_path(updated)
+    content, _ = await knowledge_service.read_file(file_path)
+
+    assert "# Updated Content" in content
+    assert "This is new content" in content
+
+    # Verify metadata was preserved
+    _, frontmatter, _ = content.split("---", 2)
+    metadata = yaml.safe_load(frontmatter)
+    assert metadata["status"] == "draft"
 
 
 @pytest.mark.asyncio
-async def test_handle_file_operation_errors(knowledge_service: KnowledgeService, monkeypatch):
-    """Should handle file operation errors gracefully."""
-
-    async def mock_write_file(*args):
-        raise FileOperationError("Test error")
-
-    monkeypatch.setattr(knowledge_service.file_ops.file_service, "write_file", mock_write_file)
-
-    with pytest.raises(FileOperationError):
-        await knowledge_service.create_entity(
-            EntitySchema(name="test", entity_type=EntityType.KNOWLEDGE, description="Test entity")
+async def test_update_entity_name(knowledge_service: KnowledgeService):
+    """Should update entity name in both DB and frontmatter."""
+    # Create test entity
+    entity = await knowledge_service.create_entity(
+        EntitySchema(
+            name="test",
+            entity_type=EntityType.KNOWLEDGE,
+            description="Test entity",
+            entity_metadata={"status": "draft"},
         )
+    )
+
+    # Update name
+    updated = await knowledge_service.update_entity(entity.path_id, name="new-name")
+
+    # Verify name was updated in DB
+    assert updated.name == "new-name"
+
+    # Verify frontmatter was updated in file
+    file_path = knowledge_service.get_entity_path(updated)
+    content, _ = await knowledge_service.read_file(file_path)
+
+    _, frontmatter, _ = content.split("---", 2)
+    metadata = yaml.safe_load(frontmatter)
+    assert metadata["id"] == entity.path_id
+
+    # And verify content uses new name for title
+    assert "# new-name" in content
 
 
 @pytest.mark.asyncio
-async def test_entity_not_found_error(knowledge_service: KnowledgeService):
-    """Should raise EntityNotFoundError for non-existent entity."""
-    with pytest.raises(EntityNotFoundError):
-        await knowledge_service.add_observations(999, ["Test observation"])
-
-
-@pytest.mark.asyncio
-async def test_cleanup_on_creation_failure(knowledge_service: KnowledgeService, monkeypatch):
-    """Should clean up DB entity if file write fails."""
-    entity_ids: List[str] = []
-
-    # Capture created entity ID
-    original_create = knowledge_service.entity_ops.entity_service.create_entity
-
-    async def mock_create_entity(*args, **kwargs):
-        entity = await original_create(*args, **kwargs)
-        entity_ids.append(entity.path_id)
-        return entity
-
-    # Force file write to fail
-    async def mock_write_file(*args):
-        raise FileOperationError("Test error")
-
-    monkeypatch.setattr(knowledge_service.entity_ops.entity_service, "create_entity", mock_create_entity)
-    monkeypatch.setattr(knowledge_service.file_ops.file_service, "write_file", mock_write_file)
-
-    # Attempt creation (should fail)
-    with pytest.raises(FileOperationError):
-        await knowledge_service.create_entity(
-            EntitySchema(name="test", entity_type=EntityType.KNOWLEDGE, description="Test entity")
+async def test_update_entity_type(knowledge_service: KnowledgeService):
+    """Should update entity type and reflect change in frontmatter."""
+    # Create test entity as note
+    entity = await knowledge_service.create_entity(
+        EntitySchema(
+            name="test",
+            entity_type=EntityType.NOTE,
+            description="Test note",
+            entity_metadata={"status": "draft"},
         )
-
-    # Verify entity was cleaned up
-    assert len(entity_ids) == 1
-    with pytest.raises(EntityNotFoundError):
-        await knowledge_service.entity_ops.entity_service.get_by_path_id(entity_ids[0])
-
-
-@pytest.mark.asyncio
-async def test_skip_failed_batch_operations(knowledge_service: KnowledgeService):
-    """Should continue processing batch operations if some fail."""
-    entities = [
-        EntitySchema(name="test-1", entity_type=EntityType.KNOWLEDGE, description="Test entity 1"),
-        EntitySchema(name="test-1", entity_type=EntityType.KNOWLEDGE, description="Duplicate name - should fail"),
-        EntitySchema(name="test-2", entity_type=EntityType.KNOWLEDGE, description="Test entity 2"),
-    ]
-
-    with pytest.raises(IntegrityError):
-        await knowledge_service.create_entities(entities)
-
-
-@pytest.mark.asyncio
-async def test_update_relations_in_files(knowledge_service: KnowledgeService):
-    """Should update both entity files when creating relations."""
-    # Create test entities
-    entity1 = await knowledge_service.create_entity(
-        EntitySchema(name="source", entity_type=EntityType.KNOWLEDGE, description="Source entity")
-    )
-    entity2 = await knowledge_service.create_entity(
-        EntitySchema(name="target", entity_type=EntityType.KNOWLEDGE, description="Target entity")
     )
 
-    # Create relation
-    relations = [
-        RelationSchema(
-            from_id=entity1.path_id,
-            to_id=entity2.path_id,
-            relation_type="connects_to",
-            context="Test connection",
-        )
-    ]
-
-    await knowledge_service.create_relations(relations)
-
-    # Verify source file contains relation
-    for entity in [entity1]:
-        file_path = knowledge_service.get_entity_path(entity)
-        content, _ = await knowledge_service.read_file(file_path)
-        assert "connects_to" in content
-
-    # Source should show outgoing relation
-    content, _ = await knowledge_service.read_file(
-        knowledge_service.get_entity_path(entity1)
+    # Update to knowledge type
+    updated = await knowledge_service.update_entity(
+        entity.path_id, entity_type=EntityType.KNOWLEDGE
     )
-    assert "target" in content
 
-    # Target should not show incoming relation
-    content, _ = await knowledge_service.read_file(
-        knowledge_service.get_entity_path(entity2)
-    )
-    assert "source" not in content
+    # Verify type was updated in DB
+    assert updated.entity_type == EntityType.KNOWLEDGE
+
+    # Verify frontmatter was updated
+    file_path = knowledge_service.get_entity_path(updated)
+    content, _ = await knowledge_service.read_file(file_path)
+
+    _, frontmatter, _ = content.split("---", 2)
+    metadata = yaml.safe_load(frontmatter)
+    assert metadata["type"] == EntityType.KNOWLEDGE
+
+    # Verify content format changed to knowledge style (structured)
+    assert "# test" in content
+    assert "Test note" in content  # Description included