knowledge_service tests passing

This commit is contained in:
phernandez
2025-01-06 20:07:12 -06:00
parent ea359d2484
commit 55c40bd6d3
10 changed files with 325 additions and 236 deletions
+7
View File
@@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import (
from basic_memory import db
from basic_memory.config import ProjectConfig, config
from basic_memory.markdown.knowledge_writer import KnowledgeWriter
from basic_memory.markdown.note_writer import NoteWriter
from basic_memory.repository.document_repository import DocumentRepository
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.observation_repository import ObservationRepository
@@ -196,6 +197,12 @@ async def get_knowledge_writer() -> KnowledgeWriter:
KnowledgeWriterDep = Annotated[KnowledgeWriter, Depends(get_knowledge_writer)]
async def get_note_writer() -> NoteWriter:
return NoteWriter()
NoteWriterDep = Annotated[NoteWriter, Depends(get_note_writer)]
async def get_knowledge_service(
entity_service: EntityServiceDep,
+6 -32
View File
@@ -1,10 +1,5 @@
"""Writer for knowledge entity markdown files."""
from typing import Optional, Dict, Any
import yaml
from loguru import logger
from basic_memory.models import Entity as EntityModel
@@ -13,34 +8,17 @@ class KnowledgeWriter:
async def format_frontmatter(self, entity: EntityModel) -> dict:
"""Generate frontmatter metadata for entity."""
return {
frontmatter = {
"id": entity.path_id,
"type": entity.entity_type,
"created": entity.created_at.isoformat(),
"modified": entity.updated_at.isoformat(),
}
if entity.entity_metadata:
frontmatter.update(entity.entity_metadata)
return frontmatter
async def format_metadata(self, metadata: Optional[Dict[str, Any]] = None) -> str:
"""Format metadata section as YAML block."""
if not metadata:
return ""
try:
yaml_block = yaml.dump(metadata, sort_keys=False)
return (
"# Metadata\n"
"<!-- anything below this line is for AI -->\n\n"
"```yml\n"
f"{yaml_block}"
"```\n"
)
except Exception as e:
logger.warning(f"Failed to format metadata YAML: {e}")
return "" # Skip metadata on error
async def format_content(
self, entity: EntityModel, metadata: Optional[Dict[str, Any]] = None
) -> str:
async def format_content(self, entity: EntityModel, content: str) -> str:
"""Format entity content as markdown."""
sections = [
f"# {entity.name}\n",
@@ -78,9 +56,5 @@ class KnowledgeWriter:
# Outgoing relations (entity is "from")
for rel in entity.outgoing_relations:
sections.append(f"- {rel.relation_type} [[{rel.to_entity.name}]] ")
if metadata:
sections.append("\n")
sections.append(await self.format_metadata(metadata))
return "\n".join(sections)
return "\n".join(sections)
+44
View File
@@ -0,0 +1,44 @@
"""Writer for note entity markdown files."""
from typing import Optional
from basic_memory.models import Entity as EntityModel
class NoteWriter:
"""Formats notes into markdown files with frontmatter."""
async def format_frontmatter(self, entity: EntityModel) -> dict:
"""Generate frontmatter for note.
Args:
entity: The note entity to format frontmatter for
Returns:
Dictionary of frontmatter fields
"""
frontmatter = {
"id": entity.path_id,
"type": entity.entity_type,
"created": entity.created_at.isoformat(),
"modified": entity.updated_at.isoformat()
}
# Add any entity_metadata that isn't None
if entity.entity_metadata:
frontmatter.update(entity.entity_metadata)
return frontmatter
async def format_content(
self,
entity: EntityModel,
content: str,
) -> str:
"""Format note content as markdown.
Args:
entity: The note entity
content: Raw content to format
Returns:
Formatted markdown content
"""
return content.strip() # Just return the trimmed content
+2 -2
View File
@@ -256,9 +256,9 @@ class Entity(BaseModel):
@property
def path_id(self) -> PathId:
"""Get the path ID in format {type}/{snake_case_name}."""
"""Get the path ID in format {snake_case_name}."""
normalized_name = to_snake_case(self.name)
return f"{self.entity_type}/{normalized_name}"
return normalized_name
@property
def file_path(self):
@@ -1,30 +1,34 @@
"""File operations for knowledge service."""
from pathlib import Path
from typing import Tuple
from typing import Tuple, 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.models.knowledge import EntityType
from basic_memory.services.entity_service import EntityService
from basic_memory.services.exceptions import FileOperationError
from basic_memory.services.file_service import FileService
class FileOperations:
"""File operations for knowledge entities."""
"""File operations for both knowledge and note entities."""
def __init__(
self,
entity_service: EntityService,
file_service: FileService,
knowledge_writer: KnowledgeWriter,
note_writer: NoteWriter,
base_path: Path,
):
self.entity_service = entity_service
self.file_service = file_service
self.knowledge_writer = knowledge_writer
self.note_writer = note_writer
self.base_path = base_path
async def file_exists(self, path: Path) -> bool:
@@ -35,25 +39,37 @@ class FileOperations:
def get_entity_path(self, entity: EntityModel) -> Path:
"""Generate filesystem path for entity."""
return self.base_path / entity.entity_type / f"{entity.name}.md"
if entity.file_path:
return self.base_path / entity.file_path
return self.base_path / f"{entity.path_id}.md"
async def write_entity_file(self, entity: EntityModel) -> Tuple[Path, str]:
async def write_entity_file(
self,
entity: EntityModel,
content: Optional[str] = None,
) -> Tuple[Path, str]:
"""Write entity to filesystem and return path and checksum."""
try:
# Ensure we have a fresh entity with all relations loaded
# Ensure we have a fresh entity with all relations
entity = await self.entity_service.get_by_path_id(entity.path_id)
frontmatter = await self.knowledge_writer.format_frontmatter(entity)
# Format content
path = self.get_entity_path(entity)
entity_content = await self.knowledge_writer.format_content(entity)
file_content = await self.file_service.add_frontmatter(
frontmatter=frontmatter,
content=entity_content,
# Select writer based on entity type
writer = self.note_writer if entity.entity_type == EntityType.NOTE else self.knowledge_writer
# Get frontmatter and content
frontmatter = await writer.format_frontmatter(entity)
file_content = await writer.format_content(
entity=entity,
content=content or entity.description or "",
)
# Write and get checksum
return path, await self.file_service.write_file(path, file_content)
# Add frontmatter and write
content_with_frontmatter = await self.file_service.add_frontmatter(
frontmatter=frontmatter,
content=file_content
)
path = self.get_entity_path(entity)
return path, await self.file_service.write_file(path, content_with_frontmatter)
except Exception as e:
logger.error(f"Failed to write entity file: {e}")
@@ -19,6 +19,7 @@ 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:
@@ -42,6 +43,7 @@ class KnowledgeService:
relation_service: RelationService,
file_service: FileService,
knowledge_writer: KnowledgeWriter,
note_writer: NoteWriter,
base_path: Path,
):
@@ -52,6 +54,7 @@ class KnowledgeService:
entity_service=entity_service,
file_service=file_service,
knowledge_writer=knowledge_writer,
note_writer=note_writer,
base_path=base_path
)