chore: Cloud compatibility fixes and performance improvements (#454)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Paul Hernandez
2025-12-15 20:07:55 -06:00
committed by GitHub
parent 126c0495c0
commit 78673d8e51
18 changed files with 568 additions and 247 deletions
+33
View File
@@ -433,6 +433,39 @@ See the [Documentation](https://memory.basicmachines.co/) for more info, includi
- [Managing multiple Projects](https://docs.basicmemory.com/guides/cli-reference/#project)
- [Importing data from OpenAI/Claude Projects](https://docs.basicmemory.com/guides/cli-reference/#import)
## Logging
Basic Memory uses [Loguru](https://github.com/Delgan/loguru) for logging. The logging behavior varies by entry point:
| Entry Point | Default Behavior | Use Case |
|-------------|------------------|----------|
| CLI commands | File only | Prevents log output from interfering with command output |
| MCP server | File only | Stdout would corrupt the JSON-RPC protocol |
| API server | File (local) or stdout (cloud) | Docker/cloud deployments use stdout |
**Log file location:** `~/.basic-memory/basic-memory.log` (10MB rotation, 10 days retention)
### Environment Variables
| Variable | Default | Description |
|----------|---------|-------------|
| `BASIC_MEMORY_LOG_LEVEL` | `INFO` | Log level: DEBUG, INFO, WARNING, ERROR |
| `BASIC_MEMORY_CLOUD_MODE` | `false` | When `true`, API logs to stdout with structured context |
| `BASIC_MEMORY_ENV` | `dev` | Set to `test` for test mode (stderr only) |
### Examples
```bash
# Enable debug logging
BASIC_MEMORY_LOG_LEVEL=DEBUG basic-memory sync
# View logs
tail -f ~/.basic-memory/basic-memory.log
# Cloud/Docker mode (stdout logging with structured context)
BASIC_MEMORY_CLOUD_MODE=true uvicorn basic_memory.api.app:app
```
## Development
### Running Tests
+4 -1
View File
@@ -30,7 +30,7 @@ from basic_memory.api.v2.routers import (
prompt_router as v2_prompt,
importer_router as v2_importer,
)
from basic_memory.config import ConfigManager
from basic_memory.config import ConfigManager, init_api_logging
from basic_memory.services.initialization import initialize_file_sync, initialize_app
@@ -38,6 +38,9 @@ from basic_memory.services.initialization import initialize_file_sync, initializ
async def lifespan(app: FastAPI): # pragma: no cover
"""Lifecycle manager for the FastAPI app. Not called in stdio mcp mode"""
# Initialize logging for API (stdout in cloud mode, file otherwise)
init_api_logging()
app_config = ConfigManager().config
logger.info("Starting Basic Memory API")
+18 -15
View File
@@ -2,9 +2,9 @@
import tempfile
from pathlib import Path
from typing import Annotated
from typing import Annotated, Union
from fastapi import APIRouter, HTTPException, BackgroundTasks, Body
from fastapi import APIRouter, HTTPException, BackgroundTasks, Body, Response
from fastapi.responses import FileResponse, JSONResponse
from loguru import logger
@@ -50,7 +50,7 @@ def get_entity_ids(item: SearchIndexRow) -> set[int]:
raise ValueError(f"Unexpected type: {item.type}")
@router.get("/{identifier:path}")
@router.get("/{identifier:path}", response_model=None)
async def get_resource_content(
config: ProjectConfigDep,
link_resolver: LinkResolverDep,
@@ -61,7 +61,7 @@ async def get_resource_content(
identifier: str,
page: int = 1,
page_size: int = 10,
) -> FileResponse:
) -> Union[Response, FileResponse]:
"""Get resource content by identifier: name or permalink."""
logger.debug(f"Getting content for: {identifier}")
@@ -92,13 +92,16 @@ async def get_resource_content(
# return single response
if len(results) == 1:
entity = results[0]
file_path = Path(f"{config.home}/{entity.file_path}")
if not file_path.exists():
# Check file exists via file_service (for cloud compatibility)
if not await file_service.exists(entity.file_path):
raise HTTPException(
status_code=404,
detail=f"File not found: {file_path}",
detail=f"File not found: {entity.file_path}",
)
return FileResponse(path=file_path)
# Read content via file_service as bytes (works with both local and S3)
content = await file_service.read_file_bytes(entity.file_path)
content_type = file_service.content_type(entity.file_path)
return Response(content=content, media_type=content_type)
# for multiple files, initialize a temporary file for writing the results
with tempfile.NamedTemporaryFile(delete=False, mode="w", suffix=".md") as tmp_file:
@@ -192,7 +195,7 @@ async def write_resource(
checksum = await file_service.write_file(full_path, content_str)
# Get file info
file_stats = file_service.file_stats(full_path)
file_metadata = await file_service.get_file_metadata(full_path)
# Determine file details
file_name = Path(file_path).name
@@ -213,7 +216,7 @@ async def write_resource(
"content_type": content_type,
"file_path": file_path,
"checksum": checksum,
"updated_at": datetime.fromtimestamp(file_stats.st_mtime).astimezone(),
"updated_at": file_metadata.modified_at,
},
)
status_code = 200
@@ -225,8 +228,8 @@ async def write_resource(
content_type=content_type,
file_path=file_path,
checksum=checksum,
created_at=datetime.fromtimestamp(file_stats.st_ctime).astimezone(),
updated_at=datetime.fromtimestamp(file_stats.st_mtime).astimezone(),
created_at=file_metadata.created_at,
updated_at=file_metadata.modified_at,
)
entity = await entity_repository.add(entity)
status_code = 201
@@ -240,9 +243,9 @@ async def write_resource(
content={
"file_path": file_path,
"checksum": checksum,
"size": file_stats.st_size,
"created_at": file_stats.st_ctime,
"modified_at": file_stats.st_mtime,
"size": file_metadata.size,
"created_at": file_metadata.created_at.timestamp(),
"modified_at": file_metadata.modified_at.timestamp(),
},
)
except Exception as e: # pragma: no cover
+26 -14
View File
@@ -24,8 +24,24 @@ async def to_graph_context(
page: Optional[int] = None,
page_size: Optional[int] = None,
):
# First pass: collect all entity IDs needed for relations
entity_ids_needed: set[int] = set()
for context_item in context_result.results:
for item in [context_item.primary_result] + context_item.observations + context_item.related_results:
if item.type == SearchItemType.RELATION:
if item.from_id: # pyright: ignore
entity_ids_needed.add(item.from_id) # pyright: ignore
if item.to_id:
entity_ids_needed.add(item.to_id)
# Batch fetch all entities at once
entity_lookup: dict[int, str] = {}
if entity_ids_needed:
entities = await entity_repository.find_by_ids(list(entity_ids_needed))
entity_lookup = {e.id: e.title for e in entities}
# Helper function to convert items to summaries
async def to_summary(item: SearchIndexRow | ContextResultRow):
def to_summary(item: SearchIndexRow | ContextResultRow):
match item.type:
case SearchItemType.ENTITY:
return EntitySummary(
@@ -48,8 +64,8 @@ async def to_graph_context(
created_at=item.created_at,
)
case SearchItemType.RELATION:
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
from_title = entity_lookup.get(item.from_id) if item.from_id else None # pyright: ignore
to_title = entity_lookup.get(item.to_id) if item.to_id else None
return RelationSummary(
relation_id=item.id,
entity_id=item.entity_id, # pyright: ignore
@@ -57,9 +73,9 @@ async def to_graph_context(
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=from_title,
from_entity_id=item.from_id, # pyright: ignore
to_entity=to_entity.title if to_entity else None,
to_entity=to_title,
to_entity_id=item.to_id,
created_at=item.created_at,
)
@@ -70,23 +86,19 @@ async def to_graph_context(
hierarchical_results = []
for context_item in context_result.results:
# Process primary result
primary_result = await to_summary(context_item.primary_result)
primary_result = to_summary(context_item.primary_result)
# Process observations
observations = []
for obs in context_item.observations:
observations.append(await to_summary(obs))
# Process observations (always ObservationSummary, validated by context_service)
observations = [to_summary(obs) for obs in context_item.observations]
# Process related results
related = []
for rel in context_item.related_results:
related.append(await to_summary(rel))
related = [to_summary(rel) for rel in context_item.related_results]
# Add to hierarchical results
hierarchical_results.append(
ContextResult(
primary_result=primary_result,
observations=observations,
observations=observations, # pyright: ignore[reportArgumentType]
related_results=related,
)
)
@@ -11,8 +11,7 @@ Key differences from v1:
from pathlib import Path
from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
from fastapi import APIRouter, HTTPException, Response
from loguru import logger
from basic_memory.deps import (
@@ -30,7 +29,6 @@ from basic_memory.schemas.v2.resource import (
ResourceResponse,
)
from basic_memory.utils import validate_project_path
from datetime import datetime
router = APIRouter(prefix="/resource", tags=["resources-v2"])
@@ -42,7 +40,7 @@ async def get_resource_content(
config: ProjectConfigV2Dep,
entity_service: EntityServiceV2Dep,
file_service: FileServiceV2Dep,
) -> FileResponse:
) -> Response:
"""Get raw resource content by entity ID.
Args:
@@ -53,7 +51,7 @@ async def get_resource_content(
file_service: File service for reading file content
Returns:
FileResponse with entity content
Response with entity content
Raises:
HTTPException: 404 if entity or file not found
@@ -76,14 +74,18 @@ async def get_resource_content(
detail="Entity contains invalid file path",
)
file_path = Path(f"{config.home}/{entity.file_path}")
if not file_path.exists():
# Check file exists via file_service (for cloud compatibility)
if not await file_service.exists(entity.file_path):
raise HTTPException(
status_code=404,
detail=f"File not found: {file_path}",
detail=f"File not found: {entity.file_path}",
)
return FileResponse(path=file_path)
# Read content via file_service as bytes (works with both local and S3)
content = await file_service.read_file_bytes(entity.file_path)
content_type = file_service.content_type(entity.file_path)
return Response(content=content, media_type=content_type)
@router.post("", response_model=ResourceResponse)
@@ -143,7 +145,7 @@ async def create_resource(
checksum = await file_service.write_file(full_path, data.content)
# Get file info
file_stats = file_service.file_stats(full_path)
file_metadata = await file_service.get_file_metadata(full_path)
# Determine file details
file_name = Path(data.file_path).name
@@ -157,8 +159,8 @@ async def create_resource(
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(),
created_at=file_metadata.created_at,
updated_at=file_metadata.modified_at,
)
entity = await entity_repository.add(entity)
@@ -170,9 +172,9 @@ async def create_resource(
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,
size=file_metadata.size,
created_at=file_metadata.created_at.timestamp(),
modified_at=file_metadata.modified_at.timestamp(),
)
except HTTPException:
# Re-raise HTTP exceptions without wrapping
@@ -233,7 +235,6 @@ async def update_resource(
)
# 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
@@ -241,9 +242,9 @@ async def update_resource(
# 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()
# If old file exists, remove it via file_service (for cloud compatibility)
if await file_service.exists(entity.file_path):
await file_service.delete_file(entity.file_path)
else:
# Ensure directory exists for in-place update
new_full_path.parent.mkdir(parents=True, exist_ok=True)
@@ -252,7 +253,7 @@ async def update_resource(
checksum = await file_service.write_file(new_full_path, data.content)
# Get file info
file_stats = file_service.file_stats(new_full_path)
file_metadata = await file_service.get_file_metadata(new_full_path)
# Determine file details
file_name = Path(target_file_path).name
@@ -268,7 +269,7 @@ async def update_resource(
"content_type": content_type,
"file_path": target_file_path,
"checksum": checksum,
"updated_at": datetime.fromtimestamp(file_stats.st_mtime).astimezone(),
"updated_at": file_metadata.modified_at,
},
)
@@ -280,9 +281,9 @@ async def update_resource(
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,
size=file_metadata.size,
created_at=file_metadata.created_at.timestamp(),
modified_at=file_metadata.modified_at.timestamp(),
)
except HTTPException:
# Re-raise HTTP exceptions without wrapping
+4 -1
View File
@@ -2,7 +2,7 @@ from typing import Optional
import typer
from basic_memory.config import ConfigManager
from basic_memory.config import ConfigManager, init_cli_logging
def version_callback(value: bool) -> None:
@@ -31,6 +31,9 @@ def app_callback(
) -> None:
"""Basic Memory - Local-first personal knowledge management."""
# Initialize logging for CLI (file only, no stdout)
init_cli_logging()
# Run initialization for every command unless --version was specified
if not version and ctx.invoked_subcommand is not None:
from basic_memory.services.initialization import ensure_initialization
+3 -1
View File
@@ -6,7 +6,7 @@ import typer
from typing import Optional
from basic_memory.cli.app import app
from basic_memory.config import ConfigManager
from basic_memory.config import ConfigManager, init_mcp_logging
# Import mcp instance
from basic_memory.mcp.server import mcp as mcp_server # pragma: no cover
@@ -44,6 +44,8 @@ if not config.cloud_mode_enabled:
- streamable-http: Recommended for web deployments (default)
- sse: Server-Sent Events (for compatibility with existing clients)
"""
# Initialize logging for MCP (file only, stdout breaks protocol)
init_mcp_logging()
# Validate and set project constraint if specified
if project:
+79 -70
View File
@@ -9,10 +9,9 @@ from typing import Any, Dict, Literal, Optional, List, Tuple
from enum import Enum
from loguru import logger
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, Field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
import basic_memory
from basic_memory.utils import setup_logging, generate_permalink
@@ -214,6 +213,36 @@ class BasicMemoryConfig(BaseSettings):
# Fall back to config file value
return self.cloud_mode
@classmethod
def for_cloud_tenant(
cls,
database_url: str,
projects: Optional[Dict[str, str]] = None,
) -> "BasicMemoryConfig":
"""Create config for cloud tenant - no config.json, database is source of truth.
This factory method creates a BasicMemoryConfig suitable for cloud deployments
where:
- Database is Postgres (Neon), not SQLite
- Projects are discovered from the database, not config file
- Path validation is skipped (no local filesystem in cloud)
- Initialization sync is skipped (stateless deployment)
Args:
database_url: Postgres connection URL for tenant database
projects: Optional project mapping (usually empty, discovered from DB)
Returns:
BasicMemoryConfig configured for cloud mode
"""
return cls(
database_backend=DatabaseBackend.POSTGRES,
database_url=database_url,
projects=projects or {},
cloud_mode=True,
skip_initialization_sync=True,
)
model_config = SettingsConfigDict(
env_prefix="BASIC_MEMORY_",
extra="ignore",
@@ -230,6 +259,10 @@ class BasicMemoryConfig(BaseSettings):
def model_post_init(self, __context: Any) -> None:
"""Ensure configuration is valid after initialization."""
# Skip project initialization in cloud mode - projects are discovered from DB
if self.database_backend == DatabaseBackend.POSTGRES:
return
# Ensure at least one project exists; if none exist then create main
if not self.projects: # pragma: no cover
self.projects["main"] = str(
@@ -272,19 +305,26 @@ class BasicMemoryConfig(BaseSettings):
"""Get all configured projects as ProjectConfig objects."""
return [ProjectConfig(name=name, home=Path(path)) for name, path in self.projects.items()]
@field_validator("projects")
@classmethod
def ensure_project_paths_exists(cls, v: Dict[str, str]) -> Dict[str, str]: # pragma: no cover
"""Ensure project path exists."""
for name, path_value in v.items():
@model_validator(mode="after")
def ensure_project_paths_exists(self) -> "BasicMemoryConfig": # pragma: no cover
"""Ensure project paths exist.
Skips path creation when using Postgres backend (cloud mode) since
cloud tenants don't use local filesystem paths.
"""
# Skip path creation for cloud mode - no local filesystem
if self.database_backend == DatabaseBackend.POSTGRES:
return self
for name, path_value in self.projects.items():
path = Path(path_value)
if not Path(path).exists():
if not path.exists():
try:
path.mkdir(parents=True)
except Exception as e:
logger.error(f"Failed to create project path: {e}")
raise e
return v
return self
@property
def data_dir_path(self):
@@ -487,69 +527,38 @@ def save_basic_memory_config(file_path: Path, config: BasicMemoryConfig) -> None
logger.error(f"Failed to save config: {e}")
# setup logging to a single log file in user home directory
user_home = Path.home()
log_dir = user_home / DATA_DIR_NAME
log_dir.mkdir(parents=True, exist_ok=True)
# Logging initialization functions for different entry points
# Process info for logging
def get_process_name(): # pragma: no cover
def init_cli_logging() -> None: # pragma: no cover
"""Initialize logging for CLI commands - file only.
CLI commands should not log to stdout to avoid interfering with
command output and shell integration.
"""
get the type of process for logging
"""
import sys
log_level = os.getenv("BASIC_MEMORY_LOG_LEVEL", "INFO")
setup_logging(log_level=log_level, log_to_file=True)
if "sync" in sys.argv:
return "sync"
elif "mcp" in sys.argv:
return "mcp"
elif "cli" in sys.argv:
return "cli"
def init_mcp_logging() -> None: # pragma: no cover
"""Initialize logging for MCP server - file only.
MCP server must not log to stdout as it would corrupt the
JSON-RPC protocol communication.
"""
log_level = os.getenv("BASIC_MEMORY_LOG_LEVEL", "INFO")
setup_logging(log_level=log_level, log_to_file=True)
def init_api_logging() -> None: # pragma: no cover
"""Initialize logging for API server.
Cloud mode (BASIC_MEMORY_CLOUD_MODE=1): stdout with structured context
Local mode: file only
"""
log_level = os.getenv("BASIC_MEMORY_LOG_LEVEL", "INFO")
cloud_mode = os.getenv("BASIC_MEMORY_CLOUD_MODE", "").lower() in ("1", "true")
if cloud_mode:
setup_logging(log_level=log_level, log_to_stdout=True, structured_context=True)
else:
return "api"
process_name = get_process_name()
# Global flag to track if logging has been set up
_LOGGING_SETUP = False
# Logging
def setup_basic_memory_logging(): # pragma: no cover
"""Set up logging for basic-memory, ensuring it only happens once."""
global _LOGGING_SETUP
if _LOGGING_SETUP:
# We can't log before logging is set up
# print("Skipping duplicate logging setup")
return
# Check for console logging environment variable - accept more truthy values
console_logging_env = os.getenv("BASIC_MEMORY_CONSOLE_LOGGING", "false").lower()
console_logging = console_logging_env in ("true", "1", "yes", "on")
# Check for log level environment variable first, fall back to config
log_level = os.getenv("BASIC_MEMORY_LOG_LEVEL")
if not log_level:
config_manager = ConfigManager()
log_level = config_manager.config.log_level
config_manager = ConfigManager()
config = get_project_config()
setup_logging(
env=config_manager.config.env,
home_dir=user_home, # Use user home for logs
log_level=log_level,
log_file=f"{DATA_DIR_NAME}/basic-memory-{process_name}.log",
console=console_logging,
)
logger.info(f"Basic Memory {basic_memory.__version__} (Project: {config.project})")
_LOGGING_SETUP = True
# Set up logging
setup_basic_memory_logging()
setup_logging(log_level=log_level, log_to_file=True)
+16
View File
@@ -1,6 +1,8 @@
"""Utilities for file operations."""
import hashlib
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import re
from typing import Any, Dict, Union
@@ -13,6 +15,20 @@ from loguru import logger
from basic_memory.utils import FilePath
@dataclass
class FileMetadata:
"""File metadata for cloud-compatible file operations.
This dataclass provides a cloud-agnostic way to represent file metadata,
enabling S3FileService to return metadata from head_object responses
instead of mock stat_result with zeros.
"""
size: int
created_at: datetime
modified_at: datetime
class FileError(Exception):
"""Base exception for file operations."""
+21 -8
View File
@@ -233,8 +233,11 @@ class EntityService(BaseService[EntityModel]):
final_content = dump_frontmatter(post)
checksum = await self.file_service.write_file(file_path, final_content)
# parse entity from file
entity_markdown = await self.entity_parser.parse_file(file_path)
# parse entity from content we just wrote (avoids re-reading file for cloud compatibility)
entity_markdown = await self.entity_parser.parse_markdown_content(
file_path=file_path,
content=final_content,
)
# create entity
created = await self.create_entity_from_markdown(file_path, entity_markdown)
@@ -254,8 +257,12 @@ class EntityService(BaseService[EntityModel]):
# Convert file path string to Path
file_path = Path(entity.file_path)
# Read existing frontmatter from the file if it exists
existing_markdown = await self.entity_parser.parse_file(file_path)
# Read existing content via file_service (for cloud compatibility)
existing_content = await self.file_service.read_file_content(file_path)
existing_markdown = await self.entity_parser.parse_markdown_content(
file_path=file_path,
content=existing_content,
)
# Parse content frontmatter to check for user-specified permalink and entity_type
content_markdown = None
@@ -311,8 +318,11 @@ class EntityService(BaseService[EntityModel]):
final_content = dump_frontmatter(merged_post)
checksum = await self.file_service.write_file(file_path, final_content)
# parse entity from file
entity_markdown = await self.entity_parser.parse_file(file_path)
# parse entity from content we just wrote (avoids re-reading file for cloud compatibility)
entity_markdown = await self.entity_parser.parse_markdown_content(
file_path=file_path,
content=final_content,
)
# update entity in db
entity = await self.update_entity_and_observations(file_path, entity_markdown)
@@ -559,8 +569,11 @@ class EntityService(BaseService[EntityModel]):
# Write the updated content back to the file
checksum = await self.file_service.write_file(file_path, new_content)
# Parse the updated file to get new observations/relations
entity_markdown = await self.entity_parser.parse_file(file_path)
# Parse the content we just wrote (avoids re-reading file for cloud compatibility)
entity_markdown = await self.entity_parser.parse_markdown_content(
file_path=file_path,
content=new_content,
)
# Update entity and its relationships
entity = await self.update_entity_and_observations(file_path, entity_markdown)
+53 -7
View File
@@ -3,7 +3,7 @@
import asyncio
import hashlib
import mimetypes
from os import stat_result
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Tuple, Union
@@ -12,7 +12,7 @@ import aiofiles
import yaml
from basic_memory import file_utils
from basic_memory.file_utils import FileError, ParseError
from basic_memory.file_utils import FileError, FileMetadata, ParseError
from basic_memory.markdown.markdown_processor import MarkdownProcessor
from basic_memory.models import Entity as EntityModel
from basic_memory.schemas import Entity as EntitySchema
@@ -221,6 +221,41 @@ class FileService:
logger.exception("File read error", path=str(full_path), error=str(e))
raise FileOperationError(f"Failed to read file: {e}")
async def read_file_bytes(self, path: FilePath) -> bytes:
"""Read file content as bytes using true async I/O with aiofiles.
This method reads files in binary mode, suitable for non-text files
like images, PDFs, etc. For cloud compatibility with S3FileService.
Args:
path: Path to read (Path or string)
Returns:
File content as bytes
Raises:
FileOperationError: If read fails
"""
# Convert string to Path if needed
path_obj = self.base_path / path if isinstance(path, str) else path
full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
try:
logger.debug("Reading file bytes", operation="read_file_bytes", path=str(full_path))
async with aiofiles.open(full_path, mode="rb") as f:
content = await f.read()
logger.debug(
"File read completed",
path=str(full_path),
content_length=len(content),
)
return content
except Exception as e:
logger.exception("File read error", path=str(full_path), error=str(e))
raise FileOperationError(f"Failed to read file: {e}")
async def read_file(self, path: FilePath) -> Tuple[str, str]:
"""Read file and compute checksum using true async I/O.
@@ -382,20 +417,31 @@ class FileService:
logger.error("Failed to compute checksum", path=str(full_path), error=str(e))
raise FileError(f"Failed to compute checksum for {path}: {e}")
def file_stats(self, path: FilePath) -> stat_result:
"""Return file stats for a given path.
async def get_file_metadata(self, path: FilePath) -> FileMetadata:
"""Return file metadata for a given path.
This method is async to support cloud implementations (S3FileService)
where file metadata requires async operations (head_object).
Args:
path: Path to the file (Path or string)
Returns:
File statistics
FileMetadata with size, created_at, and modified_at
"""
# Convert string to Path if needed
path_obj = self.base_path / path if isinstance(path, str) else path
full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
# get file timestamps
return full_path.stat()
# Run blocking stat() in thread pool to maintain async compatibility
loop = asyncio.get_event_loop()
stat_result = await loop.run_in_executor(None, full_path.stat)
return FileMetadata(
size=stat_result.st_size,
created_at=datetime.fromtimestamp(stat_result.st_ctime).astimezone(),
modified_at=datetime.fromtimestamp(stat_result.st_mtime).astimezone(),
)
def content_type(self, path: FilePath) -> str:
"""Return content_type for a given path.
+5 -6
View File
@@ -24,9 +24,6 @@ from basic_memory.config import WATCH_STATUS_JSON, ConfigManager, get_project_co
from basic_memory.utils import generate_permalink
config = ConfigManager().config
class ProjectService:
"""Service for managing Basic Memory projects."""
@@ -144,6 +141,7 @@ class ProjectService:
"""
# If project_root is set, constrain all projects to that directory
project_root = self.config_manager.config.project_root
sanitized_name = None
if project_root:
base_path = Path(project_root)
@@ -200,14 +198,15 @@ class ProjectService:
f"Projects cannot share directory trees."
)
# First add to config file (this will validate the project doesn't exist)
project_config = self.config_manager.add_project(name, resolved_path)
if not self.config_manager.config.cloud_mode:
# First add to config file (this will validate the project doesn't exist)
self.config_manager.add_project(name, resolved_path)
# Then add to database
project_data = {
"name": name,
"path": resolved_path,
"permalink": generate_permalink(project_config.name),
"permalink": sanitized_name,
"is_active": True,
# Don't set is_default=False to avoid UNIQUE constraint issues
# Let it default to NULL, only set to True when explicitly making default
+27 -20
View File
@@ -642,12 +642,19 @@ class SyncService:
file_contains_frontmatter = has_frontmatter(file_content)
# Get file timestamps for tracking modification times
file_stats = self.file_service.file_stats(path)
created = datetime.fromtimestamp(file_stats.st_ctime).astimezone()
modified = datetime.fromtimestamp(file_stats.st_mtime).astimezone()
file_metadata = await self.file_service.get_file_metadata(path)
created = file_metadata.created_at
modified = file_metadata.modified_at
# entity markdown will always contain front matter, so it can be used up create/update the entity
entity_markdown = await self.entity_parser.parse_file(path)
# Parse markdown content with file metadata (avoids redundant file read/stat)
# This enables cloud implementations (S3FileService) to provide metadata from head_object
abs_path = self.file_service.base_path / path
entity_markdown = await self.entity_parser.parse_markdown_content(
file_path=abs_path,
content=file_content,
mtime=file_metadata.modified_at.timestamp(),
ctime=file_metadata.created_at.timestamp(),
)
# if the file contains frontmatter, resolve a permalink (unless disabled)
if file_contains_frontmatter and not self.app_config.disable_permalinks:
@@ -693,8 +700,8 @@ class SyncService:
"checksum": final_checksum,
"created_at": created,
"updated_at": modified,
"mtime": file_stats.st_mtime,
"size": file_stats.st_size,
"mtime": file_metadata.modified_at.timestamp(),
"size": file_metadata.size,
},
)
@@ -723,9 +730,9 @@ class SyncService:
await self.entity_service.resolve_permalink(path, skip_conflict_check=True)
# get file timestamps
file_stats = self.file_service.file_stats(path)
created = datetime.fromtimestamp(file_stats.st_ctime).astimezone()
modified = datetime.fromtimestamp(file_stats.st_mtime).astimezone()
file_metadata = await self.file_service.get_file_metadata(path)
created = file_metadata.created_at
modified = file_metadata.modified_at
# get mime type
content_type = self.file_service.content_type(path)
@@ -741,8 +748,8 @@ class SyncService:
created_at=created,
updated_at=modified,
content_type=content_type,
mtime=file_stats.st_mtime,
size=file_stats.st_size,
mtime=file_metadata.modified_at.timestamp(),
size=file_metadata.size,
)
)
return entity, checksum
@@ -758,15 +765,15 @@ class SyncService:
logger.error(f"Entity not found after constraint violation, path={path}")
raise ValueError(f"Entity not found after constraint violation: {path}")
# Re-get file stats since we're in update path
file_stats_for_update = self.file_service.file_stats(path)
# Re-get file metadata since we're in update path
file_metadata_for_update = await self.file_service.get_file_metadata(path)
updated = await self.entity_repository.update(
entity.id,
{
"file_path": path,
"checksum": checksum,
"mtime": file_stats_for_update.st_mtime,
"size": file_stats_for_update.st_size,
"mtime": file_metadata_for_update.modified_at.timestamp(),
"size": file_metadata_for_update.size,
},
)
@@ -780,8 +787,8 @@ class SyncService:
raise
else:
# Get file timestamps for updating modification time
file_stats = self.file_service.file_stats(path)
modified = datetime.fromtimestamp(file_stats.st_mtime).astimezone()
file_metadata = await self.file_service.get_file_metadata(path)
modified = file_metadata.modified_at
entity = await self.entity_repository.get_by_file_path(path)
if entity is None: # pragma: no cover
@@ -796,8 +803,8 @@ class SyncService:
"file_path": path,
"checksum": checksum,
"updated_at": modified,
"mtime": file_stats.st_mtime,
"size": file_stats.st_size,
"mtime": file_metadata.modified_at.timestamp(),
"size": file_metadata.size,
},
)
+65 -61
View File
@@ -5,9 +5,9 @@ import os
import logging
import re
import sys
from datetime import datetime
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Protocol, Union, runtime_checkable, List
from typing import Protocol, Union, runtime_checkable, List
from loguru import logger
from unidecode import unidecode
@@ -203,29 +203,35 @@ def generate_permalink(file_path: Union[Path, str, PathLike], split_extension: b
def setup_logging(
env: str,
home_dir: Path,
log_file: Optional[str] = None,
log_level: str = "INFO",
console: bool = True,
log_to_file: bool = False,
log_to_stdout: bool = False,
structured_context: bool = False,
) -> None: # pragma: no cover
"""
Configure logging for the application.
"""Configure logging with explicit settings.
This function provides a simple, explicit interface for configuring logging.
Each entry point (CLI, MCP, API) should call this with appropriate settings.
Args:
env: The environment name (dev, test, prod)
home_dir: The root directory for the application
log_file: The name of the log file to write to
log_level: The logging level to use
console: Whether to log to the console
log_level: DEBUG, INFO, WARNING, ERROR
log_to_file: Write to ~/.basic-memory/basic-memory.log with rotation
log_to_stdout: Write to stderr (for Docker/cloud deployments)
structured_context: Bind tenant_id, fly_region, etc. for cloud observability
"""
# Remove default handler and any existing handlers
logger.remove()
# Add file handler if we are not running tests and a log file is specified
if log_file and env != "test":
# Setup file logger
log_path = home_dir / log_file
# In test mode, only log to stdout regardless of settings
env = os.getenv("BASIC_MEMORY_ENV", "dev")
if env == "test":
logger.add(sys.stderr, level=log_level, backtrace=True, diagnose=True, colorize=True)
return
# Add file handler with rotation
if log_to_file:
log_path = Path.home() / ".basic-memory" / "basic-memory.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
logger.add(
str(log_path),
level=log_level,
@@ -233,42 +239,28 @@ def setup_logging(
retention="10 days",
backtrace=True,
diagnose=True,
enqueue=True,
enqueue=True, # Thread-safe async logging
colorize=False,
)
# Add console logger if requested or in test mode
if env == "test" or console:
# Add stdout handler (for Docker/cloud)
if log_to_stdout:
logger.add(sys.stderr, level=log_level, backtrace=True, diagnose=True, colorize=True)
logger.info(f"ENV: '{env}' Log level: '{log_level}' Logging to {log_file}")
# Bind environment context for structured logging (works in both local and cloud)
tenant_id = os.getenv("BASIC_MEMORY_TENANT_ID", "local")
fly_app_name = os.getenv("FLY_APP_NAME", "local")
fly_machine_id = os.getenv("FLY_MACHINE_ID", "local")
fly_region = os.getenv("FLY_REGION", "local")
logger.configure(
extra={
"tenant_id": tenant_id,
"fly_app_name": fly_app_name,
"fly_machine_id": fly_machine_id,
"fly_region": fly_region,
}
)
# Bind structured context for cloud observability
if structured_context:
logger.configure(
extra={
"tenant_id": os.getenv("BASIC_MEMORY_TENANT_ID", "local"),
"fly_app_name": os.getenv("FLY_APP_NAME", "local"),
"fly_machine_id": os.getenv("FLY_MACHINE_ID", "local"),
"fly_region": os.getenv("FLY_REGION", "local"),
}
)
# Reduce noise from third-party libraries
noisy_loggers = {
# HTTP client logs
"httpx": logging.WARNING,
# File watching logs
"watchfiles.main": logging.WARNING,
}
# Set log levels for noisy loggers
for logger_name, level in noisy_loggers.items():
logging.getLogger(logger_name).setLevel(level)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("watchfiles.main").setLevel(logging.WARNING)
def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
@@ -337,7 +329,7 @@ def normalize_file_path_for_comparison(file_path: str) -> str:
This function normalizes file paths to help detect potential conflicts:
- Converts to lowercase for case-insensitive comparison
- Normalizes Unicode characters
- Handles path separators consistently
- Converts backslashes to forward slashes for cross-platform consistency
Args:
file_path: The file path to normalize
@@ -346,19 +338,15 @@ def normalize_file_path_for_comparison(file_path: str) -> str:
Normalized file path for comparison purposes
"""
import unicodedata
from pathlib import PureWindowsPath
# Convert to lowercase for case-insensitive comparison
normalized = file_path.lower()
# Use PureWindowsPath to ensure backslashes are treated as separators
# regardless of current platform, then convert to POSIX-style
normalized = PureWindowsPath(file_path).as_posix().lower()
# Normalize Unicode characters (NFD normalization)
normalized = unicodedata.normalize("NFD", normalized)
# Replace path separators with forward slashes
normalized = normalized.replace("\\", "/")
# Remove multiple slashes
normalized = re.sub(r"/+", "/", normalized)
return normalized
@@ -442,21 +430,37 @@ def validate_project_path(path: str, project_path: Path) -> bool:
return False
def ensure_timezone_aware(dt: datetime) -> datetime:
"""Ensure a datetime is timezone-aware using system timezone.
def ensure_timezone_aware(dt: datetime, cloud_mode: bool | None = None) -> datetime:
"""Ensure a datetime is timezone-aware.
If the datetime is naive, convert it to timezone-aware using the system's local timezone.
If it's already timezone-aware, return it unchanged.
If the datetime is naive, convert it to timezone-aware. The interpretation
depends on cloud_mode:
- In cloud mode (PostgreSQL/asyncpg): naive datetimes are interpreted as UTC
- In local mode (SQLite): naive datetimes are interpreted as local time
asyncpg uses binary protocol which returns timestamps in UTC but as naive
datetimes. In cloud deployments, cloud_mode=True handles this correctly.
Args:
dt: The datetime to ensure is timezone-aware
cloud_mode: Optional explicit cloud_mode setting. If None, loads from config.
Returns:
A timezone-aware datetime
"""
if dt.tzinfo is None:
# Naive datetime - assume it's in local time and add timezone
return dt.astimezone()
# Determine cloud_mode: use explicit parameter if provided, otherwise load from config
if cloud_mode is None:
from basic_memory.config import ConfigManager
cloud_mode = ConfigManager().config.cloud_mode_enabled
if cloud_mode:
# Cloud/PostgreSQL mode: naive datetimes from asyncpg are already UTC
return dt.replace(tzinfo=timezone.utc)
else:
# Local/SQLite mode: naive datetimes are in local time
return dt.astimezone()
else:
# Already timezone-aware
return dt
+18 -14
View File
@@ -9,17 +9,19 @@ from basic_memory.mcp.async_client import create_client
def test_create_client_uses_asgi_when_no_remote_env():
"""Test that create_client uses ASGI transport when BASIC_MEMORY_USE_REMOTE_API is not set."""
# Ensure env vars are not set (pop if they exist)
"""Test that create_client uses ASGI transport when cloud mode is disabled."""
# Ensure env vars are not set and config cloud_mode is False
with patch.dict("os.environ", clear=False):
os.environ.pop("BASIC_MEMORY_USE_REMOTE_API", None)
os.environ.pop("BASIC_MEMORY_CLOUD_MODE", None)
client = create_client()
# Also patch the config's cloud_mode to ensure it's False
with patch.object(ConfigManager().config, "cloud_mode", False):
client = create_client()
assert isinstance(client, AsyncClient)
assert isinstance(client._transport, ASGITransport)
assert str(client.base_url) == "http://test"
assert isinstance(client, AsyncClient)
assert isinstance(client._transport, ASGITransport)
assert str(client.base_url) == "http://test"
def test_create_client_uses_http_when_cloud_mode_env_set():
@@ -37,16 +39,18 @@ def test_create_client_uses_http_when_cloud_mode_env_set():
def test_create_client_configures_extended_timeouts():
"""Test that create_client configures 30-second timeouts for long operations."""
# Ensure env vars are not set (pop if they exist)
# Ensure env vars are not set and config cloud_mode is False
with patch.dict("os.environ", clear=False):
os.environ.pop("BASIC_MEMORY_USE_REMOTE_API", None)
os.environ.pop("BASIC_MEMORY_CLOUD_MODE", None)
client = create_client()
# Also patch the config's cloud_mode to ensure it's False
with patch.object(ConfigManager().config, "cloud_mode", False):
client = create_client()
# Verify timeout configuration
assert isinstance(client.timeout, Timeout)
assert client.timeout.connect == 10.0 # 10 seconds for connection
assert client.timeout.read == 30.0 # 30 seconds for reading
assert client.timeout.write == 30.0 # 30 seconds for writing
assert client.timeout.pool == 30.0 # 30 seconds for pool
# Verify timeout configuration
assert isinstance(client.timeout, Timeout)
assert client.timeout.connect == 10.0 # 10 seconds for connection
assert client.timeout.read == 30.0 # 30 seconds for reading
assert client.timeout.write == 30.0 # 30 seconds for writing
assert client.timeout.pool == 30.0 # 30 seconds for pool
+74
View File
@@ -162,3 +162,77 @@ async def test_write_unicode_content(tmp_path: Path, file_service: FileService):
content, _ = await file_service.read_file(test_path)
assert content == test_content
@pytest.mark.asyncio
async def test_read_file_content(tmp_path: Path, file_service: FileService):
"""Test read_file_content returns just the content without checksum."""
test_path = tmp_path / "test.md"
test_content = "test content\nwith multiple lines"
# Write file
await file_service.write_file(test_path, test_content)
# Read content only
content = await file_service.read_file_content(test_path)
assert content == test_content
@pytest.mark.asyncio
async def test_read_file_content_missing_file(tmp_path: Path, file_service: FileService):
"""Test read_file_content raises error for missing files."""
test_path = tmp_path / "missing.md"
with pytest.raises(FileOperationError):
await file_service.read_file_content(test_path)
@pytest.mark.asyncio
async def test_read_file_bytes(tmp_path: Path, file_service: FileService):
"""Test read_file_bytes for binary file reading."""
test_path = tmp_path / "test.bin"
# Create binary content with non-UTF8 bytes
binary_content = b"\x00\x01\x02\x03\xff\xfe\xfd"
# Write binary file directly
test_path.write_bytes(binary_content)
# Read back using read_file_bytes
content = await file_service.read_file_bytes(test_path)
assert content == binary_content
@pytest.mark.asyncio
async def test_read_file_bytes_image(tmp_path: Path, file_service: FileService):
"""Test read_file_bytes with image-like binary content."""
test_path = tmp_path / "test.png"
# PNG header signature
png_header = b"\x89PNG\r\n\x1a\n"
fake_image_content = png_header + b"\x00" * 100
test_path.write_bytes(fake_image_content)
content = await file_service.read_file_bytes(test_path)
assert content == fake_image_content
assert content.startswith(png_header)
@pytest.mark.asyncio
async def test_read_file_bytes_missing_file(tmp_path: Path, file_service: FileService):
"""Test read_file_bytes raises error for missing files."""
test_path = tmp_path / "missing.bin"
with pytest.raises(FileOperationError):
await file_service.read_file_bytes(test_path)
@pytest.mark.asyncio
async def test_read_file_bytes_text_file(tmp_path: Path, file_service: FileService):
"""Test read_file_bytes can read text files as bytes."""
test_path = tmp_path / "test.txt"
text_content = "Hello, World!"
test_path.write_text(text_content)
content = await file_service.read_file_bytes(test_path)
assert content == text_content.encode("utf-8")
-5
View File
@@ -890,11 +890,6 @@ async def test_add_project_without_project_root_allows_arbitrary_paths(
if "BASIC_MEMORY_PROJECT_ROOT" in os.environ:
monkeypatch.delenv("BASIC_MEMORY_PROJECT_ROOT")
# Force reload config without project_root
from basic_memory.services import project_service as ps_module
monkeypatch.setattr(ps_module, "config", config_manager.load_config())
# Create a test directory
test_dir = Path(temp_dir) / "arbitrary-location"
test_dir.mkdir(parents=True, exist_ok=True)
+97
View File
@@ -0,0 +1,97 @@
"""Tests for timezone utilities."""
from datetime import datetime, timezone
import pytest
from basic_memory.utils import ensure_timezone_aware
class TestEnsureTimezoneAware:
"""Tests for ensure_timezone_aware function."""
def test_already_timezone_aware_returns_unchanged(self):
"""Timezone-aware datetime should be returned unchanged."""
dt = datetime(2024, 1, 15, 12, 30, 0, tzinfo=timezone.utc)
result = ensure_timezone_aware(dt)
assert result == dt
assert result.tzinfo == timezone.utc
def test_naive_datetime_cloud_mode_true_interprets_as_utc(self):
"""In cloud mode, naive datetimes should be interpreted as UTC."""
naive_dt = datetime(2024, 1, 15, 12, 30, 0)
result = ensure_timezone_aware(naive_dt, cloud_mode=True)
# Should have UTC timezone
assert result.tzinfo == timezone.utc
# Time values should be unchanged (just tagged as UTC)
assert result.year == 2024
assert result.month == 1
assert result.day == 15
assert result.hour == 12
assert result.minute == 30
def test_naive_datetime_cloud_mode_false_interprets_as_local(self):
"""In local mode, naive datetimes should be interpreted as local time."""
naive_dt = datetime(2024, 1, 15, 12, 30, 0)
result = ensure_timezone_aware(naive_dt, cloud_mode=False)
# Should have some timezone info (local)
assert result.tzinfo is not None
# The datetime should be converted to local timezone
# We can't assert exact timezone as it depends on system
def test_cloud_mode_true_does_not_shift_time(self):
"""Cloud mode should use replace() not astimezone() - time values unchanged."""
naive_dt = datetime(2024, 6, 15, 18, 0, 0) # Summer time
result = ensure_timezone_aware(naive_dt, cloud_mode=True)
# Hour should remain 18, not be shifted by timezone offset
assert result.hour == 18
assert result.tzinfo == timezone.utc
def test_explicit_cloud_mode_skips_config_loading(self):
"""When cloud_mode is explicitly passed, config should not be loaded."""
# This test verifies we can call ensure_timezone_aware without
# triggering ConfigManager import when cloud_mode is explicit
naive_dt = datetime(2024, 1, 15, 12, 30, 0)
# Should work without any config setup
result_cloud = ensure_timezone_aware(naive_dt, cloud_mode=True)
assert result_cloud.tzinfo == timezone.utc
result_local = ensure_timezone_aware(naive_dt, cloud_mode=False)
assert result_local.tzinfo is not None
def test_none_cloud_mode_falls_back_to_config(self):
"""When cloud_mode is None, should load from config."""
from unittest.mock import patch, MagicMock
naive_dt = datetime(2024, 1, 15, 12, 30, 0)
# Mock ConfigManager to return cloud_mode_enabled=True
mock_config = MagicMock()
mock_config.cloud_mode_enabled = True
with patch("basic_memory.config.ConfigManager") as mock_manager:
mock_manager.return_value.config = mock_config
result = ensure_timezone_aware(naive_dt, cloud_mode=None)
# Should have used cloud mode (UTC)
assert result.tzinfo == timezone.utc
def test_asyncpg_naive_utc_scenario(self):
"""Simulate asyncpg returning naive datetime that's actually UTC.
asyncpg binary protocol returns timestamps in UTC but as naive datetimes.
In cloud mode, we interpret these as UTC rather than local time.
"""
# Simulate what asyncpg returns: a naive datetime that's actually UTC
asyncpg_result = datetime(2024, 1, 15, 18, 30, 0) # 6:30 PM UTC
# In cloud mode, interpret as UTC
cloud_result = ensure_timezone_aware(asyncpg_result, cloud_mode=True)
assert cloud_result == datetime(2024, 1, 15, 18, 30, 0, tzinfo=timezone.utc)
# The hour should remain 18, not shifted
assert cloud_result.hour == 18