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

feat: Implement API v2 with ID-based endpoints (Phase 1) (#441)

Browse Source 
Signed-off-by: Joe P <joe@basicmemory.com>
Signed-off-by: phernandez <paul@basicmachines.co>
Signed-off-by: Claude <noreply@anthropic.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>
Co-authored-by: phernandez <paul@basicmachines.co>
This commit is contained in:
jope-bm
2025-11-27 09:35:55 -07:00
committed by GitHub
parent 9b7bbc7116
commit 28cc5225a7
44 changed files with 4875 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -52,4 +52,5 @@ ENV/
# claude action
claude-output
**/.claude/settings.local.json
**/.claude/settings.local.json
.mcp.json
CLAUDE.md
+1
View File
@@ -264,5 +264,6 @@ With GitHub integration, the development workflow includes:
2. **Contribution tracking** - All of Claude's contributions are properly attributed in the Git history
3. **Branch management** - Claude can create feature branches for implementations
4. **Documentation maintenance** - Claude can keep documentation updated as the code evolves
5. **Code Commits**: ALWAYS sign off commits with `git commit -s`
This level of integration represents a new paradigm in AI-human collaboration, where the AI assistant becomes a full-fledged team member rather than just a tool for generating code snippets.
src/basic_memory/api/app.py
+22 -3
View File
@@ -20,6 +20,16 @@ from basic_memory.api.routers import (
search,
prompt_router,
)
from basic_memory.api.v2.routers import (
knowledge_router as v2_knowledge,
project_router as v2_project,
memory_router as v2_memory,
search_router as v2_search,
resource_router as v2_resource,
directory_router as v2_directory,
prompt_router as v2_prompt,
importer_router as v2_importer,
)
from basic_memory.config import ConfigManager
from basic_memory.services.initialization import initialize_file_sync, initialize_app
@@ -66,8 +76,7 @@ app = FastAPI(
lifespan=lifespan,
)
# Include routers
# Include v1 routers
app.include_router(knowledge.router, prefix="/{project}")
app.include_router(memory.router, prefix="/{project}")
app.include_router(resource.router, prefix="/{project}")
@@ -77,7 +86,17 @@ app.include_router(directory_router.router, prefix="/{project}")
app.include_router(prompt_router.router, prefix="/{project}")
app.include_router(importer_router.router, prefix="/{project}")
# Project resource router works accross projects
# Include v2 routers (ID-based paths)
app.include_router(v2_knowledge, prefix="/v2/projects/{project_id}")
app.include_router(v2_memory, prefix="/v2/projects/{project_id}")
app.include_router(v2_search, prefix="/v2/projects/{project_id}")
app.include_router(v2_resource, prefix="/v2/projects/{project_id}")
app.include_router(v2_directory, prefix="/v2/projects/{project_id}")
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")
# Project resource router works across projects
app.include_router(project.project_resource_router)
app.include_router(management.router)
src/basic_memory/api/routers/knowledge_router.py
+13 -2
View File
@@ -1,4 +1,11 @@
"""Router for knowledge graph operations."""
"""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
@@ -25,7 +32,11 @@ from basic_memory.schemas import (
from basic_memory.schemas.request import EditEntityRequest, MoveEntityRequest
from basic_memory.schemas.base import Permalink, Entity
router = APIRouter(prefix="/knowledge", tags=["knowledge"])
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:
src/basic_memory/api/routers/project_router.py
+50 -8
View File
@@ -50,6 +50,7 @@ async def get_project(
) # pragma: no cover
return ProjectItem(
id=found_project.id,
name=found_project.name,
path=normalize_project_path(found_project.path),
is_default=found_project.is_default or False,
@@ -80,9 +81,17 @@ async def update_project(
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(
name=name,
path=project_service.projects.get(name, ""),
id=old_project.id,
name=old_project.name,
path=old_project.path,
is_default=old_project.is_default or False,
)
if path:
@@ -91,14 +100,21 @@ async def update_project(
await project_service.update_project(name, is_active=is_active)
# Get updated project info
updated_path = path if path else project_service.projects.get(name, "")
updated_project = await project_service.get_project(name)
if not updated_project:
raise HTTPException(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(name=name, path=updated_path),
new_project=ProjectItem(
id=updated_project.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))
@@ -186,6 +202,7 @@ async def list_projects(
project_items = [
ProjectItem(
id=project.id,
name=project.name,
path=normalize_project_path(project.path),
is_default=project.is_default or False,
@@ -232,6 +249,7 @@ async def add_project(
status="success",
default=existing_project.is_default or False,
new_project=ProjectItem(
id=existing_project.id,
name=existing_project.name,
path=existing_project.path,
is_default=existing_project.is_default or False,
@@ -250,12 +268,20 @@ async def 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 '{project_data.name}' added successfully",
status="success",
default=project_data.set_default,
new_project=ProjectItem(
name=project_data.name, path=project_data.path, is_default=project_data.set_default
id=new_project.id,
name=new_project.name,
path=new_project.path,
is_default=new_project.is_default or False,
),
)
except ValueError as e: # pragma: no cover
@@ -306,7 +332,12 @@ async def remove_project(
message=f"Project '{name}' removed successfully",
status="success",
default=False,
old_project=ProjectItem(name=old_project.name, path=old_project.path),
old_project=ProjectItem(
id=old_project.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
@@ -349,8 +380,14 @@ async def set_default_project(
message=f"Project '{name}' set as default successfully",
status="success",
default=True,
old_project=ProjectItem(name=default_name, path=default_project.path),
old_project=ProjectItem(
id=default_project.id,
name=default_name,
path=default_project.path,
is_default=False,
),
new_project=ProjectItem(
id=new_default_project.id,
name=name,
path=new_default_project.path,
is_default=True,
@@ -378,7 +415,12 @@ async def get_default_project(
status_code=404, detail=f"Default Project: '{default_name}' does not exist"
)
return ProjectItem(name=default_project.name, path=default_project.path, is_default=True)
return ProjectItem(
id=default_project.id,
name=default_project.name,
path=default_project.path,
is_default=True,
)
# Synchronize projects between config and database
src/basic_memory/api/routers/utils.py
+25
View File
@@ -29,6 +29,7 @@ async def to_graph_context(
match item.type:
case SearchItemType.ENTITY:
return EntitySummary(
entity_id=item.id,
title=item.title, # pyright: ignore
permalink=item.permalink,
content=item.content,
@@ -37,6 +38,8 @@ async def to_graph_context(
)
case SearchItemType.OBSERVATION:
return ObservationSummary(
observation_id=item.id,
entity_id=item.entity_id, # pyright: ignore
title=item.title, # pyright: ignore
file_path=item.file_path,
category=item.category, # pyright: ignore
@@ -48,12 +51,16 @@ async def to_graph_context(
from_entity = await entity_repository.find_by_id(item.from_id) # pyright: ignore
to_entity = await entity_repository.find_by_id(item.to_id) if item.to_id else None
return RelationSummary(
relation_id=item.id,
entity_id=item.entity_id, # pyright: ignore
title=item.title, # pyright: ignore
file_path=item.file_path,
permalink=item.permalink, # pyright: ignore
relation_type=item.relation_type, # pyright: ignore
from_entity=from_entity.title if from_entity else None,
from_entity_id=item.from_id, # pyright: ignore
to_entity=to_entity.title if to_entity else None,
to_entity_id=item.to_id,
created_at=item.created_at,
)
case _: # pragma: no cover
@@ -111,6 +118,21 @@ async def to_search_results(entity_service: EntityService, results: List[SearchI
search_results = []
for r in results:
entities = await entity_service.get_entities_by_id([r.entity_id, r.from_id, r.to_id]) # pyright: ignore
# Determine which IDs to set based on type
entity_id = None
observation_id = None
relation_id = None
if r.type == SearchItemType.ENTITY:
entity_id = r.id
elif r.type == SearchItemType.OBSERVATION:
observation_id = r.id
entity_id = r.entity_id # Parent entity
elif r.type == SearchItemType.RELATION:
relation_id = r.id
entity_id = r.entity_id # Parent entity
search_results.append(
SearchResult(
title=r.title, # pyright: ignore
@@ -121,6 +143,9 @@ async def to_search_results(entity_service: EntityService, results: List[SearchI
content=r.content,
file_path=r.file_path,
metadata=r.metadata,
entity_id=entity_id,
observation_id=observation_id,
relation_id=relation_id,
category=r.category,
from_entity=entities[0].permalink if entities else None,
to_entity=entities[1].permalink if len(entities) > 1 else None,
src/basic_memory/api/v2/__init__.py
+35
View File
@@ -0,0 +1,35 @@
"""API v2 module - ID-based entity references.
Version 2 of the Basic Memory API uses integer entity IDs as the primary
identifier for improved performance and stability.
Key changes from v1:
- Entity lookups use integer IDs instead of paths/permalinks
- Direct database queries instead of cascading resolution
- Stable references that don't change with file moves
- Better caching support
All v2 routers are registered with the /v2 prefix.
"""
from basic_memory.api.v2.routers import (
knowledge_router,
memory_router,
project_router,
resource_router,
search_router,
directory_router,
prompt_router,
importer_router,
)
__all__ = [
"knowledge_router",
"memory_router",
"project_router",
"resource_router",
"search_router",
"directory_router",
"prompt_router",
"importer_router",
]
src/basic_memory/api/v2/routers/__init__.py
+21
View File
@@ -0,0 +1,21 @@
"""V2 API routers."""
from basic_memory.api.v2.routers.knowledge_router import router as knowledge_router
from basic_memory.api.v2.routers.project_router import router as project_router
from basic_memory.api.v2.routers.memory_router import router as memory_router
from basic_memory.api.v2.routers.search_router import router as search_router
from basic_memory.api.v2.routers.resource_router import router as resource_router
from basic_memory.api.v2.routers.directory_router import router as directory_router
from basic_memory.api.v2.routers.prompt_router import router as prompt_router
from basic_memory.api.v2.routers.importer_router import router as importer_router
__all__ = [
"knowledge_router",
"project_router",
"memory_router",
"search_router",
"resource_router",
"directory_router",
"prompt_router",
"importer_router",
]
src/basic_memory/api/v2/routers/directory_router.py
+93
View File
@@ -0,0 +1,93 @@
"""V2 Directory Router - ID-based directory tree operations.
This router provides directory structure browsing for projects using
integer project IDs instead of name-based identifiers.
Key improvements:
- Direct project lookup via integer primary keys
- Consistent with other v2 endpoints
- Better performance through indexed queries
"""
from typing import List, Optional
from fastapi import APIRouter, Query
from basic_memory.deps import DirectoryServiceV2Dep, ProjectIdPathDep
from basic_memory.schemas.directory import DirectoryNode
router = APIRouter(prefix="/directory", tags=["directory-v2"])
@router.get("/tree", response_model=DirectoryNode, response_model_exclude_none=True)
async def get_directory_tree(
directory_service: DirectoryServiceV2Dep,
project_id: ProjectIdPathDep,
):
"""Get hierarchical directory structure from the knowledge base.
Args:
directory_service: Service for directory operations
project_id: Numeric project ID
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: DirectoryServiceV2Dep,
project_id: ProjectIdPathDep,
):
"""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: Numeric project ID
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: DirectoryServiceV2Dep,
project_id: ProjectIdPathDep,
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: Numeric project ID
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
src/basic_memory/api/v2/routers/importer_router.py
+182
View File
@@ -0,0 +1,182 @@
"""V2 Import Router - ID-based data import operations.
This router uses v2 dependencies for consistent project ID handling.
Import endpoints use project_id in the path for consistency with other v2 endpoints.
"""
import json
import logging
from fastapi import APIRouter, Form, HTTPException, UploadFile, status
from basic_memory.deps import (
ChatGPTImporterV2Dep,
ClaudeConversationsImporterV2Dep,
ClaudeProjectsImporterV2Dep,
MemoryJsonImporterV2Dep,
ProjectIdPathDep,
)
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-v2"])
@router.post("/chatgpt", response_model=ChatImportResult)
async def import_chatgpt(
project_id: ProjectIdPathDep,
importer: ChatGPTImporterV2Dep,
file: UploadFile,
folder: str = Form("conversations"),
) -> ChatImportResult:
"""Import conversations from ChatGPT JSON export.
Args:
project_id: Validated numeric project ID from URL path
file: The ChatGPT conversations.json file.
folder: The folder to place the files in.
importer: ChatGPT importer instance.
Returns:
ChatImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
logger.info(f"V2 Importing ChatGPT conversations for project {project_id}")
return await import_file(importer, file, folder)
@router.post("/claude/conversations", response_model=ChatImportResult)
async def import_claude_conversations(
project_id: ProjectIdPathDep,
importer: ClaudeConversationsImporterV2Dep,
file: UploadFile,
folder: str = Form("conversations"),
) -> ChatImportResult:
"""Import conversations from Claude conversations.json export.
Args:
project_id: Validated numeric project ID from URL path
file: The Claude conversations.json file.
folder: The folder to place the files in.
importer: Claude conversations importer instance.
Returns:
ChatImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
logger.info(f"V2 Importing Claude conversations for project {project_id}")
return await import_file(importer, file, folder)
@router.post("/claude/projects", response_model=ProjectImportResult)
async def import_claude_projects(
project_id: ProjectIdPathDep,
importer: ClaudeProjectsImporterV2Dep,
file: UploadFile,
folder: str = Form("projects"),
) -> ProjectImportResult:
"""Import projects from Claude projects.json export.
Args:
project_id: Validated numeric project ID from URL path
file: The Claude projects.json file.
folder: The base folder to place the files in.
importer: Claude projects importer instance.
Returns:
ProjectImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
logger.info(f"V2 Importing Claude projects for project {project_id}")
return await import_file(importer, file, folder)
@router.post("/memory-json", response_model=EntityImportResult)
async def import_memory_json(
project_id: ProjectIdPathDep,
importer: MemoryJsonImporterV2Dep,
file: UploadFile,
folder: str = Form("conversations"),
) -> EntityImportResult:
"""Import entities and relations from a memory.json file.
Args:
project_id: Validated numeric project ID from URL path
file: The memory.json file.
folder: Optional destination folder within the project.
importer: Memory JSON importer instance.
Returns:
EntityImportResult with import statistics.
Raises:
HTTPException: If import fails.
"""
logger.info(f"V2 Importing memory.json for project {project_id}")
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, 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",
)
except Exception as e:
logger.exception("V2 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):
"""Helper function to import a file using an importer instance.
Args:
importer: The importer instance to use
file: The file to import
destination_folder: Destination folder for imported content
Returns:
Import result from the importer
Raises:
HTTPException: If import fails
"""
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("V2 Import failed")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Import failed: {str(e)}",
)
src/basic_memory/api/v2/routers/knowledge_router.py
+415
View File
@@ -0,0 +1,415 @@
"""V2 Knowledge Router - ID-based entity operations.
This router provides ID-based CRUD operations for entities, replacing the
path-based identifiers used in v1 with direct integer ID lookups.
Key improvements:
- Direct database lookups via integer primary keys
- Stable references that don't change with file moves
- Better performance through indexed queries
- Simplified caching strategies
"""
from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends, Response
from loguru import logger
from basic_memory.deps import (
EntityServiceV2Dep,
SearchServiceV2Dep,
LinkResolverV2Dep,
ProjectConfigV2Dep,
AppConfigDep,
SyncServiceV2Dep,
EntityRepositoryV2Dep,
ProjectIdPathDep,
)
from basic_memory.schemas import DeleteEntitiesResponse
from basic_memory.schemas.base import Entity
from basic_memory.schemas.request import EditEntityRequest
from basic_memory.schemas.v2 import (
EntityResolveRequest,
EntityResolveResponse,
EntityResponseV2,
MoveEntityRequestV2,
)
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:
# 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:
# Log but don't fail - this is a background task
logger.warning(
f"Background: Failed to resolve relations for entity {entity_permalink}: {e}"
)
## Resolution endpoint
@router.post("/resolve", response_model=EntityResolveResponse)
async def resolve_identifier(
project_id: ProjectIdPathDep,
data: EntityResolveRequest,
link_resolver: LinkResolverV2Dep,
) -> EntityResolveResponse:
"""Resolve a string identifier (permalink, title, or path) to an entity ID.
This endpoint provides a bridge between v1-style identifiers and v2 entity IDs.
Use this to convert existing references to the new ID-based format.
Args:
data: Request containing the identifier to resolve
Returns:
Entity ID and metadata about how it was resolved
Raises:
HTTPException: 404 if identifier cannot be resolved
Example:
POST /v2/{project}/knowledge/resolve
{"identifier": "specs/search"}
Returns:
{
"entity_id": 123,
"permalink": "specs/search",
"file_path": "specs/search.md",
"title": "Search Specification",
"resolution_method": "permalink"
}
"""
logger.info(f"API v2 request: resolve_identifier for '{data.identifier}'")
# Try to resolve the identifier
entity = await link_resolver.resolve_link(data.identifier)
if not entity:
raise HTTPException(
status_code=404, detail=f"Could not resolve identifier: '{data.identifier}'"
)
# Determine resolution method
resolution_method = "search" # default
if data.identifier.isdigit():
resolution_method = "id"
elif entity.permalink == data.identifier:
resolution_method = "permalink"
elif entity.title == data.identifier:
resolution_method = "title"
elif entity.file_path == data.identifier:
resolution_method = "path"
result = EntityResolveResponse(
entity_id=entity.id,
permalink=entity.permalink,
file_path=entity.file_path,
title=entity.title,
resolution_method=resolution_method,
)
logger.info(
f"API v2 response: resolved '{data.identifier}' to entity_id={result.entity_id} via {resolution_method}"
)
return result
## Read endpoints
@router.get("/entities/{entity_id}", response_model=EntityResponseV2)
async def get_entity_by_id(
project_id: ProjectIdPathDep,
entity_id: int,
entity_repository: EntityRepositoryV2Dep,
) -> EntityResponseV2:
"""Get an entity by its numeric ID.
This is the primary entity retrieval method in v2, using direct database
lookups for maximum performance.
Args:
entity_id: Numeric entity ID
Returns:
Complete entity with observations and relations
Raises:
HTTPException: 404 if entity not found
"""
logger.info(f"API v2 request: get_entity_by_id entity_id={entity_id}")
entity = await entity_repository.get_by_id(entity_id)
if not entity:
raise HTTPException(status_code=404, detail=f"Entity {entity_id} not found")
result = EntityResponseV2.model_validate(entity)
logger.info(f"API v2 response: entity_id={entity_id}, title='{result.title}'")
return result
## Create endpoints
@router.post("/entities", response_model=EntityResponseV2)
async def create_entity(
project_id: ProjectIdPathDep,
data: Entity,
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2Dep,
search_service: SearchServiceV2Dep,
) -> EntityResponseV2:
"""Create a new entity.
Args:
data: Entity data to create
Returns:
Created entity with generated ID
"""
logger.info(
"API v2 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 = EntityResponseV2.model_validate(entity)
logger.info(
f"API v2 response: endpoint='create_entity' id={entity.id}, title={result.title}, permalink={result.permalink}, status_code=201"
)
return result
## Update endpoints
@router.put("/entities/{entity_id}", response_model=EntityResponseV2)
async def update_entity_by_id(
project_id: ProjectIdPathDep,
entity_id: int,
data: Entity,
response: Response,
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2Dep,
search_service: SearchServiceV2Dep,
sync_service: SyncServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
) -> EntityResponseV2:
"""Update an entity by ID.
If the entity doesn't exist, it will be created (upsert behavior).
Args:
entity_id: Numeric entity ID
data: Updated entity data
Returns:
Updated entity
"""
logger.info(f"API v2 request: update_entity_by_id entity_id={entity_id}")
# Check if entity exists
existing = await entity_repository.get_by_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(
resolve_relations_background, sync_service, entity.id, entity.permalink or ""
)
result = EntityResponseV2.model_validate(entity)
logger.info(
f"API v2 response: entity_id={entity_id}, created={created}, status_code={response.status_code}"
)
return result
@router.patch("/entities/{entity_id}", response_model=EntityResponseV2)
async def edit_entity_by_id(
project_id: ProjectIdPathDep,
entity_id: int,
data: EditEntityRequest,
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2Dep,
search_service: SearchServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
) -> EntityResponseV2:
"""Edit an existing entity by ID using operations like append, prepend, etc.
Args:
entity_id: Numeric entity ID
data: Edit operation details
Returns:
Updated entity
Raises:
HTTPException: 404 if entity not found, 400 if edit fails
"""
logger.info(
f"API v2 request: edit_entity_by_id entity_id={entity_id}, operation='{data.operation}'"
)
# Verify entity exists
entity = await entity_repository.get_by_id(entity_id)
if not entity:
raise HTTPException(status_code=404, detail=f"Entity {entity_id} not found")
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,
)
# Reindex
await search_service.index_entity(updated_entity, background_tasks=background_tasks)
result = EntityResponseV2.model_validate(updated_entity)
logger.info(
f"API v2 response: entity_id={entity_id}, operation='{data.operation}', status_code=200"
)
return result
except Exception as e:
logger.error(f"Error editing entity {entity_id}: {e}")
raise HTTPException(status_code=400, detail=str(e))
## Delete endpoints
@router.delete("/entities/{entity_id}", response_model=DeleteEntitiesResponse)
async def delete_entity_by_id(
project_id: ProjectIdPathDep,
entity_id: int,
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
search_service=Depends(lambda: None), # Optional for now
) -> DeleteEntitiesResponse:
"""Delete an entity by ID.
Args:
entity_id: Numeric entity ID
Returns:
Deletion status
Note: Returns deleted=False if entity doesn't exist (idempotent)
"""
logger.info(f"API v2 request: delete_entity_by_id entity_id={entity_id}")
entity = await entity_repository.get_by_id(entity_id)
if entity is None:
logger.info(f"API v2 response: entity_id={entity_id} not found, deleted=False")
return DeleteEntitiesResponse(deleted=False)
# Delete the entity
deleted = await entity_service.delete_entity(entity_id)
# Remove from search index if search service available
if search_service:
background_tasks.add_task(search_service.handle_delete, entity)
logger.info(f"API v2 response: entity_id={entity_id}, deleted={deleted}")
return DeleteEntitiesResponse(deleted=deleted)
## Move endpoint
@router.put("/entities/{entity_id}/move", response_model=EntityResponseV2)
async def move_entity(
project_id: ProjectIdPathDep,
entity_id: int,
data: MoveEntityRequestV2,
background_tasks: BackgroundTasks,
entity_service: EntityServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
project_config: ProjectConfigV2Dep,
app_config: AppConfigDep,
search_service: SearchServiceV2Dep,
) -> EntityResponseV2:
"""Move an entity to a new file location.
V2 API uses entity ID in the URL path for stable references.
The entity ID will remain stable after the move.
Args:
project_id: Project ID from URL path
entity_id: Entity ID from URL path (primary identifier)
data: Move request with destination path only
Returns:
Updated entity with new file path
"""
logger.info(
f"API v2 request: move_entity entity_id={entity_id}, destination='{data.destination_path}'"
)
try:
# First, get the entity by ID to verify it exists
entity = await entity_repository.find_by_id(entity_id)
if not entity:
raise HTTPException(status_code=404, detail=f"Entity not found: {entity_id}")
# Move the entity using its current file path as identifier
moved_entity = await entity_service.move_entity(
identifier=entity.file_path, # Use file path for resolution
destination_path=data.destination_path,
project_config=project_config,
app_config=app_config,
)
# Reindex at new location
reindexed_entity = await entity_service.link_resolver.resolve_link(data.destination_path)
if reindexed_entity:
await search_service.index_entity(reindexed_entity, background_tasks=background_tasks)
result = EntityResponseV2.model_validate(moved_entity)
logger.info(
f"API v2 response: moved entity_id={moved_entity.id} to '{data.destination_path}'"
)
return result
except HTTPException:
raise
except Exception as e:
logger.error(f"Error moving entity: {e}")
raise HTTPException(status_code=400, detail=str(e))
src/basic_memory/api/v2/routers/memory_router.py
+130
View File
@@ -0,0 +1,130 @@
"""V2 routes for memory:// URI operations.
This router uses integer project IDs for stable, efficient routing.
V1 uses string-based project names which are less efficient and less stable.
"""
from typing import Annotated, Optional
from fastapi import APIRouter, Query
from loguru import logger
from basic_memory.deps import ContextServiceV2Dep, EntityRepositoryV2Dep, ProjectIdPathDep
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
# Note: No prefix here - it's added during registration as /v2/{project_id}/memory
router = APIRouter(tags=["memory"])
@router.get("/memory/recent", response_model=GraphContext)
async def recent(
project_id: ProjectIdPathDep,
context_service: ContextServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
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:
"""Get recent activity context for a project.
Args:
project_id: Validated numeric project ID from URL path
context_service: Context service scoped to project
entity_repository: Entity repository scoped to project
type: Types of items to include (entities, relations, observations)
depth: How many levels of related entities to include
timeframe: Time window for recent activity (e.g., "7d", "1 week")
page: Page number for pagination
page_size: Number of items per page
max_related: Maximum related entities to include per item
Returns:
GraphContext with recent activity and related entities
"""
# return all types by default
types = (
[SearchItemType.ENTITY, SearchItemType.RELATION, SearchItemType.OBSERVATION]
if not type
else type
)
logger.debug(
f"V2 Getting recent context for project {project_id}: `{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"V2 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("/memory/{uri:path}", response_model=GraphContext)
async def get_memory_context(
project_id: ProjectIdPathDep,
context_service: ContextServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
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.
V2 supports both legacy path-based URIs and new ID-based URIs:
- Legacy: memory://path/to/note
- ID-based: memory://id/123 or memory://123
Args:
project_id: Validated numeric project ID from URL path
context_service: Context service scoped to project
entity_repository: Entity repository scoped to project
uri: Memory URI path (e.g., "id/123", "123", or "path/to/note")
depth: How many levels of related entities to include
timeframe: Optional time window for filtering related content
page: Page number for pagination
page_size: Number of items per page
max_related: Maximum related entities to include
Returns:
GraphContext with the entity and its related context
"""
logger.debug(
f"V2 Getting context for project {project_id}, 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
)
src/basic_memory/api/v2/routers/project_router.py
+264
View File
@@ -0,0 +1,264 @@
"""V2 Project Router - ID-based project management operations.
This router provides ID-based CRUD operations for projects, replacing the
name-based identifiers used in v1 with direct integer ID lookups.
Key improvements:
- Direct database lookups via integer primary keys
- Stable references that don't change with project renames
- Better performance through indexed queries
- Consistent with v2 entity operations
"""
import os
from typing import Optional
from fastapi import APIRouter, HTTPException, Body, Query
from loguru import logger
from basic_memory.deps import (
ProjectServiceDep,
ProjectRepositoryDep,
ProjectIdPathDep,
)
from basic_memory.schemas.project_info import (
ProjectItem,
ProjectStatusResponse,
)
from basic_memory.utils import normalize_project_path
router = APIRouter(prefix="/projects", tags=["project_management-v2"])
@router.get("/{project_id}", response_model=ProjectItem)
async def get_project_by_id(
project_id: ProjectIdPathDep,
project_repository: ProjectRepositoryDep,
) -> ProjectItem:
"""Get project by its numeric ID.
This is the primary project retrieval method in v2, using direct database
lookups for maximum performance.
Args:
project_id: Numeric project ID
Returns:
Project information
Raises:
HTTPException: 404 if project not found
Example:
GET /v2/projects/3
"""
logger.info(f"API v2 request: get_project_by_id for project_id={project_id}")
project = await project_repository.get_by_id(project_id)
if not project:
raise HTTPException(status_code=404, detail=f"Project with ID {project_id} not found")
return ProjectItem(
id=project.id,
name=project.name,
path=normalize_project_path(project.path),
is_default=project.is_default or False,
)
@router.patch("/{project_id}", response_model=ProjectStatusResponse)
async def update_project_by_id(
project_id: ProjectIdPathDep,
project_service: ProjectServiceDep,
project_repository: ProjectRepositoryDep,
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 by ID.
Args:
project_id: Numeric project ID
path: Optional new absolute path for the project
is_active: Optional status update for the project
Returns:
Response confirming the project was updated
Raises:
HTTPException: 400 if validation fails, 404 if project not found
Example:
PATCH /v2/projects/3
{"path": "/new/path"}
"""
logger.info(f"API v2 request: update_project_by_id for project_id={project_id}")
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_repository.get_by_id(project_id)
if not old_project:
raise HTTPException(status_code=404, detail=f"Project with ID {project_id} not found")
old_project_info = ProjectItem(
id=old_project.id,
name=old_project.name,
path=old_project.path,
is_default=old_project.is_default or False,
)
# Update using project name (service layer still uses names internally)
if path:
await project_service.move_project(old_project.name, path)
elif is_active is not None:
await project_service.update_project(old_project.name, is_active=is_active)
# Get updated project info
updated_project = await project_repository.get_by_id(project_id)
if not updated_project:
raise HTTPException(
status_code=404, detail=f"Project with ID {project_id} not found after update"
)
return ProjectStatusResponse(
message=f"Project '{updated_project.name}' updated successfully",
status="success",
default=(old_project.name == project_service.default_project),
old_project=old_project_info,
new_project=ProjectItem(
id=updated_project.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))
@router.delete("/{project_id}", response_model=ProjectStatusResponse)
async def delete_project_by_id(
project_id: ProjectIdPathDep,
project_service: ProjectServiceDep,
project_repository: ProjectRepositoryDep,
delete_notes: bool = Query(
False, description="If True, delete project directory from filesystem"
),
) -> ProjectStatusResponse:
"""Delete a project by ID.
Args:
project_id: Numeric project ID
delete_notes: If True, delete the project directory from the filesystem
Returns:
Response confirming the project was deleted
Raises:
HTTPException: 400 if trying to delete default project, 404 if not found
Example:
DELETE /v2/projects/3?delete_notes=false
"""
logger.info(
f"API v2 request: delete_project_by_id for project_id={project_id}, delete_notes={delete_notes}"
)
try:
old_project = await project_repository.get_by_id(project_id)
if not old_project:
raise HTTPException(status_code=404, detail=f"Project with ID {project_id} not found")
# Check if trying to delete the default project
if old_project.name == project_service.default_project:
available_projects = await project_service.list_projects()
other_projects = [p.name for p in available_projects if p.id != project_id]
detail = f"Cannot delete default project '{old_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)
# Delete using project name (service layer still uses names internally)
await project_service.remove_project(old_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,
name=old_project.name,
path=old_project.path,
is_default=old_project.is_default or False,
),
new_project=None,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
@router.put("/{project_id}/default", response_model=ProjectStatusResponse)
async def set_default_project_by_id(
project_id: ProjectIdPathDep,
project_service: ProjectServiceDep,
project_repository: ProjectRepositoryDep,
) -> ProjectStatusResponse:
"""Set a project as the default project by ID.
Args:
project_id: Numeric project ID to set as default
Returns:
Response confirming the project was set as default
Raises:
HTTPException: 404 if project not found
Example:
PUT /v2/projects/3/default
"""
logger.info(f"API v2 request: set_default_project_by_id for project_id={project_id}")
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:
raise HTTPException(
status_code=404, detail=f"Default Project: '{default_name}' does not exist"
)
# Get the new default project
new_default_project = await project_repository.get_by_id(project_id)
if not new_default_project:
raise HTTPException(status_code=404, detail=f"Project with ID {project_id} not found")
# Set as default using project name (service layer still uses names internally)
await project_service.set_default_project(new_default_project.name)
return ProjectStatusResponse(
message=f"Project '{new_default_project.name}' set as default successfully",
status="success",
default=True,
old_project=ProjectItem(
id=default_project.id,
name=default_name,
path=default_project.path,
is_default=False,
),
new_project=ProjectItem(
id=new_default_project.id,
name=new_default_project.name,
path=new_default_project.path,
is_default=True,
),
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
src/basic_memory/api/v2/routers/prompt_router.py
+270
View File
@@ -0,0 +1,270 @@
"""V2 Prompt Router - ID-based prompt generation operations.
This router uses v2 dependencies for consistent project ID handling.
Prompt endpoints are action-based (not resource-based), so they don't
have entity IDs in URLs - they generate formatted prompts from queries.
"""
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 (
ContextServiceV2Dep,
EntityRepositoryV2Dep,
SearchServiceV2Dep,
EntityServiceV2Dep,
ProjectIdPathDep,
)
from basic_memory.schemas.prompt import (
ContinueConversationRequest,
SearchPromptRequest,
PromptResponse,
PromptMetadata,
)
from basic_memory.schemas.search import SearchItemType, SearchQuery
router = APIRouter(prefix="/prompt", tags=["prompt-v2"])
@router.post("/continue-conversation", response_model=PromptResponse)
async def continue_conversation(
project_id: ProjectIdPathDep,
search_service: SearchServiceV2Dep,
entity_service: EntityServiceV2Dep,
context_service: ContextServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
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:
project_id: Validated numeric project ID from URL path
request: The request parameters
Returns:
Formatted continuation prompt with context
"""
logger.info(
f"V2 Generating continue conversation prompt for project {project_id}, "
f"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(
project_id: ProjectIdPathDep,
search_service: SearchServiceV2Dep,
entity_service: EntityServiceV2Dep,
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:
project_id: Validated numeric project ID from URL path
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"V2 Generating search prompt for project {project_id}, "
f"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)}",
)
src/basic_memory/api/v2/routers/resource_router.py
+292
View File
@@ -0,0 +1,292 @@
"""V2 Resource Router - ID-based resource content operations.
This router uses entity IDs for all operations, with file paths in request bodies
when needed. This is consistent with v2's ID-first design.
Key differences from v1:
- Uses integer entity IDs in URL paths instead of file paths
- File paths are in request bodies for create/update operations
- More RESTful: POST for create, PUT for update, GET for read
"""
from pathlib import Path
from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
from loguru import logger
from basic_memory.deps import (
ProjectConfigV2Dep,
EntityServiceV2Dep,
FileServiceV2Dep,
EntityRepositoryV2Dep,
SearchServiceV2Dep,
ProjectIdPathDep,
)
from basic_memory.models.knowledge import Entity as EntityModel
from basic_memory.schemas.v2.resource import (
CreateResourceRequest,
UpdateResourceRequest,
ResourceResponse,
)
from basic_memory.utils import validate_project_path
from datetime import datetime
router = APIRouter(prefix="/resource", tags=["resources-v2"])
@router.get("/{entity_id}")
async def get_resource_content(
project_id: ProjectIdPathDep,
entity_id: int,
config: ProjectConfigV2Dep,
entity_service: EntityServiceV2Dep,
file_service: FileServiceV2Dep,
) -> FileResponse:
"""Get raw resource content by entity ID.
Args:
project_id: Validated numeric project ID from URL path
entity_id: Numeric entity ID
config: Project configuration
entity_service: Entity service for fetching entity data
file_service: File service for reading file content
Returns:
FileResponse with entity content
Raises:
HTTPException: 404 if entity or file not found
"""
logger.debug(f"V2 Getting content for project {project_id}, entity_id: {entity_id}")
# Get entity by ID
entities = await entity_service.get_entities_by_id([entity_id])
if not entities:
raise HTTPException(status_code=404, detail=f"Entity {entity_id} not found")
entity = entities[0]
# Validate entity file path to prevent path traversal
project_path = Path(config.home)
if not validate_project_path(entity.file_path, project_path):
logger.error(f"Invalid file path in entity {entity.id}: {entity.file_path}")
raise HTTPException(
status_code=500,
detail="Entity contains invalid file path",
)
file_path = Path(f"{config.home}/{entity.file_path}")
if not file_path.exists():
raise HTTPException(
status_code=404,
detail=f"File not found: {file_path}",
)
return FileResponse(path=file_path)
@router.post("", response_model=ResourceResponse)
async def create_resource(
project_id: ProjectIdPathDep,
data: CreateResourceRequest,
config: ProjectConfigV2Dep,
file_service: FileServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
search_service: SearchServiceV2Dep,
) -> ResourceResponse:
"""Create a new resource file.
Args:
project_id: Validated numeric project ID from URL path
data: Create resource request with file_path and content
config: Project configuration
file_service: File service for writing files
entity_repository: Entity repository for creating entities
search_service: Search service for indexing
Returns:
ResourceResponse with file information including entity_id
Raises:
HTTPException: 400 for invalid file paths, 409 if file already exists
"""
try:
# Validate path to prevent path traversal attacks
project_path = Path(config.home)
if not validate_project_path(data.file_path, project_path):
logger.warning(
f"Invalid file path attempted: {data.file_path} in project {config.name}"
)
raise HTTPException(
status_code=400,
detail=f"Invalid file path: {data.file_path}. "
"Path must be relative and stay within project boundaries.",
)
# Check if entity already exists
existing_entity = await entity_repository.get_by_file_path(data.file_path)
if existing_entity:
raise HTTPException(
status_code=409,
detail=f"Resource already exists at {data.file_path} with entity_id {existing_entity.id}. "
f"Use PUT /resource/{existing_entity.id} to update it.",
)
# Get full file path
full_path = Path(f"{config.home}/{data.file_path}")
# Ensure parent directory exists
full_path.parent.mkdir(parents=True, exist_ok=True)
# Write content to file
checksum = await file_service.write_file(full_path, data.content)
# Get file info
file_stats = file_service.file_stats(full_path)
# Determine file details
file_name = Path(data.file_path).name
content_type = file_service.content_type(full_path)
entity_type = "canvas" if data.file_path.endswith(".canvas") else "file"
# Create a new entity model
entity = EntityModel(
title=file_name,
entity_type=entity_type,
content_type=content_type,
file_path=data.file_path,
checksum=checksum,
created_at=datetime.fromtimestamp(file_stats.st_ctime).astimezone(),
updated_at=datetime.fromtimestamp(file_stats.st_mtime).astimezone(),
)
entity = await entity_repository.add(entity)
# Index the file for search
await search_service.index_entity(entity) # pyright: ignore
# Return success response
return ResourceResponse(
entity_id=entity.id,
file_path=data.file_path,
checksum=checksum,
size=file_stats.st_size,
created_at=file_stats.st_ctime,
modified_at=file_stats.st_mtime,
)
except HTTPException:
# Re-raise HTTP exceptions without wrapping
raise
except Exception as e: # pragma: no cover
logger.error(f"Error creating resource {data.file_path}: {e}")
raise HTTPException(status_code=500, detail=f"Failed to create resource: {str(e)}")
@router.put("/{entity_id}", response_model=ResourceResponse)
async def update_resource(
project_id: ProjectIdPathDep,
entity_id: int,
data: UpdateResourceRequest,
config: ProjectConfigV2Dep,
file_service: FileServiceV2Dep,
entity_repository: EntityRepositoryV2Dep,
search_service: SearchServiceV2Dep,
) -> ResourceResponse:
"""Update an existing resource by entity ID.
Can update content and optionally move the file to a new path.
Args:
project_id: Validated numeric project ID from URL path
entity_id: Entity ID of the resource to update
data: Update resource request with content and optional new file_path
config: Project configuration
file_service: File service for writing files
entity_repository: Entity repository for updating entities
search_service: Search service for indexing
Returns:
ResourceResponse with updated file information
Raises:
HTTPException: 404 if entity not found, 400 for invalid paths
"""
try:
# Get existing entity
entity = await entity_repository.get_by_id(entity_id)
if not entity:
raise HTTPException(status_code=404, detail=f"Entity {entity_id} not found")
# Determine target file path
target_file_path = data.file_path if data.file_path else entity.file_path
# Validate path to prevent path traversal attacks
project_path = Path(config.home)
if not validate_project_path(target_file_path, project_path):
logger.warning(
f"Invalid file path attempted: {target_file_path} in project {config.name}"
)
raise HTTPException(
status_code=400,
detail=f"Invalid file path: {target_file_path}. "
"Path must be relative and stay within project boundaries.",
)
# Get full paths
old_full_path = Path(f"{config.home}/{entity.file_path}")
new_full_path = Path(f"{config.home}/{target_file_path}")
# If moving file, handle the move
if data.file_path and data.file_path != entity.file_path:
# Ensure new parent directory exists
new_full_path.parent.mkdir(parents=True, exist_ok=True)
# If old file exists, remove it
if old_full_path.exists():
old_full_path.unlink()
else:
# Ensure directory exists for in-place update
new_full_path.parent.mkdir(parents=True, exist_ok=True)
# Write content to target file
checksum = await file_service.write_file(new_full_path, data.content)
# Get file info
file_stats = file_service.file_stats(new_full_path)
# Determine file details
file_name = Path(target_file_path).name
content_type = file_service.content_type(new_full_path)
entity_type = "canvas" if target_file_path.endswith(".canvas") else "file"
# Update entity
updated_entity = await entity_repository.update(
entity_id,
{
"title": file_name,
"entity_type": entity_type,
"content_type": content_type,
"file_path": target_file_path,
"checksum": checksum,
"updated_at": datetime.fromtimestamp(file_stats.st_mtime).astimezone(),
},
)
# Index the updated file for search
await search_service.index_entity(updated_entity) # pyright: ignore
# Return success response
return ResourceResponse(
entity_id=entity_id,
file_path=target_file_path,
checksum=checksum,
size=file_stats.st_size,
created_at=file_stats.st_ctime,
modified_at=file_stats.st_mtime,
)
except HTTPException:
# Re-raise HTTP exceptions without wrapping
raise
except Exception as e: # pragma: no cover
logger.error(f"Error updating resource {entity_id}: {e}")
raise HTTPException(status_code=500, detail=f"Failed to update resource: {str(e)}")
src/basic_memory/api/v2/routers/search_router.py
+73
View File
@@ -0,0 +1,73 @@
"""V2 router for search operations.
This router uses integer project IDs for stable, efficient routing.
V1 uses string-based project names which are less efficient and less stable.
"""
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 SearchServiceV2Dep, EntityServiceV2Dep, ProjectIdPathDep
# Note: No prefix here - it's added during registration as /v2/{project_id}/search
router = APIRouter(tags=["search"])
@router.post("/search/", response_model=SearchResponse)
async def search(
project_id: ProjectIdPathDep,
query: SearchQuery,
search_service: SearchServiceV2Dep,
entity_service: EntityServiceV2Dep,
page: int = 1,
page_size: int = 10,
):
"""Search across all knowledge and documents in a project.
V2 uses integer project IDs for improved performance and stability.
Args:
project_id: Validated numeric project ID from URL path
query: Search query parameters (text, filters, etc.)
search_service: Search service scoped to project
entity_service: Entity service scoped to project
page: Page number for pagination
page_size: Number of results per page
Returns:
SearchResponse with paginated search results
"""
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("/search/reindex")
async def reindex(
project_id: ProjectIdPathDep,
background_tasks: BackgroundTasks,
search_service: SearchServiceV2Dep,
):
"""Recreate and populate the search index for a project.
This is a background operation that rebuilds the search index
from scratch. Useful after bulk updates or if the index becomes
corrupted.
Args:
project_id: Validated numeric project ID from URL path
background_tasks: FastAPI background tasks handler
search_service: Search service scoped to project
Returns:
Status message indicating reindex has been initiated
"""
await search_service.reindex_all(background_tasks=background_tasks)
return {"status": "ok", "message": "Reindex initiated"}
src/basic_memory/deps.py
+280
View File
@@ -76,6 +76,34 @@ async def get_project_config(
ProjectConfigDep = Annotated[ProjectConfig, Depends(get_project_config)] # pragma: no cover
async def get_project_config_v2(
project_id: "ProjectIdPathDep", project_repository: "ProjectRepositoryDep"
) -> ProjectConfig: # pragma: no cover
"""Get the project config for v2 API (uses integer project_id from path).
Args:
project_id: The validated numeric project ID from the URL path
project_repository: Repository for project operations
Returns:
The resolved project config
Raises:
HTTPException: If project is not found
"""
project_obj = await project_repository.get_by_id(project_id)
if project_obj:
return ProjectConfig(name=project_obj.name, home=pathlib.Path(project_obj.path))
# Not found (this should not happen since ProjectIdPathDep already validates existence)
raise HTTPException( # pragma: no cover
status_code=status.HTTP_404_NOT_FOUND, detail=f"Project with ID {project_id} not found."
)
ProjectConfigV2Dep = Annotated[ProjectConfig, Depends(get_project_config_v2)] # pragma: no cover
## sqlalchemy
@@ -130,6 +158,38 @@ ProjectRepositoryDep = Annotated[ProjectRepository, Depends(get_project_reposito
ProjectPathDep = Annotated[str, Path()] # Use Path dependency to extract from URL
async def validate_project_id(
project_id: int,
project_repository: ProjectRepositoryDep,
) -> int:
"""Validate that a numeric project ID exists in the database.
This is used for v2 API endpoints that take project IDs as integers in the path.
The project_id parameter will be automatically extracted from the URL path by FastAPI.
Args:
project_id: The numeric project ID from the URL path
project_repository: Repository for project operations
Returns:
The validated project ID
Raises:
HTTPException: If project with that ID is not found
"""
project_obj = await project_repository.get_by_id(project_id)
if not project_obj:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Project with ID {project_id} not found.",
)
return project_id
# V2 API: Validated integer project ID from path
ProjectIdPathDep = Annotated[int, Depends(validate_project_id)]
async def get_project_id(
project_repository: ProjectRepositoryDep,
project: ProjectPathDep,
@@ -188,6 +248,17 @@ async def get_entity_repository(
EntityRepositoryDep = Annotated[EntityRepository, Depends(get_entity_repository)]
async def get_entity_repository_v2(
session_maker: SessionMakerDep,
project_id: ProjectIdPathDep,
) -> EntityRepository:
"""Create an EntityRepository instance for v2 API (uses integer project_id from path)."""
return EntityRepository(session_maker, project_id=project_id)
EntityRepositoryV2Dep = Annotated[EntityRepository, Depends(get_entity_repository_v2)]
async def get_observation_repository(
session_maker: SessionMakerDep,
project_id: ProjectIdDep,
@@ -199,6 +270,19 @@ async def get_observation_repository(
ObservationRepositoryDep = Annotated[ObservationRepository, Depends(get_observation_repository)]
async def get_observation_repository_v2(
session_maker: SessionMakerDep,
project_id: ProjectIdPathDep,
) -> ObservationRepository:
"""Create an ObservationRepository instance for v2 API."""
return ObservationRepository(session_maker, project_id=project_id)
ObservationRepositoryV2Dep = Annotated[
ObservationRepository, Depends(get_observation_repository_v2)
]
async def get_relation_repository(
session_maker: SessionMakerDep,
project_id: ProjectIdDep,
@@ -210,6 +294,17 @@ async def get_relation_repository(
RelationRepositoryDep = Annotated[RelationRepository, Depends(get_relation_repository)]
async def get_relation_repository_v2(
session_maker: SessionMakerDep,
project_id: ProjectIdPathDep,
) -> RelationRepository:
"""Create a RelationRepository instance for v2 API."""
return RelationRepository(session_maker, project_id=project_id)
RelationRepositoryV2Dep = Annotated[RelationRepository, Depends(get_relation_repository_v2)]
async def get_search_repository(
session_maker: SessionMakerDep,
project_id: ProjectIdDep,
@@ -225,6 +320,17 @@ async def get_search_repository(
SearchRepositoryDep = Annotated[SearchRepository, Depends(get_search_repository)]
async def get_search_repository_v2(
session_maker: SessionMakerDep,
project_id: ProjectIdPathDep,
) -> SearchRepository:
"""Create a SearchRepository instance for v2 API."""
return create_search_repository(session_maker, project_id=project_id)
SearchRepositoryV2Dep = Annotated[SearchRepository, Depends(get_search_repository_v2)]
# ProjectInfoRepository is deprecated and will be removed in a future version.
# Use ProjectRepository instead, which has the same functionality plus more project-specific operations.
@@ -238,6 +344,13 @@ async def get_entity_parser(project_config: ProjectConfigDep) -> EntityParser:
EntityParserDep = Annotated["EntityParser", Depends(get_entity_parser)]
async def get_entity_parser_v2(project_config: ProjectConfigV2Dep) -> EntityParser:
return EntityParser(project_config.home)
EntityParserV2Dep = Annotated["EntityParser", Depends(get_entity_parser_v2)]
async def get_markdown_processor(entity_parser: EntityParserDep) -> MarkdownProcessor:
return MarkdownProcessor(entity_parser)
@@ -245,6 +358,13 @@ async def get_markdown_processor(entity_parser: EntityParserDep) -> MarkdownProc
MarkdownProcessorDep = Annotated[MarkdownProcessor, Depends(get_markdown_processor)]
async def get_markdown_processor_v2(entity_parser: EntityParserV2Dep) -> MarkdownProcessor:
return MarkdownProcessor(entity_parser)
MarkdownProcessorV2Dep = Annotated[MarkdownProcessor, Depends(get_markdown_processor_v2)]
async def get_file_service(
project_config: ProjectConfigDep, markdown_processor: MarkdownProcessorDep
) -> FileService:
@@ -259,6 +379,20 @@ async def get_file_service(
FileServiceDep = Annotated[FileService, Depends(get_file_service)]
async def get_file_service_v2(
project_config: ProjectConfigV2Dep, markdown_processor: MarkdownProcessorV2Dep
) -> FileService:
logger.debug(
f"Creating FileService for project: {project_config.name}, base_path: {project_config.home}"
)
file_service = FileService(project_config.home, markdown_processor)
logger.debug(f"Created FileService for project: {file_service} ")
return file_service
FileServiceV2Dep = Annotated[FileService, Depends(get_file_service_v2)]
async def get_entity_service(
entity_repository: EntityRepositoryDep,
observation_repository: ObservationRepositoryDep,
@@ -283,6 +417,30 @@ async def get_entity_service(
EntityServiceDep = Annotated[EntityService, Depends(get_entity_service)]
async def get_entity_service_v2(
entity_repository: EntityRepositoryV2Dep,
observation_repository: ObservationRepositoryV2Dep,
relation_repository: RelationRepositoryV2Dep,
entity_parser: EntityParserV2Dep,
file_service: FileServiceV2Dep,
link_resolver: "LinkResolverV2Dep",
app_config: AppConfigDep,
) -> EntityService:
"""Create EntityService for v2 API."""
return EntityService(
entity_repository=entity_repository,
observation_repository=observation_repository,
relation_repository=relation_repository,
entity_parser=entity_parser,
file_service=file_service,
link_resolver=link_resolver,
app_config=app_config,
)
EntityServiceV2Dep = Annotated[EntityService, Depends(get_entity_service_v2)]
async def get_search_service(
search_repository: SearchRepositoryDep,
entity_repository: EntityRepositoryDep,
@@ -295,6 +453,18 @@ async def get_search_service(
SearchServiceDep = Annotated[SearchService, Depends(get_search_service)]
async def get_search_service_v2(
search_repository: SearchRepositoryV2Dep,
entity_repository: EntityRepositoryV2Dep,
file_service: FileServiceV2Dep,
) -> SearchService:
"""Create SearchService for v2 API."""
return SearchService(search_repository, entity_repository, file_service)
SearchServiceV2Dep = Annotated[SearchService, Depends(get_search_service_v2)]
async def get_link_resolver(
entity_repository: EntityRepositoryDep, search_service: SearchServiceDep
) -> LinkResolver:
@@ -304,6 +474,15 @@ async def get_link_resolver(
LinkResolverDep = Annotated[LinkResolver, Depends(get_link_resolver)]
async def get_link_resolver_v2(
entity_repository: EntityRepositoryV2Dep, search_service: SearchServiceV2Dep
) -> LinkResolver:
return LinkResolver(entity_repository=entity_repository, search_service=search_service)
LinkResolverV2Dep = Annotated[LinkResolver, Depends(get_link_resolver_v2)]
async def get_context_service(
search_repository: SearchRepositoryDep,
entity_repository: EntityRepositoryDep,
@@ -319,6 +498,22 @@ async def get_context_service(
ContextServiceDep = Annotated[ContextService, Depends(get_context_service)]
async def get_context_service_v2(
search_repository: SearchRepositoryV2Dep,
entity_repository: EntityRepositoryV2Dep,
observation_repository: ObservationRepositoryV2Dep,
) -> ContextService:
"""Create ContextService for v2 API."""
return ContextService(
search_repository=search_repository,
entity_repository=entity_repository,
observation_repository=observation_repository,
)
ContextServiceV2Dep = Annotated[ContextService, Depends(get_context_service_v2)]
async def get_sync_service(
app_config: AppConfigDep,
entity_service: EntityServiceDep,
@@ -348,6 +543,32 @@ async def get_sync_service(
SyncServiceDep = Annotated[SyncService, Depends(get_sync_service)]
async def get_sync_service_v2(
app_config: AppConfigDep,
entity_service: EntityServiceV2Dep,
entity_parser: EntityParserV2Dep,
entity_repository: EntityRepositoryV2Dep,
relation_repository: RelationRepositoryV2Dep,
project_repository: ProjectRepositoryDep,
search_service: SearchServiceV2Dep,
file_service: FileServiceV2Dep,
) -> SyncService: # pragma: no cover
"""Create SyncService for v2 API."""
return SyncService(
app_config=app_config,
entity_service=entity_service,
entity_parser=entity_parser,
entity_repository=entity_repository,
relation_repository=relation_repository,
project_repository=project_repository,
search_service=search_service,
file_service=file_service,
)
SyncServiceV2Dep = Annotated[SyncService, Depends(get_sync_service_v2)]
async def get_project_service(
project_repository: ProjectRepositoryDep,
) -> ProjectService:
@@ -370,6 +591,18 @@ async def get_directory_service(
DirectoryServiceDep = Annotated[DirectoryService, Depends(get_directory_service)]
async def get_directory_service_v2(
entity_repository: EntityRepositoryV2Dep,
) -> DirectoryService:
"""Create DirectoryService for v2 API (uses integer project_id from path)."""
return DirectoryService(
entity_repository=entity_repository,
)
DirectoryServiceV2Dep = Annotated[DirectoryService, Depends(get_directory_service_v2)]
# Import
@@ -413,3 +646,50 @@ async def get_memory_json_importer(
MemoryJsonImporterDep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer)]
# V2 Import dependencies
async def get_chatgpt_importer_v2(
project_config: ProjectConfigV2Dep, markdown_processor: MarkdownProcessorV2Dep
) -> ChatGPTImporter:
"""Create ChatGPTImporter with v2 dependencies."""
return ChatGPTImporter(project_config.home, markdown_processor)
ChatGPTImporterV2Dep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2)]
async def get_claude_conversations_importer_v2(
project_config: ProjectConfigV2Dep, markdown_processor: MarkdownProcessorV2Dep
) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with v2 dependencies."""
return ClaudeConversationsImporter(project_config.home, markdown_processor)
ClaudeConversationsImporterV2Dep = Annotated[
ClaudeConversationsImporter, Depends(get_claude_conversations_importer_v2)
]
async def get_claude_projects_importer_v2(
project_config: ProjectConfigV2Dep, markdown_processor: MarkdownProcessorV2Dep
) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with v2 dependencies."""
return ClaudeProjectsImporter(project_config.home, markdown_processor)
ClaudeProjectsImporterV2Dep = Annotated[
ClaudeProjectsImporter, Depends(get_claude_projects_importer_v2)
]
async def get_memory_json_importer_v2(
project_config: ProjectConfigV2Dep, markdown_processor: MarkdownProcessorV2Dep
) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with v2 dependencies."""
return MemoryJsonImporter(project_config.home, markdown_processor)
MemoryJsonImporterV2Dep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer_v2)]
src/basic_memory/importers/claude_conversations_importer.py
+4 -1
View File
@@ -40,10 +40,13 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
chats_imported = 0
for chat in conversations:
# Get name, providing default for unnamed conversations
chat_name = chat.get("name") or f"Conversation {chat.get('uuid', 'untitled')}"
# Convert to entity
entity = self._format_chat_content(
base_path=folder_path,
name=chat["name"],
name=chat_name,
messages=chat["chat_messages"],
created_at=chat["created_at"],
modified_at=chat["updated_at"],
src/basic_memory/models/knowledge.py
+1 -1
View File
@@ -129,7 +129,7 @@ class Entity(Base):
return value
def __repr__(self) -> str:
return f"Entity(id={self.id}, name='{self.title}', type='{self.entity_type}'"
return f"Entity(id={self.id}, name='{self.title}', type='{self.entity_type}', checksum='{self.checksum}')"
class Observation(Base):
src/basic_memory/repository/entity_repository.py
+12
View File
@@ -32,6 +32,18 @@ class EntityRepository(Repository[Entity]):
"""
super().__init__(session_maker, Entity, project_id=project_id)
async def get_by_id(self, entity_id: int) -> Optional[Entity]:
"""Get entity by numeric ID.
Args:
entity_id: Numeric entity ID
Returns:
Entity if found, None otherwise
"""
async with db.scoped_session(self.session_maker) as session:
return await self.select_by_id(session, entity_id)
async def get_by_permalink(self, permalink: str) -> Optional[Entity]:
"""Get entity by permalink.
src/basic_memory/repository/postgres_search_repository.py
+65
View File
@@ -311,3 +311,68 @@ class PostgresSearchRepository(SearchRepositoryBase):
)
return results
async def bulk_index_items(self, search_index_rows: List[SearchIndexRow]) -> None:
"""Index multiple items in a single batch operation using UPSERT.
Uses INSERT ... ON CONFLICT DO UPDATE to handle re-indexing of existing
entities (e.g., during forward reference resolution) without requiring
a separate delete operation. This eliminates race conditions between
delete and insert operations in separate transactions.
Args:
search_index_rows: List of SearchIndexRow objects to index
"""
if not search_index_rows:
return
async with db.scoped_session(self.session_maker) as session:
# When using text() raw SQL, always serialize JSON to string
# Both SQLite (TEXT) and Postgres (JSONB) accept JSON strings in raw SQL
# The database driver/column type will handle conversion
insert_data_list = []
for row in search_index_rows:
insert_data = row.to_insert(serialize_json=True)
insert_data["project_id"] = self.project_id
insert_data_list.append(insert_data)
# Use UPSERT (INSERT ... ON CONFLICT) to handle re-indexing
# Primary key is (id, type, project_id)
# This handles race conditions during forward reference resolution
# where an entity might be re-indexed before the delete commits
# Syntax works for both SQLite 3.24+ and PostgreSQL
await session.execute(
text("""
INSERT INTO search_index (
id, title, content_stems, content_snippet, permalink, file_path, type, metadata,
from_id, to_id, relation_type,
entity_id, category,
created_at, updated_at,
project_id
) VALUES (
:id, :title, :content_stems, :content_snippet, :permalink, :file_path, :type, :metadata,
:from_id, :to_id, :relation_type,
:entity_id, :category,
:created_at, :updated_at,
:project_id
)
ON CONFLICT (id, type, project_id) DO UPDATE SET
title = EXCLUDED.title,
content_stems = EXCLUDED.content_stems,
content_snippet = EXCLUDED.content_snippet,
permalink = EXCLUDED.permalink,
file_path = EXCLUDED.file_path,
metadata = EXCLUDED.metadata,
from_id = EXCLUDED.from_id,
to_id = EXCLUDED.to_id,
relation_type = EXCLUDED.relation_type,
entity_id = EXCLUDED.entity_id,
category = EXCLUDED.category,
created_at = EXCLUDED.created_at,
updated_at = EXCLUDED.updated_at
"""),
insert_data_list,
)
logger.debug(f"Bulk indexed {len(search_index_rows)} rows")
await session.commit()
src/basic_memory/repository/project_repository.py
+12
View File
@@ -49,6 +49,18 @@ class ProjectRepository(Repository[Project]):
query = self.select().where(Project.path == Path(path).as_posix())
return await self.find_one(query)
async def get_by_id(self, project_id: int) -> Optional[Project]:
"""Get project by numeric ID.
Args:
project_id: Numeric project ID
Returns:
Project if found, None otherwise
"""
async with db.scoped_session(self.session_maker) as session:
return await self.select_by_id(session, project_id)
async def get_default_project(self) -> Optional[Project]:
"""Get the default project (the one marked as is_default=True)."""
query = self.select().where(Project.is_default.is_not(None))
src/basic_memory/schemas/memory.py
+7
View File
@@ -124,6 +124,7 @@ class EntitySummary(BaseModel):
"""Simplified entity representation."""
type: Literal["entity"] = "entity"
entity_id: int # Database ID for v2 API consistency
permalink: Optional[str]
title: str
content: Optional[str] = None
@@ -141,12 +142,16 @@ class RelationSummary(BaseModel):
"""Simplified relation representation."""
type: Literal["relation"] = "relation"
relation_id: int # Database ID for v2 API consistency
entity_id: Optional[int] = None # ID of the entity this relation belongs to
title: str
file_path: str
permalink: str
relation_type: str
from_entity: Optional[str] = None
from_entity_id: Optional[int] = None # ID of source entity
to_entity: Optional[str] = None
to_entity_id: Optional[int] = None # ID of target entity
created_at: Annotated[
datetime, Field(json_schema_extra={"type": "string", "format": "date-time"})
]
@@ -160,6 +165,8 @@ class ObservationSummary(BaseModel):
"""Simplified observation representation."""
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
title: str
file_path: str
permalink: str
src/basic_memory/schemas/project_info.py
+1
View File
@@ -173,6 +173,7 @@ class ProjectWatchStatus(BaseModel):
class ProjectItem(BaseModel):
"""Simple representation of a project."""
id: int
name: str
path: str
is_default: bool = False
src/basic_memory/schemas/search.py
+5
View File
@@ -97,6 +97,11 @@ class SearchResult(BaseModel):
metadata: Optional[dict] = None
# IDs for v2 API consistency
entity_id: Optional[int] = None # Entity ID (always present for entities)
observation_id: Optional[int] = None # Observation ID (for observation results)
relation_id: Optional[int] = None # Relation ID (for relation results)
# Type-specific fields
category: Optional[str] = None # For observations
from_entity: Optional[Permalink] = None # For relations
src/basic_memory/schemas/v2/__init__.py
+23
View File
@@ -0,0 +1,23 @@
"""V2 API schemas - ID-based entity references."""
from basic_memory.schemas.v2.entity import (
EntityResolveRequest,
EntityResolveResponse,
EntityResponseV2,
MoveEntityRequestV2,
)
from basic_memory.schemas.v2.resource import (
CreateResourceRequest,
UpdateResourceRequest,
ResourceResponse,
)
__all__ = [
"EntityResolveRequest",
"EntityResolveResponse",
"EntityResponseV2",
"MoveEntityRequestV2",
"CreateResourceRequest",
"UpdateResourceRequest",
"ResourceResponse",
]
src/basic_memory/schemas/v2/entity.py
+96
View File
@@ -0,0 +1,96 @@
"""V2 entity schemas with ID-first design."""
from datetime import datetime
from typing import Dict, List, Literal, Optional
from pydantic import BaseModel, Field, ConfigDict
from basic_memory.schemas.response import ObservationResponse, RelationResponse
class EntityResolveRequest(BaseModel):
"""Request to resolve a string identifier to an entity ID.
Supports resolution of:
- Permalinks (e.g., "specs/search")
- Titles (e.g., "Search Specification")
- File paths (e.g., "specs/search.md")
"""
identifier: str = Field(
...,
description="Entity identifier to resolve (permalink, title, or file path)",
min_length=1,
max_length=500,
)
class EntityResolveResponse(BaseModel):
"""Response from identifier resolution.
Returns the entity ID and associated metadata for the resolved entity.
"""
entity_id: int = Field(..., description="Numeric entity ID (primary identifier)")
permalink: Optional[str] = Field(None, description="Entity permalink")
file_path: str = Field(..., description="Relative file path")
title: str = Field(..., description="Entity title")
resolution_method: Literal["id", "permalink", "title", "path", "search"] = Field(
..., description="How the identifier was resolved"
)
class MoveEntityRequestV2(BaseModel):
"""V2 request schema for moving an entity to a new file location.
In V2 API, the entity ID is provided in the URL path, so this request
only needs the destination path.
"""
destination_path: str = Field(
...,
description="New file path for the entity (relative to project root)",
min_length=1,
max_length=500,
)
class EntityResponseV2(BaseModel):
"""V2 entity response with ID as the primary field.
This response format emphasizes the entity ID as the primary identifier,
with all other fields (permalink, file_path) as secondary metadata.
"""
# ID first - this is the primary identifier in v2
id: int = Field(..., description="Numeric entity ID (primary identifier)")
# Core entity fields
title: str = Field(..., description="Entity title")
entity_type: str = Field(..., description="Entity type")
content_type: str = Field(default="text/markdown", description="Content MIME type")
# Secondary identifiers (for compatibility and convenience)
permalink: Optional[str] = Field(None, description="Entity permalink (may change)")
file_path: str = Field(..., description="Relative file path (may change)")
# Content and metadata
content: Optional[str] = Field(None, description="Entity content")
entity_metadata: Optional[Dict] = Field(None, description="Entity metadata")
# Relationships
observations: List[ObservationResponse] = Field(
default_factory=list, description="Entity observations"
)
relations: List[RelationResponse] = Field(default_factory=list, description="Entity relations")
# Timestamps
created_at: datetime = Field(..., description="Creation timestamp")
updated_at: datetime = Field(..., description="Last update timestamp")
# V2-specific metadata
api_version: Literal["v2"] = Field(
default="v2", description="API version (always 'v2' for this response)"
)
model_config = ConfigDict(from_attributes=True)
src/basic_memory/schemas/v2/resource.py
+46
View File
@@ -0,0 +1,46 @@
"""V2 resource schemas for file content operations."""
from pydantic import BaseModel, Field
class CreateResourceRequest(BaseModel):
"""Request to create a new resource file.
File path is required for new resources since we need to know where
to create the file.
"""
file_path: str = Field(
...,
description="Path to create the file, relative to project root",
min_length=1,
max_length=500,
)
content: str = Field(..., description="File content to write")
class UpdateResourceRequest(BaseModel):
"""Request to update an existing resource by entity ID.
Only content is required - the file path is already known from the entity.
Optionally can update the file_path to move the file.
"""
content: str = Field(..., description="File content to write")
file_path: str | None = Field(
None,
description="Optional new file path to move the resource",
min_length=1,
max_length=500,
)
class ResourceResponse(BaseModel):
"""Response from resource operations."""
entity_id: int = Field(..., description="Entity ID of the resource")
file_path: str = Field(..., description="File path of the resource")
checksum: str = Field(..., description="File content checksum")
size: int = Field(..., description="File size in bytes")
created_at: float = Field(..., description="Creation timestamp")
modified_at: float = Field(..., description="Modification timestamp")
test-int/mcp/test_project_management_integration.py
+2 -1
View File
@@ -77,7 +77,8 @@ async def test_create_project_basic_operation(mcp_server, app, test_project):
assert "test-new-project" in create_text
assert "Project Details:" in create_text
assert "Name: test-new-project" in create_text
assert "Path: /tmp/test-new-project" in create_text
# Check path contains project name (platform-independent)
assert "Path:" in create_text and "test-new-project" in create_text
assert "Project is now available for use" in create_text
# Verify project appears in project list
test-int/mcp/test_write_note_integration.py
-2
View File
@@ -437,9 +437,7 @@ async def test_write_note_project_path_validation(mcp_server, app, test_project)
project_with_tilde = ProjectItem(
id=1,
name="Test BiSync", # Name differs from path structure
description="Test",
path="~/Documents/Test BiSync", # Path with tilde
is_active=True,
is_default=False,
)
tests/api/test_continue_conversation_template.py
+3
View File
@@ -18,6 +18,7 @@ def template_loader():
def entity_summary():
"""Create a sample EntitySummary for testing."""
return EntitySummary(
entity_id=1,
title="Test Entity",
permalink="test/entity",
type=SearchItemType.ENTITY,
@@ -34,6 +35,8 @@ def context_with_results(entity_summary):
# Create an observation for the entity
observation = ObservationSummary(
observation_id=1,
entity_id=1,
title="Test Observation",
permalink="test/entity/observations/1",
category="test",
tests/api/v2/__init__.py
+1
View File
@@ -0,0 +1 @@
"""V2 API tests."""
tests/api/v2/conftest.py
+21
View File
@@ -0,0 +1,21 @@
"""Fixtures for V2 API tests."""
import pytest
from basic_memory.models import Project
@pytest.fixture
def v2_project_url(test_project: Project) -> str:
"""Create a URL prefix for v2 project-scoped routes using project ID.
This helps tests generate the correct URL for v2 project-scoped routes
which use integer project IDs instead of permalinks.
"""
return f"/v2/projects/{test_project.id}"
@pytest.fixture
def v2_projects_url() -> str:
"""Base URL for v2 project management endpoints."""
return "/v2/projects"
tests/api/v2/test_directory_router.py
+129
View File
@@ -0,0 +1,129 @@
"""Tests for V2 directory API routes (ID-based endpoints)."""
import pytest
from httpx import AsyncClient
from basic_memory.models import Project
from basic_memory.schemas.directory import DirectoryNode
@pytest.mark.asyncio
async def test_get_directory_tree(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test getting directory tree via v2 endpoint."""
response = await client.get(f"{v2_project_url}/directory/tree")
assert response.status_code == 200
tree = DirectoryNode.model_validate(response.json())
assert tree.type == "directory"
@pytest.mark.asyncio
async def test_get_directory_structure(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test getting directory structure (folders only) via v2 endpoint."""
response = await client.get(f"{v2_project_url}/directory/structure")
assert response.status_code == 200
structure = DirectoryNode.model_validate(response.json())
assert structure.type == "directory"
# Structure should only contain directories, not files
if structure.children:
for child in structure.children:
assert child.type == "directory"
@pytest.mark.asyncio
async def test_list_directory_default(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test listing directory contents with default parameters via v2 endpoint."""
response = await client.get(f"{v2_project_url}/directory/list")
assert response.status_code == 200
nodes = response.json()
assert isinstance(nodes, list)
@pytest.mark.asyncio
async def test_list_directory_with_depth(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test listing directory with custom depth via v2 endpoint."""
response = await client.get(f"{v2_project_url}/directory/list?depth=2")
assert response.status_code == 200
nodes = response.json()
assert isinstance(nodes, list)
@pytest.mark.asyncio
async def test_list_directory_with_glob(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test listing directory with file name glob filter via v2 endpoint."""
response = await client.get(f"{v2_project_url}/directory/list?file_name_glob=*.md")
assert response.status_code == 200
nodes = response.json()
assert isinstance(nodes, list)
# All file nodes should have .md extension
for node in nodes:
if node.get("type") == "file":
assert node.get("path", "").endswith(".md")
@pytest.mark.asyncio
async def test_list_directory_with_custom_path(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test listing a specific directory path via v2 endpoint."""
response = await client.get(f"{v2_project_url}/directory/list?dir_name=/")
assert response.status_code == 200
nodes = response.json()
assert isinstance(nodes, list)
@pytest.mark.asyncio
async def test_directory_invalid_project_id(
client: AsyncClient,
):
"""Test directory endpoints with invalid project ID return 404."""
# Test tree endpoint
response = await client.get("/v2/projects/999999/directory/tree")
assert response.status_code == 404
# Test structure endpoint
response = await client.get("/v2/projects/999999/directory/structure")
assert response.status_code == 404
# Test list endpoint
response = await client.get("/v2/projects/999999/directory/list")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_v2_directory_endpoints_use_project_id_not_name(
client: AsyncClient, test_project: Project
):
"""Verify v2 directory endpoints require project ID, not name."""
# Try using project name instead of ID - should fail
response = await client.get(f"/v2/projects/{test_project.name}/directory/tree")
# Should get validation error or 404 because name is not a valid integer
assert response.status_code in [404, 422]
tests/api/v2/test_importer_router.py
+530
View File
@@ -0,0 +1,530 @@
"""Tests for V2 importer API routes (ID-based endpoints)."""
import json
from pathlib import Path
import pytest
from httpx import AsyncClient
from basic_memory.models import Project
from basic_memory.schemas.importer import (
ChatImportResult,
EntityImportResult,
ProjectImportResult,
)
@pytest.fixture
def chatgpt_json_content():
"""Sample ChatGPT conversation data for testing."""
return [
{
"title": "Test Conversation",
"create_time": 1736616594.24054,
"update_time": 1736616603.164995,
"mapping": {
"root": {"id": "root", "message": None, "parent": None, "children": ["msg1"]},
"msg1": {
"id": "msg1",
"message": {
"id": "msg1",
"author": {"role": "user", "name": None, "metadata": {}},
"create_time": 1736616594.24054,
"content": {
"content_type": "text",
"parts": ["Hello, this is a test message"],
},
"status": "finished_successfully",
"metadata": {},
},
"parent": "root",
"children": ["msg2"],
},
"msg2": {
"id": "msg2",
"message": {
"id": "msg2",
"author": {"role": "assistant", "name": None, "metadata": {}},
"create_time": 1736616603.164995,
"content": {"content_type": "text", "parts": ["This is a test response"]},
"status": "finished_successfully",
"metadata": {},
},
"parent": "msg1",
"children": [],
},
},
}
]
@pytest.fixture
def claude_conversations_json_content():
"""Sample Claude conversations data for testing."""
return [
{
"uuid": "test-uuid",
"name": "Test Conversation",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"updated_at": "2025-01-05T20:56:39.477600+00:00",
"chat_messages": [
{
"uuid": "msg-1",
"text": "Hello, this is a test",
"sender": "human",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"content": [{"type": "text", "text": "Hello, this is a test"}],
},
{
"uuid": "msg-2",
"text": "Response to test",
"sender": "assistant",
"created_at": "2025-01-05T20:55:40.123456+00:00",
"content": [{"type": "text", "text": "Response to test"}],
},
],
}
]
@pytest.fixture
def claude_projects_json_content():
"""Sample Claude projects data for testing."""
return [
{
"uuid": "test-uuid",
"name": "Test Project",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"updated_at": "2025-01-05T20:56:39.477600+00:00",
"prompt_template": "# Test Prompt\n\nThis is a test prompt.",
"docs": [
{
"uuid": "doc-uuid-1",
"filename": "Test Document",
"content": "# Test Document\n\nThis is test content.",
"created_at": "2025-01-05T20:56:39.477600+00:00",
},
{
"uuid": "doc-uuid-2",
"filename": "Another Document",
"content": "# Another Document\n\nMore test content.",
"created_at": "2025-01-05T20:56:39.477600+00:00",
},
],
}
]
@pytest.fixture
def memory_json_content():
"""Sample memory.json data for testing."""
return [
{
"type": "entity",
"name": "test_entity",
"entityType": "test",
"observations": ["Test observation 1", "Test observation 2"],
},
{
"type": "relation",
"from": "test_entity",
"to": "related_entity",
"relationType": "test_relation",
},
]
async def create_test_upload_file(tmp_path, content):
"""Create a test file for upload."""
file_path = tmp_path / "test_import.json"
with open(file_path, "w", encoding="utf-8") as f:
json.dump(content, f)
return file_path
@pytest.mark.asyncio
async def test_import_chatgpt(
project_config,
client: AsyncClient,
tmp_path,
chatgpt_json_content,
file_service,
v2_project_url: str,
):
"""Test importing ChatGPT conversations via v2 endpoint."""
# Create a test file
file_path = await create_test_upload_file(tmp_path, chatgpt_json_content)
# Create a multipart form with the file
with open(file_path, "rb") as f:
files = {"file": ("conversations.json", f, "application/json")}
data = {"folder": "test_chatgpt"}
# Send request
response = await client.post(f"{v2_project_url}/import/chatgpt", files=files, data=data)
# Check response
assert response.status_code == 200
result = ChatImportResult.model_validate(response.json())
assert result.success is True
assert result.conversations == 1
assert result.messages == 2
# Verify files were created
conv_path = Path("test_chatgpt") / "20250111-Test_Conversation.md"
assert await file_service.exists(conv_path)
content, _ = await file_service.read_file(conv_path)
assert "# Test Conversation" in content
assert "Hello, this is a test message" in content
assert "This is a test response" in content
@pytest.mark.asyncio
async def test_import_chatgpt_invalid_file(client: AsyncClient, tmp_path, v2_project_url: str):
"""Test importing invalid ChatGPT file via v2 endpoint."""
# Create invalid file
file_path = tmp_path / "invalid.json"
with open(file_path, "w") as f:
f.write("This is not JSON")
# Create multipart form with invalid file
with open(file_path, "rb") as f:
files = {"file": ("invalid.json", f, "application/json")}
data = {"folder": "test_chatgpt"}
# Send request - this should return an error
response = await client.post(f"{v2_project_url}/import/chatgpt", files=files, data=data)
# Check response
assert response.status_code == 500
assert "Import failed" in response.json()["detail"]
@pytest.mark.asyncio
async def test_import_claude_conversations(
client: AsyncClient,
tmp_path,
claude_conversations_json_content,
file_service,
v2_project_url: str,
):
"""Test importing Claude conversations via v2 endpoint."""
# Create a test file
file_path = await create_test_upload_file(tmp_path, claude_conversations_json_content)
# Create a multipart form with the file
with open(file_path, "rb") as f:
files = {"file": ("conversations.json", f, "application/json")}
data = {"folder": "test_claude_conversations"}
# Send request
response = await client.post(
f"{v2_project_url}/import/claude/conversations", files=files, data=data
)
# Check response
assert response.status_code == 200
result = ChatImportResult.model_validate(response.json())
assert result.success is True
assert result.conversations == 1
assert result.messages == 2
# Verify files were created
conv_path = Path("test_claude_conversations") / "20250105-Test_Conversation.md"
assert await file_service.exists(conv_path)
content, _ = await file_service.read_file(conv_path)
assert "# Test Conversation" in content
assert "Hello, this is a test" in content
assert "Response to test" in content
@pytest.mark.asyncio
async def test_import_claude_conversations_invalid_file(
client: AsyncClient, tmp_path, v2_project_url: str
):
"""Test importing invalid Claude conversations file via v2 endpoint."""
# Create invalid file
file_path = tmp_path / "invalid.json"
with open(file_path, "w") as f:
f.write("This is not JSON")
# Create multipart form with invalid file
with open(file_path, "rb") as f:
files = {"file": ("invalid.json", f, "application/json")}
data = {"folder": "test_claude_conversations"}
# Send request - this should return an error
response = await client.post(
f"{v2_project_url}/import/claude/conversations", files=files, data=data
)
# Check response
assert response.status_code == 500
assert "Import failed" in response.json()["detail"]
@pytest.mark.asyncio
async def test_import_claude_projects(
client: AsyncClient, tmp_path, claude_projects_json_content, file_service, v2_project_url: str
):
"""Test importing Claude projects via v2 endpoint."""
# Create a test file
file_path = await create_test_upload_file(tmp_path, claude_projects_json_content)
# Create a multipart form with the file
with open(file_path, "rb") as f:
files = {"file": ("projects.json", f, "application/json")}
data = {"folder": "test_claude_projects"}
# Send request
response = await client.post(
f"{v2_project_url}/import/claude/projects", files=files, data=data
)
# Check response
assert response.status_code == 200
result = ProjectImportResult.model_validate(response.json())
assert result.success is True
assert result.documents == 2
assert result.prompts == 1
# Verify files were created
project_dir = Path("test_claude_projects") / "Test_Project"
assert await file_service.exists(project_dir / "prompt-template.md")
assert await file_service.exists(project_dir / "docs" / "Test_Document.md")
assert await file_service.exists(project_dir / "docs" / "Another_Document.md")
# Check content
prompt_content, _ = await file_service.read_file(project_dir / "prompt-template.md")
assert "# Test Prompt" in prompt_content
doc_content, _ = await file_service.read_file(project_dir / "docs" / "Test_Document.md")
assert "# Test Document" in doc_content
assert "This is test content" in doc_content
@pytest.mark.asyncio
async def test_import_claude_projects_invalid_file(
client: AsyncClient, tmp_path, v2_project_url: str
):
"""Test importing invalid Claude projects file via v2 endpoint."""
# Create invalid file
file_path = tmp_path / "invalid.json"
with open(file_path, "w") as f:
f.write("This is not JSON")
# Create multipart form with invalid file
with open(file_path, "rb") as f:
files = {"file": ("invalid.json", f, "application/json")}
data = {"folder": "test_claude_projects"}
# Send request - this should return an error
response = await client.post(
f"{v2_project_url}/import/claude/projects", files=files, data=data
)
# Check response
assert response.status_code == 500
assert "Import failed" in response.json()["detail"]
@pytest.mark.asyncio
async def test_import_memory_json(
client: AsyncClient, tmp_path, memory_json_content, file_service, v2_project_url: str
):
"""Test importing memory.json file via v2 endpoint."""
# Create a test file
json_file = tmp_path / "memory.json"
with open(json_file, "w", encoding="utf-8") as f:
for entity in memory_json_content:
f.write(json.dumps(entity) + "\n")
# Create a multipart form with the file
with open(json_file, "rb") as f:
files = {"file": ("memory.json", f, "application/json")}
data = {"folder": "test_memory_json"}
# Send request
response = await client.post(f"{v2_project_url}/import/memory-json", files=files, data=data)
# Check response
assert response.status_code == 200
result = EntityImportResult.model_validate(response.json())
assert result.success is True
assert result.entities == 1
assert result.relations == 1
# Verify files were created
entity_path = Path("test_memory_json") / "test" / "test_entity.md"
assert await file_service.exists(entity_path)
# Check content
content, _ = await file_service.read_file(entity_path)
assert "Test observation 1" in content
assert "Test observation 2" in content
assert "test_relation [[related_entity]]" in content
@pytest.mark.asyncio
async def test_import_memory_json_without_folder(
client: AsyncClient, tmp_path, memory_json_content, file_service, v2_project_url: str
):
"""Test importing memory.json file without specifying a destination folder."""
# Create a test file
json_file = tmp_path / "memory.json"
with open(json_file, "w", encoding="utf-8") as f:
for entity in memory_json_content:
f.write(json.dumps(entity) + "\n")
# Create a multipart form with the file
with open(json_file, "rb") as f:
files = {"file": ("memory.json", f, "application/json")}
# Send request without destination_folder
response = await client.post(f"{v2_project_url}/import/memory-json", files=files)
# Check response
assert response.status_code == 200
result = EntityImportResult.model_validate(response.json())
assert result.success is True
assert result.entities == 1
assert result.relations == 1
# Verify files were created in the default directory
entity_path = Path("conversations") / "test" / "test_entity.md"
assert await file_service.exists(entity_path)
@pytest.mark.asyncio
async def test_import_memory_json_invalid_file(client: AsyncClient, tmp_path, v2_project_url: str):
"""Test importing invalid memory.json file via v2 endpoint."""
# Create invalid file
file_path = tmp_path / "invalid.json"
with open(file_path, "w") as f:
f.write("This is not JSON")
# Create multipart form with invalid file
with open(file_path, "rb") as f:
files = {"file": ("invalid.json", f, "application/json")}
data = {"folder": "test_memory_json"}
# Send request - this should return an error
response = await client.post(f"{v2_project_url}/import/memory-json", files=files, data=data)
# Check response
assert response.status_code == 500
assert "Import failed" in response.json()["detail"]
@pytest.mark.asyncio
async def test_v2_import_endpoints_use_project_id_not_name(
client: AsyncClient, tmp_path, test_project: Project, chatgpt_json_content
):
"""Verify v2 import endpoints require project ID, not name."""
# Create a test file
file_path = await create_test_upload_file(tmp_path, chatgpt_json_content)
# Try using project name instead of ID - should fail
with open(file_path, "rb") as f:
files = {"file": ("conversations.json", f, "application/json")}
data = {"folder": "test"}
response = await client.post(
f"/v2/projects/{test_project.name}/import/chatgpt",
files=files,
data=data,
)
# Should get validation error or 404 because name is not a valid integer
assert response.status_code in [404, 422]
@pytest.mark.asyncio
async def test_import_invalid_project_id(client: AsyncClient, tmp_path, chatgpt_json_content):
"""Test import endpoints with invalid project ID return 404."""
# Create a test file
file_path = await create_test_upload_file(tmp_path, chatgpt_json_content)
# Test all import endpoints
endpoints = [
"/import/chatgpt",
"/import/claude/conversations",
"/import/claude/projects",
"/import/memory-json",
]
for endpoint in endpoints:
with open(file_path, "rb") as f:
files = {"file": ("test.json", f, "application/json")}
data = {"folder": "test"}
response = await client.post(
f"/v2/projects/999999{endpoint}",
files=files,
data=data,
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_import_missing_file(client: AsyncClient, v2_project_url: str):
"""Test importing with missing file via v2 endpoint."""
# Send a request without a file
response = await client.post(f"{v2_project_url}/import/chatgpt", data={"folder": "test_folder"})
# Check that the request was rejected
assert response.status_code in [400, 422] # Either bad request or unprocessable entity
@pytest.mark.asyncio
async def test_import_empty_file(client: AsyncClient, tmp_path, v2_project_url: str):
"""Test importing an empty file via v2 endpoint."""
# Create an empty file
file_path = tmp_path / "empty.json"
with open(file_path, "w") as f:
f.write("")
# Create multipart form with empty file
with open(file_path, "rb") as f:
files = {"file": ("empty.json", f, "application/json")}
data = {"folder": "test_chatgpt"}
# Send request
response = await client.post(f"{v2_project_url}/import/chatgpt", files=files, data=data)
# Check response
assert response.status_code == 500
assert "Import failed" in response.json()["detail"]
@pytest.mark.asyncio
async def test_import_malformed_json(client: AsyncClient, tmp_path, v2_project_url: str):
"""Test importing malformed JSON for all v2 import endpoints."""
# Create malformed JSON file
file_path = tmp_path / "malformed.json"
with open(file_path, "w") as f:
f.write('{"incomplete": "json"') # Missing closing brace
# Test all import endpoints
endpoints = [
(f"{v2_project_url}/import/chatgpt", {"folder": "test"}),
(f"{v2_project_url}/import/claude/conversations", {"folder": "test"}),
(f"{v2_project_url}/import/claude/projects", {"folder": "test"}),
(f"{v2_project_url}/import/memory-json", {"folder": "test"}),
]
for endpoint, data in endpoints:
# Create multipart form with malformed JSON
with open(file_path, "rb") as f:
files = {"file": ("malformed.json", f, "application/json")}
# Send request
response = await client.post(endpoint, files=files, data=data)
# Check response
assert response.status_code == 500
assert "Import failed" in response.json()["detail"]
tests/api/v2/test_knowledge_router.py
+407
View File
@@ -0,0 +1,407 @@
"""Tests for V2 knowledge graph API routes (ID-based endpoints)."""
import pytest
from httpx import AsyncClient
from basic_memory.models import Project
from basic_memory.schemas import DeleteEntitiesResponse
from basic_memory.schemas.v2 import EntityResponseV2, EntityResolveResponse
@pytest.mark.asyncio
async def test_resolve_identifier_by_permalink(
client: AsyncClient, test_graph, v2_project_url, test_project: Project, entity_repository
):
"""Test resolving an identifier by permalink returns correct entity ID."""
# test_graph fixture creates some test entities
# We'll use one of them to test resolution
# Create an entity first
entity_data = {
"title": "TestResolve",
"folder": "test",
"content": "Test content for resolve",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=entity_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
entity_id = created_entity.id
# Now resolve it by permalink
resolve_data = {"identifier": created_entity.permalink}
response = await client.post(f"{v2_project_url}/knowledge/resolve", json=resolve_data)
assert response.status_code == 200
resolved = EntityResolveResponse.model_validate(response.json())
assert resolved.entity_id == entity_id
assert resolved.permalink == created_entity.permalink
assert resolved.resolution_method == "permalink"
@pytest.mark.asyncio
async def test_resolve_identifier_not_found(client: AsyncClient, v2_project_url):
"""Test resolving a non-existent identifier returns 404."""
resolve_data = {"identifier": "nonexistent/entity"}
response = await client.post(f"{v2_project_url}/knowledge/resolve", json=resolve_data)
assert response.status_code == 404
assert "Could not resolve identifier" in response.json()["detail"]
@pytest.mark.asyncio
async def test_get_entity_by_id(client: AsyncClient, test_graph, v2_project_url, entity_repository):
"""Test getting an entity by its numeric ID."""
# Create an entity first
entity_data = {
"title": "TestGetById",
"folder": "test",
"content": "Test content for get by ID",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=entity_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
entity_id = created_entity.id
# Get it by ID using v2 endpoint
response = await client.get(f"{v2_project_url}/knowledge/entities/{entity_id}")
assert response.status_code == 200
entity = EntityResponseV2.model_validate(response.json())
assert entity.id == entity_id
assert entity.title == "TestGetById"
assert entity.api_version == "v2"
@pytest.mark.asyncio
async def test_get_entity_by_id_not_found(client: AsyncClient, v2_project_url):
"""Test getting a non-existent entity by ID returns 404."""
response = await client.get(f"{v2_project_url}/knowledge/entities/999999")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_create_entity(client: AsyncClient, file_service, v2_project_url):
"""Test creating an entity via v2 endpoint."""
data = {
"title": "TestV2Entity",
"folder": "test",
"entity_type": "test",
"content_type": "text/markdown",
"content": "TestContent for V2",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=data)
assert response.status_code == 200
entity = EntityResponseV2.model_validate(response.json())
# V2 endpoints must return id field
assert entity.id is not None
assert isinstance(entity.id, int)
assert entity.api_version == "v2"
assert entity.permalink == "test/test-v2-entity"
assert entity.file_path == "test/TestV2Entity.md"
assert entity.entity_type == data["entity_type"]
# Verify file was created
file_path = file_service.get_entity_path(entity)
file_content, _ = await file_service.read_file(file_path)
assert data["content"] in file_content
@pytest.mark.asyncio
async def test_create_entity_with_observations_and_relations(
client: AsyncClient, file_service, v2_project_url
):
"""Test creating an entity with observations and relations via v2."""
data = {
"title": "TestV2Complex",
"folder": "test",
"content": """
# TestV2Complex
## Observations
- [note] This is a test observation #tag1 (context)
- related to [[OtherEntity]]
""",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=data)
assert response.status_code == 200
entity = EntityResponseV2.model_validate(response.json())
# V2 endpoints must return id field
assert entity.id is not None
assert isinstance(entity.id, int)
assert entity.api_version == "v2"
assert len(entity.observations) == 1
assert entity.observations[0].category == "note"
assert entity.observations[0].content == "This is a test observation #tag1"
assert entity.observations[0].tags == ["tag1"]
assert len(entity.relations) == 1
assert entity.relations[0].relation_type == "related to"
@pytest.mark.asyncio
async def test_update_entity_by_id(
client: AsyncClient, file_service, v2_project_url, entity_repository
):
"""Test updating an entity by ID using PUT (replace)."""
# Create an entity first
create_data = {
"title": "TestUpdate",
"folder": "test",
"content": "Original content",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=create_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
original_id = created_entity.id
# Update it by ID
update_data = {
"title": "TestUpdate",
"folder": "test",
"content": "Updated content via V2",
}
response = await client.put(
f"{v2_project_url}/knowledge/entities/{original_id}",
json=update_data,
)
assert response.status_code == 200
updated_entity = EntityResponseV2.model_validate(response.json())
# V2 update must return id field
assert updated_entity.id is not None
assert isinstance(updated_entity.id, int)
assert updated_entity.api_version == "v2"
# Verify file was updated
file_path = file_service.get_entity_path(updated_entity)
file_content, _ = await file_service.read_file(file_path)
assert "Updated content via V2" in file_content
assert "Original content" not in file_content
@pytest.mark.asyncio
async def test_edit_entity_by_id_append(
client: AsyncClient, file_service, v2_project_url, entity_repository
):
"""Test editing an entity by ID using PATCH (append operation)."""
# Create an entity first
create_data = {
"title": "TestEdit",
"folder": "test",
"content": "# TestEdit\n\nOriginal content",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=create_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
original_id = created_entity.id
# Edit it by appending
edit_data = {
"operation": "append",
"content": "\n\n## New Section\n\nAppended content",
}
response = await client.patch(
f"{v2_project_url}/knowledge/entities/{original_id}",
json=edit_data,
)
assert response.status_code == 200
edited_entity = EntityResponseV2.model_validate(response.json())
# V2 patch must return id field
assert edited_entity.id is not None
assert isinstance(edited_entity.id, int)
assert edited_entity.api_version == "v2"
# Verify file has both original and appended content
file_path = file_service.get_entity_path(edited_entity)
file_content, _ = await file_service.read_file(file_path)
assert "Original content" in file_content
assert "Appended content" in file_content
@pytest.mark.asyncio
async def test_edit_entity_by_id_find_replace(
client: AsyncClient, file_service, v2_project_url, entity_repository
):
"""Test editing an entity by ID using PATCH (find/replace operation)."""
# Create an entity first
create_data = {
"title": "TestFindReplace",
"folder": "test",
"content": "# TestFindReplace\n\nOld text that will be replaced",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=create_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
original_id = created_entity.id
# Edit using find/replace
edit_data = {
"operation": "find_replace",
"find_text": "Old text",
"content": "New text",
}
response = await client.patch(
f"{v2_project_url}/knowledge/entities/{original_id}",
json=edit_data,
)
assert response.status_code == 200
edited_entity = EntityResponseV2.model_validate(response.json())
# V2 patch must return id field
assert edited_entity.id is not None
assert isinstance(edited_entity.id, int)
assert edited_entity.api_version == "v2"
# Verify replacement
file_path = file_service.get_entity_path(created_entity)
file_content, _ = await file_service.read_file(file_path)
assert "New text" in file_content
assert "Old text" not in file_content
@pytest.mark.asyncio
async def test_delete_entity_by_id(
client: AsyncClient, file_service, v2_project_url, entity_repository
):
"""Test deleting an entity by ID."""
# Create an entity first
create_data = {
"title": "TestDelete",
"folder": "test",
"content": "Content to be deleted",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=create_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
entity_id = created_entity.id
# Delete it by ID
response = await client.delete(f"{v2_project_url}/knowledge/entities/{entity_id}")
assert response.status_code == 200
delete_response = DeleteEntitiesResponse.model_validate(response.json())
assert delete_response.deleted is True
# Verify it's gone - trying to get it should return 404
response = await client.get(f"{v2_project_url}/knowledge/entities/{entity_id}")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_delete_entity_by_id_not_found(client: AsyncClient, v2_project_url):
"""Test deleting a non-existent entity returns deleted=False (idempotent)."""
response = await client.delete(f"{v2_project_url}/knowledge/entities/999999")
# Delete is idempotent - returns 200 with deleted=False
assert response.status_code == 200
delete_response = DeleteEntitiesResponse.model_validate(response.json())
assert delete_response.deleted is False
@pytest.mark.asyncio
async def test_move_entity(client: AsyncClient, file_service, v2_project_url, entity_repository):
"""Test moving an entity to a new location."""
# Create an entity first
create_data = {
"title": "TestMove",
"folder": "test",
"content": "Content to be moved",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=create_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id
assert created_entity.id is not None
original_id = created_entity.id
# Move it to a new folder (V2 uses entity ID in path)
move_data = {
"destination_path": "moved/MovedEntity.md",
}
response = await client.put(
f"{v2_project_url}/knowledge/entities/{created_entity.id}/move", json=move_data
)
assert response.status_code == 200
moved_entity = EntityResponseV2.model_validate(response.json())
# V2 move must return id field
assert moved_entity.id is not None
assert isinstance(moved_entity.id, int)
assert moved_entity.api_version == "v2"
# ID should remain the same (stable reference)
assert moved_entity.id == original_id
assert moved_entity.file_path == "moved/MovedEntity.md"
@pytest.mark.asyncio
async def test_v2_endpoints_use_project_id_not_name(client: AsyncClient, test_project: Project):
"""Verify v2 endpoints require project ID, not name."""
# Try using project name instead of ID - should fail
response = await client.get(f"/v2/{test_project.name}/knowledge/entities/1")
# Should get validation error or 404 because name is not a valid integer
assert response.status_code in [404, 422]
@pytest.mark.asyncio
async def test_entity_response_v2_has_api_version(
client: AsyncClient, v2_project_url, entity_repository
):
"""Test that EntityResponseV2 includes api_version field."""
# Create an entity
entity_data = {
"title": "TestApiVersion",
"folder": "test",
"content": "Test content",
}
response = await client.post(f"{v2_project_url}/knowledge/entities", json=entity_data)
assert response.status_code == 200
created_entity = EntityResponseV2.model_validate(response.json())
# V2 create must return id and api_version
assert created_entity.id is not None
assert created_entity.api_version == "v2"
entity_id = created_entity.id
# Get it via v2 endpoint
response = await client.get(f"{v2_project_url}/knowledge/entities/{entity_id}")
assert response.status_code == 200
entity_v2 = EntityResponseV2.model_validate(response.json())
assert entity_v2.api_version == "v2"
assert entity_v2.id == entity_id
tests/api/v2/test_memory_router.py
+301
View File
@@ -0,0 +1,301 @@
"""Tests for v2 memory router endpoints."""
import pytest
from httpx import AsyncClient
from pathlib import Path
from basic_memory.models import Project
async def create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
):
"""Helper to create an entity with file and index it."""
# Create file
test_content = f"# {entity_data['title']}\n\nTest content"
file_path = Path(test_project.path) / entity_data["file_path"]
file_path.parent.mkdir(parents=True, exist_ok=True)
await file_service.write_file(file_path, test_content)
# Create entity
entity = await entity_repository.create(entity_data)
# Index for search
await search_service.index_entity(entity)
return entity
@pytest.mark.asyncio
async def test_get_recent_context(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test getting recent activity context."""
entity_data = {
"title": "Recent Test Entity",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "recent_test.md",
"checksum": "abc123",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get recent context
response = await client.get(f"{v2_project_url}/memory/recent")
assert response.status_code == 200
data = response.json()
# Verify response structure (GraphContext uses 'results' not 'entities')
assert "results" in data
assert "metadata" in data
assert "page" in data
assert "page_size" in data
@pytest.mark.asyncio
async def test_get_recent_context_with_pagination(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test recent context with pagination parameters."""
# Create multiple test entities
for i in range(5):
entity_data = {
"title": f"Entity {i}",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": f"entity_{i}.md",
"checksum": f"checksum{i}",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get recent context with pagination
response = await client.get(
f"{v2_project_url}/memory/recent", params={"page": 1, "page_size": 3}
)
assert response.status_code == 200
data = response.json()
assert "results" in data
assert data["page"] == 1
assert data["page_size"] == 3
@pytest.mark.asyncio
async def test_get_recent_context_with_type_filter(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test filtering recent context by type."""
# Create a test entity
entity_data = {
"title": "Filtered Entity",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "filtered.md",
"checksum": "xyz789",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get recent context filtered by type
response = await client.get(f"{v2_project_url}/memory/recent", params={"type": ["entity"]})
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_get_recent_context_with_timeframe(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test recent context with custom timeframe."""
response = await client.get(f"{v2_project_url}/memory/recent", params={"timeframe": "1d"})
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_get_recent_context_invalid_project_id(
client: AsyncClient,
):
"""Test getting recent context with invalid project ID returns 404."""
response = await client.get("/v2/projects/999999/memory/recent")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_get_memory_context_by_permalink(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test getting context for a specific memory URI (permalink)."""
# Create a test entity
entity_data = {
"title": "Context Test",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "context_test.md",
"checksum": "def456",
"permalink": "context-test",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get context for this entity
response = await client.get(f"{v2_project_url}/memory/context-test")
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_get_memory_context_by_id(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test getting context using ID-based memory URI."""
# Create a test entity
entity_data = {
"title": "ID Context Test",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "id_context_test.md",
"checksum": "ghi789",
}
created_entity = await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get context using ID format (memory://id/123 or memory://123)
response = await client.get(f"{v2_project_url}/memory/id/{created_entity.id}")
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_get_memory_context_with_depth(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test getting context with depth parameter."""
# Create a test entity
entity_data = {
"title": "Depth Test",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "depth_test.md",
"checksum": "jkl012",
"permalink": "depth-test",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get context with depth
response = await client.get(f"{v2_project_url}/memory/depth-test", params={"depth": 2})
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_get_memory_context_not_found(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test getting context for non-existent memory URI returns 404."""
response = await client.get(f"{v2_project_url}/memory/nonexistent-uri")
# Note: This might return 200 with empty results depending on implementation
# Adjust assertion based on actual behavior
assert response.status_code in [200, 404]
@pytest.mark.asyncio
async def test_get_memory_context_with_timeframe(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test getting context with timeframe filter."""
# Create a test entity
entity_data = {
"title": "Timeframe Test",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "timeframe_test.md",
"checksum": "mno345",
"permalink": "timeframe-test",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Get context with timeframe
response = await client.get(
f"{v2_project_url}/memory/timeframe-test", params={"timeframe": "7d"}
)
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_v2_memory_endpoints_use_project_id_not_name(
client: AsyncClient,
test_project: Project,
):
"""Test that v2 memory endpoints reject string project names."""
# Try to use project name instead of ID - should fail
response = await client.get(f"/v2/{test_project.name}/memory/recent")
# FastAPI path validation should reject non-integer project_id
assert response.status_code in [404, 422]
tests/api/v2/test_project_router.py
+251
View File
@@ -0,0 +1,251 @@
"""Tests for V2 project management API routes (ID-based endpoints)."""
import tempfile
from pathlib import Path
import pytest
from httpx import AsyncClient
from basic_memory.models import Project
from basic_memory.schemas.project_info import ProjectItem, ProjectStatusResponse
@pytest.mark.asyncio
async def test_get_project_by_id(client: AsyncClient, test_project: Project, v2_projects_url):
"""Test getting a project by its numeric ID."""
response = await client.get(f"{v2_projects_url}/{test_project.id}")
assert response.status_code == 200
project = ProjectItem.model_validate(response.json())
assert project.id == test_project.id
assert project.name == test_project.name
assert project.path == test_project.path
assert project.is_default == (test_project.is_default or False)
@pytest.mark.asyncio
async def test_get_project_by_id_not_found(client: AsyncClient, v2_projects_url):
"""Test getting a non-existent project by ID returns 404."""
response = await client.get(f"{v2_projects_url}/999999")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_update_project_path_by_id(
client: AsyncClient, test_project: Project, v2_projects_url
):
"""Test updating a project's path by ID."""
with tempfile.TemporaryDirectory() as tmpdir:
new_path = str(Path(tmpdir) / "new-project-location")
Path(new_path).mkdir(parents=True, exist_ok=True)
update_data = {"path": new_path}
response = await client.patch(
f"{v2_projects_url}/{test_project.id}",
json=update_data,
)
assert response.status_code == 200
status_response = ProjectStatusResponse.model_validate(response.json())
assert status_response.status == "success"
assert status_response.new_project.id == test_project.id
# Normalize paths for cross-platform comparison (Windows uses backslashes, API returns forward slashes)
assert Path(status_response.new_project.path) == Path(new_path)
assert status_response.old_project.id == test_project.id
@pytest.mark.asyncio
async def test_update_project_invalid_path(
client: AsyncClient, test_project: Project, v2_projects_url
):
"""Test updating with a relative path returns 400."""
update_data = {"path": "relative/path"}
response = await client.patch(
f"{v2_projects_url}/{test_project.id}",
json=update_data,
)
assert response.status_code == 400
assert "absolute" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_update_project_not_found(client: AsyncClient, v2_projects_url):
"""Test updating a non-existent project returns 404."""
update_data = {"path": "/tmp/new-path"}
response = await client.patch(
f"{v2_projects_url}/999999",
json=update_data,
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_set_default_project_by_id(
client: AsyncClient, test_project: Project, v2_projects_url, project_repository, project_service
):
"""Test setting a project as default by ID."""
# Create a second project to test setting default
await project_service.add_project("second-project", "/tmp/second-project")
# Get the created project from the repository to get its ID
created_project = await project_repository.get_by_name("second-project")
assert created_project is not None
# Set the second project as default
response = await client.put(f"{v2_projects_url}/{created_project.id}/default")
assert response.status_code == 200
status_response = ProjectStatusResponse.model_validate(response.json())
assert status_response.status == "success"
assert status_response.default is True
assert status_response.new_project.id == created_project.id
assert status_response.new_project.is_default is True
assert status_response.old_project.id == test_project.id
assert status_response.old_project.is_default is False
@pytest.mark.asyncio
async def test_set_default_project_not_found(client: AsyncClient, v2_projects_url):
"""Test setting a non-existent project as default returns 404."""
response = await client.put(f"{v2_projects_url}/999999/default")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_delete_project_by_id(
client: AsyncClient, test_project: Project, v2_projects_url, project_repository, project_service
):
"""Test deleting a project by ID."""
# Create a second project since we can't delete the default
await project_service.add_project("to-delete", "/tmp/to-delete")
# Get the created project from the repository to get its ID
created_project = await project_repository.get_by_name("to-delete")
assert created_project is not None
# Delete it
response = await client.delete(f"{v2_projects_url}/{created_project.id}")
assert response.status_code == 200
status_response = ProjectStatusResponse.model_validate(response.json())
assert status_response.status == "success"
assert status_response.old_project.id == created_project.id
assert status_response.new_project is None
# Verify it's deleted - trying to get it should return 404
response = await client.get(f"{v2_projects_url}/{created_project.id}")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_delete_project_with_delete_notes_param(
client: AsyncClient, test_project: Project, v2_projects_url, project_repository, project_service
):
"""Test deleting a project with delete_notes parameter."""
# Create a project in a temp directory
with tempfile.TemporaryDirectory() as tmpdir:
project_path = Path(tmpdir) / "test-delete-notes"
project_path.mkdir(parents=True, exist_ok=True)
# Create a test file in the project
test_file = project_path / "test.md"
test_file.write_text("Test content")
await project_service.add_project("delete-with-notes", str(project_path))
# Get the created project from the repository to get its ID
created_project = await project_repository.get_by_name("delete-with-notes")
assert created_project is not None
# Delete with delete_notes=true
response = await client.delete(f"{v2_projects_url}/{created_project.id}?delete_notes=true")
assert response.status_code == 200
# Verify directory was deleted
assert not project_path.exists()
@pytest.mark.asyncio
async def test_delete_default_project_fails(
client: AsyncClient, test_project: Project, v2_projects_url
):
"""Test that deleting the default project returns 400."""
# test_project is the default project
response = await client.delete(f"{v2_projects_url}/{test_project.id}")
assert response.status_code == 400
assert "default project" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_delete_project_not_found(client: AsyncClient, v2_projects_url):
"""Test deleting a non-existent project returns 404."""
response = await client.delete(f"{v2_projects_url}/999999")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_v2_project_endpoints_use_id_not_name(
client: AsyncClient, test_project: Project, v2_projects_url
):
"""Verify v2 project endpoints require project ID, not name."""
# Try using project name instead of ID - should fail
response = await client.get(f"{v2_projects_url}/{test_project.name}")
# Should get 404 or 422 because name is not a valid integer
assert response.status_code in [404, 422]
@pytest.mark.asyncio
async def test_project_id_stability_after_rename(
client: AsyncClient, test_project: Project, v2_projects_url, project_repository
):
"""Test that project ID remains stable even after renaming."""
original_id = test_project.id
original_name = test_project.name
# Get project by ID
response = await client.get(f"{v2_projects_url}/{original_id}")
assert response.status_code == 200
project_before = ProjectItem.model_validate(response.json())
assert project_before.id == original_id
assert project_before.name == original_name
# Even if we renamed the project (not testing rename here, just the concept),
# the ID would stay the same. This test demonstrates the stability.
# Re-fetch by same ID
response = await client.get(f"{v2_projects_url}/{original_id}")
assert response.status_code == 200
project_after = ProjectItem.model_validate(response.json())
assert project_after.id == original_id
@pytest.mark.asyncio
async def test_update_project_active_status(
client: AsyncClient, test_project: Project, v2_projects_url, project_repository, project_service
):
"""Test updating a project's active status by ID."""
# Create a non-default project
await project_service.add_project("test-active", "/tmp/test-active")
# Get the created project from the repository to get its ID
created_project = await project_repository.get_by_name("test-active")
assert created_project is not None
# Update active status
update_data = {"is_active": False}
response = await client.patch(
f"{v2_projects_url}/{created_project.id}",
json=update_data,
)
assert response.status_code == 200
status_response = ProjectStatusResponse.model_validate(response.json())
assert status_response.status == "success"
tests/api/v2/test_prompt_router.py
+212
View File
@@ -0,0 +1,212 @@
"""Tests for V2 prompt router endpoints (ID-based)."""
import pytest
import pytest_asyncio
from httpx import AsyncClient
from basic_memory.models import Project
from basic_memory.services.context_service import ContextService
@pytest_asyncio.fixture
async def context_service(entity_repository, search_service, observation_repository):
"""Create a real context service for testing."""
return ContextService(entity_repository, search_service, observation_repository)
@pytest.mark.asyncio
async def test_continue_conversation_endpoint(
client: AsyncClient,
entity_service,
search_service,
context_service,
entity_repository,
test_graph,
v2_project_url: str,
):
"""Test the v2 continue_conversation endpoint with real services."""
# Create request data
request_data = {
"topic": "Root", # This should match our test entity in test_graph
"timeframe": "7d",
"depth": 1,
"related_items_limit": 2,
}
# Call the endpoint
response = await client.post(
f"{v2_project_url}/prompt/continue-conversation", json=request_data
)
# Verify response
assert response.status_code == 200
result = response.json()
assert "prompt" in result
assert "context" in result
# Check content of context
context = result["context"]
assert context["topic"] == "Root"
assert context["timeframe"] == "7d"
assert context["has_results"] is True
assert len(context["hierarchical_results"]) > 0
# Check content of prompt
prompt = result["prompt"]
assert "Continuing conversation on: Root" in prompt
assert "memory retrieval session" in prompt
@pytest.mark.asyncio
async def test_continue_conversation_without_topic(
client: AsyncClient,
entity_service,
search_service,
context_service,
entity_repository,
test_graph,
v2_project_url: str,
):
"""Test v2 continue_conversation without topic - should use recent activity."""
request_data = {"timeframe": "1d", "depth": 1, "related_items_limit": 2}
response = await client.post(
f"{v2_project_url}/prompt/continue-conversation", json=request_data
)
assert response.status_code == 200
result = response.json()
assert "Recent Activity" in result["context"]["topic"]
@pytest.mark.asyncio
async def test_search_prompt_endpoint(
client: AsyncClient, entity_service, search_service, test_graph, v2_project_url: str
):
"""Test the v2 search_prompt endpoint with real services."""
# Create request data
request_data = {
"query": "Root", # This should match our test entity
"timeframe": "7d",
}
# Call the endpoint
response = await client.post(f"{v2_project_url}/prompt/search", json=request_data)
# Verify response
assert response.status_code == 200
result = response.json()
assert "prompt" in result
assert "context" in result
# Check content of context
context = result["context"]
assert context["query"] == "Root"
assert context["timeframe"] == "7d"
assert context["has_results"] is True
assert len(context["results"]) > 0
# Check content of prompt
prompt = result["prompt"]
assert 'Search Results for: "Root"' in prompt
assert "This is a memory search session" in prompt
@pytest.mark.asyncio
async def test_search_prompt_no_results(
client: AsyncClient, entity_service, search_service, v2_project_url: str
):
"""Test the v2 search_prompt endpoint with a query that returns no results."""
# Create request data with a query that shouldn't match anything
request_data = {"query": "NonExistentQuery12345", "timeframe": "7d"}
# Call the endpoint
response = await client.post(f"{v2_project_url}/prompt/search", json=request_data)
# Verify response
assert response.status_code == 200
result = response.json()
# Check content of context
context = result["context"]
assert context["query"] == "NonExistentQuery12345"
assert context["has_results"] is False
assert len(context["results"]) == 0
# Check content of prompt
prompt = result["prompt"]
assert 'Search Results for: "NonExistentQuery12345"' in prompt
assert "I couldn't find any results for this query" in prompt
assert "Opportunity to Capture Knowledge" in prompt
@pytest.mark.asyncio
async def test_error_handling(client: AsyncClient, monkeypatch, v2_project_url: str):
"""Test error handling in v2 endpoints by breaking the template loader."""
# Patch the template loader to raise an exception
def mock_render(*args, **kwargs):
raise Exception("Template error")
# Apply the patch
monkeypatch.setattr("basic_memory.api.template_loader.TemplateLoader.render", mock_render)
# Test continue_conversation error handling
response = await client.post(
f"{v2_project_url}/prompt/continue-conversation",
json={"topic": "test error", "timeframe": "7d"},
)
assert response.status_code == 500
assert "detail" in response.json()
assert "Template error" in response.json()["detail"]
# Test search_prompt error handling
response = await client.post(
f"{v2_project_url}/prompt/search", json={"query": "test error", "timeframe": "7d"}
)
assert response.status_code == 500
assert "detail" in response.json()
assert "Template error" in response.json()["detail"]
@pytest.mark.asyncio
async def test_v2_prompt_endpoints_use_project_id_not_name(
client: AsyncClient, test_project: Project
):
"""Verify v2 prompt endpoints require project ID, not name."""
# Try using project name instead of ID - should fail
response = await client.post(
f"/v2/projects/{test_project.name}/prompt/continue-conversation",
json={"topic": "test", "timeframe": "7d"},
)
# Should get validation error or 404 because name is not a valid integer
assert response.status_code in [404, 422]
# Also test search endpoint
response = await client.post(
f"/v2/projects/{test_project.name}/prompt/search",
json={"query": "test", "timeframe": "7d"},
)
assert response.status_code in [404, 422]
@pytest.mark.asyncio
async def test_prompt_invalid_project_id(client: AsyncClient):
"""Test prompt endpoints with invalid project ID return 404."""
# Test continue-conversation
response = await client.post(
"/v2/projects/999999/prompt/continue-conversation",
json={"topic": "test", "timeframe": "7d"},
)
assert response.status_code == 404
# Test search
response = await client.post(
"/v2/projects/999999/prompt/search",
json={"query": "test", "timeframe": "7d"},
)
assert response.status_code == 404
tests/api/v2/test_resource_router.py
+267
View File
@@ -0,0 +1,267 @@
"""Tests for V2 resource API routes (ID-based endpoints)."""
import pytest
from httpx import AsyncClient
from basic_memory.models import Project
from basic_memory.schemas.v2.resource import ResourceResponse
@pytest.mark.asyncio
async def test_create_resource(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test creating a new resource via v2 POST endpoint."""
create_data = {
"file_path": "test-resources/test-file.md",
"content": "# Test Resource\n\nThis is test content.",
}
response = await client.post(
f"{v2_project_url}/resource",
json=create_data,
)
assert response.status_code == 200
result = ResourceResponse.model_validate(response.json())
# V2 must return entity_id
assert result.entity_id is not None
assert isinstance(result.entity_id, int)
assert result.file_path == "test-resources/test-file.md"
assert result.checksum is not None
@pytest.mark.asyncio
async def test_create_resource_duplicate_fails(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test that creating a resource at an existing path returns 409."""
create_data = {
"file_path": "duplicate-test.md",
"content": "First version",
}
# Create first time - should succeed
response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert response.status_code == 200
# Try to create again - should fail with 409
response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert response.status_code == 409
assert "already exists" in response.json()["detail"]
@pytest.mark.asyncio
async def test_get_resource_by_id(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test getting resource content by entity ID."""
# First create a resource
test_content = "# Test Resource\n\nThis is test content."
create_data = {
"file_path": "test-get.md",
"content": test_content,
}
create_response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert create_response.status_code == 200
created = ResourceResponse.model_validate(create_response.json())
# Now get it by entity ID
response = await client.get(f"{v2_project_url}/resource/{created.entity_id}")
assert response.status_code == 200
# Normalize line endings for cross-platform compatibility
assert test_content.replace("\n", "") in response.text.replace("\r\n", "").replace("\n", "")
@pytest.mark.asyncio
async def test_get_resource_not_found(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test getting a non-existent resource returns 404."""
response = await client.get(f"{v2_project_url}/resource/999999")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_update_resource(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test updating resource content by entity ID."""
# Create a resource
create_data = {
"file_path": "test-update.md",
"content": "Original content",
}
create_response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert create_response.status_code == 200
created = ResourceResponse.model_validate(create_response.json())
# Update it
update_data = {
"content": "Updated content",
}
response = await client.put(
f"{v2_project_url}/resource/{created.entity_id}",
json=update_data,
)
assert response.status_code == 200
result = ResourceResponse.model_validate(response.json())
assert result.entity_id == created.entity_id
assert result.file_path == "test-update.md"
# Verify content was updated
get_response = await client.get(f"{v2_project_url}/resource/{created.entity_id}")
assert "Updated content" in get_response.text
@pytest.mark.asyncio
async def test_update_resource_and_move(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test updating resource content and moving it to a new path."""
# Create a resource
create_data = {
"file_path": "original-location.md",
"content": "Original content",
}
create_response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert create_response.status_code == 200
created = ResourceResponse.model_validate(create_response.json())
# Update content and move file
update_data = {
"content": "Updated content in new location",
"file_path": "moved/new-location.md",
}
response = await client.put(
f"{v2_project_url}/resource/{created.entity_id}",
json=update_data,
)
assert response.status_code == 200
result = ResourceResponse.model_validate(response.json())
assert result.entity_id == created.entity_id
assert result.file_path == "moved/new-location.md"
# Verify content at new location
get_response = await client.get(f"{v2_project_url}/resource/{created.entity_id}")
assert "Updated content in new location" in get_response.text
@pytest.mark.asyncio
async def test_update_resource_not_found(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test updating a non-existent resource returns 404."""
update_data = {
"content": "New content",
}
response = await client.put(
f"{v2_project_url}/resource/999999",
json=update_data,
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_create_resource_invalid_path(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test creating a resource with path traversal attempt fails."""
create_data = {
"file_path": "../../../etc/passwd",
"content": "malicious content",
}
response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert response.status_code == 400
assert "Invalid file path" in response.json()["detail"]
@pytest.mark.asyncio
async def test_update_resource_invalid_path(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test updating a resource with path traversal attempt fails."""
# Create a valid resource first
create_data = {
"file_path": "valid.md",
"content": "Valid content",
}
create_response = await client.post(f"{v2_project_url}/resource", json=create_data)
assert create_response.status_code == 200
created = ResourceResponse.model_validate(create_response.json())
# Try to move it to an invalid path
update_data = {
"content": "Updated content",
"file_path": "../../../etc/passwd",
}
response = await client.put(
f"{v2_project_url}/resource/{created.entity_id}",
json=update_data,
)
assert response.status_code == 400
assert "Invalid file path" in response.json()["detail"]
@pytest.mark.asyncio
async def test_resource_invalid_project_id(
client: AsyncClient,
):
"""Test resource endpoints with invalid project ID return 404."""
# Test create
response = await client.post(
"/v2/projects/999999/resource",
json={"file_path": "test.md", "content": "test"},
)
assert response.status_code == 404
# Test get
response = await client.get("/v2/projects/999999/resource/1")
assert response.status_code == 404
# Test update
response = await client.put(
"/v2/projects/999999/resource/1",
json={"content": "test"},
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_v2_resource_endpoints_use_project_id_not_name(
client: AsyncClient, test_project: Project
):
"""Verify v2 resource endpoints require project ID, not name."""
# Try using project name instead of ID - should fail
response = await client.get(f"/v2/projects/{test_project.name}/resource/1")
# Should get validation error or 404 because name is not a valid integer
assert response.status_code in [404, 422]
tests/api/v2/test_search_router.py
+289
View File
@@ -0,0 +1,289 @@
"""Tests for v2 search router endpoints."""
import pytest
from httpx import AsyncClient
from pathlib import Path
from basic_memory.models import Project
async def create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
):
"""Helper to create an entity with file and index it."""
# Create file
test_content = f"# {entity_data['title']}\n\nTest content"
file_path = Path(test_project.path) / entity_data["file_path"]
file_path.parent.mkdir(parents=True, exist_ok=True)
await file_service.write_file(file_path, test_content)
# Create entity
entity = await entity_repository.create(entity_data)
# Index for search
await search_service.index_entity(entity)
return entity
@pytest.mark.asyncio
async def test_search_entities(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test searching for entities."""
# Create a test entity
entity_data = {
"title": "Searchable Entity",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "searchable.md",
"checksum": "search123",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Search for the entity
response = await client.post(f"{v2_project_url}/search/", json={"search_text": "Searchable"})
assert response.status_code == 200
data = response.json()
# Verify response structure
assert "results" in data
assert "current_page" in data
assert "page_size" in data
@pytest.mark.asyncio
async def test_search_with_pagination(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test search with pagination parameters."""
# Create multiple test entities
for i in range(5):
entity_data = {
"title": f"Search Entity {i}",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": f"search_{i}.md",
"checksum": f"searchsum{i}",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Search with pagination
response = await client.post(
f"{v2_project_url}/search/",
json={"search_text": "Search Entity"},
params={"page": 1, "page_size": 3},
)
assert response.status_code == 200
data = response.json()
assert data["current_page"] == 1
assert data["page_size"] == 3
@pytest.mark.asyncio
async def test_search_by_permalink(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test searching by permalink."""
# Create a test entity with permalink
entity_data = {
"title": "Permalink Search",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "permalink_search.md",
"checksum": "perm123",
"permalink": "permalink-search",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Search by permalink
response = await client.post(
f"{v2_project_url}/search/", json={"permalink": "permalink-search"}
)
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_search_by_title(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test searching by title."""
# Create a test entity
entity_data = {
"title": "Unique Title For Search",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "unique_title.md",
"checksum": "title123",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Search by title
response = await client.post(f"{v2_project_url}/search/", json={"title": "Unique Title"})
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_search_with_type_filter(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test searching with entity type filter."""
# Create test entities of different types
for entity_type in ["note", "document"]:
entity_data = {
"title": f"Type {entity_type}",
"entity_type": entity_type,
"content_type": "text/markdown",
"file_path": f"type_{entity_type}.md",
"checksum": f"type{entity_type}",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Search with type filter
response = await client.post(
f"{v2_project_url}/search/", json={"search_text": "Type", "types": ["note"]}
)
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_search_with_date_filter(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
entity_repository,
search_service,
file_service,
):
"""Test searching with date filter."""
# Create a test entity
entity_data = {
"title": "Date Filtered",
"entity_type": "note",
"content_type": "text/markdown",
"file_path": "date_filtered.md",
"checksum": "date123",
}
await create_test_entity(
test_project, entity_data, entity_repository, search_service, file_service
)
# Search with date filter
response = await client.post(
f"{v2_project_url}/search/",
json={"search_text": "Date Filtered", "after_date": "2024-01-01T00:00:00Z"},
)
assert response.status_code == 200
data = response.json()
assert "results" in data
@pytest.mark.asyncio
async def test_search_empty_query(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test search with empty query."""
response = await client.post(f"{v2_project_url}/search/", json={})
# Empty query should still be valid (returns all)
assert response.status_code in [200, 422]
@pytest.mark.asyncio
async def test_search_invalid_project_id(
client: AsyncClient,
):
"""Test searching with invalid project ID returns 404."""
response = await client.post("/v2/projects/999999/search/", json={"search_text": "test"})
assert response.status_code == 404
@pytest.mark.asyncio
async def test_reindex(
client: AsyncClient,
test_project: Project,
v2_project_url: str,
):
"""Test reindexing search index."""
response = await client.post(f"{v2_project_url}/search/reindex")
assert response.status_code == 200
data = response.json()
# Verify response structure
assert "status" in data
assert data["status"] == "ok"
assert "message" in data
@pytest.mark.asyncio
async def test_reindex_invalid_project_id(
client: AsyncClient,
):
"""Test reindexing with invalid project ID returns 404."""
response = await client.post("/v2/projects/999999/search/reindex")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_v2_search_endpoints_use_project_id_not_name(
client: AsyncClient,
test_project: Project,
):
"""Test that v2 search endpoints reject string project names."""
# Try to use project name instead of ID - should fail
response = await client.post(f"/v2/{test_project.name}/search/", json={"search_text": "test"})
# FastAPI path validation should reject non-integer project_id
assert response.status_code in [404, 422]
tests/cli/test_project_add_with_local_path.py
+1
View File
@@ -55,6 +55,7 @@ def mock_api_client():
"default": False,
"old_project": None,
"new_project": {
"id": 1,
"name": "test-project",
"path": "/test-project",
"is_default": False,
tests/mcp/test_prompts.py
+1
View File
@@ -116,6 +116,7 @@ def test_prompt_context_with_file_path_no_permalink():
# Create a mock context with a file that has no permalink (like a binary file)
test_entity = EntitySummary(
entity_id=1,
type="entity",
title="Test File",
permalink=None, # No permalink
tests/schemas/test_memory_serialization.py
+20 -1
View File
@@ -22,6 +22,7 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 10, 30, 0)
entity = EntitySummary(
entity_id=1,
permalink="test/entity",
title="Test Entity",
file_path="test/entity.md",
@@ -41,6 +42,8 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 15, 45, 30)
relation = RelationSummary(
relation_id=1,
entity_id=1,
title="Test Relation",
file_path="test/relation.md",
permalink="test/relation",
@@ -63,6 +66,8 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 20, 15, 45)
observation = ObservationSummary(
observation_id=1,
entity_id=1,
title="Test Observation",
file_path="test/observation.md",
permalink="test/observation",
@@ -100,6 +105,7 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 9, 30, 15)
entity = EntitySummary(
entity_id=1,
permalink="test/entity",
title="Test Entity",
file_path="test/entity.md",
@@ -107,6 +113,8 @@ class TestDateTimeSerialization:
)
observation = ObservationSummary(
observation_id=1,
entity_id=1,
title="Test Observation",
file_path="test/observation.md",
permalink="test/observation",
@@ -131,6 +139,7 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 14, 20, 10)
entity = EntitySummary(
entity_id=1,
permalink="test/entity",
title="Test Entity",
file_path="test/entity.md",
@@ -159,6 +168,7 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 10, 30, 0, 123456)
entity = EntitySummary(
entity_id=1,
permalink="test/entity",
title="Test Entity",
file_path="test/entity.md",
@@ -176,6 +186,7 @@ class TestDateTimeSerialization:
test_datetime = datetime(2023, 12, 8, 10, 30, 0)
entity = EntitySummary(
entity_id=1,
permalink="test/entity",
title="Test Entity",
file_path="test/entity.md",
@@ -212,10 +223,16 @@ class TestDateTimeSerialization:
if model_class == EntitySummary:
instance = model_class(
permalink="test", title="Test", file_path="test.md", created_at=test_datetime
entity_id=1,
permalink="test",
title="Test",
file_path="test.md",
created_at=test_datetime,
)
elif model_class == RelationSummary:
instance = model_class(
relation_id=1,
entity_id=1,
title="Test",
file_path="test.md",
permalink="test",
@@ -224,6 +241,8 @@ class TestDateTimeSerialization:
)
elif model_class == ObservationSummary:
instance = model_class(
observation_id=1,
entity_id=1,
title="Test",
file_path="test.md",
permalink="test",