feat: fast edit entities, refactors for webui, enhance search (#532)

Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
Paul Hernandez
2026-01-31 15:16:52 -06:00
committed by GitHub
parent e3ced49d9d
commit 530cbac73f
85 changed files with 1849 additions and 7120 deletions
@@ -0,0 +1,152 @@
"""Add structured metadata indexes for entity frontmatter
Revision ID: d7e8f9a0b1c2
Revises: g9a0b3c4d5e6
Create Date: 2026-01-31 12:00:00.000000
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy import text
def column_exists(connection, table: str, column: str) -> bool:
"""Check if a column exists in a table (idempotent migration support)."""
if connection.dialect.name == "postgresql":
result = connection.execute(
text(
"SELECT 1 FROM information_schema.columns "
"WHERE table_name = :table AND column_name = :column"
),
{"table": table, "column": column},
)
return result.fetchone() is not None
# SQLite
result = connection.execute(text(f"PRAGMA table_info({table})"))
columns = [row[1] for row in result]
return column in columns
def index_exists(connection, index_name: str) -> bool:
"""Check if an index exists (idempotent migration support)."""
if connection.dialect.name == "postgresql":
result = connection.execute(
text("SELECT 1 FROM pg_indexes WHERE indexname = :index_name"),
{"index_name": index_name},
)
return result.fetchone() is not None
# SQLite
result = connection.execute(
text("SELECT 1 FROM sqlite_master WHERE type='index' AND name = :index_name"),
{"index_name": index_name},
)
return result.fetchone() is not None
# revision identifiers, used by Alembic.
revision: str = "d7e8f9a0b1c2"
down_revision: Union[str, None] = "6830751f5fb6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Add JSONB/GiN indexes for Postgres and generated columns for SQLite."""
connection = op.get_bind()
dialect = connection.dialect.name
if dialect == "postgresql":
# Ensure JSONB for efficient indexing
result = connection.execute(
text(
"SELECT data_type FROM information_schema.columns "
"WHERE table_name = 'entity' AND column_name = 'entity_metadata'"
)
).fetchone()
if result and result[0] != "jsonb":
op.execute(
"ALTER TABLE entity ALTER COLUMN entity_metadata "
"TYPE jsonb USING entity_metadata::jsonb"
)
# General JSONB GIN index
op.execute(
"CREATE INDEX IF NOT EXISTS idx_entity_metadata_gin "
"ON entity USING GIN (entity_metadata jsonb_path_ops)"
)
# Common field indexes
op.execute(
"CREATE INDEX IF NOT EXISTS idx_entity_tags_json "
"ON entity USING GIN ((entity_metadata -> 'tags'))"
)
op.execute(
"CREATE INDEX IF NOT EXISTS idx_entity_frontmatter_type "
"ON entity ((entity_metadata ->> 'type'))"
)
op.execute(
"CREATE INDEX IF NOT EXISTS idx_entity_frontmatter_status "
"ON entity ((entity_metadata ->> 'status'))"
)
return
# SQLite: add generated columns for common frontmatter fields
if not column_exists(connection, "entity", "tags_json"):
op.add_column(
"entity",
sa.Column(
"tags_json",
sa.Text(),
sa.Computed("json_extract(entity_metadata, '$.tags')", persisted=True),
),
)
if not column_exists(connection, "entity", "frontmatter_status"):
op.add_column(
"entity",
sa.Column(
"frontmatter_status",
sa.Text(),
sa.Computed("json_extract(entity_metadata, '$.status')", persisted=True),
),
)
if not column_exists(connection, "entity", "frontmatter_type"):
op.add_column(
"entity",
sa.Column(
"frontmatter_type",
sa.Text(),
sa.Computed("json_extract(entity_metadata, '$.type')", persisted=True),
),
)
# Index generated columns
if not index_exists(connection, "idx_entity_tags_json"):
op.create_index("idx_entity_tags_json", "entity", ["tags_json"])
if not index_exists(connection, "idx_entity_frontmatter_status"):
op.create_index("idx_entity_frontmatter_status", "entity", ["frontmatter_status"])
if not index_exists(connection, "idx_entity_frontmatter_type"):
op.create_index("idx_entity_frontmatter_type", "entity", ["frontmatter_type"])
def downgrade() -> None:
"""Best-effort downgrade (drop indexes, revert JSONB on Postgres)."""
connection = op.get_bind()
dialect = connection.dialect.name
if dialect == "postgresql":
op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_status")
op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_type")
op.execute("DROP INDEX IF EXISTS idx_entity_tags_json")
op.execute("DROP INDEX IF EXISTS idx_entity_metadata_gin")
op.execute(
"ALTER TABLE entity ALTER COLUMN entity_metadata TYPE json USING entity_metadata::json"
)
return
# SQLite: drop indexes (dropping generated columns requires table rebuild)
op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_status")
op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_type")
op.execute("DROP INDEX IF EXISTS idx_entity_tags_json")
+3 -23
View File
@@ -8,17 +8,6 @@ from loguru import logger
from basic_memory import __version__ as version
from basic_memory.api.container import ApiContainer, set_container
from basic_memory.api.routers import (
directory_router,
importer_router,
knowledge,
management,
memory,
project,
resource,
search,
prompt_router,
)
from basic_memory.api.v2.routers import (
knowledge_router as v2_knowledge,
project_router as v2_project,
@@ -90,19 +79,10 @@ app.include_router(v2_prompt, prefix="/v2/projects/{project_id}")
app.include_router(v2_importer, prefix="/v2/projects/{project_id}")
app.include_router(v2_project, prefix="/v2")
# Include v1 routers (/{project} is a catch-all, must come after specific prefixes)
app.include_router(knowledge.router, prefix="/{project}")
app.include_router(memory.router, prefix="/{project}")
app.include_router(resource.router, prefix="/{project}")
app.include_router(search.router, prefix="/{project}")
app.include_router(project.project_router, prefix="/{project}")
app.include_router(directory_router.router, prefix="/{project}")
app.include_router(prompt_router.router, prefix="/{project}")
app.include_router(importer_router.router, prefix="/{project}")
# Legacy web app proxy paths (compat with /proxy/projects/projects)
app.include_router(v2_project, prefix="/proxy/projects")
# Project resource router works across projects
app.include_router(project.project_resource_router)
app.include_router(management.router)
# V2 routers are the only public API surface
@app.exception_handler(Exception)
-11
View File
@@ -1,11 +0,0 @@
"""API routers."""
from . import knowledge_router as knowledge
from . import management_router as management
from . import memory_router as memory
from . import project_router as project
from . import resource_router as resource
from . import search_router as search
from . import prompt_router as prompt
__all__ = ["knowledge", "management", "memory", "project", "resource", "search", "prompt"]
@@ -1,84 +0,0 @@
"""Router for directory tree operations."""
from typing import List, Optional
from fastapi import APIRouter, Query
from basic_memory.deps import DirectoryServiceDep, ProjectIdDep
from basic_memory.schemas.directory import DirectoryNode
router = APIRouter(prefix="/directory", tags=["directory"])
@router.get("/tree", response_model=DirectoryNode, response_model_exclude_none=True)
async def get_directory_tree(
directory_service: DirectoryServiceDep,
project_id: ProjectIdDep,
):
"""Get hierarchical directory structure from the knowledge base.
Args:
directory_service: Service for directory operations
project_id: ID of the current project
Returns:
DirectoryNode representing the root of the hierarchical tree structure
"""
# Get a hierarchical directory tree for the specific project
tree = await directory_service.get_directory_tree()
# Return the hierarchical tree
return tree
@router.get("/structure", response_model=DirectoryNode, response_model_exclude_none=True)
async def get_directory_structure(
directory_service: DirectoryServiceDep,
project_id: ProjectIdDep,
):
"""Get folder structure for navigation (no files).
Optimized endpoint for folder tree navigation. Returns only directory nodes
without file metadata. For full tree with files, use /directory/tree.
Args:
directory_service: Service for directory operations
project_id: ID of the current project
Returns:
DirectoryNode tree containing only folders (type="directory")
"""
structure = await directory_service.get_directory_structure()
return structure
@router.get("/list", response_model=List[DirectoryNode], response_model_exclude_none=True)
async def list_directory(
directory_service: DirectoryServiceDep,
project_id: ProjectIdDep,
dir_name: str = Query("/", description="Directory path to list"),
depth: int = Query(1, ge=1, le=10, description="Recursion depth (1-10)"),
file_name_glob: Optional[str] = Query(
None, description="Glob pattern for filtering file names"
),
):
"""List directory contents with filtering and depth control.
Args:
directory_service: Service for directory operations
project_id: ID of the current project
dir_name: Directory path to list (default: root "/")
depth: Recursion depth (1-10, default: 1 for immediate children only)
file_name_glob: Optional glob pattern for filtering file names (e.g., "*.md", "*meeting*")
Returns:
List of DirectoryNode objects matching the criteria
"""
# Get directory listing with filtering
nodes = await directory_service.list_directory(
dir_name=dir_name,
depth=depth,
file_name_glob=file_name_glob,
)
return nodes
@@ -1,152 +0,0 @@
"""Import router for Basic Memory API."""
import json
import logging
from fastapi import APIRouter, Form, HTTPException, UploadFile, status
from basic_memory.deps import (
ChatGPTImporterDep,
ClaudeConversationsImporterDep,
ClaudeProjectsImporterDep,
MemoryJsonImporterDep,
)
from basic_memory.importers import Importer
from basic_memory.schemas.importer import (
ChatImportResult,
EntityImportResult,
ProjectImportResult,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/import", tags=["import"])
@router.post("/chatgpt", response_model=ChatImportResult)
async def import_chatgpt(
importer: ChatGPTImporterDep,
file: UploadFile,
directory: str = Form("conversations"),
) -> ChatImportResult:
"""Import conversations from ChatGPT JSON export.
Args:
file: The ChatGPT conversations.json file.
directory: The directory to place the files in.
markdown_processor: MarkdownProcessor instance.
Returns:
ChatImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
return await import_file(importer, file, directory)
@router.post("/claude/conversations", response_model=ChatImportResult)
async def import_claude_conversations(
importer: ClaudeConversationsImporterDep,
file: UploadFile,
directory: str = Form("conversations"),
) -> ChatImportResult:
"""Import conversations from Claude conversations.json export.
Args:
file: The Claude conversations.json file.
directory: The directory to place the files in.
markdown_processor: MarkdownProcessor instance.
Returns:
ChatImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
return await import_file(importer, file, directory)
@router.post("/claude/projects", response_model=ProjectImportResult)
async def import_claude_projects(
importer: ClaudeProjectsImporterDep,
file: UploadFile,
directory: str = Form("projects"),
) -> ProjectImportResult:
"""Import projects from Claude projects.json export.
Args:
file: The Claude projects.json file.
directory: The directory to place the files in.
markdown_processor: MarkdownProcessor instance.
Returns:
ProjectImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
return await import_file(importer, file, directory)
@router.post("/memory-json", response_model=EntityImportResult)
async def import_memory_json(
importer: MemoryJsonImporterDep,
file: UploadFile,
directory: str = Form("conversations"),
) -> EntityImportResult:
"""Import entities and relations from a memory.json file.
Args:
file: The memory.json file.
directory: Optional destination directory within the project.
markdown_processor: MarkdownProcessor instance.
Returns:
EntityImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
try:
file_data = []
file_bytes = await file.read()
file_str = file_bytes.decode("utf-8")
for line in file_str.splitlines():
json_data = json.loads(line)
file_data.append(json_data)
result = await importer.import_data(file_data, directory)
if not result.success: # pragma: no cover
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=result.error_message or "Import failed",
)
except Exception as e:
logger.exception("Import failed")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Import failed: {str(e)}",
)
return result
async def import_file(importer: Importer, file: UploadFile, destination_folder: str):
try:
# Process file
json_data = json.load(file.file)
result = await importer.import_data(json_data, destination_folder)
if not result.success: # pragma: no cover
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=result.error_message or "Import failed",
)
return result
except Exception as e:
logger.exception("Import failed")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Import failed: {str(e)}",
)
@@ -1,364 +0,0 @@
"""Router for knowledge graph operations.
⚠️ DEPRECATED: This v1 API is deprecated and will be removed on June 30, 2026.
Please migrate to /v2/{project}/knowledge endpoints which use entity IDs instead
of path-based identifiers for improved performance and stability.
Migration guide: See docs/migration/v1-to-v2.md
"""
from typing import Annotated
from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends, Query, Response
from loguru import logger
from basic_memory.deps import (
EntityServiceDep,
get_search_service,
SearchServiceDep,
LinkResolverDep,
ProjectPathDep,
FileServiceDep,
ProjectConfigDep,
AppConfigDep,
SyncServiceDep,
)
from basic_memory.schemas import (
EntityListResponse,
EntityResponse,
DeleteEntitiesResponse,
DeleteEntitiesRequest,
)
from basic_memory.schemas.request import EditEntityRequest, MoveEntityRequest, MoveDirectoryRequest
from basic_memory.schemas.response import DirectoryMoveResult
from basic_memory.schemas.base import Permalink, Entity
router = APIRouter(
prefix="/knowledge",
tags=["knowledge"],
deprecated=True, # Marks entire router as deprecated in OpenAPI docs
)
async def resolve_relations_background(sync_service, entity_id: int, entity_permalink: str) -> None:
"""Background task to resolve relations for a specific entity.
This runs asynchronously after the API response is sent, preventing
long delays when creating entities with many relations.
"""
try:
# Only resolve relations for the newly created entity
await sync_service.resolve_relations(entity_id=entity_id)
logger.debug(
f"Background: Resolved relations for entity {entity_permalink} (id={entity_id})"
)
except Exception as e: # pragma: no cover
# Log but don't fail - this is a background task.
# Avoid forcing synthetic failures just for coverage.
logger.warning( # pragma: no cover
f"Background: Failed to resolve relations for entity {entity_permalink}: {e}"
)
## Create endpoints
@router.post("/entities", response_model=EntityResponse)
async def create_entity(
data: Entity,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
search_service: SearchServiceDep,
) -> EntityResponse:
"""Create an entity."""
logger.info(
"API request", endpoint="create_entity", entity_type=data.entity_type, title=data.title
)
entity = await entity_service.create_entity(data)
# reindex
await search_service.index_entity(entity, background_tasks=background_tasks)
result = EntityResponse.model_validate(entity)
logger.info(
f"API response: endpoint='create_entity' title={result.title}, permalink={result.permalink}, status_code=201"
)
return result
@router.put("/entities/{permalink:path}", response_model=EntityResponse)
async def create_or_update_entity(
project: ProjectPathDep,
permalink: Permalink,
data: Entity,
response: Response,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
search_service: SearchServiceDep,
file_service: FileServiceDep,
sync_service: SyncServiceDep,
) -> EntityResponse:
"""Create or update an entity. If entity exists, it will be updated, otherwise created."""
logger.info(
f"API request: create_or_update_entity for {project=}, {permalink=}, {data.entity_type=}, {data.title=}"
)
# Validate permalink matches
if data.permalink != permalink:
logger.warning(
f"API validation error: creating/updating entity with permalink mismatch - url={permalink}, data={data.permalink}",
)
raise HTTPException(
status_code=400,
detail=f"Entity permalink {data.permalink} must match URL path: '{permalink}'",
)
# Try create_or_update operation
entity, created = await entity_service.create_or_update_entity(data)
response.status_code = 201 if created else 200
# reindex
await search_service.index_entity(entity, background_tasks=background_tasks)
# Schedule relation resolution as a background task for new entities
# This prevents blocking the API response while resolving potentially many relations
if created:
background_tasks.add_task(
resolve_relations_background, sync_service, entity.id, entity.permalink or ""
)
result = EntityResponse.model_validate(entity)
logger.info(
f"API response: {result.title=}, {result.permalink=}, {created=}, status_code={response.status_code}"
)
return result
@router.patch("/entities/{identifier:path}", response_model=EntityResponse)
async def edit_entity(
identifier: str,
data: EditEntityRequest,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
search_service: SearchServiceDep,
) -> EntityResponse:
"""Edit an existing entity using various operations like append, prepend, find_replace, or replace_section.
This endpoint allows for targeted edits without requiring the full entity content.
"""
logger.info(
f"API request: endpoint='edit_entity', identifier='{identifier}', operation='{data.operation}'"
)
try:
# Edit the entity using the service
entity = await entity_service.edit_entity(
identifier=identifier,
operation=data.operation,
content=data.content,
section=data.section,
find_text=data.find_text,
expected_replacements=data.expected_replacements,
)
# Reindex the updated entity
await search_service.index_entity(entity, background_tasks=background_tasks)
# Return the updated entity response
result = EntityResponse.model_validate(entity)
logger.info(
"API response",
endpoint="edit_entity",
identifier=identifier,
operation=data.operation,
permalink=result.permalink,
status_code=200,
)
return result
except Exception as e:
logger.error(f"Error editing entity: {e}")
raise HTTPException(status_code=400, detail=str(e))
@router.post("/move")
async def move_entity(
data: MoveEntityRequest,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
project_config: ProjectConfigDep,
app_config: AppConfigDep,
search_service: SearchServiceDep,
) -> EntityResponse:
"""Move an entity to a new file location with project consistency.
This endpoint moves a note to a different path while maintaining project
consistency and optionally updating permalinks based on configuration.
"""
logger.info(
f"API request: endpoint='move_entity', identifier='{data.identifier}', destination='{data.destination_path}'"
)
try:
# Move the entity using the service
moved_entity = await entity_service.move_entity(
identifier=data.identifier,
destination_path=data.destination_path,
project_config=project_config,
app_config=app_config,
)
# Get the moved entity to reindex it
entity = await entity_service.link_resolver.resolve_link(data.destination_path)
if entity:
await search_service.index_entity(entity, background_tasks=background_tasks)
logger.info(
"API response",
endpoint="move_entity",
identifier=data.identifier,
destination=data.destination_path,
status_code=200,
)
result = EntityResponse.model_validate(moved_entity)
return result
except Exception as e:
logger.error(f"Error moving entity: {e}")
raise HTTPException(status_code=400, detail=str(e))
@router.post("/move-directory")
async def move_directory(
data: MoveDirectoryRequest,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
project_config: ProjectConfigDep,
app_config: AppConfigDep,
search_service: SearchServiceDep,
) -> DirectoryMoveResult:
"""Move all entities in a directory to a new location.
This endpoint moves all files within a source directory to a destination
directory, updating database records and optionally updating permalinks.
"""
logger.info(
f"API request: endpoint='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, background_tasks=background_tasks)
logger.info(
f"API response: endpoint='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))
## Read endpoints
@router.get("/entities/{identifier:path}", response_model=EntityResponse)
async def get_entity(
entity_service: EntityServiceDep,
link_resolver: LinkResolverDep,
identifier: str,
) -> EntityResponse:
"""Get a specific entity by file path or permalink..
Args:
identifier: Entity file path or permalink
:param entity_service: EntityService
:param link_resolver: LinkResolver
"""
logger.info(f"request: get_entity with identifier={identifier}")
entity = await link_resolver.resolve_link(identifier)
if not entity:
raise HTTPException(status_code=404, detail=f"Entity {identifier} not found")
result = EntityResponse.model_validate(entity)
return result
@router.get("/entities", response_model=EntityListResponse)
async def get_entities(
entity_service: EntityServiceDep,
permalink: Annotated[list[str] | None, Query()] = None,
) -> EntityListResponse:
"""Open specific entities"""
logger.info(f"request: get_entities with permalinks={permalink}")
entities = await entity_service.get_entities_by_permalinks(permalink) if permalink else []
result = EntityListResponse(
entities=[EntityResponse.model_validate(entity) for entity in entities]
)
return result
## Delete endpoints
@router.delete("/entities/{identifier:path}", response_model=DeleteEntitiesResponse)
async def delete_entity(
identifier: str,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
link_resolver: LinkResolverDep,
search_service=Depends(get_search_service),
) -> DeleteEntitiesResponse:
"""Delete a single entity and remove from search index."""
logger.info(f"request: delete_entity with identifier={identifier}")
entity = await link_resolver.resolve_link(identifier)
if entity is None:
return DeleteEntitiesResponse(deleted=False)
# Delete the entity
deleted = await entity_service.delete_entity(entity.permalink or entity.id)
# Remove from search index (entity, observations, and relations)
background_tasks.add_task(search_service.handle_delete, entity)
result = DeleteEntitiesResponse(deleted=deleted)
return result
@router.post("/entities/delete", response_model=DeleteEntitiesResponse)
async def delete_entities(
data: DeleteEntitiesRequest,
background_tasks: BackgroundTasks,
entity_service: EntityServiceDep,
search_service=Depends(get_search_service),
) -> DeleteEntitiesResponse:
"""Delete entities and remove from search index."""
logger.info(f"request: delete_entities with data={data}")
deleted = False
# Remove each deleted entity from search index
for permalink in data.permalinks:
deleted = await entity_service.delete_entity(permalink)
background_tasks.add_task(search_service.delete_by_permalink, permalink)
result = DeleteEntitiesResponse(deleted=deleted)
return result
@@ -1,80 +0,0 @@
"""Management router for basic-memory API."""
import asyncio
from fastapi import APIRouter, Request
from loguru import logger
from pydantic import BaseModel
from basic_memory.config import ConfigManager
from basic_memory.deps import SyncServiceDep, ProjectRepositoryDep
router = APIRouter(prefix="/management", tags=["management"])
class WatchStatusResponse(BaseModel):
"""Response model for watch status."""
running: bool
"""Whether the watch service is currently running."""
@router.get("/watch/status", response_model=WatchStatusResponse)
async def get_watch_status(request: Request) -> WatchStatusResponse:
"""Get the current status of the watch service."""
return WatchStatusResponse(
running=request.app.state.watch_task is not None and not request.app.state.watch_task.done()
)
@router.post("/watch/start", response_model=WatchStatusResponse)
async def start_watch_service(
request: Request, project_repository: ProjectRepositoryDep, sync_service: SyncServiceDep
) -> WatchStatusResponse:
"""Start the watch service if it's not already running."""
# needed because of circular imports from sync -> app
from basic_memory.sync import WatchService
from basic_memory.sync.background_sync import create_background_sync_task
if request.app.state.watch_task is not None and not request.app.state.watch_task.done():
# Watch service is already running
return WatchStatusResponse(running=True)
app_config = ConfigManager().config
# Create and start a new watch service
logger.info("Starting watch service via management API")
# Get services needed for the watch task
watch_service = WatchService(
app_config=app_config,
project_repository=project_repository,
)
# Create and store the task
watch_task = create_background_sync_task(sync_service, watch_service)
request.app.state.watch_task = watch_task
return WatchStatusResponse(running=True)
@router.post("/watch/stop", response_model=WatchStatusResponse)
async def stop_watch_service(request: Request) -> WatchStatusResponse: # pragma: no cover
"""Stop the watch service if it's running."""
if request.app.state.watch_task is None or request.app.state.watch_task.done():
# Watch service is not running
return WatchStatusResponse(running=False)
# Cancel the running task
logger.info("Stopping watch service via management API")
request.app.state.watch_task.cancel()
# Wait for it to be properly cancelled
try:
await request.app.state.watch_task
except asyncio.CancelledError:
pass
request.app.state.watch_task = None
return WatchStatusResponse(running=False)
@@ -1,90 +0,0 @@
"""Routes for memory:// URI operations."""
from typing import Annotated, Optional
from fastapi import APIRouter, Query
from loguru import logger
from basic_memory.deps import ContextServiceDep, EntityRepositoryDep
from basic_memory.schemas.base import TimeFrame, parse_timeframe
from basic_memory.schemas.memory import (
GraphContext,
normalize_memory_url,
)
from basic_memory.schemas.search import SearchItemType
from basic_memory.api.routers.utils import to_graph_context
router = APIRouter(prefix="/memory", tags=["memory"])
@router.get("/recent", response_model=GraphContext)
async def recent(
context_service: ContextServiceDep,
entity_repository: EntityRepositoryDep,
type: Annotated[list[SearchItemType] | None, Query()] = None,
depth: int = 1,
timeframe: TimeFrame = "7d",
page: int = 1,
page_size: int = 10,
max_related: int = 10,
) -> GraphContext:
# return all types by default
types = (
[SearchItemType.ENTITY, SearchItemType.RELATION, SearchItemType.OBSERVATION]
if not type
else type
)
logger.debug(
f"Getting recent context: `{types}` depth: `{depth}` timeframe: `{timeframe}` page: `{page}` page_size: `{page_size}` max_related: `{max_related}`"
)
# Parse timeframe
since = parse_timeframe(timeframe)
limit = page_size
offset = (page - 1) * page_size
# Build context
context = await context_service.build_context(
types=types, depth=depth, since=since, limit=limit, offset=offset, max_related=max_related
)
recent_context = await to_graph_context(
context, entity_repository=entity_repository, page=page, page_size=page_size
)
logger.debug(f"Recent context: {recent_context.model_dump_json()}")
return recent_context
# get_memory_context needs to be declared last so other paths can match
@router.get("/{uri:path}", response_model=GraphContext)
async def get_memory_context(
context_service: ContextServiceDep,
entity_repository: EntityRepositoryDep,
uri: str,
depth: int = 1,
timeframe: Optional[TimeFrame] = None,
page: int = 1,
page_size: int = 10,
max_related: int = 10,
) -> GraphContext:
"""Get rich context from memory:// URI."""
# add the project name from the config to the url as the "host
# Parse URI
logger.debug(
f"Getting context for URI: `{uri}` depth: `{depth}` timeframe: `{timeframe}` page: `{page}` page_size: `{page_size}` max_related: `{max_related}`"
)
memory_url = normalize_memory_url(uri)
# Parse timeframe
since = parse_timeframe(timeframe) if timeframe else None
limit = page_size
offset = (page - 1) * page_size
# Build context
context = await context_service.build_context(
memory_url, depth=depth, since=since, limit=limit, offset=offset, max_related=max_related
)
return await to_graph_context(
context, entity_repository=entity_repository, page=page, page_size=page_size
)
@@ -1,472 +0,0 @@
"""Router for project management."""
import os
from fastapi import APIRouter, HTTPException, Path, Body, BackgroundTasks, Response, Query
from typing import Optional
from loguru import logger
from basic_memory.deps import (
ProjectConfigDep,
ProjectServiceDep,
ProjectPathDep,
SyncServiceDep,
)
from basic_memory.schemas import ProjectInfoResponse, SyncReportResponse
from basic_memory.schemas.project_info import (
ProjectList,
ProjectItem,
ProjectInfoRequest,
ProjectStatusResponse,
)
from basic_memory.utils import normalize_project_path
# Router for resources in a specific project
# The ProjectPathDep is used in the path as a prefix, so the request path is like /{project}/project/info
project_router = APIRouter(prefix="/project", tags=["project"])
# Router for managing project resources
project_resource_router = APIRouter(prefix="/projects", tags=["project_management"])
@project_router.get("/info", response_model=ProjectInfoResponse)
async def get_project_info(
project_service: ProjectServiceDep,
project: ProjectPathDep,
) -> ProjectInfoResponse:
"""Get comprehensive information about the specified Basic Memory project."""
return await project_service.get_project_info(project)
@project_router.get("/item", response_model=ProjectItem)
async def get_project(
project_service: ProjectServiceDep,
project: ProjectPathDep,
) -> ProjectItem:
"""Get bassic info about the specified Basic Memory project."""
found_project = await project_service.get_project(project)
if not found_project:
raise HTTPException(
status_code=404, detail=f"Project: '{project}' does not exist"
) # pragma: no cover
return ProjectItem(
id=found_project.id,
external_id=found_project.external_id,
name=found_project.name,
path=normalize_project_path(found_project.path),
is_default=found_project.is_default or False,
)
# Update a project
@project_router.patch("/{name}", response_model=ProjectStatusResponse)
async def update_project(
project_service: ProjectServiceDep,
name: str = Path(..., description="Name of the project to update"),
path: Optional[str] = Body(None, description="New absolute path for the project"),
is_active: Optional[bool] = Body(None, description="Status of the project (active/inactive)"),
) -> ProjectStatusResponse:
"""Update a project's information in configuration and database.
Args:
name: The name of the project to update
path: Optional new absolute path for the project
is_active: Optional status update for the project
Returns:
Response confirming the project was updated
"""
try:
# Validate that path is absolute if provided
if path and not os.path.isabs(path):
raise HTTPException(status_code=400, detail="Path must be absolute")
# Get original project info for the response
old_project = await project_service.get_project(name)
if not old_project:
raise HTTPException(
status_code=400, detail=f"Project '{name}' not found in configuration"
)
old_project_info = ProjectItem(
id=old_project.id,
external_id=old_project.external_id,
name=old_project.name,
path=old_project.path,
is_default=old_project.is_default or False,
)
if path:
await project_service.move_project(name, path)
elif is_active is not None:
await project_service.update_project(name, is_active=is_active)
# Get updated project info
updated_project = await project_service.get_project(name)
if not updated_project:
raise HTTPException( # pragma: no cover
status_code=404, detail=f"Project '{name}' not found after update"
)
return ProjectStatusResponse(
message=f"Project '{name}' updated successfully",
status="success",
default=(name == project_service.default_project),
old_project=old_project_info,
new_project=ProjectItem(
id=updated_project.id,
external_id=updated_project.external_id,
name=updated_project.name,
path=updated_project.path,
is_default=updated_project.is_default or False,
),
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) # pragma: no cover
# Sync project filesystem
@project_router.post("/sync")
async def sync_project(
background_tasks: BackgroundTasks,
sync_service: SyncServiceDep,
project_config: ProjectConfigDep,
force_full: bool = Query(
False, description="Force full scan, bypassing watermark optimization"
),
run_in_background: bool = Query(True, description="Run in background"),
):
"""Force project filesystem sync to database.
Scans the project directory and updates the database with any new or modified files.
Args:
background_tasks: FastAPI background tasks
sync_service: Sync service for this project
project_config: Project configuration
force_full: If True, force a full scan even if watermark exists
run_in_background: If True, run sync in background and return immediately
Returns:
Response confirming sync was initiated (background) or SyncReportResponse (foreground)
"""
if run_in_background:
background_tasks.add_task(
sync_service.sync, project_config.home, project_config.name, force_full=force_full
)
logger.info(
f"Filesystem sync initiated for project: {project_config.name} (force_full={force_full})"
)
return {
"status": "sync_started",
"message": f"Filesystem sync initiated for project '{project_config.name}'",
}
else:
report = await sync_service.sync(
project_config.home, project_config.name, force_full=force_full
)
logger.info(
f"Filesystem sync completed for project: {project_config.name} (force_full={force_full})"
)
return SyncReportResponse.from_sync_report(report)
@project_router.post("/status", response_model=SyncReportResponse)
async def project_sync_status(
sync_service: SyncServiceDep,
project_config: ProjectConfigDep,
) -> SyncReportResponse:
"""Scan directory for changes compared to database state.
Args:
sync_service: Sync service for this project
project_config: Project configuration
Returns:
Scan report with details on files that need syncing
"""
logger.info(f"Scanning filesystem for project: {project_config.name}") # pragma: no cover
sync_report = await sync_service.scan(project_config.home) # pragma: no cover
return SyncReportResponse.from_sync_report(sync_report) # pragma: no cover
# List all available projects
@project_resource_router.get("/projects", response_model=ProjectList)
async def list_projects(
project_service: ProjectServiceDep,
) -> ProjectList:
"""List all configured projects.
Returns:
A list of all projects with metadata
"""
projects = await project_service.list_projects()
default_project = project_service.default_project
project_items = [
ProjectItem(
id=project.id,
external_id=project.external_id,
name=project.name,
path=normalize_project_path(project.path),
is_default=project.is_default or False,
)
for project in projects
]
return ProjectList(
projects=project_items,
default_project=default_project,
)
# Add a new project
@project_resource_router.post("/projects", response_model=ProjectStatusResponse, status_code=201)
async def add_project(
response: Response,
project_data: ProjectInfoRequest,
project_service: ProjectServiceDep,
) -> ProjectStatusResponse:
"""Add a new project to configuration and database.
Args:
project_data: The project name and path, with option to set as default
Returns:
Response confirming the project was added
"""
# Check if project already exists before attempting to add
existing_project = await project_service.get_project(project_data.name)
if existing_project:
# Project exists - check if paths match for true idempotency
# Normalize paths for comparison (resolve symlinks, etc.)
from pathlib import Path
requested_path = Path(project_data.path).resolve()
existing_path = Path(existing_project.path).resolve()
if requested_path == existing_path:
# Same name, same path - return 200 OK (idempotent)
response.status_code = 200
return ProjectStatusResponse( # pyright: ignore [reportCallIssue]
message=f"Project '{project_data.name}' already exists",
status="success",
default=existing_project.is_default or False,
new_project=ProjectItem(
id=existing_project.id,
external_id=existing_project.external_id,
name=existing_project.name,
path=existing_project.path,
is_default=existing_project.is_default or False,
),
)
else:
# Same name, different path - this is an error
raise HTTPException(
status_code=400,
detail=f"Project '{project_data.name}' already exists with different path. Existing: {existing_project.path}, Requested: {project_data.path}",
)
try: # pragma: no cover
# The service layer now handles cloud mode validation and path sanitization
await project_service.add_project(
project_data.name, project_data.path, set_default=project_data.set_default
)
# Fetch the newly created project to get its ID
new_project = await project_service.get_project(project_data.name)
if not new_project:
raise HTTPException(status_code=500, detail="Failed to retrieve newly created project")
return ProjectStatusResponse( # pyright: ignore [reportCallIssue]
message=f"Project '{new_project.name}' added successfully",
status="success",
default=project_data.set_default,
new_project=ProjectItem(
id=new_project.id,
external_id=new_project.external_id,
name=new_project.name,
path=new_project.path,
is_default=new_project.is_default or False,
),
)
except ValueError as e: # pragma: no cover
raise HTTPException(status_code=400, detail=str(e))
# Remove a project
@project_resource_router.delete("/{name}", response_model=ProjectStatusResponse)
async def remove_project(
project_service: ProjectServiceDep,
name: str = Path(..., description="Name of the project to remove"),
delete_notes: bool = Query(
False, description="If True, delete project directory from filesystem"
),
) -> ProjectStatusResponse:
"""Remove a project from configuration and database.
Args:
name: The name of the project to remove
delete_notes: If True, delete the project directory from the filesystem
Returns:
Response confirming the project was removed
"""
try:
old_project = await project_service.get_project(name)
if not old_project: # pragma: no cover
raise HTTPException(
status_code=404, detail=f"Project: '{name}' does not exist"
) # pragma: no cover
# Check if trying to delete the default project
# In cloud mode, database is source of truth; in local mode, check config
config_default = project_service.default_project
db_default = await project_service.repository.get_default_project()
# Use database default if available, otherwise fall back to config default
default_project_name = db_default.name if db_default else config_default
if name == default_project_name:
available_projects = await project_service.list_projects()
other_projects = [p.name for p in available_projects if p.name != name]
detail = f"Cannot delete default project '{name}'. "
if other_projects:
detail += (
f"Set another project as default first. Available: {', '.join(other_projects)}"
)
else:
detail += "This is the only project in your configuration."
raise HTTPException(status_code=400, detail=detail)
await project_service.remove_project(name, delete_notes=delete_notes)
return ProjectStatusResponse(
message=f"Project '{old_project.name}' removed successfully",
status="success",
default=False,
old_project=ProjectItem(
id=old_project.id,
external_id=old_project.external_id,
name=old_project.name,
path=old_project.path,
is_default=old_project.is_default or False,
),
new_project=None,
)
except ValueError as e: # pragma: no cover
raise HTTPException(status_code=400, detail=str(e))
# Set a project as default
@project_resource_router.put("/{name}/default", response_model=ProjectStatusResponse)
async def set_default_project(
project_service: ProjectServiceDep,
name: str = Path(..., description="Name of the project to set as default"),
) -> ProjectStatusResponse:
"""Set a project as the default project.
Args:
name: The name of the project to set as default
Returns:
Response confirming the project was set as default
"""
try:
# Get the old default project
default_name = project_service.default_project
default_project = await project_service.get_project(default_name)
if not default_project: # pragma: no cover
raise HTTPException( # pragma: no cover
status_code=404, detail=f"Default Project: '{default_name}' does not exist"
)
# get the new project
new_default_project = await project_service.get_project(name)
if not new_default_project: # pragma: no cover
raise HTTPException(
status_code=404, detail=f"Project: '{name}' does not exist"
) # pragma: no cover
await project_service.set_default_project(name)
return ProjectStatusResponse(
message=f"Project '{name}' set as default successfully",
status="success",
default=True,
old_project=ProjectItem(
id=default_project.id,
external_id=default_project.external_id,
name=default_name,
path=default_project.path,
is_default=False,
),
new_project=ProjectItem(
id=new_default_project.id,
external_id=new_default_project.external_id,
name=name,
path=new_default_project.path,
is_default=True,
),
)
except ValueError as e: # pragma: no cover
raise HTTPException(status_code=400, detail=str(e))
# Get the default project
@project_resource_router.get("/default", response_model=ProjectItem)
async def get_default_project(
project_service: ProjectServiceDep,
) -> ProjectItem:
"""Get the default project.
Returns:
Response with project default information
"""
# Get the default project
# In cloud mode, database is source of truth; in local mode, check config
config_default = project_service.default_project
db_default = await project_service.repository.get_default_project()
# Use database default if available, otherwise fall back to config default
default_name = db_default.name if db_default else config_default
default_project = await project_service.get_project(default_name)
if not default_project: # pragma: no cover
raise HTTPException( # pragma: no cover
status_code=404, detail=f"Default Project: '{default_name}' does not exist"
)
return ProjectItem(
id=default_project.id,
external_id=default_project.external_id,
name=default_project.name,
path=default_project.path,
is_default=True,
)
# Synchronize projects between config and database
@project_resource_router.post("/config/sync", response_model=ProjectStatusResponse)
async def synchronize_projects(
project_service: ProjectServiceDep,
) -> ProjectStatusResponse:
"""Synchronize projects between configuration file and database.
Ensures that all projects in the configuration file exist in the database
and vice versa.
Returns:
Response confirming synchronization was completed
"""
try: # pragma: no cover
await project_service.synchronize_projects()
return ProjectStatusResponse( # pyright: ignore [reportCallIssue]
message="Projects synchronized successfully between configuration and database",
status="success",
default=False,
)
except ValueError as e: # pragma: no cover
raise HTTPException(status_code=400, detail=str(e))
@@ -1,260 +0,0 @@
"""Router for prompt-related operations.
This router is responsible for rendering various prompts using Handlebars templates.
It centralizes all prompt formatting logic that was previously in the MCP prompts.
"""
from datetime import datetime, timezone
from fastapi import APIRouter, HTTPException, status
from loguru import logger
from basic_memory.api.routers.utils import to_graph_context, to_search_results
from basic_memory.api.template_loader import template_loader
from basic_memory.schemas.base import parse_timeframe
from basic_memory.deps import (
ContextServiceDep,
EntityRepositoryDep,
SearchServiceDep,
EntityServiceDep,
)
from basic_memory.schemas.prompt import (
ContinueConversationRequest,
SearchPromptRequest,
PromptResponse,
PromptMetadata,
)
from basic_memory.schemas.search import SearchItemType, SearchQuery
router = APIRouter(prefix="/prompt", tags=["prompt"])
@router.post("/continue-conversation", response_model=PromptResponse)
async def continue_conversation(
search_service: SearchServiceDep,
entity_service: EntityServiceDep,
context_service: ContextServiceDep,
entity_repository: EntityRepositoryDep,
request: ContinueConversationRequest,
) -> PromptResponse:
"""Generate a prompt for continuing a conversation.
This endpoint takes a topic and/or timeframe and generates a prompt with
relevant context from the knowledge base.
Args:
request: The request parameters
Returns:
Formatted continuation prompt with context
"""
logger.info(
f"Generating continue conversation prompt, topic: {request.topic}, timeframe: {request.timeframe}"
)
since = parse_timeframe(request.timeframe) if request.timeframe else None
# Initialize search results
search_results = []
# Get data needed for template
if request.topic:
query = SearchQuery(text=request.topic, after_date=request.timeframe)
results = await search_service.search(query, limit=request.search_items_limit)
search_results = await to_search_results(entity_service, results)
# Build context from results
all_hierarchical_results = []
for result in search_results:
if hasattr(result, "permalink") and result.permalink:
# Get hierarchical context using the new dataclass-based approach
context_result = await context_service.build_context(
result.permalink,
depth=request.depth,
since=since,
max_related=request.related_items_limit,
include_observations=True, # Include observations for entities
)
# Process results into the schema format
graph_context = await to_graph_context(
context_result, entity_repository=entity_repository
)
# Add results to our collection (limit to top results for each permalink)
if graph_context.results:
all_hierarchical_results.extend(graph_context.results[:3])
# Limit to a reasonable number of total results
all_hierarchical_results = all_hierarchical_results[:10]
template_context = {
"topic": request.topic,
"timeframe": request.timeframe,
"hierarchical_results": all_hierarchical_results,
"has_results": len(all_hierarchical_results) > 0,
}
else:
# If no topic, get recent activity
context_result = await context_service.build_context(
types=[SearchItemType.ENTITY],
depth=request.depth,
since=since,
max_related=request.related_items_limit,
include_observations=True,
)
recent_context = await to_graph_context(context_result, entity_repository=entity_repository)
hierarchical_results = recent_context.results[:5] # Limit to top 5 recent items
template_context = {
"topic": f"Recent Activity from ({request.timeframe})",
"timeframe": request.timeframe,
"hierarchical_results": hierarchical_results,
"has_results": len(hierarchical_results) > 0,
}
try:
# Render template
rendered_prompt = await template_loader.render(
"prompts/continue_conversation.hbs", template_context
)
# Calculate metadata
# Count items of different types
observation_count = 0
relation_count = 0
entity_count = 0
# Get the hierarchical results from the template context
hierarchical_results_for_count = template_context.get("hierarchical_results", [])
# For topic-based search
if request.topic:
for item in hierarchical_results_for_count:
if hasattr(item, "observations"):
observation_count += len(item.observations) if item.observations else 0
if hasattr(item, "related_results"):
for related in item.related_results or []:
if hasattr(related, "type"):
if related.type == "relation":
relation_count += 1
elif related.type == "entity": # pragma: no cover
entity_count += 1 # pragma: no cover
# For recent activity
else:
for item in hierarchical_results_for_count:
if hasattr(item, "observations"):
observation_count += len(item.observations) if item.observations else 0
if hasattr(item, "related_results"):
for related in item.related_results or []:
if hasattr(related, "type"):
if related.type == "relation":
relation_count += 1
elif related.type == "entity": # pragma: no cover
entity_count += 1 # pragma: no cover
# Build metadata
metadata = {
"query": request.topic,
"timeframe": request.timeframe,
"search_count": len(search_results)
if request.topic
else 0, # Original search results count
"context_count": len(hierarchical_results_for_count),
"observation_count": observation_count,
"relation_count": relation_count,
"total_items": (
len(hierarchical_results_for_count)
+ observation_count
+ relation_count
+ entity_count
),
"search_limit": request.search_items_limit,
"context_depth": request.depth,
"related_limit": request.related_items_limit,
"generated_at": datetime.now(timezone.utc).isoformat(),
}
prompt_metadata = PromptMetadata(**metadata)
return PromptResponse(
prompt=rendered_prompt, context=template_context, metadata=prompt_metadata
)
except Exception as e:
logger.error(f"Error rendering continue conversation template: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error rendering prompt template: {str(e)}",
)
@router.post("/search", response_model=PromptResponse)
async def search_prompt(
search_service: SearchServiceDep,
entity_service: EntityServiceDep,
request: SearchPromptRequest,
page: int = 1,
page_size: int = 10,
) -> PromptResponse:
"""Generate a prompt for search results.
This endpoint takes a search query and formats the results into a helpful
prompt with context and suggestions.
Args:
request: The search parameters
page: The page number for pagination
page_size: The number of results per page, defaults to 10
Returns:
Formatted search results prompt with context
"""
logger.info(f"Generating search prompt, query: {request.query}, timeframe: {request.timeframe}")
limit = page_size
offset = (page - 1) * page_size
query = SearchQuery(text=request.query, after_date=request.timeframe)
results = await search_service.search(query, limit=limit, offset=offset)
search_results = await to_search_results(entity_service, results)
template_context = {
"query": request.query,
"timeframe": request.timeframe,
"results": search_results,
"has_results": len(search_results) > 0,
"result_count": len(search_results),
}
try:
# Render template
rendered_prompt = await template_loader.render("prompts/search.hbs", template_context)
# Build metadata
metadata = {
"query": request.query,
"timeframe": request.timeframe,
"search_count": len(search_results),
"context_count": len(search_results),
"observation_count": 0, # Search results don't include observations
"relation_count": 0, # Search results don't include relations
"total_items": len(search_results),
"search_limit": limit,
"context_depth": 0, # No context depth for basic search
"related_limit": 0, # No related items for basic search
"generated_at": datetime.now(timezone.utc).isoformat(),
}
prompt_metadata = PromptMetadata(**metadata)
return PromptResponse(
prompt=rendered_prompt, context=template_context, metadata=prompt_metadata
)
except Exception as e:
logger.error(f"Error rendering search template: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error rendering prompt template: {str(e)}",
)
@@ -1,252 +0,0 @@
"""Routes for getting entity content."""
import tempfile
import uuid
from pathlib import Path
from typing import Annotated, Union
from fastapi import APIRouter, HTTPException, BackgroundTasks, Body, Response
from fastapi.responses import FileResponse, JSONResponse
from loguru import logger
from basic_memory.deps import (
ProjectConfigDep,
LinkResolverDep,
SearchServiceDep,
EntityServiceDep,
FileServiceDep,
EntityRepositoryDep,
)
from basic_memory.repository.search_repository import SearchIndexRow
from basic_memory.schemas.memory import normalize_memory_url
from basic_memory.schemas.search import SearchQuery, SearchItemType
from basic_memory.models.knowledge import Entity as EntityModel
from datetime import datetime
router = APIRouter(prefix="/resource", tags=["resources"])
def _mtime_to_datetime(entity: EntityModel) -> datetime:
"""Convert entity mtime (file modification time) to datetime.
Returns the file's actual modification time, falling back to updated_at
if mtime is not available.
"""
if entity.mtime: # pragma: no cover
return datetime.fromtimestamp(entity.mtime).astimezone() # pragma: no cover
return entity.updated_at
def get_entity_ids(item: SearchIndexRow) -> set[int]:
match item.type:
case SearchItemType.ENTITY:
return {item.id}
case SearchItemType.OBSERVATION:
return {item.entity_id} # pyright: ignore [reportReturnType]
case SearchItemType.RELATION:
from_entity = item.from_id
to_entity = item.to_id # pyright: ignore [reportReturnType]
return {from_entity, to_entity} if to_entity else {from_entity} # pyright: ignore [reportReturnType]
case _: # pragma: no cover
raise ValueError(f"Unexpected type: {item.type}")
@router.get("/{identifier:path}", response_model=None)
async def get_resource_content(
config: ProjectConfigDep,
link_resolver: LinkResolverDep,
search_service: SearchServiceDep,
entity_service: EntityServiceDep,
file_service: FileServiceDep,
background_tasks: BackgroundTasks,
identifier: str,
page: int = 1,
page_size: int = 10,
) -> Union[Response, FileResponse]:
"""Get resource content by identifier: name or permalink."""
logger.debug(f"Getting content for: {identifier}")
# Find single entity by permalink
entity = await link_resolver.resolve_link(identifier)
results = [entity] if entity else []
# pagination for multiple results
limit = page_size
offset = (page - 1) * page_size
# search using the identifier as a permalink
if not results:
# if the identifier contains a wildcard, use GLOB search
query = (
SearchQuery(permalink_match=identifier)
if "*" in identifier
else SearchQuery(permalink=identifier)
)
search_results = await search_service.search(query, limit, offset)
if not search_results:
raise HTTPException(status_code=404, detail=f"Resource not found: {identifier}")
# get the deduplicated entities related to the search results
entity_ids = {id for result in search_results for id in get_entity_ids(result)}
results = await entity_service.get_entities_by_id(list(entity_ids))
# return single response
if len(results) == 1:
entity = results[0]
# Check file exists via file_service (for cloud compatibility)
if not await file_service.exists(entity.file_path):
raise HTTPException(
status_code=404,
detail=f"File not found: {entity.file_path}",
)
# Read content via file_service as bytes (works with both local and S3)
content = await file_service.read_file_bytes(entity.file_path)
content_type = file_service.content_type(entity.file_path)
return Response(content=content, media_type=content_type)
# for multiple files, initialize a temporary file for writing the results
with tempfile.NamedTemporaryFile(delete=False, mode="w", suffix=".md") as tmp_file:
temp_file_path = tmp_file.name
for result in results:
# Read content for each entity
content = await file_service.read_entity_content(result)
memory_url = normalize_memory_url(result.permalink)
modified_date = _mtime_to_datetime(result).isoformat()
checksum = result.checksum[:8] if result.checksum else ""
# Prepare the delimited content
response_content = f"--- {memory_url} {modified_date} {checksum}\n"
response_content += f"\n{content}\n"
response_content += "\n"
# Write content directly to the temporary file in append mode
tmp_file.write(response_content)
# Ensure all content is written to disk
tmp_file.flush()
# Schedule the temporary file to be deleted after the response
background_tasks.add_task(cleanup_temp_file, temp_file_path)
# Return the file response
return FileResponse(path=temp_file_path)
def cleanup_temp_file(file_path: str):
"""Delete the temporary file."""
try:
Path(file_path).unlink() # Deletes the file
logger.debug(f"Temporary file deleted: {file_path}")
except Exception as e: # pragma: no cover
logger.error(f"Error deleting temporary file {file_path}: {e}")
@router.put("/{file_path:path}")
async def write_resource(
config: ProjectConfigDep,
file_service: FileServiceDep,
entity_repository: EntityRepositoryDep,
search_service: SearchServiceDep,
file_path: str,
content: Annotated[str, Body()],
) -> JSONResponse:
"""Write content to a file in the project.
This endpoint allows writing content directly to a file in the project.
Also creates an entity record and indexes the file for search.
Args:
file_path: Path to write to, relative to project root
request: Contains the content to write
Returns:
JSON response with file information
"""
try:
# Get content from request body
# Defensive type checking: ensure content is a string
# FastAPI should validate this, but if a dict somehow gets through
# (e.g., via JSON body parsing), we need to catch it here
if isinstance(content, dict):
logger.error( # pragma: no cover
f"Error writing resource {file_path}: "
f"content is a dict, expected string. Keys: {list(content.keys())}"
)
raise HTTPException( # pragma: no cover
status_code=400,
detail="content must be a string, not a dict. "
"Ensure request body is sent as raw string content, not JSON object.",
)
# Ensure it's UTF-8 string content
if isinstance(content, bytes): # pragma: no cover
content_str = content.decode("utf-8")
else:
content_str = str(content)
# Cloud compatibility: do not assume a local filesystem path structure.
# Delegate directory creation + writes to the configured FileService (local or S3).
await file_service.ensure_directory(Path(file_path).parent)
checksum = await file_service.write_file(file_path, content_str)
# Get file info
file_metadata = await file_service.get_file_metadata(file_path)
# Determine file details
file_name = Path(file_path).name
content_type = file_service.content_type(file_path)
entity_type = "canvas" if file_path.endswith(".canvas") else "file"
# Check if entity already exists
existing_entity = await entity_repository.get_by_file_path(file_path)
if existing_entity:
# Update existing entity
entity = await entity_repository.update(
existing_entity.id,
{
"title": file_name,
"entity_type": entity_type,
"content_type": content_type,
"file_path": file_path,
"checksum": checksum,
"updated_at": file_metadata.modified_at,
},
)
status_code = 200
else:
# Create a new entity model
# Explicitly set external_id to ensure NOT NULL constraint is satisfied (fixes #512)
entity = EntityModel(
external_id=str(uuid.uuid4()),
title=file_name,
entity_type=entity_type,
content_type=content_type,
file_path=file_path,
checksum=checksum,
created_at=file_metadata.created_at,
updated_at=file_metadata.modified_at,
)
entity = await entity_repository.add(entity)
status_code = 201
# Index the file for search
await search_service.index_entity(entity) # pyright: ignore
# Return success response
return JSONResponse(
status_code=status_code,
content={
"file_path": file_path,
"checksum": checksum,
"size": file_metadata.size,
"created_at": file_metadata.created_at.timestamp(),
"modified_at": file_metadata.modified_at.timestamp(),
},
)
except Exception as e: # pragma: no cover
logger.error(f"Error writing resource {file_path}: {e}")
raise HTTPException(status_code=500, detail=f"Failed to write resource: {str(e)}")
@@ -1,36 +0,0 @@
"""Router for search operations."""
from fastapi import APIRouter, BackgroundTasks
from basic_memory.api.routers.utils import to_search_results
from basic_memory.schemas.search import SearchQuery, SearchResponse
from basic_memory.deps import SearchServiceDep, EntityServiceDep
router = APIRouter(prefix="/search", tags=["search"])
@router.post("/", response_model=SearchResponse)
async def search(
query: SearchQuery,
search_service: SearchServiceDep,
entity_service: EntityServiceDep,
page: int = 1,
page_size: int = 10,
):
"""Search across all knowledge and documents."""
limit = page_size
offset = (page - 1) * page_size
results = await search_service.search(query, limit=limit, offset=offset)
search_results = await to_search_results(entity_service, results)
return SearchResponse(
results=search_results,
current_page=page,
page_size=page_size,
)
@router.post("/reindex")
async def reindex(background_tasks: BackgroundTasks, search_service: SearchServiceDep):
"""Recreate and populate the search index."""
await search_service.reindex_all(background_tasks=background_tasks)
return {"status": "ok", "message": "Reindex initiated"}
@@ -10,7 +10,7 @@ Key improvements:
- Simplified caching strategies
"""
from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends, Response, Path
from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends, Response, Path, Query
from loguru import logger
from basic_memory.deps import (
@@ -19,9 +19,10 @@ from basic_memory.deps import (
LinkResolverV2ExternalDep,
ProjectConfigV2ExternalDep,
AppConfigDep,
SyncServiceV2ExternalDep,
EntityRepositoryV2ExternalDep,
ProjectExternalIdPathDep,
TaskSchedulerDep,
FileServiceV2ExternalDep,
)
from basic_memory.schemas import DeleteEntitiesResponse
from basic_memory.schemas.base import Entity
@@ -38,26 +39,6 @@ from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteRe
router = APIRouter(prefix="/knowledge", tags=["knowledge-v2"])
async def resolve_relations_background(sync_service, entity_id: int, entity_permalink: str) -> None:
"""Background task to resolve relations for a specific entity.
This runs asynchronously after the API response is sent, preventing
long delays when creating entities with many relations.
"""
try: # pragma: no cover
# Only resolve relations for the newly created entity
await sync_service.resolve_relations(entity_id=entity_id) # pragma: no cover
logger.debug( # pragma: no cover
f"Background: Resolved relations for entity {entity_permalink} (id={entity_id})"
)
except Exception as e: # pragma: no cover
# Log but don't fail - this is a background task
logger.warning( # pragma: no cover
f"Background: Failed to resolve relations for entity {entity_permalink}: {e}"
)
## Resolution endpoint
@@ -186,24 +167,43 @@ async def create_entity(
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
task_scheduler: TaskSchedulerDep,
file_service: FileServiceV2ExternalDep,
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)
Created entity with generated external_id (UUID) and file content
"""
logger.info(
"API v2 request", endpoint="create_entity", entity_type=data.entity_type, title=data.title
)
entity = await entity_service.create_entity(data)
if fast:
entity = await entity_service.fast_write_entity(data)
task_scheduler.schedule(
"reindex_entity",
entity_id=entity.id,
project_id=project_id,
)
else:
entity = await entity_service.create_entity(data)
await search_service.index_entity(entity, background_tasks=background_tasks)
# reindex
await search_service.index_entity(entity, background_tasks=background_tasks)
result = EntityResponseV2.model_validate(entity)
if fast:
result = result.model_copy(update={"observations": [], "relations": []})
# Always read and return file content
content = await file_service.read_file_content(entity.file_path)
result = result.model_copy(update={"content": content})
logger.info(
f"API v2 response: endpoint='create_entity' external_id={entity.external_id}, title={result.title}, permalink={result.permalink}, status_code=201"
@@ -222,9 +222,13 @@ async def update_entity_by_id(
project_id: ProjectExternalIdPathDep,
entity_service: EntityServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
sync_service: SyncServiceV2ExternalDep,
entity_repository: EntityRepositoryV2ExternalDep,
task_scheduler: TaskSchedulerDep,
file_service: FileServiceV2ExternalDep,
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.
@@ -233,30 +237,55 @@ 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
Updated entity with file content
"""
logger.info(f"API v2 request: update_entity_by_id entity_id={entity_id}")
# Check if entity exists
# Check if entity exists (external_id is the source of truth for v2)
existing = await entity_repository.get_by_external_id(entity_id)
created = existing is None
# Perform update or create
entity, _ = await entity_service.create_or_update_entity(data)
response.status_code = 201 if created else 200
# reindex
await search_service.index_entity(entity, background_tasks=background_tasks)
# Schedule relation resolution for new entities
if created:
background_tasks.add_task( # pragma: no cover
resolve_relations_background, sync_service, entity.id, entity.permalink or ""
if fast:
entity = await entity_service.fast_write_entity(data, external_id=entity_id)
response.status_code = 200 if existing else 201
task_scheduler.schedule(
"reindex_entity",
entity_id=entity.id,
project_id=project_id,
resolve_relations=created,
)
else:
if existing:
# Update the existing entity in-place to avoid path-based duplication
entity = await entity_service.update_entity(existing, data)
response.status_code = 200
else:
# Create new entity, then bind external_id to the requested UUID
entity = await entity_service.create_entity(data)
if entity.external_id != entity_id:
entity = await entity_repository.update(
entity.id,
{"external_id": entity_id},
)
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, background_tasks=background_tasks)
result = EntityResponseV2.model_validate(entity)
if fast:
result = result.model_copy(update={"observations": [], "relations": []})
# Always read and return file content
content = await file_service.read_file_content(entity.file_path)
result = result.model_copy(update={"content": content})
logger.info(
f"API v2 response: external_id={entity_id}, created={created}, status_code={response.status_code}"
@@ -272,16 +301,22 @@ async def edit_entity_by_id(
entity_service: EntityServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
entity_repository: EntityRepositoryV2ExternalDep,
task_scheduler: TaskSchedulerDep,
file_service: FileServiceV2ExternalDep,
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
Updated entity with file content
Raises:
HTTPException: 404 if entity not found, 400 if edit fails
@@ -298,21 +333,41 @@ async def edit_entity_by_id(
)
try:
# Edit using the entity's permalink or path
identifier = entity.permalink or entity.file_path
updated_entity = await entity_service.edit_entity(
identifier=identifier,
operation=data.operation,
content=data.content,
section=data.section,
find_text=data.find_text,
expected_replacements=data.expected_replacements,
)
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,
)
task_scheduler.schedule(
"reindex_entity",
entity_id=updated_entity.id,
project_id=project_id,
)
else:
# Edit using the entity's permalink or path
identifier = entity.permalink or entity.file_path
updated_entity = await entity_service.edit_entity(
identifier=identifier,
operation=data.operation,
content=data.content,
section=data.section,
find_text=data.find_text,
expected_replacements=data.expected_replacements,
)
# Reindex
await search_service.index_entity(updated_entity, background_tasks=background_tasks)
await search_service.index_entity(updated_entity, background_tasks=background_tasks)
result = EntityResponseV2.model_validate(updated_entity)
if fast:
result = result.model_copy(update={"observations": [], "relations": []})
# Always read and return file content
content = await file_service.read_file_content(updated_entity.file_path)
result = result.model_copy(update={"content": content})
logger.info(
f"API v2 response: external_id={entity_id}, operation='{data.operation}', status_code=200"
@@ -16,7 +16,7 @@ from basic_memory.schemas.memory import (
normalize_memory_url,
)
from basic_memory.schemas.search import SearchItemType
from basic_memory.api.routers.utils import to_graph_context
from basic_memory.api.v2.utils import to_graph_context
# Note: No prefix here - it's added during registration as /v2/{project_id}/memory
router = APIRouter(tags=["memory"])
@@ -19,9 +19,17 @@ from loguru import logger
from basic_memory.deps import (
ProjectServiceDep,
ProjectRepositoryDep,
ProjectConfigV2ExternalDep,
SyncServiceV2ExternalDep,
TaskSchedulerDep,
ProjectExternalIdPathDep,
)
from basic_memory.schemas import SyncReportResponse
from basic_memory.schemas.project_info import (
ProjectItem,
ProjectList,
ProjectInfoRequest,
ProjectInfoResponse,
ProjectStatusResponse,
)
from basic_memory.schemas.v2 import ProjectResolveRequest, ProjectResolveResponse
@@ -30,6 +38,177 @@ from basic_memory.utils import normalize_project_path, generate_permalink
router = APIRouter(prefix="/projects", tags=["project_management-v2"])
@router.get("/", response_model=ProjectList)
async def list_projects(
project_service: ProjectServiceDep,
) -> ProjectList:
"""List all configured projects.
Returns:
A list of all projects with metadata
"""
projects = await project_service.list_projects()
default_project = project_service.default_project
project_items = [
ProjectItem(
id=project.id,
external_id=project.external_id,
name=project.name,
path=normalize_project_path(project.path),
is_default=project.is_default or False,
)
for project in projects
]
return ProjectList(
projects=project_items,
default_project=default_project,
)
@router.post("/", response_model=ProjectStatusResponse, status_code=201)
async def add_project(
project_data: ProjectInfoRequest,
project_service: ProjectServiceDep,
) -> ProjectStatusResponse:
"""Add a new project to configuration and database.
Args:
project_data: The project name and path, with option to set as default
Returns:
Response confirming the project was added
"""
# Check if project already exists before attempting to add
existing_project = await project_service.get_project(project_data.name)
if existing_project:
# Project exists - check if paths match for true idempotency
# Normalize paths for comparison (resolve symlinks, etc.)
from pathlib import Path
requested_path = Path(project_data.path).resolve()
existing_path = Path(existing_project.path).resolve()
if requested_path == existing_path:
# Same name, same path - return 200 OK (idempotent)
return ProjectStatusResponse( # pyright: ignore [reportCallIssue]
message=f"Project '{project_data.name}' already exists",
status="success",
default=existing_project.is_default or False,
new_project=ProjectItem(
id=existing_project.id,
external_id=existing_project.external_id,
name=existing_project.name,
path=existing_project.path,
is_default=existing_project.is_default or False,
),
)
else:
# Same name, different path - this is an error
raise HTTPException(
status_code=400,
detail=(
f"Project '{project_data.name}' already exists with different path. "
f"Existing: {existing_project.path}, Requested: {project_data.path}"
),
)
try: # pragma: no cover
# The service layer handles cloud mode validation and path sanitization
await project_service.add_project(
project_data.name, project_data.path, set_default=project_data.set_default
)
# Fetch the newly created project to get its ID
new_project = await project_service.get_project(project_data.name)
if not new_project:
raise HTTPException(status_code=500, detail="Failed to retrieve newly created project")
return ProjectStatusResponse( # pyright: ignore [reportCallIssue]
message=f"Project '{new_project.name}' added successfully",
status="success",
default=project_data.set_default,
new_project=ProjectItem(
id=new_project.id,
external_id=new_project.external_id,
name=new_project.name,
path=new_project.path,
is_default=new_project.is_default or False,
),
)
except ValueError as e: # pragma: no cover
raise HTTPException(status_code=400, detail=str(e))
@router.post("/config/sync", response_model=ProjectStatusResponse)
async def synchronize_projects(
project_service: ProjectServiceDep,
) -> ProjectStatusResponse:
"""Synchronize projects between configuration file and database."""
try: # pragma: no cover
await project_service.synchronize_projects()
return ProjectStatusResponse( # pyright: ignore [reportCallIssue]
message="Projects synchronized successfully between configuration and database",
status="success",
default=False,
)
except ValueError as e: # pragma: no cover
raise HTTPException(status_code=400, detail=str(e))
@router.post("/{project_id}/sync")
async def sync_project(
sync_service: SyncServiceV2ExternalDep,
project_config: ProjectConfigV2ExternalDep,
task_scheduler: TaskSchedulerDep,
project_internal_id: ProjectExternalIdPathDep,
force_full: bool = Query(
False, description="Force full scan, bypassing watermark optimization"
),
run_in_background: bool = Query(True, description="Run in background"),
):
"""Force project filesystem sync to database."""
if run_in_background:
task_scheduler.schedule(
"sync_project",
project_id=project_internal_id,
force_full=force_full,
)
logger.info(
f"Filesystem sync initiated for project: {project_config.name} (force_full={force_full})"
)
return {
"status": "sync_started",
"message": f"Filesystem sync initiated for project '{project_config.name}'",
}
report = await sync_service.sync(
project_config.home, project_config.name, force_full=force_full
)
logger.info(
f"Filesystem sync completed for project: {project_config.name} (force_full={force_full})"
)
return SyncReportResponse.from_sync_report(report)
@router.post("/{project_id}/status", response_model=SyncReportResponse)
async def get_project_status(
sync_service: SyncServiceV2ExternalDep,
project_config: ProjectConfigV2ExternalDep,
project_id: str = Path(..., description="Project external ID (UUID)"),
force_full: bool = Query(
False, description="Force full scan, bypassing watermark optimization"
),
) -> SyncReportResponse:
"""Get sync status of files vs database for a project."""
logger.info(f"API v2 request: get_project_status for project_id={project_id}")
report = await sync_service.scan(project_config.home, force_full=force_full)
return SyncReportResponse.from_sync_report(report)
@router.post("/resolve", response_model=ProjectResolveResponse)
async def resolve_project_identifier(
data: ProjectResolveRequest,
@@ -147,6 +326,22 @@ async def get_project_by_id(
)
@router.get("/{project_id}/info", response_model=ProjectInfoResponse)
async def get_project_info_by_id(
project_service: ProjectServiceDep,
project_repository: ProjectRepositoryDep,
project_id: str = Path(..., description="Project external ID (UUID)"),
) -> ProjectInfoResponse:
"""Get detailed project information by external ID."""
logger.info(f"API v2 request: get_project_info_by_id for project_id={project_id}")
project = await project_repository.get_by_external_id(project_id)
if not project:
raise HTTPException(
status_code=404, detail=f"Project with external_id '{project_id}' not found"
)
return await project_service.get_project_info(project.name)
@router.patch("/{project_id}", response_model=ProjectStatusResponse)
async def update_project_by_id(
project_service: ProjectServiceDep,
@@ -9,7 +9,7 @@ from datetime import datetime, timezone
from fastapi import APIRouter, HTTPException, status, Path
from loguru import logger
from basic_memory.api.routers.utils import to_graph_context, to_search_results
from basic_memory.api.v2.utils import to_graph_context, to_search_results
from basic_memory.api.template_loader import template_loader
from basic_memory.schemas.base import parse_timeframe
from basic_memory.deps import (
@@ -4,11 +4,16 @@ This router uses external_id UUIDs for stable, API-friendly routing.
V1 uses string-based project names which are less efficient and less stable.
"""
from fastapi import APIRouter, BackgroundTasks, Path
from fastapi import APIRouter, Path
from basic_memory.api.routers.utils import to_search_results
from basic_memory.api.v2.utils import to_search_results
from basic_memory.schemas.search import SearchQuery, SearchResponse
from basic_memory.deps import SearchServiceV2ExternalDep, EntityServiceV2ExternalDep
from basic_memory.deps import (
SearchServiceV2ExternalDep,
EntityServiceV2ExternalDep,
TaskSchedulerDep,
ProjectExternalIdPathDep,
)
# Note: No prefix here - it's added during registration as /v2/{project_id}/search
router = APIRouter(tags=["search"])
@@ -51,9 +56,8 @@ async def search(
@router.post("/search/reindex")
async def reindex(
background_tasks: BackgroundTasks,
search_service: SearchServiceV2ExternalDep,
project_id: str = Path(..., description="Project external UUID"),
task_scheduler: TaskSchedulerDep,
project_id: ProjectExternalIdPathDep,
):
"""Recreate and populate the search index for a project.
@@ -63,11 +67,10 @@ async def reindex(
Args:
project_id: Project external UUID from URL path
background_tasks: FastAPI background tasks handler
search_service: Search service scoped to project
task_scheduler: Task scheduler for background work
Returns:
Status message indicating reindex has been initiated
"""
await search_service.reindex_all(background_tasks=background_tasks)
task_scheduler.schedule("reindex_project", project_id=project_id)
return {"status": "ok", "message": "Reindex initiated"}
@@ -24,29 +24,42 @@ async def to_graph_context(
page: Optional[int] = None,
page_size: Optional[int] = None,
):
# First pass: collect all entity IDs needed for relations
# First pass: collect all entity IDs needed for external_id lookup
# This includes: entity primary results, observation parent entities, relation from/to entities
entity_ids_needed: set[int] = set()
for context_item in context_result.results:
for item in (
[context_item.primary_result] + context_item.observations + context_item.related_results
):
if item.type == SearchItemType.RELATION:
if item.type == SearchItemType.ENTITY:
# Entity's own ID for its external_id
entity_ids_needed.add(item.id)
elif item.type == SearchItemType.OBSERVATION:
# Parent entity ID for entity_external_id
if item.entity_id: # pyright: ignore
entity_ids_needed.add(item.entity_id) # pyright: ignore
elif item.type == SearchItemType.RELATION:
# Source and target entity IDs for external_ids
if item.from_id: # pyright: ignore
entity_ids_needed.add(item.from_id) # pyright: ignore
if item.to_id:
entity_ids_needed.add(item.to_id)
# Batch fetch all entities at once
entity_lookup: dict[int, str] = {}
# Batch fetch all entities at once - get both title and external_id
entity_title_lookup: dict[int, str] = {}
entity_external_id_lookup: dict[int, str] = {}
if entity_ids_needed:
entities = await entity_repository.find_by_ids(list(entity_ids_needed))
entity_lookup = {e.id: e.title for e in entities}
for e in entities:
entity_title_lookup[e.id] = e.title
entity_external_id_lookup[e.id] = e.external_id
# Helper function to convert items to summaries
def to_summary(item: SearchIndexRow | ContextResultRow):
match item.type:
case SearchItemType.ENTITY:
return EntitySummary(
external_id=entity_external_id_lookup.get(item.id, ""),
entity_id=item.id,
title=item.title, # pyright: ignore
permalink=item.permalink,
@@ -55,10 +68,14 @@ async def to_graph_context(
created_at=item.created_at,
)
case SearchItemType.OBSERVATION:
entity_ext_id = None
if item.entity_id: # pyright: ignore
entity_ext_id = entity_external_id_lookup.get(item.entity_id) # pyright: ignore
return ObservationSummary(
observation_id=item.id,
entity_id=item.entity_id, # pyright: ignore
title=item.title, # pyright: ignore
entity_external_id=entity_ext_id,
title=entity_title_lookup.get(item.entity_id), # pyright: ignore
file_path=item.file_path,
category=item.category, # pyright: ignore
content=item.content, # pyright: ignore
@@ -66,8 +83,10 @@ async def to_graph_context(
created_at=item.created_at,
)
case SearchItemType.RELATION:
from_title = entity_lookup.get(item.from_id) if item.from_id else None # pyright: ignore
to_title = entity_lookup.get(item.to_id) if item.to_id else None
from_title = entity_title_lookup.get(item.from_id) if item.from_id else None # pyright: ignore
to_title = entity_title_lookup.get(item.to_id) if item.to_id else None
from_ext_id = entity_external_id_lookup.get(item.from_id) if item.from_id else None # pyright: ignore
to_ext_id = entity_external_id_lookup.get(item.to_id) if item.to_id else None
return RelationSummary(
relation_id=item.id,
entity_id=item.entity_id, # pyright: ignore
@@ -77,8 +96,10 @@ async def to_graph_context(
relation_type=item.relation_type, # pyright: ignore
from_entity=from_title,
from_entity_id=item.from_id, # pyright: ignore
from_entity_external_id=from_ext_id,
to_entity=to_title,
to_entity_id=item.to_id,
to_entity_external_id=to_ext_id,
created_at=item.created_at,
)
case _: # pragma: no cover
@@ -30,7 +30,7 @@ async def fetch_cloud_projects(
config = config_manager.config
host_url = config.cloud_host.rstrip("/")
response = await api_request(method="GET", url=f"{host_url}/proxy/projects/projects")
response = await api_request(method="GET", url=f"{host_url}/proxy/v2/projects/")
return CloudProjectList.model_validate(response.json())
except Exception as e:
@@ -66,7 +66,7 @@ async def create_cloud_project(
response = await api_request(
method="POST",
url=f"{host_url}/proxy/projects/projects",
url=f"{host_url}/proxy/v2/projects/",
headers={"Content-Type": "application/json"},
json_data=project_data.model_dump(),
)
@@ -58,7 +58,7 @@ async def run_sync(
try:
async with get_client() as client:
project_item = await get_active_project(client, project, None)
url = f"{project_item.project_url}/project/sync"
url = f"/v2/projects/{project_item.external_id}/sync"
params = []
if force_full:
params.append("force_full=true")
@@ -92,7 +92,7 @@ async def get_project_info(project: str):
try:
async with get_client() as client:
project_item = await get_active_project(client, project, None)
response = await call_get(client, f"{project_item.project_url}/project/info")
response = await call_get(client, f"/v2/projects/{project_item.external_id}/info")
return ProjectInfoResponse.model_validate(response.json())
except (ToolError, ValueError) as e:
console.print(f"[red]Sync failed: {e}[/red]")
+25 -17
View File
@@ -17,6 +17,7 @@ from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.tools.utils import call_delete, call_get, call_patch, call_post, call_put
from basic_memory.schemas.project_info import ProjectList, ProjectStatusResponse
from basic_memory.schemas.v2 import ProjectResolveResponse
from basic_memory.utils import generate_permalink, normalize_project_path
# Import rclone commands for project sync
@@ -65,7 +66,7 @@ def list_projects(
async def _list_projects():
async with get_client() as client:
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
return ProjectList.model_validate(response.json())
try:
@@ -167,7 +168,7 @@ def add_project(
"local_sync_path": local_sync_path,
"set_default": set_default,
}
response = await call_post(client, "/projects/projects", json=data)
response = await call_post(client, "/v2/projects/", json=data)
return ProjectStatusResponse.model_validate(response.json())
else:
# Local mode: path is required
@@ -181,7 +182,7 @@ def add_project(
async def _add_project():
async with get_client() as client:
data = {"name": name, "path": resolved_path, "set_default": set_default}
response = await call_post(client, "/projects/projects", json=data)
response = await call_post(client, "/v2/projects/", json=data)
return ProjectStatusResponse.model_validate(response.json())
try:
@@ -234,7 +235,7 @@ def setup_project_sync(
async def _verify_project_exists():
"""Verify the project exists on cloud by listing all projects."""
async with get_client() as client:
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
project_list = response.json()
project_names = [p["name"] for p in project_list["projects"]]
if name not in project_names:
@@ -433,7 +434,7 @@ def synchronize_projects(
async def _sync_config():
async with get_client() as client:
response = await call_post(client, "/projects/config/sync")
response = await call_post(client, "/v2/projects/config/sync")
return ProjectStatusResponse.model_validate(response.json())
try:
@@ -475,10 +476,15 @@ def move_project(
async def _move_project():
async with get_client() as client:
data = {"path": resolved_path}
project_permalink = generate_permalink(name)
# TODO fix route to use ProjectPathDep
response = await call_patch(client, f"/{name}/project/{project_permalink}", json=data)
resolve_response = await call_post(
client,
"/v2/projects/resolve",
json={"identifier": name},
)
project_info = ProjectResolveResponse.model_validate(resolve_response.json())
response = await call_patch(
client, f"/v2/projects/{project_info.external_id}", json=data
)
return ProjectStatusResponse.model_validate(response.json())
try:
@@ -530,7 +536,7 @@ def sync_project_command(
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
@@ -571,9 +577,10 @@ def sync_project_command(
async def _trigger_db_sync():
async with get_client() as client:
permalink = generate_permalink(name)
response = await call_post(
client, f"/{permalink}/project/sync?force_full=true", json={}
client,
f"/v2/projects/{project_data.external_id}/sync?force_full=true",
json={},
)
return response.json()
@@ -621,7 +628,7 @@ def bisync_project_command(
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
@@ -669,9 +676,10 @@ def bisync_project_command(
async def _trigger_db_sync():
async with get_client() as client:
permalink = generate_permalink(name)
response = await call_post(
client, f"/{permalink}/project/sync?force_full=true", json={}
client,
f"/v2/projects/{project_data.external_id}/sync?force_full=true",
json={},
)
return response.json()
@@ -715,7 +723,7 @@ def check_project_command(
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
@@ -816,7 +824,7 @@ def ls_project_command(
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
+1 -1
View File
@@ -146,7 +146,7 @@ async def run_status(project: Optional[str] = None, verbose: bool = False): # p
try:
async with get_client() as client:
project_item = await get_active_project(client, project, None)
response = await call_post(client, f"{project_item.project_url}/project/status")
response = await call_post(client, f"/v2/projects/{project_item.external_id}/status")
sync_report = SyncReportResponse.model_validate(response.json())
display_changes(project_item.name, "Status", sync_report, verbose)
+66 -9
View File
@@ -1,5 +1,6 @@
"""CLI tool commands for Basic Memory."""
import json
import sys
from typing import Annotated, List, Optional
@@ -288,7 +289,10 @@ def recent_activity(
@tool_app.command("search-notes")
def search_notes(
query: str,
query: Annotated[
Optional[str],
typer.Argument(help="Search query string (optional when using metadata filters)"),
] = "",
permalink: Annotated[bool, typer.Option("--permalink", help="Search permalink values")] = False,
title: Annotated[bool, typer.Option("--title", help="Search title values")] = False,
project: Annotated[
@@ -301,6 +305,26 @@ def search_notes(
Optional[str],
typer.Option("--after_date", help="Search results after date, eg. '2d', '1 week'"),
] = None,
tags: Annotated[
Optional[List[str]],
typer.Option("--tag", help="Filter by frontmatter tag (repeatable)"),
] = None,
status: Annotated[
Optional[str],
typer.Option("--status", help="Filter by frontmatter status"),
] = None,
note_types: Annotated[
Optional[List[str]],
typer.Option("--type", help="Filter by frontmatter type (repeatable)"),
] = None,
meta: Annotated[
Optional[List[str]],
typer.Option("--meta", help="Filter by frontmatter key=value (repeatable)"),
] = None,
filter_json: Annotated[
Optional[str],
typer.Option("--filter", help="JSON metadata filter (advanced)"),
] = None,
page: int = 1,
page_size: int = 10,
local: bool = typer.Option(
@@ -335,26 +359,59 @@ def search_notes(
)
raise typer.Exit(1)
# Build metadata filters from --filter and --meta
metadata_filters = {}
if filter_json:
try:
metadata_filters = json.loads(filter_json)
if not isinstance(metadata_filters, dict):
raise ValueError("Metadata filter JSON must be an object")
except json.JSONDecodeError as e:
typer.echo(f"Invalid JSON for --filter: {e}", err=True)
raise typer.Exit(1)
if meta:
for item in meta:
if "=" not in item:
typer.echo(
f"Invalid --meta entry '{item}'. Use key=value format.",
err=True,
)
raise typer.Exit(1)
key, value = item.split("=", 1)
key = key.strip()
if not key:
typer.echo(f"Invalid --meta entry '{item}'.", err=True)
raise typer.Exit(1)
metadata_filters[key] = value
if not metadata_filters:
metadata_filters = None
# set search type
search_type = ("permalink" if permalink else None,)
search_type = ("permalink_match" if permalink and "*" in query else None,)
search_type = ("title" if title else None,)
search_type = "text" if search_type is None else search_type
search_type = "text"
if permalink:
search_type = "permalink"
if query and "*" in query:
search_type = "permalink"
if title:
search_type = "title"
with force_routing(local=local, cloud=cloud):
results = run_with_cleanup(
mcp_search.fn(
query,
query or "",
project_name,
search_type=search_type,
page=page,
after_date=after_date,
page_size=page_size,
types=note_types,
metadata_filters=metadata_filters,
tags=tags,
status=status,
)
)
# Use json module for more controlled serialization
import json
results_dict = results.model_dump(exclude_none=True)
print(json.dumps(results_dict, indent=2, ensure_ascii=True, default=str))
except ValueError as e:
+4
View File
@@ -91,6 +91,8 @@ from basic_memory.deps.services import (
FileServiceV2Dep,
get_file_service_v2_external,
FileServiceV2ExternalDep,
get_task_scheduler,
TaskSchedulerDep,
get_search_service,
SearchServiceDep,
get_search_service_v2,
@@ -227,6 +229,8 @@ __all__ = [
"FileServiceV2Dep",
"get_file_service_v2_external",
"FileServiceV2ExternalDep",
"get_task_scheduler",
"TaskSchedulerDep",
"get_search_service",
"SearchServiceDep",
"get_search_service_v2",
+83 -2
View File
@@ -7,7 +7,8 @@ This module provides service-layer dependencies:
- SyncService, ProjectService, DirectoryService
"""
from typing import Annotated
import asyncio
from typing import Annotated, Any, Callable, Coroutine, Mapping, Protocol
from fastapi import Depends
from loguru import logger
@@ -43,7 +44,6 @@ from basic_memory.services.link_resolver import LinkResolver
from basic_memory.services.search_service import SearchService
from basic_memory.sync import SyncService
# --- Entity Parser ---
@@ -430,6 +430,87 @@ async def get_sync_service_v2_external(
SyncServiceV2ExternalDep = Annotated[SyncService, Depends(get_sync_service_v2_external)]
# --- Background Task Scheduler ---
class TaskScheduler(Protocol):
def schedule(self, task_name: str, **payload: Any) -> None:
"""Schedule a background task by name."""
def _log_task_failure(completed: asyncio.Task) -> None:
try:
completed.result()
except Exception as exc: # pragma: no cover
logger.exception("Background task failed", error=str(exc))
class LocalTaskScheduler:
"""Default scheduler that runs tasks in-process via asyncio.create_task."""
def __init__(
self,
handlers: Mapping[str, Callable[..., Coroutine[Any, Any, None]]],
) -> None:
self._handlers = handlers
def schedule(self, task_name: str, **payload: Any) -> None:
handler = self._handlers.get(task_name)
# Trigger: task name is not registered
# Why: avoid silently dropping background work
# Outcome: fail fast to surface misconfiguration
if not handler:
raise ValueError(f"Unknown task name: {task_name}")
task = asyncio.create_task(handler(**payload))
task.add_done_callback(_log_task_failure)
async def get_task_scheduler(
entity_service: EntityServiceV2ExternalDep,
sync_service: SyncServiceV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
project_config: ProjectConfigV2ExternalDep,
) -> TaskScheduler:
"""Create a scheduler that maps task specs to coroutines."""
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)
async def _resolve_relations(entity_id: int, **_: Any) -> None:
await sync_service.resolve_relations(entity_id=entity_id)
async def _sync_project(force_full: bool = False, **_: Any) -> None:
await sync_service.sync(
project_config.home,
project_config.name,
force_full=force_full,
)
async def _reindex_project(**_: Any) -> None:
await search_service.reindex_all()
return LocalTaskScheduler(
{
"reindex_entity": _reindex_entity,
"resolve_relations": _resolve_relations,
"sync_project": _sync_project,
"reindex_project": _reindex_project,
}
)
TaskSchedulerDep = Annotated[TaskScheduler, Depends(get_task_scheduler)]
# --- Project Service ---
+4 -3
View File
@@ -9,6 +9,7 @@ from frontmatter import Post
from basic_memory.file_utils import has_frontmatter, remove_frontmatter, parse_frontmatter
from basic_memory.markdown import EntityMarkdown
from basic_memory.markdown.entity_parser import normalize_frontmatter_metadata
from basic_memory.models import Entity
from basic_memory.models import Observation as ObservationModel
@@ -58,9 +59,9 @@ def entity_model_from_markdown(
model.created_at = markdown.created
model.updated_at = markdown.modified
# Handle metadata - ensure all values are strings and filter None
metadata = markdown.frontmatter.metadata or {}
model.entity_metadata = {k: str(v) for k, v in metadata.items() if v is not None}
# Handle metadata - normalize values and filter None (preserve structured data)
metadata = normalize_frontmatter_metadata(markdown.frontmatter.metadata or {})
model.entity_metadata = {k: v for k, v in metadata.items() if v is not None}
# Get project_id from entity if not provided
obs_project_id = project_id or (model.project_id if hasattr(model, "project_id") else None)
+23 -3
View File
@@ -43,7 +43,9 @@ class KnowledgeClient:
# --- Entity CRUD Operations ---
async def create_entity(self, entity_data: dict[str, Any]) -> EntityResponse:
async def create_entity(
self, entity_data: dict[str, Any], *, fast: bool | None = None
) -> EntityResponse:
"""Create a new entity.
Args:
@@ -55,14 +57,22 @@ class KnowledgeClient:
Raises:
ToolError: If the request fails
"""
params = {"fast": fast} if fast is not None else None
response = await call_post(
self.http_client,
f"{self._base_path}/entities",
json=entity_data,
params=params,
)
return EntityResponse.model_validate(response.json())
async def update_entity(self, entity_id: str, entity_data: dict[str, Any]) -> EntityResponse:
async def update_entity(
self,
entity_id: str,
entity_data: dict[str, Any],
*,
fast: bool | None = None,
) -> EntityResponse:
"""Update an existing entity (full replacement).
Args:
@@ -75,10 +85,12 @@ class KnowledgeClient:
Raises:
ToolError: If the request fails
"""
params = {"fast": fast} if fast is not None else None
response = await call_put(
self.http_client,
f"{self._base_path}/entities/{entity_id}",
json=entity_data,
params=params,
)
return EntityResponse.model_validate(response.json())
@@ -100,7 +112,13 @@ class KnowledgeClient:
)
return EntityResponse.model_validate(response.json())
async def patch_entity(self, entity_id: str, patch_data: dict[str, Any]) -> EntityResponse:
async def patch_entity(
self,
entity_id: str,
patch_data: dict[str, Any],
*,
fast: bool | None = None,
) -> EntityResponse:
"""Partially update an entity.
Args:
@@ -113,10 +131,12 @@ class KnowledgeClient:
Raises:
ToolError: If the request fails
"""
params = {"fast": fast} if fast is not None else None
response = await call_patch(
self.http_client,
f"{self._base_path}/entities/{entity_id}",
json=patch_data,
params=params,
)
return EntityResponse.model_validate(response.json())
+2 -2
View File
@@ -47,7 +47,7 @@ class ProjectClient:
"""
response = await call_get(
self.http_client,
"/projects/projects",
"/v2/projects/",
)
return ProjectList.model_validate(response.json())
@@ -65,7 +65,7 @@ class ProjectClient:
"""
response = await call_post(
self.http_client,
"/projects/projects",
"/v2/projects/",
json=project_data,
)
return ProjectStatusResponse.model_validate(response.json())
+17 -6
View File
@@ -19,7 +19,7 @@ from fastmcp import Context
from basic_memory.config import ConfigManager
from basic_memory.project_resolver import ProjectResolver
from basic_memory.schemas.project_info import ProjectItem, ProjectList
from basic_memory.utils import generate_permalink
from basic_memory.schemas.v2 import ProjectResolveResponse
async def resolve_project_parameter(
@@ -78,7 +78,7 @@ async def get_project_names(client: AsyncClient, headers: HeaderTypes | None = N
# Deferred import to avoid circular dependency with tools
from basic_memory.mcp.tools.utils import call_get
response = await call_get(client, "/projects/projects", headers=headers)
response = await call_get(client, "/v2/projects/", headers=headers)
project_list = ProjectList.model_validate(response.json())
return [project.name for project in project_list.projects]
@@ -104,7 +104,7 @@ async def get_active_project(
HTTPError: If project doesn't exist or is inaccessible
"""
# Deferred import to avoid circular dependency with tools
from basic_memory.mcp.tools.utils import call_get
from basic_memory.mcp.tools.utils import call_post
resolved_project = await resolve_project_parameter(project)
if not resolved_project:
@@ -126,9 +126,20 @@ async def get_active_project(
# Validate project exists by calling API
logger.debug(f"Validating project: {project}")
permalink = generate_permalink(project)
response = await call_get(client, f"/{permalink}/project/item", headers=headers)
active_project = ProjectItem.model_validate(response.json())
response = await call_post(
client,
"/v2/projects/resolve",
json={"identifier": project},
headers=headers,
)
resolved = ProjectResolveResponse.model_validate(response.json())
active_project = ProjectItem(
id=resolved.project_id,
external_id=resolved.external_id,
name=resolved.name,
path=resolved.path,
is_default=resolved.is_default,
)
# Cache in context if available
if context:
@@ -9,8 +9,9 @@ from typing import Annotated, Optional
from loguru import logger
from pydantic import Field
from basic_memory.config import get_project_config
from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_post
from basic_memory.schemas.prompt import ContinueConversationRequest
@@ -42,17 +43,18 @@ async def continue_conversation(
logger.info(f"Continuing session, topic: {topic}, timeframe: {timeframe}")
async with get_client() as client:
config = ConfigManager().config
active_project = await get_active_project(client, project=config.default_project)
# Create request model
request = ContinueConversationRequest( # pyright: ignore [reportCallIssue]
topic=topic, timeframe=timeframe
)
project_url = get_project_config().project_url
# Call the prompt API endpoint
response = await call_post(
client,
f"{project_url}/prompt/continue-conversation",
f"/v2/projects/{active_project.external_id}/prompt/continue-conversation",
json=request.model_dump(exclude_none=True),
)
@@ -46,9 +46,7 @@ async def recent_activity_prompt(
# Call the tool function - it returns a well-formatted string
# Pass type as string values (not enum) to match the tool's expected input
activity_summary = await recent_activity.fn(
project=project, timeframe=timeframe, type="entity"
)
activity_summary = await recent_activity.fn(project=project, timeframe=timeframe, type="entity")
# Build the prompt response
# The tool already returns formatted markdown, so we use it directly
@@ -92,7 +90,7 @@ async def recent_activity_prompt(
- summarizes [[Recent Work]]
''',
folder="insights",
project="{project or 'default'}"
project="{project or "default"}"
)
```
""")
+8 -4
View File
@@ -8,8 +8,9 @@ from typing import Annotated, Optional
from loguru import logger
from pydantic import Field
from basic_memory.config import get_project_config
from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_post
from basic_memory.schemas.prompt import SearchPromptRequest
@@ -41,14 +42,17 @@ async def search_prompt(
logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}")
async with get_client() as client:
config = ConfigManager().config
active_project = await get_active_project(client, project=config.default_project)
# Create request model
request = SearchPromptRequest(query=query, timeframe=timeframe)
project_url = get_project_config().project_url
# Call the prompt API endpoint
response = await call_post(
client, f"{project_url}/prompt/search", json=request.model_dump(exclude_none=True)
client,
f"/v2/projects/{active_project.external_id}/prompt/search",
json=request.model_dump(exclude_none=True),
)
# Extract the rendered prompt from the response
@@ -62,10 +62,9 @@ async def project_info(
async with get_client() as client:
project_config = await get_active_project(client, project, context)
project_url = project_config.permalink
# Call the API endpoint
response = await call_get(client, f"{project_url}/project/info")
response = await call_get(client, f"/v2/projects/{project_config.external_id}/info")
# Convert response to ProjectInfoResponse
return ProjectInfoResponse.model_validate(response.json())
+2 -1
View File
@@ -13,7 +13,7 @@ from basic_memory.mcp.tools.recent_activity import recent_activity
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.mcp.tools.view_note import view_note
from basic_memory.mcp.tools.write_note import write_note
from basic_memory.mcp.tools.search import search_notes
from basic_memory.mcp.tools.search import search_notes, search_by_metadata
from basic_memory.mcp.tools.canvas import canvas
from basic_memory.mcp.tools.list_directory import list_directory
from basic_memory.mcp.tools.edit_note import edit_note
@@ -42,6 +42,7 @@ __all__ = [
"read_note",
"recent_activity",
"search",
"search_by_metadata",
"search_notes",
"view_note",
"write_note",
+1 -1
View File
@@ -256,7 +256,7 @@ async def edit_note(
edit_data["expected_replacements"] = str(expected_replacements)
# Call the PATCH endpoint
result = await knowledge_client.patch_entity(entity_id, edit_data)
result = await knowledge_client.patch_entity(entity_id, edit_data, fast=False)
# Format summary
summary = [
+3 -1
View File
@@ -488,7 +488,9 @@ move_note("{identifier}", "notes/{destination_path.split("/")[-1] if "/" in dest
return "\n".join(result_lines)
except Exception as e: # pragma: no cover
logger.error(f"Directory move failed for '{identifier}' to '{destination_path}': {e}")
logger.error(
f"Directory move failed for '{identifier}' to '{destination_path}': {e}"
)
return f"""# Directory Move Failed
Error moving directory '{identifier}' to '{destination_path}': {str(e)}
@@ -144,7 +144,7 @@ async def recent_activity(
)
# Get list of all projects
response = await call_get(client, "/projects/projects")
response = await call_get(client, "/v2/projects/")
project_list = ProjectList.model_validate(response.json())
projects_activity = {}
+113 -1
View File
@@ -1,7 +1,7 @@
"""Search tools for Basic Memory MCP server."""
from textwrap import dedent
from typing import List, Optional
from typing import List, Optional, Dict, Any
from loguru import logger
from fastmcp import Context
@@ -207,6 +207,9 @@ async def search_notes(
types: List[str] | None = None,
entity_types: List[str] | None = None,
after_date: Optional[str] = None,
metadata_filters: Optional[Dict[str, Any]] = None,
tags: Optional[List[str]] = None,
status: Optional[str] = None,
context: Context | None = None,
) -> SearchResponse | str:
"""Search across all content in the knowledge base with comprehensive syntax support.
@@ -248,6 +251,27 @@ async def search_notes(
- `search_notes("research", "query", entity_types=["observation"])` - Filter by entity type
- `search_notes("team-docs", "query", after_date="2024-01-01")` - Recent content only
- `search_notes("my-project", "query", after_date="1 week")` - Relative date filtering
- `search_notes("my-project", "query", tags=["security"])` - Filter by frontmatter tags
- `search_notes("my-project", "query", status="in-progress")` - Filter by frontmatter status
- `search_notes("my-project", "query", metadata_filters={"priority": {"$in": ["high"]}})`
### Structured Metadata Filters
Filters are exact matches on frontmatter metadata. Supported forms:
- Equality: `{"status": "in-progress"}`
- Array contains (all): `{"tags": ["security", "oauth"]}`
- Operators:
- `$in`: `{"priority": {"$in": ["high", "critical"]}}`
- `$gt`, `$gte`, `$lt`, `$lte`: `{"schema.confidence": {"$gt": 0.7}}`
- `$between`: `{"schema.confidence": {"$between": [0.3, 0.6]}}`
- Nested keys use dot notation (e.g., `"schema.confidence"`).
### Filter-only Searches
You can pass an empty query string when only using structured filters:
- `search_notes("my-project", "", metadata_filters={"type": "spec"})`
### Convenience Filters
`tags` and `status` are shorthand for metadata_filters. If the same key exists in
metadata_filters, that value wins.
### Advanced Pattern Examples
- `search_notes("work-project", "project AND (meeting OR discussion)")` - Complex boolean logic
@@ -265,6 +289,9 @@ async def search_notes(
types: Optional list of note types to search (e.g., ["note", "person"])
entity_types: Optional list of entity types to filter by (e.g., ["entity", "observation"])
after_date: Optional date filter for recent content (e.g., "1 week", "2d", "2024-01-01")
metadata_filters: Optional structured frontmatter filters (e.g., {"status": "in-progress"})
tags: Optional tag filter (frontmatter tags); shorthand for metadata_filters["tags"]
status: Optional status filter (frontmatter status); shorthand for metadata_filters["status"]
context: Optional FastMCP context for performance caching.
Returns:
@@ -355,6 +382,12 @@ async def search_notes(
search_query.types = types
if after_date:
search_query.after_date = after_date
if metadata_filters:
search_query.metadata_filters = metadata_filters
if tags:
search_query.tags = tags
if status:
search_query.status = status
async with get_client() as client:
active_project = await get_active_project(client, project, context)
@@ -387,3 +420,82 @@ async def search_notes(
logger.error(f"Search failed for query '{query}': {e}, project: {active_project.name}")
# Return formatted error message as string for better user experience
return _format_search_error_response(active_project.name, str(e), query, search_type)
@mcp.tool(
description="Search entities by structured frontmatter metadata.",
)
async def search_by_metadata(
filters: Dict[str, Any],
project: Optional[str] = None,
limit: int = 20,
offset: int = 0,
context: Context | None = None,
) -> SearchResponse | str:
"""Search entities by structured frontmatter metadata.
Args:
filters: Dictionary of metadata filters (e.g., {"status": "in-progress"})
project: Project name to search in. Optional - server will resolve using hierarchy.
limit: Maximum number of results to return
offset: Number of results to skip (for pagination)
context: Optional FastMCP context for performance caching.
Returns:
SearchResponse with results, or helpful error guidance if search fails
"""
if limit <= 0:
return "# Error\n\n`limit` must be greater than 0."
# Build a structured-only search query
search_query = SearchQuery()
search_query.metadata_filters = filters
search_query.entity_types = [SearchItemType.ENTITY]
# Convert offset/limit to page/page_size (API uses paging)
page_size = limit
page = (offset // limit) + 1
offset_within_page = offset % limit
async with get_client() as client:
active_project = await get_active_project(client, project, context)
logger.info(
f"Structured search in project {active_project.name} filters={filters} limit={limit} offset={offset}"
)
try:
from basic_memory.mcp.clients import SearchClient
search_client = SearchClient(client, active_project.external_id)
result = await search_client.search(
search_query.model_dump(),
page=page,
page_size=page_size,
)
# Apply offset within page, fetch next page if needed
if offset_within_page:
remaining = result.results[offset_within_page:]
if len(remaining) < limit:
next_page = page + 1
extra = await search_client.search(
search_query.model_dump(),
page=next_page,
page_size=page_size,
)
remaining.extend(extra.results[: max(0, limit - len(remaining))])
result = SearchResponse(
results=remaining[:limit],
current_page=page,
page_size=page_size,
)
return result
except Exception as e:
logger.error(
f"Metadata search failed for filters '{filters}': {e}, project: {active_project.name}"
)
return _format_search_error_response(
active_project.name, str(e), str(filters), "metadata"
)
+4 -2
View File
@@ -159,7 +159,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())
result = await knowledge_client.create_entity(entity.model_dump(), fast=False)
action = "Created"
except Exception as e:
# If creation failed due to conflict (already exists), try to update
@@ -175,7 +175,9 @@ async def write_note(
"Entity permalink is required for updates"
) # pragma: no cover
entity_id = await knowledge_client.resolve_entity(entity.permalink)
result = await knowledge_client.update_entity(entity_id, entity.model_dump())
result = await knowledge_client.update_entity(
entity_id, entity.model_dump(), fast=False
)
action = "Updated"
except Exception as update_error: # pragma: no cover
# Re-raise the original error if update also fails
@@ -0,0 +1,131 @@
"""Helpers for parsing structured metadata filters for search."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
import re
from typing import Any, Iterable, List
_KEY_RE = re.compile(r"^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$")
_NUMERIC_RE = re.compile(r"^-?\d+(\.\d+)?$")
@dataclass(frozen=True)
class ParsedMetadataFilter:
"""Normalized metadata filter for SQL generation."""
path_parts: List[str]
op: str
value: Any
comparison: str | None = None # "numeric" or "text" for comparisons
def _is_numeric_value(value: Any) -> bool:
if isinstance(value, bool):
return False
if isinstance(value, (int, float)):
return True
if isinstance(value, str):
return bool(_NUMERIC_RE.match(value.strip()))
return False
def _is_numeric_collection(values: Iterable[Any]) -> bool:
return all(_is_numeric_value(v) for v in values)
def _normalize_scalar(value: Any) -> Any:
if isinstance(value, datetime):
return value.isoformat()
if isinstance(value, date):
return value.isoformat()
if isinstance(value, bool):
return str(value)
if isinstance(value, (int, float)):
return str(value)
return value
def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter]:
"""Parse metadata filters into normalized clauses.
Supported forms:
- {"status": "in-progress"}
- {"tags": ["security", "oauth"]} # array contains all
- {"priority": {"$in": ["high", "critical"]}}
- {"schema.confidence": {"$gt": 0.7}}
- {"schema.confidence": {"$between": [0.3, 0.6]}}
"""
parsed: List[ParsedMetadataFilter] = []
for raw_key, raw_value in (filters or {}).items():
if not isinstance(raw_key, str) or not raw_key.strip():
raise ValueError("metadata filter keys must be non-empty strings")
key = raw_key.strip()
if not _KEY_RE.match(key):
raise ValueError(f"Unsupported metadata filter key: {raw_key}")
path_parts = key.split(".")
# Operator form
if isinstance(raw_value, dict):
if len(raw_value) != 1:
raise ValueError(f"Invalid metadata filter for '{raw_key}': {raw_value}")
op, value = next(iter(raw_value.items()))
if op == "$in":
if not isinstance(value, list) or not value:
raise ValueError(f"$in requires a non-empty list for '{raw_key}'")
parsed.append(
ParsedMetadataFilter(path_parts, "in", [_normalize_scalar(v) for v in value])
)
continue
if op in {"$gt", "$gte", "$lt", "$lte"}:
normalized = _normalize_scalar(value)
comparison = "numeric" if _is_numeric_value(normalized) else "text"
parsed.append(
ParsedMetadataFilter(path_parts, op.lstrip("$"), normalized, comparison)
)
continue
if op == "$between":
if not isinstance(value, list) or len(value) != 2:
raise ValueError(f"$between requires [min, max] for '{raw_key}'")
normalized = [_normalize_scalar(v) for v in value]
comparison = "numeric" if _is_numeric_collection(normalized) else "text"
parsed.append(ParsedMetadataFilter(path_parts, "between", normalized, comparison))
continue
raise ValueError(f"Unsupported operator '{op}' in metadata filter for '{raw_key}'")
# Array contains (all)
if isinstance(raw_value, list):
if not raw_value:
raise ValueError(f"Empty list not allowed for metadata filter '{raw_key}'")
parsed.append(
ParsedMetadataFilter(
path_parts, "contains", [_normalize_scalar(v) for v in raw_value]
)
)
continue
# Simple equality
parsed.append(ParsedMetadataFilter(path_parts, "eq", _normalize_scalar(raw_value)))
return parsed
def build_sqlite_json_path(parts: List[str]) -> str:
"""Build a SQLite JSON path for json_extract/json_each."""
path = "$"
for part in parts:
path += f'."{part}"'
return path
def build_postgres_json_path(parts: List[str]) -> str:
"""Build a Postgres JSON path for #>>/#> operators."""
return "{" + ",".join(parts) + "}"
@@ -12,6 +12,10 @@ from sqlalchemy import text
from basic_memory import db
from basic_memory.repository.search_index_row import SearchIndexRow
from basic_memory.repository.search_repository_base import SearchRepositoryBase
from basic_memory.repository.metadata_filters import (
parse_metadata_filters,
build_postgres_json_path,
)
from basic_memory.schemas.search import SearchItemType
@@ -215,6 +219,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
types: Optional[List[str]] = None,
after_date: Optional[datetime] = None,
search_item_types: Optional[List[SearchItemType]] = None,
metadata_filters: Optional[dict] = None,
limit: int = 10,
offset: int = 0,
) -> List[SearchIndexRow]:
@@ -222,6 +227,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
conditions = []
params = {}
order_by_clause = ""
from_clause = "search_index"
# Handle text search for title and content using tsvector
if search_text:
@@ -233,18 +239,22 @@ class PostgresSearchRepository(SearchRepositoryBase):
processed_text = self._prepare_search_term(search_text.strip())
params["text"] = processed_text
# Use @@ operator for tsvector matching
conditions.append("textsearchable_index_col @@ to_tsquery('english', :text)")
conditions.append(
"search_index.textsearchable_index_col @@ to_tsquery('english', :text)"
)
# Handle title search
if title:
title_text = self._prepare_search_term(title.strip(), is_prefix=False)
params["title_text"] = title_text
conditions.append("to_tsvector('english', title) @@ to_tsquery('english', :title_text)")
conditions.append(
"to_tsvector('english', search_index.title) @@ to_tsquery('english', :title_text)"
)
# Handle permalink exact search
if permalink:
params["permalink"] = permalink
conditions.append("permalink = :permalink")
conditions.append("search_index.permalink = :permalink")
# Handle permalink pattern match
if permalink_match:
@@ -255,14 +265,14 @@ class PostgresSearchRepository(SearchRepositoryBase):
# Convert * to % for SQL LIKE
permalink_pattern = permalink_text.replace("*", "%")
params["permalink"] = permalink_pattern
conditions.append("permalink LIKE :permalink")
conditions.append("search_index.permalink LIKE :permalink")
else:
conditions.append("permalink = :permalink")
conditions.append("search_index.permalink = :permalink")
# Handle search item type filter
if search_item_types:
type_list = ", ".join(f"'{t.value}'" for t in search_item_types)
conditions.append(f"type IN ({type_list})")
conditions.append(f"search_index.type IN ({type_list})")
# Handle entity type filter using JSONB containment
if types:
@@ -270,19 +280,91 @@ class PostgresSearchRepository(SearchRepositoryBase):
type_conditions = []
for entity_type in types:
# Create JSONB containment condition for each type
type_conditions.append(f'metadata @> \'{{"entity_type": "{entity_type}"}}\'')
type_conditions.append(
f'search_index.metadata @> \'{{"entity_type": "{entity_type}"}}\''
)
conditions.append(f"({' OR '.join(type_conditions)})")
# Handle date filter
if after_date:
params["after_date"] = after_date
conditions.append("created_at > :after_date")
conditions.append("search_index.created_at > :after_date")
# order by most recent first
order_by_clause = ", updated_at DESC"
order_by_clause = ", search_index.updated_at DESC"
# Handle structured metadata filters (frontmatter)
if metadata_filters:
parsed_filters = parse_metadata_filters(metadata_filters)
from_clause = "search_index JOIN entity ON search_index.entity_id = entity.id"
metadata_expr = "entity.entity_metadata::jsonb"
for idx, filt in enumerate(parsed_filters):
path = build_postgres_json_path(filt.path_parts)
text_expr = f"({metadata_expr} #>> '{path}')"
json_expr = f"({metadata_expr} #> '{path}')"
if filt.op == "eq":
value_param = f"meta_val_{idx}"
params[value_param] = filt.value
conditions.append(f"{text_expr} = :{value_param}")
continue
if filt.op == "in":
placeholders = []
for j, val in enumerate(filt.value):
value_param = f"meta_val_{idx}_{j}"
params[value_param] = val
placeholders.append(f":{value_param}")
conditions.append(f"{text_expr} IN ({', '.join(placeholders)})")
continue
if filt.op == "contains":
import json as _json
base_param = f"meta_val_{idx}"
tag_conditions = []
# Require all values to be present
for j, val in enumerate(filt.value):
tag_param = f"{base_param}_{j}"
params[tag_param] = _json.dumps([val])
like_param = f"{base_param}_{j}_like"
params[like_param] = f'%"{val}"%'
like_param_single = f"{base_param}_{j}_like_single"
params[like_param_single] = f"%'{val}'%"
tag_conditions.append(
f"({json_expr} @> :{tag_param}::jsonb "
f"OR {text_expr} LIKE :{like_param} "
f"OR {text_expr} LIKE :{like_param_single})"
)
conditions.append(" AND ".join(tag_conditions))
continue
if filt.op in {"gt", "gte", "lt", "lte", "between"}:
if filt.comparison == "numeric":
numeric_expr = (
f"CASE WHEN ({text_expr}) ~ '^-?\\\\d+(\\\\.\\\\d+)?$' "
f"THEN ({text_expr})::double precision END"
)
compare_expr = numeric_expr
else:
compare_expr = text_expr
if filt.op == "between":
min_param = f"meta_val_{idx}_min"
max_param = f"meta_val_{idx}_max"
params[min_param] = filt.value[0]
params[max_param] = filt.value[1]
conditions.append(f"{compare_expr} BETWEEN :{min_param} AND :{max_param}")
else:
value_param = f"meta_val_{idx}"
params[value_param] = filt.value
operator = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="}[filt.op]
conditions.append(f"{compare_expr} {operator} :{value_param}")
continue
# Always filter by project_id
params["project_id"] = self.project_id
conditions.append("project_id = :project_id")
conditions.append("search_index.project_id = :project_id")
# set limit and offset
params["limit"] = limit
@@ -294,31 +376,33 @@ class PostgresSearchRepository(SearchRepositoryBase):
# Build SQL with ts_rank() for scoring
# Note: If no text search, score will be NULL, so we use COALESCE to default to 0
if search_text and search_text.strip() and search_text.strip() != "*":
score_expr = "ts_rank(textsearchable_index_col, to_tsquery('english', :text))"
score_expr = (
"ts_rank(search_index.textsearchable_index_col, to_tsquery('english', :text))"
)
else:
score_expr = "0"
sql = f"""
SELECT
project_id,
id,
title,
permalink,
file_path,
type,
metadata,
from_id,
to_id,
relation_type,
entity_id,
content_snippet,
category,
created_at,
updated_at,
search_index.project_id,
search_index.id,
search_index.title,
search_index.permalink,
search_index.file_path,
search_index.type,
search_index.metadata,
search_index.from_id,
search_index.to_id,
search_index.relation_type,
search_index.entity_id,
search_index.content_snippet,
search_index.category,
search_index.created_at,
search_index.updated_at,
{score_expr} as score
FROM search_index
FROM {from_clause}
WHERE {where_clause}
ORDER BY score DESC, id ASC {order_by_clause}
ORDER BY score DESC, search_index.id ASC {order_by_clause}
LIMIT :limit
OFFSET :offset
"""
@@ -40,6 +40,7 @@ class SearchRepository(Protocol):
types: Optional[List[str]] = None,
after_date: Optional[datetime] = None,
search_item_types: Optional[List[SearchItemType]] = None,
metadata_filters: Optional[dict] = None,
limit: int = 10,
offset: int = 0,
) -> List[SearchIndexRow]:
@@ -78,6 +78,7 @@ class SearchRepositoryBase(ABC):
types: Optional[List[str]] = None,
after_date: Optional[datetime] = None,
search_item_types: Optional[List[SearchItemType]] = None,
metadata_filters: Optional[Dict[str, Any]] = None,
limit: int = 10,
offset: int = 0,
) -> List[SearchIndexRow]:
@@ -91,6 +92,7 @@ class SearchRepositoryBase(ABC):
types: Filter by entity types (from metadata.entity_type)
after_date: Filter by created_at > after_date
search_item_types: Filter by SearchItemType (ENTITY, OBSERVATION, RELATION)
metadata_filters: Structured frontmatter metadata filters
limit: Maximum results to return
offset: Number of results to skip
@@ -13,6 +13,7 @@ from basic_memory import db
from basic_memory.models.search import CREATE_SEARCH_INDEX
from basic_memory.repository.search_index_row import SearchIndexRow
from basic_memory.repository.search_repository_base import SearchRepositoryBase
from basic_memory.repository.metadata_filters import parse_metadata_filters, build_sqlite_json_path
from basic_memory.schemas.search import SearchItemType
@@ -26,6 +27,17 @@ class SQLiteSearchRepository(SearchRepositoryBase):
- Prefix wildcard matching with *
"""
def __init__(self, session_maker, project_id: int):
super().__init__(session_maker, project_id)
self._entity_columns: set[str] | None = None
async def _get_entity_columns(self) -> set[str]:
if self._entity_columns is None:
async with db.scoped_session(self.session_maker) as session:
result = await session.execute(text("PRAGMA table_info(entity)"))
self._entity_columns = {row[1] for row in result.fetchall()}
return self._entity_columns
async def init_search_index(self):
"""Create FTS5 virtual table for search if it doesn't exist.
@@ -287,6 +299,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
types: Optional[List[str]] = None,
after_date: Optional[datetime] = None,
search_item_types: Optional[List[SearchItemType]] = None,
metadata_filters: Optional[dict] = None,
limit: int = 10,
offset: int = 0,
) -> List[SearchIndexRow]:
@@ -294,6 +307,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
conditions = []
params = {}
order_by_clause = ""
from_clause = "search_index"
# Handle text search for title and content
if search_text:
@@ -305,18 +319,20 @@ class SQLiteSearchRepository(SearchRepositoryBase):
# Use _prepare_search_term to handle both Boolean and non-Boolean queries
processed_text = self._prepare_search_term(search_text.strip())
params["text"] = processed_text
conditions.append("(title MATCH :text OR content_stems MATCH :text)")
conditions.append(
"(search_index.title MATCH :text OR search_index.content_stems MATCH :text)"
)
# Handle title match search
if title:
title_text = self._prepare_search_term(title.strip(), is_prefix=False)
params["title_text"] = title_text
conditions.append("title MATCH :title_text")
conditions.append("search_index.title MATCH :title_text")
# Handle permalink exact search
if permalink:
params["permalink"] = permalink
conditions.append("permalink = :permalink")
conditions.append("search_index.permalink = :permalink")
# Handle permalink match search, supports *
if permalink_match:
@@ -325,38 +341,122 @@ class SQLiteSearchRepository(SearchRepositoryBase):
permalink_text = permalink_match.lower().strip()
params["permalink"] = permalink_text
if "*" in permalink_match:
conditions.append("permalink GLOB :permalink")
conditions.append("search_index.permalink GLOB :permalink")
else:
# For exact matches without *, we can use FTS5 MATCH
# but only prepare the term if it doesn't look like a path
if "/" in permalink_text:
conditions.append("permalink = :permalink")
conditions.append("search_index.permalink = :permalink")
else:
permalink_text = self._prepare_search_term(permalink_text, is_prefix=False)
params["permalink"] = permalink_text
conditions.append("permalink MATCH :permalink")
conditions.append("search_index.permalink MATCH :permalink")
# Handle entity type filter
if search_item_types:
type_list = ", ".join(f"'{t.value}'" for t in search_item_types)
conditions.append(f"type IN ({type_list})")
conditions.append(f"search_index.type IN ({type_list})")
# Handle type filter
if types:
type_list = ", ".join(f"'{t}'" for t in types)
conditions.append(f"json_extract(metadata, '$.entity_type') IN ({type_list})")
conditions.append(
f"json_extract(search_index.metadata, '$.entity_type') IN ({type_list})"
)
# Handle date filter using datetime() for proper comparison
if after_date:
params["after_date"] = after_date
conditions.append("datetime(created_at) > datetime(:after_date)")
conditions.append("datetime(search_index.created_at) > datetime(:after_date)")
# order by most recent first
order_by_clause = ", updated_at DESC"
order_by_clause = ", search_index.updated_at DESC"
# Handle structured metadata filters (frontmatter)
if metadata_filters:
parsed_filters = parse_metadata_filters(metadata_filters)
from_clause = "search_index JOIN entity ON search_index.entity_id = entity.id"
entity_columns = await self._get_entity_columns()
for idx, filt in enumerate(parsed_filters):
path_param = f"meta_path_{idx}"
extract_expr = None
use_tags_column = False
if filt.path_parts == ["status"] and "frontmatter_status" in entity_columns:
extract_expr = "entity.frontmatter_status"
elif filt.path_parts == ["type"] and "frontmatter_type" in entity_columns:
extract_expr = "entity.frontmatter_type"
elif filt.path_parts == ["tags"] and "tags_json" in entity_columns:
extract_expr = "entity.tags_json"
use_tags_column = True
if extract_expr is None:
params[path_param] = build_sqlite_json_path(filt.path_parts)
extract_expr = f"json_extract(entity.entity_metadata, :{path_param})"
if filt.op == "eq":
value_param = f"meta_val_{idx}"
params[value_param] = filt.value
conditions.append(f"{extract_expr} = :{value_param}")
continue
if filt.op == "in":
placeholders = []
for j, val in enumerate(filt.value):
value_param = f"meta_val_{idx}_{j}"
params[value_param] = val
placeholders.append(f":{value_param}")
conditions.append(f"{extract_expr} IN ({', '.join(placeholders)})")
continue
if filt.op == "contains":
tag_conditions = []
for j, val in enumerate(filt.value):
value_param = f"meta_val_{idx}_{j}"
params[value_param] = val
like_param = f"{value_param}_like"
params[like_param] = f'%"{val}"%'
like_param_single = f"{value_param}_like_single"
params[like_param_single] = f"%'{val}'%"
json_each_expr = (
"json_each(entity.tags_json)"
if use_tags_column
else f"json_each(entity.entity_metadata, :{path_param})"
)
tag_conditions.append(
"("
f"EXISTS (SELECT 1 FROM {json_each_expr} WHERE value = :{value_param}) "
f"OR {extract_expr} LIKE :{like_param} "
f"OR {extract_expr} LIKE :{like_param_single}"
")"
)
conditions.append(" AND ".join(tag_conditions))
continue
if filt.op in {"gt", "gte", "lt", "lte", "between"}:
compare_expr = (
f"CAST({extract_expr} AS REAL)"
if filt.comparison == "numeric"
else extract_expr
)
if filt.op == "between":
min_param = f"meta_val_{idx}_min"
max_param = f"meta_val_{idx}_max"
params[min_param] = filt.value[0]
params[max_param] = filt.value[1]
conditions.append(f"{compare_expr} BETWEEN :{min_param} AND :{max_param}")
else:
value_param = f"meta_val_{idx}"
params[value_param] = filt.value
operator = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="}[filt.op]
conditions.append(f"{compare_expr} {operator} :{value_param}")
continue
# Always filter by project_id
params["project_id"] = self.project_id
conditions.append("project_id = :project_id")
conditions.append("search_index.project_id = :project_id")
# set limit on search query
params["limit"] = limit
@@ -367,23 +467,23 @@ class SQLiteSearchRepository(SearchRepositoryBase):
sql = f"""
SELECT
project_id,
id,
title,
permalink,
file_path,
type,
metadata,
from_id,
to_id,
relation_type,
entity_id,
content_snippet,
category,
created_at,
updated_at,
search_index.project_id,
search_index.id,
search_index.title,
search_index.permalink,
search_index.file_path,
search_index.type,
search_index.metadata,
search_index.from_id,
search_index.to_id,
search_index.relation_type,
search_index.entity_id,
search_index.content_snippet,
search_index.category,
search_index.created_at,
search_index.updated_at,
bm25(search_index) as score
FROM search_index
FROM {from_clause}
WHERE {where_clause}
ORDER BY score ASC {order_by_clause}
LIMIT :limit
+1 -1
View File
@@ -25,7 +25,7 @@ class CloudProject(BaseModel):
class CloudProjectList(BaseModel):
"""Response from /proxy/projects/projects endpoint."""
"""Response from /proxy/v2/projects endpoint."""
projects: list[CloudProject] = Field(default_factory=list, description="List of cloud projects")
+4
View File
@@ -124,6 +124,7 @@ class EntitySummary(BaseModel):
"""Simplified entity representation."""
type: Literal["entity"] = "entity"
external_id: str # UUID for v2 API routing
entity_id: int # Database ID for v2 API consistency
permalink: Optional[str]
title: str
@@ -150,8 +151,10 @@ class RelationSummary(BaseModel):
relation_type: str
from_entity: Optional[str] = None
from_entity_id: Optional[int] = None # ID of source entity
from_entity_external_id: Optional[str] = None # UUID of source entity for v2 API routing
to_entity: Optional[str] = None
to_entity_id: Optional[int] = None # ID of target entity
to_entity_external_id: Optional[str] = None # UUID of target entity for v2 API routing
created_at: Annotated[
datetime, Field(json_schema_extra={"type": "string", "format": "date-time"})
]
@@ -167,6 +170,7 @@ class ObservationSummary(BaseModel):
type: Literal["observation"] = "observation"
observation_id: int # Database ID for v2 API consistency
entity_id: Optional[int] = None # ID of the entity this observation belongs to
entity_external_id: Optional[str] = None # UUID of parent entity for v2 API routing
title: str
file_path: str
permalink: str
+22 -6
View File
@@ -6,7 +6,7 @@ The search system supports three primary modes:
3. Full-text search across content
"""
from typing import Optional, List, Union
from typing import Optional, List, Union, Any
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, field_validator
@@ -29,11 +29,15 @@ class SearchQuery(BaseModel):
- permalink: Exact permalink match
- permalink_match: Path pattern with *
- text: Full-text search of title/content (supports boolean operators: AND, OR, NOT)
- title: Title only search
Optionally filter results by:
- types: Limit to specific item types
- entity_types: Limit to specific entity types
- types: Limit to specific entity types (frontmatter "type")
- entity_types: Limit to search item types (entity/observation/relation)
- after_date: Only items after date
- metadata_filters: Structured frontmatter filters (field -> value)
- tags: Convenience frontmatter tag filter
- status: Convenience frontmatter status filter
Boolean search examples:
- "python AND flask" - Find items with both terms
@@ -52,6 +56,9 @@ class SearchQuery(BaseModel):
types: Optional[List[str]] = None # Filter by type
entity_types: Optional[List[SearchItemType]] = None # Filter by entity type
after_date: Optional[Union[datetime, str]] = None # Time-based filter
metadata_filters: Optional[dict[str, Any]] = None # Structured frontmatter filters
tags: Optional[List[str]] = None # Convenience tag filter
status: Optional[str] = None # Convenience status filter
@field_validator("after_date")
@classmethod
@@ -62,14 +69,23 @@ class SearchQuery(BaseModel):
return v
def no_criteria(self) -> bool:
text_is_empty = self.text is None or (isinstance(self.text, str) and not self.text.strip())
metadata_is_empty = not self.metadata_filters
tags_is_empty = not self.tags
status_is_empty = self.status is None or (isinstance(self.status, str) and not self.status)
types_is_empty = not self.types
entity_types_is_empty = not self.entity_types
return (
self.permalink is None
and self.permalink_match is None
and self.title is None
and self.text is None
and text_is_empty
and self.after_date is None
and self.types is None
and self.entity_types is None
and types_is_empty
and entity_types_is_empty
and metadata_is_empty
and tags_is_empty
and status_is_empty
)
def has_boolean_operators(self) -> bool:
+215 -43
View File
@@ -1,5 +1,6 @@
"""Service for managing entities in the database."""
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Sequence, Tuple, Union
@@ -17,7 +18,7 @@ from basic_memory.file_utils import (
dump_frontmatter,
)
from basic_memory.markdown import EntityMarkdown
from basic_memory.markdown.entity_parser import EntityParser
from basic_memory.markdown.entity_parser import EntityParser, normalize_frontmatter_metadata
from basic_memory.markdown.utils import entity_model_from_markdown, schema_to_markdown
from basic_memory.models import Entity as EntityModel
from basic_memory.models import Observation, Relation
@@ -167,6 +168,25 @@ class EntityService(BaseService[EntityModel]):
return permalink
def _build_frontmatter_markdown(
self, title: str, entity_type: str, permalink: str
) -> EntityMarkdown:
"""Build a minimal EntityMarkdown object for permalink resolution."""
from basic_memory.markdown.schemas import EntityFrontmatter
frontmatter_metadata = {
"title": title,
"type": entity_type,
"permalink": permalink,
}
frontmatter_obj = EntityFrontmatter(metadata=frontmatter_metadata)
return EntityMarkdown(
frontmatter=frontmatter_obj,
content="",
observations=[],
relations=[],
)
async def create_or_update_entity(self, schema: EntitySchema) -> Tuple[EntityModel, bool]:
"""Create new entity or update existing one.
Returns: (entity, is_new) where is_new is True if a new entity was created
@@ -210,20 +230,8 @@ class EntityService(BaseService[EntityModel]):
schema.entity_type = content_frontmatter["type"]
if "permalink" in content_frontmatter:
# Create a minimal EntityMarkdown object for permalink resolution
from basic_memory.markdown.schemas import EntityFrontmatter
frontmatter_metadata = {
"title": schema.title,
"type": schema.entity_type,
"permalink": content_frontmatter["permalink"],
}
frontmatter_obj = EntityFrontmatter(metadata=frontmatter_metadata)
content_markdown = EntityMarkdown(
frontmatter=frontmatter_obj,
content="", # content not needed for permalink resolution
observations=[],
relations=[],
content_markdown = self._build_frontmatter_markdown(
schema.title, schema.entity_type, content_frontmatter["permalink"]
)
# Get unique permalink (prioritizing content frontmatter) unless disabled
@@ -248,11 +256,8 @@ class EntityService(BaseService[EntityModel]):
content=final_content,
)
# create entity
created = await self.create_entity_from_markdown(file_path, entity_markdown)
# add relations
entity = await self.update_entity_relations(created.file_path, entity_markdown)
# create entity and relations
entity = await self.upsert_entity_from_markdown(file_path, entity_markdown, is_new=True)
# Set final checksum to mark complete
return await self.repository.update(entity.id, {"checksum": checksum})
@@ -283,20 +288,8 @@ class EntityService(BaseService[EntityModel]):
schema.entity_type = content_frontmatter["type"]
if "permalink" in content_frontmatter:
# Create a minimal EntityMarkdown object for permalink resolution
from basic_memory.markdown.schemas import EntityFrontmatter
frontmatter_metadata = {
"title": schema.title,
"type": schema.entity_type,
"permalink": content_frontmatter["permalink"],
}
frontmatter_obj = EntityFrontmatter(metadata=frontmatter_metadata)
content_markdown = EntityMarkdown(
frontmatter=frontmatter_obj,
content="", # content not needed for permalink resolution
observations=[],
relations=[],
content_markdown = self._build_frontmatter_markdown(
schema.title, schema.entity_type, content_frontmatter["permalink"]
)
# Check if we need to update the permalink based on content frontmatter (unless disabled)
@@ -333,17 +326,179 @@ class EntityService(BaseService[EntityModel]):
content=final_content,
)
# update entity in db
entity = await self.update_entity_and_observations(file_path, entity_markdown)
# add relations
await self.update_entity_relations(file_path.as_posix(), entity_markdown)
# update entity and relations
entity = await self.upsert_entity_from_markdown(file_path, entity_markdown, is_new=False)
# Set final checksum to match file
entity = await self.repository.update(entity.id, {"checksum": checksum})
return entity
async def fast_write_entity(
self,
schema: EntitySchema,
external_id: Optional[str] = None,
) -> EntityModel:
"""Write file and upsert a minimal entity row for fast responses."""
logger.debug(
"Fast-writing entity",
title=schema.title,
external_id=external_id,
content_type=schema.content_type,
)
# --- Identity & File Path ---
existing = await self.repository.get_by_external_id(external_id) if external_id else None
# Trigger: external_id already exists
# Why: avoid duplicate entities when title-derived paths change
# Outcome: update in-place and keep the existing file path
file_path = Path(existing.file_path) if existing else Path(schema.file_path)
if not existing and await self.file_service.exists(file_path):
raise EntityCreationError(
f"file for entity {schema.directory}/{schema.title} already exists: {file_path}"
)
# --- Frontmatter Overrides ---
content_markdown = None
if schema.content and has_frontmatter(schema.content):
content_frontmatter = parse_frontmatter(schema.content)
if "type" in content_frontmatter:
schema.entity_type = content_frontmatter["type"]
if "permalink" in content_frontmatter:
content_markdown = self._build_frontmatter_markdown(
schema.title, schema.entity_type, content_frontmatter["permalink"]
)
# --- Permalink Resolution ---
if self.app_config and self.app_config.disable_permalinks:
schema._permalink = ""
else:
if existing and not (content_markdown and content_markdown.frontmatter.permalink):
schema._permalink = existing.permalink or await self.resolve_permalink(
file_path, skip_conflict_check=True
)
else:
schema._permalink = await self.resolve_permalink(
file_path, content_markdown, skip_conflict_check=True
)
# --- File Write ---
post = await schema_to_markdown(schema)
final_content = dump_frontmatter(post)
checksum = await self.file_service.write_file(file_path, final_content)
# --- Minimal DB Upsert ---
metadata = normalize_frontmatter_metadata(post.metadata or {})
entity_metadata = {k: v for k, v in metadata.items() if v is not None}
update_data = {
"title": schema.title,
"entity_type": schema.entity_type,
"file_path": file_path.as_posix(),
"content_type": schema.content_type,
"entity_metadata": entity_metadata or None,
"permalink": schema.permalink,
"checksum": checksum,
"updated_at": datetime.now().astimezone(),
}
if existing:
updated = await self.repository.update(existing.id, update_data)
if not updated:
raise ValueError(f"Failed to update entity in database: {existing.id}")
return updated
create_data = {
**update_data,
"external_id": external_id,
}
return await self.repository.create(create_data)
async def fast_edit_entity(
self,
entity: EntityModel,
operation: str,
content: str,
section: Optional[str] = None,
find_text: Optional[str] = None,
expected_replacements: int = 1,
) -> EntityModel:
"""Edit an entity quickly and defer full indexing to background."""
logger.debug(f"Fast editing entity: {entity.external_id}, operation: {operation}")
# --- File Edit ---
file_path = Path(entity.file_path)
current_content, _ = await self.file_service.read_file(file_path)
new_content = self.apply_edit_operation(
current_content, operation, content, section, find_text, expected_replacements
)
checksum = await self.file_service.write_file(file_path, new_content)
# --- Frontmatter Overrides ---
update_data = {
"checksum": checksum,
"updated_at": datetime.now().astimezone(),
}
content_markdown = None
if has_frontmatter(new_content):
content_frontmatter = parse_frontmatter(new_content)
if "title" in content_frontmatter:
update_data["title"] = content_frontmatter["title"]
if "type" in content_frontmatter:
update_data["entity_type"] = content_frontmatter["type"]
if "permalink" in content_frontmatter:
content_markdown = self._build_frontmatter_markdown(
update_data.get("title", entity.title),
update_data.get("entity_type", entity.entity_type),
content_frontmatter["permalink"],
)
metadata = normalize_frontmatter_metadata(content_frontmatter or {})
update_data["entity_metadata"] = {k: v for k, v in metadata.items() if v is not None}
# --- Permalink Resolution ---
if self.app_config and self.app_config.disable_permalinks:
update_data["permalink"] = None
elif content_markdown and content_markdown.frontmatter.permalink:
update_data["permalink"] = await self.resolve_permalink(
file_path, content_markdown, skip_conflict_check=True
)
updated = await self.repository.update(entity.id, update_data)
if not updated:
raise ValueError(f"Failed to update entity in database: {entity.id}")
return updated
async def reindex_entity(self, entity_id: int) -> None:
"""Parse file content and rebuild observations/relations/search for an entity."""
entity = await self.repository.find_by_id(entity_id)
if not entity:
raise EntityNotFoundError(f"Entity not found: {entity_id}")
# --- Full Parse ---
file_path = Path(entity.file_path)
content = await self.file_service.read_file_content(file_path)
entity_markdown = await self.entity_parser.parse_markdown_content(
file_path=file_path,
content=content,
)
# --- DB Reindex ---
updated = await self.upsert_entity_from_markdown(file_path, entity_markdown, is_new=False)
checksum = await self.file_service.compute_checksum(file_path)
updated = await self.repository.update(updated.id, {"checksum": checksum})
if not updated:
raise ValueError(f"Failed to update entity in database: {entity.id}")
# --- Search Reindex ---
if self.search_service:
await self.search_service.index_entity_data(updated, content=content)
async def delete_entity(self, permalink_or_id: str | int) -> bool:
"""Delete entity and its file."""
logger.debug(f"Deleting entity: {permalink_or_id}")
@@ -465,6 +620,20 @@ class EntityService(BaseService[EntityModel]):
db_entity,
)
async def upsert_entity_from_markdown(
self,
file_path: Path,
markdown: EntityMarkdown,
*,
is_new: bool,
) -> EntityModel:
"""Create/update entity and relations from parsed markdown."""
if is_new:
created = await self.create_entity_from_markdown(file_path, markdown)
else:
created = await self.update_entity_and_observations(file_path, markdown)
return await self.update_entity_relations(created.file_path, markdown)
async def update_entity_relations(
self,
path: str,
@@ -589,8 +758,7 @@ class EntityService(BaseService[EntityModel]):
)
# Update entity and its relationships
entity = await self.update_entity_and_observations(file_path, entity_markdown)
await self.update_entity_relations(file_path.as_posix(), entity_markdown)
entity = await self.upsert_entity_from_markdown(file_path, entity_markdown, is_new=False)
# Set final checksum to match file
entity = await self.repository.update(entity.id, {"checksum": checksum})
@@ -927,7 +1095,9 @@ class EntityService(BaseService[EntityModel]):
old_path = entity.file_path
# Replace only the first occurrence of the source directory prefix
if old_path.startswith(f"{source_directory}/"):
new_path = old_path.replace(f"{source_directory}/", f"{destination_directory}/", 1)
new_path = old_path.replace(
f"{source_directory}/", f"{destination_directory}/", 1
)
else: # pragma: no cover
# Entity is directly in the source directory (shouldn't happen with prefix match)
new_path = f"{destination_directory}/{old_path}"
@@ -1016,7 +1186,9 @@ class EntityService(BaseService[EntityModel]):
logger.debug(f"Deleted entity: {file_path}")
else: # pragma: no cover
failed_deletes += 1
errors.append(DirectoryDeleteError(path=file_path, error="Delete returned False"))
errors.append(
DirectoryDeleteError(path=file_path, error="Delete returned False")
)
logger.warning(f"Delete returned False for entity: {file_path}")
except Exception as e: # pragma: no cover
+11 -1
View File
@@ -2,7 +2,7 @@
import ast
from datetime import datetime
from typing import List, Optional, Set
from typing import List, Optional, Set, Dict, Any
from dateparser import parse
@@ -95,6 +95,15 @@ class SearchService:
else None
)
# Merge structured metadata filters (explicit + convenience fields)
metadata_filters: Optional[Dict[str, Any]] = None
if query.metadata_filters or query.tags or query.status:
metadata_filters = dict(query.metadata_filters or {})
if query.tags:
metadata_filters.setdefault("tags", query.tags)
if query.status:
metadata_filters.setdefault("status", query.status)
# search
results = await self.repository.search(
search_text=query.text,
@@ -104,6 +113,7 @@ class SearchService:
types=query.types,
search_item_types=query.entity_types,
after_date=after_date,
metadata_filters=metadata_filters,
limit=limit,
offset=offset,
)
+5 -13
View File
@@ -685,19 +685,11 @@ class SyncService:
entity_markdown.frontmatter.metadata["permalink"] = permalink
await self.file_service.update_frontmatter(path, {"permalink": permalink})
# if the file is new, create an entity
if new:
# Create entity with final permalink
logger.debug(f"Creating new entity from markdown, path={path}")
await self.entity_service.create_entity_from_markdown(Path(path), entity_markdown)
# otherwise we need to update the entity and observations
else:
logger.debug(f"Updating entity from markdown, path={path}")
await self.entity_service.update_entity_and_observations(Path(path), entity_markdown)
# Update relations and search index
entity = await self.entity_service.update_entity_relations(path, entity_markdown)
# Create/update entity and relations in one path
logger.debug(f"{'Creating' if new else 'Updating'} entity from markdown, path={path}")
entity = await self.entity_service.upsert_entity_from_markdown(
Path(path), entity_markdown, is_new=new
)
# After updating relations, we need to compute the checksum again
# This is necessary for files with wikilinks to ensure consistent checksums