Compare commits

..

34 Commits

Author SHA1 Message Date
semantic-release 9437a5f83b chore(release): 0.12.3 [skip ci] 2025-04-17 16:04:31 +00:00
phernandez 3c1cc346df fix: modify recent_activity args to be strings instead of enums
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-17 10:56:46 -05:00
phernandez 81616ab42e update CLAUDE.md
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-17 10:56:46 -05:00
phernandez 73ea91fe0d fix: add extra logic for permalink generation with mixed Latin unicode and Chinese characters
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-17 10:00:40 -05:00
andyxinweiminicloud 03d4e97b90 Fix: preserve Chinese characters in permalinks 2025-04-17 09:03:35 -05:00
Paul Hernandez 98622a7a47 Update README.md
fix example json for passing --project arg

Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>
2025-04-16 07:50:12 -05:00
Amadeusz Wieczorek 54dfa08aba Update name of function search_notes in error output (#92)
Signed-off-by: Amadeusz Wieczorek <git@amadeusw.com>
2025-04-14 11:06:15 -05:00
semantic-release 9433065a57 chore(release): 0.12.2 [skip ci] 2025-04-08 16:27:42 +00:00
Paul Hernandez 2934176331 fix: utf8 for all file reads/write/open instead of default platform encoding (#91)
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-08 10:55:01 -05:00
semantic-release ac89eb47cb chore(release): 0.12.1 [skip ci] 2025-04-07 22:47:23 +00:00
Paul Hernandez 78a3412bcf fix: run migrations and sync when starting mcp (#88) 2025-04-07 17:45:26 -05:00
Paul Hernandez fa314b5d2b Update README.md
Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>
2025-04-06 20:29:00 -05:00
Paul Hernandez d2ac62a27c Update README.md
Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>
2025-04-06 20:25:06 -05:00
semantic-release 7ce7a29f35 chore(release): 0.12.0 [skip ci] 2025-04-06 23:44:17 +00:00
Paul Hernandez 3f4d9e4d87 fix: write_note preserves frontmatter fields in content (#84) 2025-04-06 18:39:51 -05:00
Paul Hernandez 00c8633cfc feat: add watch to mcp process (#83) 2025-04-06 18:33:10 -05:00
Paul Hernandez 617e60bda4 feat: permalink enhancements (#82)
- Avoiding "useless permalink values" for files without metadata
- Enable permalinks to be updated on move via config setting
2025-04-06 14:54:59 -05:00
github-actions[bot] 6c19c9edf5 fix: [BUG] # character accumulation in markdown frontmatter tags prop (#79) 2025-04-05 23:07:53 -05:00
github-actions[bot] 9bff1f732e fix: [BUG] write_note Tool Fails to Update Existing Files in Some Situations. (#80) 2025-04-05 22:31:40 -05:00
Paul Hernandez 248214cb11 fix: set default mcp log level to ERROR (#81) 2025-04-05 22:27:29 -05:00
Jason Noble 40ea28b0bf docs: Updated basicmachines.co links to be https (#69) 2025-04-05 22:22:37 -05:00
Marc Baiza 43cbb7b38c docs: Add VS Code instructions to README (#76) 2025-04-05 22:18:56 -05:00
github-actions[bot] 7930ddb291 fix: [BUG] Some notes never exit "modified" status (#77) 2025-04-05 22:03:38 -05:00
phernandez 1844e58210 Add pull_request_target trigger to test workflow with explanatory comment 2025-04-05 20:30:05 -05:00
phernandez 7fbb5ebfc5 Update claude-code-github-action to v0.11.0 with backtick handling fix 2025-04-05 20:04:18 -05:00
phernandez 0e5a465b10 add claude-output to .gitignore
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-05 19:45:07 -05:00
github-actions[bot] 9d581cee13 fix: [BUG] Cursor has errors calling search tool (#78) 2025-04-05 19:12:23 -05:00
phernandez f58852954b Update to claude-code-github-action v0.10.0 to check comment author for org membership 2025-04-05 12:54:33 -05:00
phernandez edd426b99f Update to claude-code-github-action v0.9.0 with organization membership check fix 2025-04-05 12:42:35 -05:00
phernandez 6b4a421315 Update to claude-code-github-action v0.8.0 with PERSONAL_ACCESS_TOKEN support 2025-04-05 12:13:05 -05:00
phernandez 4632405446 Fix claude-code-actions.yml to use v0.7.0 of basicmachines-co action 2025-04-05 12:01:04 -05:00
phernandez 3a6d5ff58d update claude code actions
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-04 23:31:08 -05:00
phernandez 03de85ec32 add claude code github action workflow
Signed-off-by: phernandez <paul@basicmachines.co>
2025-04-04 21:20:40 -05:00
phernandez 084c91ecce update basic-memory dep version
Signed-off-by: phernandez <paul@basicmachines.co>
2025-03-30 17:46:44 -05:00
78 changed files with 1855 additions and 725 deletions
+16
View File
@@ -0,0 +1,16 @@
name: Claude Code Integration
on:
issue_comment:
types: [ created ]
pull_request_review_comment:
types: [ created ]
jobs:
claude-integration:
uses: basicmachines-co/claude-code-github-action/.github/workflows/claude-full.yml@v0.11.0
with:
issue-label: 'claude-fix' # Optional: customize the trigger label
secrets:
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
PERSONAL_ACCESS_TOKEN: ${{ secrets.CLAUDE_TOKEN }}
+6
View File
@@ -5,6 +5,12 @@ on:
branches: [ "main" ]
pull_request:
branches: [ "main" ]
# pull_request_target runs on the BASE of the PR, not the merge result.
# It has write permissions and access to secrets.
# It's useful for PRs from forks or automated PRs but requires careful use for security reasons.
# See: https://docs.github.com/en/actions/using-workflows/events-that-trigger-workflows#pull_request_target
pull_request_target:
branches: [ "main" ]
jobs:
test:
+5 -1
View File
@@ -47,4 +47,8 @@ ENV/
# obsidian docs:
/docs/.obsidian/
/examples/.obsidian/
/examples/.basic-memory/
/examples/.basic-memory/
# claude action
claude-output
+85
View File
@@ -1,6 +1,91 @@
# CHANGELOG
## v0.12.3 (2025-04-17)
### Bug Fixes
- Add extra logic for permalink generation with mixed Latin unicode and Chinese characters
([`73ea91f`](https://github.com/basicmachines-co/basic-memory/commit/73ea91fe0d1f7ab89b99a1b691d59fe608b7fcbb))
Signed-off-by: phernandez <paul@basicmachines.co>
- Modify recent_activity args to be strings instead of enums
([`3c1cc34`](https://github.com/basicmachines-co/basic-memory/commit/3c1cc346df519e703fae6412d43a92c7232c6226))
Signed-off-by: phernandez <paul@basicmachines.co>
## v0.12.2 (2025-04-08)
### Bug Fixes
- Utf8 for all file reads/write/open instead of default platform encoding
([#91](https://github.com/basicmachines-co/basic-memory/pull/91),
[`2934176`](https://github.com/basicmachines-co/basic-memory/commit/29341763318408ea8f1e954a41046c4185f836c6))
Signed-off-by: phernandez <paul@basicmachines.co>
## v0.12.1 (2025-04-07)
### Bug Fixes
- Run migrations and sync when starting mcp
([#88](https://github.com/basicmachines-co/basic-memory/pull/88),
[`78a3412`](https://github.com/basicmachines-co/basic-memory/commit/78a3412bcff83b46e78e26f8b9fce42ed9e05991))
## v0.12.0 (2025-04-06)
### Bug Fixes
- [bug] `#` character accumulation in markdown frontmatter tags prop
([#79](https://github.com/basicmachines-co/basic-memory/pull/79),
[`6c19c9e`](https://github.com/basicmachines-co/basic-memory/commit/6c19c9edf5131054ba201a109b37f15c83ef150c))
- [bug] Cursor has errors calling search tool
([#78](https://github.com/basicmachines-co/basic-memory/pull/78),
[`9d581ce`](https://github.com/basicmachines-co/basic-memory/commit/9d581cee133f9dde4a0a85118868227390c84161))
- [bug] Some notes never exit "modified" status
([#77](https://github.com/basicmachines-co/basic-memory/pull/77),
[`7930ddb`](https://github.com/basicmachines-co/basic-memory/commit/7930ddb2919057be30ceac8c4c19da6aaa1d3e92))
- [bug] write_note Tool Fails to Update Existing Files in Some Situations.
([#80](https://github.com/basicmachines-co/basic-memory/pull/80),
[`9bff1f7`](https://github.com/basicmachines-co/basic-memory/commit/9bff1f732e71bc60f88b5c2ce3db5a2aa60b8e28))
- Set default mcp log level to ERROR
([#81](https://github.com/basicmachines-co/basic-memory/pull/81),
[`248214c`](https://github.com/basicmachines-co/basic-memory/commit/248214cb114a269ca60ff6398e382f9e2495ad8e))
- Write_note preserves frontmatter fields in content
([#84](https://github.com/basicmachines-co/basic-memory/pull/84),
[`3f4d9e4`](https://github.com/basicmachines-co/basic-memory/commit/3f4d9e4d872ebc0ed719c61b24d803c14a9db5e6))
### Documentation
- Add VS Code instructions to README
([#76](https://github.com/basicmachines-co/basic-memory/pull/76),
[`43cbb7b`](https://github.com/basicmachines-co/basic-memory/commit/43cbb7b38cc0482ac0a41b6759320e3588186e43))
- Updated basicmachines.co links to be https
([#69](https://github.com/basicmachines-co/basic-memory/pull/69),
[`40ea28b`](https://github.com/basicmachines-co/basic-memory/commit/40ea28b0bfc60012924a69ecb76511daa4c7d133))
### Features
- Add watch to mcp process ([#83](https://github.com/basicmachines-co/basic-memory/pull/83),
[`00c8633`](https://github.com/basicmachines-co/basic-memory/commit/00c8633cfcee75ff640ff8fe81dafeb956281a94))
- Permalink enhancements ([#82](https://github.com/basicmachines-co/basic-memory/pull/82),
[`617e60b`](https://github.com/basicmachines-co/basic-memory/commit/617e60bda4a590678a5f551f10a73e7b47e3b13e))
- Avoiding "useless permalink values" for files without metadata - Enable permalinks to be updated
on move via config setting
## v0.11.0 (2025-03-29)
### Bug Fixes
+15 -1
View File
@@ -172,4 +172,18 @@ With GitHub integration, the development workflow includes:
4. **Documentation maintenance** - Claude can keep documentation updated as the code evolves
With this integration, the AI assistant is a full-fledged team member rather than just a tool for generating code
snippets.
snippets.
### Basic Memory Pro
Basic Memory Pro is a desktop GUI application that wraps the basic-memory CLI/MCP tools:
- Built with Tauri (Rust), React (TypeScript), and a Python FastAPI sidecar
- Provides visual knowledge graph exploration and project management
- Uses the same core codebase but adds a desktop-friendly interface
- Project configuration is shared between CLI and Pro versions
- Multiple project support with visual switching interface
local repo: /Users/phernandez/dev/basicmachines/basic-memory-pro
github: https://github.com/basicmachines-co/basic-memory-pro
+43 -12
View File
@@ -13,8 +13,8 @@ Basic Memory lets you build persistent knowledge through natural conversations w
Claude, while keeping everything in simple Markdown files on your computer. It uses the Model Context Protocol (MCP) to
enable any compatible LLM to read and write to your local knowledge base.
- Website: http://basicmachines.co
- Documentation: http://memory.basicmachines.co
- Website: https://basicmachines.co
- Documentation: https://memory.basicmachines.co
## Pick up your conversation right where you left off
@@ -153,7 +153,7 @@ The note embeds semantic content and links to other topics via simple Markdown f
3. You see this file on your computer in real time in the current project directory (default `~/$HOME/basic-memory`).
- Realtime sync can be enabled via running `basic-memory sync --watch`
- Realtime sync is enabled by default with the v0.12.0 version
4. In a chat with the LLM, you can reference a topic:
@@ -262,6 +262,43 @@ Examples of relations:
- documented_in [[Coffee Journal]]
```
## Using with VS Code
For one-click installation, click one of the install buttons below...
[![Install with UV in VS Code](https://img.shields.io/badge/VS_Code-UV-0098FF?style=flat-square&logo=visualstudiocode&logoColor=white)](https://insiders.vscode.dev/redirect/mcp/install?name=basic-memory&config=%7B%22command%22%3A%22uvx%22%2C%22args%22%3A%5B%22basic-memory%22%2C%22mcp%22%5D%7D) [![Install with UV in VS Code Insiders](https://img.shields.io/badge/VS_Code_Insiders-UV-24bfa5?style=flat-square&logo=visualstudiocode&logoColor=white)](https://insiders.vscode.dev/redirect/mcp/install?name=basic-memory&config=%7B%22command%22%3A%22uvx%22%2C%22args%22%3A%5B%22basic-memory%22%2C%22mcp%22%5D%7D&quality=insiders)
You can use Basic Memory with VS Code to easily retrieve and store information while coding. Click the installation buttons above for one-click setup, or follow the manual installation instructions below.
### Manual Installation
Add the following JSON block to your User Settings (JSON) file in VS Code. You can do this by pressing `Ctrl + Shift + P` and typing `Preferences: Open User Settings (JSON)`.
```json
{
"mcp": {
"servers": {
"basic-memory": {
"command": "uvx",
"args": ["basic-memory", "mcp"]
}
}
}
}
```
Optionally, you can add it to a file called `.vscode/mcp.json` in your workspace. This will allow you to share the configuration with others.
```json
{
"servers": {
"basic-memory": {
"command": "uvx",
"args": ["basic-memory", "mcp"]
}
}
}
```
## Using with Claude Desktop
Basic Memory is built using the MCP (Model Context Protocol) and works with the Claude desktop app (https://claude.ai/):
@@ -296,9 +333,9 @@ config:
"command": "uvx",
"args": [
"basic-memory",
"mcp",
"--project",
"your-project-name"
"your-project-name",
"mcp"
]
}
}
@@ -307,13 +344,7 @@ config:
2. Sync your knowledge:
```bash
# One-time sync of local knowledge updates
basic-memory sync
# Run realtime sync process (recommended)
basic-memory sync --watch
```
Basic Memory will sync the files in your project in real time if you make manual edits.
3. In Claude Desktop, the LLM can now use these tools:
View File
+13
View File
@@ -0,0 +1,13 @@
Starting issue-fix mode at Sat Apr 5 18:02:54 UTC 2025
Fetching issue #75 details
Using repository: basicmachines-co/basic-memory
Checking if phernandez is a member of organization basicmachines-co
User phernandez is a member of organization basicmachines-co. Proceeding with Claude fix.
Creating a new branch: fix-issue-75-20250405180254
From https://github.com/basicmachines-co/basic-memory
* branch main -> FETCH_HEAD
Switched to a new branch 'fix-issue-75-20250405180254'
branch 'fix-issue-75-20250405180254' set up to track 'origin/main'.
Prompt saved to ./claude-output/claude_prompt_75.txt for debugging
Running Claude to fix the issue...
Committing changes...
+35
View File
@@ -0,0 +1,35 @@
Let's summarize the changes we've made to fix issue #75:
1. We updated the `search_notes` tool in `/src/basic_memory/mcp/tools/search.py` to accept primitive types as parameters instead of a complex Pydantic `SearchQuery` object. This makes it easier for LLMs like Cursor to make proper tool calls.
2. We converted the internal implementation to create a SearchQuery object from the primitive parameters, maintaining backward compatibility with the existing API.
3. We updated tests in `/tests/mcp/test_tool_search.py` to use the new function signature with primitive parameters.
4. We updated code in `/src/basic_memory/mcp/tools/read_note.py` to use the new function signature when making calls to `search_notes`.
5. We updated code in `/src/basic_memory/mcp/prompts/search.py` to use the new function signature when making calls to `search_notes`.
These changes should make it easier for Cursor and other LLMs to use the search_notes tool by eliminating the complex Pydantic object parameter in favor of simple primitive parameters.
---SUMMARY---
Fixed issue #75 where Cursor was having errors calling the search_notes tool. The problem was that the search_notes tool was expecting a complex Pydantic object (SearchQuery) as input, which was confusing Cursor.
Changes:
1. Modified the search_notes tool to accept primitive types (strings, lists, etc.) as parameters instead of a complex Pydantic object
2. Updated the implementation to create a SearchQuery object internally from these primitive parameters
3. Updated all call sites in the codebase that were using the old function signature
4. Updated tests to use the new function signature
The fix makes it easier for LLMs like Cursor to make proper calls to the search_notes tool, which will resolve the reported error messages:
- "Parameter 'query' must be of type undefined, got object"
- "Parameter 'query' must be of type undefined, got string"
- "Invalid type for parameter 'query' in tool search_notes"
Files modified:
- src/basic_memory/mcp/tools/search.py
- src/basic_memory/mcp/tools/read_note.py
- src/basic_memory/mcp/prompts/search.py
- tests/mcp/test_tool_search.py
- tests/mcp/test_tool_read_note.py
---END SUMMARY---
+49
View File
@@ -0,0 +1,49 @@
You are Claude, an AI assistant tasked with fixing issues in a GitHub repository.
Issue #75: [BUG] Cursor has errors calling search tool
Issue Description:
## Bug Description
> Cursor cannot figure out how to structure the parameters for that tool call. No matter what Cursor seems to try it gets the errors.
>
> ```Looking at the error messages more carefully:
> - When I pass an object: "Parameter 'query' must be of type undefined, got object"
> - When I pass a string: "Parameter 'query' must be of type undefined, got string"
>
>
>
> and then it reports: "Invalid type for parameter 'query' in tool search_notes"
> Any chance you can give me some guidance with this?
>
## Steps To Reproduce
Steps to reproduce the behavior:
try using search tool in Cursor.
## Possible Solution
The tool args should probably be plain text and not json to make it easier to call.
Additional Instructions from User Comment:
let make a PR to implement option #1.
Your task is to:
1. Analyze the issue carefully to understand the problem
2. Look through the repository to identify the relevant files that need to be modified
3. Make precise changes to fix the issue
4. Use the Edit tool to modify files directly when needed
5. Be minimal in your changes - only modify what's necessary to fix the issue
After making changes, provide a summary of what you did in this format:
---SUMMARY---
[Your detailed summary of changes, including which files were modified and how]
---END SUMMARY---
Remember:
- Be specific in your changes
- Only modify files that are necessary to fix the issue
- Follow existing code style and conventions
- Make the minimal changes needed to resolve the issue
+16
View File
@@ -29,6 +29,22 @@ Options:
- `--watch`: Continuously monitor for changes
- `--verbose`: Show detailed output
**Note**:
As of the v0.12.0 release syncing will occur in real time when the mcp process starts.
- The real time sync means that it is no longer necessary to run the `basic-memory sync --watch` process in a a terminal to sync changes to the db (so the AI can see them). This will be done automatically.
This behavior can be changed via the config. The config file for Basic Memory is in the home directory under `.basic-memory/config.json`.
To change the properties, set the following values:
```
~/.basic-memory/config.json
{
"sync_changes": false,
}
```
Thanks for using Basic Memory!
### import
Imports external knowledge sources:
+9 -20
View File
@@ -89,22 +89,11 @@ Replace `/absolute/path/to/uvx` with the actual path you found in Step 1.
Close and reopen Claude Desktop for the changes to take effect.
### 3. Start the Sync Service
### 3. Sync changes in real time
> Note the sync service is optional. You can run it if you want your local change to be available in Claude Desktop
Start the sync service to monitor your files for changes:
```bash
# One-time sync
basic-memory sync
# For continuous monitoring (recommended)
basic-memory sync --watch
```
The `--watch` flag enables automatic detection of file changes, updating your knowledge in real time.
> **Note**: The service will sync changes from your project directory in real time so they available for the AI assistant.
To disable realtime sync, you can update the config. See [[CLI Reference#sync]].
### 4. Staying Updated
To update Basic Memory when new versions are released:
@@ -145,8 +134,8 @@ If Claude cannot find Basic Memory tools:
1. **Check absolute paths**: Ensure you're using complete absolute paths to uvx in the Claude Desktop configuration
2. **Verify installation**: Run `basic-memory --version` in Terminal to confirm Basic Memory is installed
3. **Restart applications**: Restart both Terminal and Claude Desktop after making configuration changes
4. **Check sync status**: Ensure `basic-memory sync --watch` is running
4. **Check sync status**: You can view the sync status by running `basic-memory status
.
#### Permission Issues
If you encounter permission errors:
@@ -282,15 +271,15 @@ basic-memory import claude conversations
basic-memory import chatgpt
```
After importing, run `basic-memory sync` to index everything.
After importing, the changes will be synced. Initial syncs may take a few moments. You can see info about your project by running `basic-memrory project info`.
## Quick Tips
- Keep `basic-memory sync --watch` running in a terminal window
- Basic Memory will sync changes from your project in real time.
- Use special prompts (Continue Conversation, Recent Activity, Search) to start contextual discussions
- Build connections between notes for a richer knowledge graph
- Use direct memory:// URLs when you need precise context
- Use git to version control your knowledge base
- Use direct `memory://` URLs with a permalink when you need precise context. See [[User Guide#Using memory // URLs]]
- Use git to version control your knowledge base (git integration is on the roadmap)
- Review and edit AI-generated notes for accuracy
## Next Steps
+13 -1
View File
@@ -144,7 +144,19 @@ permalink: auth-approaches-2024
---
```
If not specified, one will be generated automatically from the title.
If not specified, one will be generated automatically from the title, if the note has has a frontmatter section.
By default a notes' permalink value will not change if the file is moved. It's a **stable** identifier :). But if you'd rather permalinks are always updated when a file moves, you can set the config setting in the global config.
The config file for Basic Memory is in the home directory under `.basic-memory/config.json`.
To change the behavior, set the following value:
```
~/.basic-memory/config.json
{
"update_permalinks_on_move": true
}
```
### Using memory:// URLs
+1 -1
View File
@@ -44,7 +44,7 @@ def update_claude_config():
# Load existing config or create new
if config_path.exists():
config = json.loads(config_path.read_text())
config = json.loads(config_path.read_text(encoding="utf-8"))
else:
config = {"mcpServers": {}}
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "basic-memory"
version = "0.11.0"
version = "0.12.3"
description = "Local-first knowledge management combining Zettelkasten with knowledge graphs"
readme = "README.md"
requires-python = ">=3.12.1"
+1 -1
View File
@@ -1,3 +1,3 @@
"""basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
__version__ = "0.11.0"
__version__ = "0.12.3"
+11 -3
View File
@@ -7,16 +7,24 @@ from fastapi.exception_handlers import http_exception_handler
from loguru import logger
from basic_memory import db
from basic_memory.config import config as app_config
from basic_memory.api.routers import knowledge, search, memory, resource, project_info
from basic_memory.api.routers import knowledge, memory, project_info, resource, search
from basic_memory.config import config as project_config
from basic_memory.services.initialization import initialize_app
@asynccontextmanager
async def lifespan(app: FastAPI): # pragma: no cover
"""Lifecycle manager for the FastAPI app."""
await db.run_migrations(app_config)
# Initialize database and file sync services
watch_task = await initialize_app(project_config)
# proceed with startup
yield
logger.info("Shutting down Basic Memory API")
if watch_task:
watch_task.cancel()
await db.shutdown_db()
+12 -7
View File
@@ -1,18 +1,17 @@
import asyncio
from typing import Optional
import typer
from basic_memory import db
from basic_memory.config import config
def version_callback(value: bool) -> None:
"""Show version and exit."""
if value: # pragma: no cover
import basic_memory
from basic_memory.config import config
typer.echo(f"Basic Memory version: {basic_memory.__version__}")
typer.echo(f"Current project: {config.project}")
typer.echo(f"Project path: {config.home}")
raise typer.Exit()
@@ -21,11 +20,12 @@ app = typer.Typer(name="basic-memory")
@app.callback()
def app_callback(
ctx: typer.Context,
project: Optional[str] = typer.Option(
None,
"--project",
"-p",
help="Specify which project to use",
help="Specify which project to use 1",
envvar="BASIC_MEMORY_PROJECT",
),
version: Optional[bool] = typer.Option(
@@ -38,6 +38,7 @@ def app_callback(
),
) -> None:
"""Basic Memory - Local-first personal knowledge management."""
# We use the project option to set the BASIC_MEMORY_PROJECT environment variable
# The config module will pick this up when loading
if project: # pragma: no cover
@@ -57,9 +58,13 @@ def app_callback(
config = new_config
# Run migrations for every command unless --version was specified
if not version and ctx.invoked_subcommand is not None:
from basic_memory.config import config
from basic_memory.services.initialization import ensure_initialize_database
ensure_initialize_database(config)
# Run database migrations
asyncio.run(db.run_migrations(config))
# Register sub-command groups
import_app = typer.Typer(help="Import data from various sources")
@@ -38,7 +38,7 @@ async def process_memory_json(
read_task = progress.add_task("Reading memory.json...", total=None)
# First pass - collect entities and relations
with open(json_path) as f:
with open(json_path, encoding="utf-8") as f:
lines = f.readlines()
progress.update(read_task, total=len(lines))
+18 -9
View File
@@ -1,10 +1,7 @@
"""MCP server command."""
from loguru import logger
import basic_memory
from basic_memory.cli.app import app
from basic_memory.config import config
# Import mcp instance
from basic_memory.mcp.server import mcp as mcp_server # pragma: no cover
@@ -15,12 +12,24 @@ import basic_memory.mcp.tools # noqa: F401 # pragma: no cover
@app.command()
def mcp(): # pragma: no cover
"""Run the MCP server for Claude Desktop integration."""
home_dir = config.home
project_name = config.project
"""Run the MCP server"""
from basic_memory.config import config
import asyncio
from basic_memory.services.initialization import initialize_database
logger.info(f"Starting Basic Memory MCP server {basic_memory.__version__}")
logger.info(f"Project: {project_name}")
logger.info(f"Project directory: {home_dir}")
# First, run just the database migrations synchronously
asyncio.run(initialize_database(config))
# Load config to check if sync is enabled
from basic_memory.config import config_manager
basic_memory_config = config_manager.load_config()
if basic_memory_config.sync_changes:
# For now, we'll just log that sync will be handled by the MCP server
from loguru import logger
logger.info("File sync will be handled by the MCP server")
# Start the MCP server
mcp_server.run()
+9 -8
View File
@@ -70,6 +70,7 @@ async def get_sync_service(): # pragma: no cover
# Create sync service
sync_service = SyncService(
config=config,
entity_service=entity_service,
entity_parser=entity_parser,
entity_repository=entity_repository,
@@ -178,14 +179,14 @@ async def run_sync(verbose: bool = False, watch: bool = False, console_status: b
)
# full sync - no progress bars in watch mode
await sync_service.sync(config.home, show_progress=False)
await sync_service.sync(config.home)
# watch changes
await watch_service.run() # pragma: no cover
else:
# one time sync - use progress bars for better UX
# one time sync
logger.info("Running one-time sync")
knowledge_changes = await sync_service.sync(config.home, show_progress=True)
knowledge_changes = await sync_service.sync(config.home)
# Log results
duration_ms = int((time.time() - start_time) * 1000)
@@ -236,11 +237,11 @@ def sync(
if not isinstance(e, typer.Exit):
logger.exception(
"Sync command failed",
project=config.project,
error=str(e),
error_type=type(e).__name__,
watch_mode=watch,
directory=str(config.home),
f"project={config.project},"
f"error={str(e)},"
f"error_type={type(e).__name__},"
f"watch_mode={watch},"
f"directory={str(config.home)}",
)
typer.echo(f"Error during sync: {e}", err=True)
raise typer.Exit(1)
+28 -15
View File
@@ -2,31 +2,29 @@
import asyncio
import sys
from typing import Optional, List, Annotated
from typing import Annotated, List, Optional
import typer
from loguru import logger
from rich import print as rprint
from basic_memory.cli.app import app
from basic_memory.mcp.tools import build_context as mcp_build_context
from basic_memory.mcp.tools import read_note as mcp_read_note
from basic_memory.mcp.tools import recent_activity as mcp_recent_activity
from basic_memory.mcp.tools import search_notes as mcp_search
from basic_memory.mcp.tools import write_note as mcp_write_note
# Import prompts
from basic_memory.mcp.prompts.continue_conversation import (
continue_conversation as mcp_continue_conversation,
)
from basic_memory.mcp.prompts.recent_activity import (
recent_activity_prompt as recent_activity_prompt,
)
from basic_memory.mcp.tools import build_context as mcp_build_context
from basic_memory.mcp.tools import read_note as mcp_read_note
from basic_memory.mcp.tools import recent_activity as mcp_recent_activity
from basic_memory.mcp.tools import search_notes as mcp_search
from basic_memory.mcp.tools import write_note as mcp_write_note
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.memory import MemoryUrl
from basic_memory.schemas.search import SearchQuery, SearchItemType
from basic_memory.schemas.search import SearchItemType
tool_app = typer.Typer()
app.add_typer(tool_app, name="tool", help="Access to MCP tools via CLI")
@@ -198,13 +196,28 @@ def search_notes(
raise typer.Abort()
try:
search_query = SearchQuery(
permalink_match=query if permalink else None,
text=query if not (permalink or title) else None,
title=query if title else None,
after_date=after_date,
if permalink and title: # pragma: no cover
typer.echo(
"Use either --permalink or --title, not both. Exiting.",
err=True,
)
raise typer.Exit(1)
# set search type
search_type = ("permalink" if permalink else None,)
search_type = ("permalink_match" if permalink and "*" in query else None,)
search_type = ("title" if title else None,)
search_type = "text" if search_type is None else search_type
results = asyncio.run(
mcp_search(
query,
search_type=search_type,
page=page,
after_date=after_date,
page_size=page_size,
)
)
results = asyncio.run(mcp_search(query=search_query, page=page, page_size=page_size))
# Use json module for more controlled serialization
import json
+12 -44
View File
@@ -1,58 +1,26 @@
"""Main CLI entry point for basic-memory.""" # pragma: no cover
from basic_memory.cli.app import app # pragma: no cover
import typer
# Register commands
from basic_memory.cli.commands import ( # noqa: F401 # pragma: no cover
status,
sync,
db,
import_memory_json,
mcp,
import_chatgpt,
import_claude_conversations,
import_claude_projects,
import_chatgpt,
tool,
import_memory_json,
mcp,
project,
status,
sync,
tool,
)
# Version command
@app.callback(invoke_without_command=True)
def main(
ctx: typer.Context,
project: str = typer.Option( # noqa
"main",
"--project",
"-p",
help="Specify which project to use",
envvar="BASIC_MEMORY_PROJECT",
),
version: bool = typer.Option(
False,
"--version",
"-V",
help="Show version information and exit.",
is_eager=True,
),
):
"""Basic Memory - Local-first personal knowledge management system."""
if version: # pragma: no cover
from basic_memory import __version__
from basic_memory.config import config
typer.echo(f"Basic Memory v{__version__}")
typer.echo(f"Current project: {config.project}")
typer.echo(f"Project path: {config.home}")
raise typer.Exit()
# Handle project selection via environment variable
if project:
import os
os.environ["BASIC_MEMORY_PROJECT"] = project
from basic_memory.config import config
from basic_memory.services.initialization import ensure_initialization
if __name__ == "__main__": # pragma: no cover
# Run initialization if we are starting as a module
ensure_initialization(config)
# start the app
app()
+30 -6
View File
@@ -35,10 +35,14 @@ class ProjectConfig(BaseSettings):
# Watch service configuration
sync_delay: int = Field(
default=500, description="Milliseconds to wait after changes before syncing", gt=0
default=1000, description="Milliseconds to wait after changes before syncing", gt=0
)
log_level: str = "DEBUG"
# update permalinks on move
update_permalinks_on_move: bool = Field(
default=False,
description="Whether to update permalinks when files are moved or renamed. default (False)",
)
model_config = SettingsConfigDict(
env_prefix="BASIC_MEMORY_",
@@ -77,6 +81,18 @@ class BasicMemoryConfig(BaseSettings):
description="Name of the default project to use",
)
log_level: str = "INFO"
update_permalinks_on_move: bool = Field(
default=False,
description="Whether to update permalinks when files are moved or renamed. default (False)",
)
sync_changes: bool = Field(
default=True,
description="Whether to sync changes in real time. default (True)",
)
model_config = SettingsConfigDict(
env_prefix="BASIC_MEMORY_",
extra="ignore",
@@ -194,9 +210,14 @@ def get_project_config(project_name: Optional[str] = None) -> ProjectConfig:
"BASIC_MEMORY_PROJECT", project_name or config_manager.default_project
)
update_permalinks_on_move = config_manager.load_config().update_permalinks_on_move
try:
project_path = config_manager.get_project_path(actual_project_name)
return ProjectConfig(home=project_path, project=actual_project_name)
return ProjectConfig(
home=project_path,
project=actual_project_name,
update_permalinks_on_move=update_permalinks_on_move,
)
except ValueError: # pragma: no cover
logger.warning(f"Project '{actual_project_name}' not found, using default")
project_path = config_manager.get_project_path(config_manager.default_project)
@@ -225,8 +246,10 @@ def get_process_name(): # pragma: no cover
return "sync"
elif "mcp" in sys.argv:
return "mcp"
else:
elif "cli" in sys.argv:
return "cli"
else:
return "api"
process_name = get_process_name()
@@ -234,6 +257,7 @@ process_name = get_process_name()
# Global flag to track if logging has been set up
_LOGGING_SETUP = False
def setup_basic_memory_logging(): # pragma: no cover
"""Set up logging for basic-memory, ensuring it only happens once."""
global _LOGGING_SETUP
@@ -245,12 +269,12 @@ def setup_basic_memory_logging(): # pragma: no cover
setup_logging(
env=config.env,
home_dir=user_home, # Use user home for logs
log_level=config.log_level,
log_level=config_manager.load_config().log_level,
log_file=f"{DATA_DIR_NAME}/basic-memory-{process_name}.log",
console=False,
)
logger.info(f"Starting Basic Memory {basic_memory.__version__} (Project: {config.project})")
logger.info(f"Basic Memory {basic_memory.__version__} (Project: {config.project})")
_LOGGING_SETUP = True
+3 -1
View File
@@ -146,7 +146,9 @@ async def engine_session_factory(
_session_maker = None
async def run_migrations(app_config: ProjectConfig, database_type=DatabaseType.FILESYSTEM):
async def run_migrations(
app_config: ProjectConfig, database_type=DatabaseType.FILESYSTEM
): # pragma: no cover
"""Run any pending alembic migrations."""
logger.info("Running database migrations...")
try:
+3
View File
@@ -104,6 +104,9 @@ def has_frontmatter(content: str) -> bool:
Returns:
True if content has valid frontmatter markers (---), False otherwise
"""
if not content:
return False
content = content.strip()
if not content.startswith("---"):
return False
+17 -8
View File
@@ -92,27 +92,36 @@ class EntityParser:
async def parse_file(self, path: Path | str) -> EntityMarkdown:
"""Parse markdown file into EntityMarkdown."""
absolute_path = self.base_path / path
# Parse frontmatter and content using python-frontmatter
post = frontmatter.load(str(absolute_path))
# TODO move to api endpoint to check if absolute path was requested
# Check if the path is already absolute
if (
isinstance(path, Path)
and path.is_absolute()
or (isinstance(path, str) and Path(path).is_absolute())
):
absolute_path = Path(path)
else:
absolute_path = self.base_path / path
# Parse frontmatter and content using python-frontmatter
file_content = absolute_path.read_text(encoding="utf-8")
return await self.parse_file_content(absolute_path, file_content)
async def parse_file_content(self, absolute_path, file_content):
post = frontmatter.loads(file_content)
# Extract file stat info
file_stats = absolute_path.stat()
metadata = post.metadata
metadata["title"] = post.metadata.get("title", absolute_path.name)
metadata["title"] = post.metadata.get("title", absolute_path.stem)
metadata["type"] = post.metadata.get("type", "note")
tags = parse_tags(post.metadata.get("tags", [])) # pyright: ignore
if tags:
metadata["tags"] = tags
# frontmatter
entity_frontmatter = EntityFrontmatter(
metadata=post.metadata,
)
entity_content = parse(post.content)
return EntityMarkdown(
frontmatter=entity_frontmatter,
content=post.content,
+21 -13
View File
@@ -1,14 +1,14 @@
"""Utilities for converting between markdown and entity models."""
from pathlib import Path
from typing import Optional, Any
from typing import Any, Optional
from frontmatter import Post
from basic_memory.file_utils import has_frontmatter, remove_frontmatter
from basic_memory.file_utils import has_frontmatter, remove_frontmatter, parse_frontmatter
from basic_memory.markdown import EntityMarkdown
from basic_memory.models import Entity, Observation as ObservationModel
from basic_memory.utils import generate_permalink
from basic_memory.models import Entity
from basic_memory.models import Observation as ObservationModel
def entity_model_from_markdown(
@@ -32,16 +32,13 @@ def entity_model_from_markdown(
if not markdown.created or not markdown.modified: # pragma: no cover
raise ValueError("Both created and modified dates are required in markdown")
# Generate permalink if not provided
permalink = markdown.frontmatter.permalink or generate_permalink(file_path)
# Create or update entity
model = entity or Entity()
# Update basic fields
model.title = markdown.frontmatter.title
model.entity_type = markdown.frontmatter.type
model.permalink = permalink
model.permalink = markdown.frontmatter.permalink
model.file_path = str(file_path)
model.content_type = "text/markdown"
model.created_at = markdown.created
@@ -77,22 +74,33 @@ async def schema_to_markdown(schema: Any) -> Post:
"""
# Extract content and metadata
content = schema.content or ""
frontmatter_metadata = dict(schema.entity_metadata or {})
entity_metadata = dict(schema.entity_metadata or {})
# if the content contains frontmatter, remove it and merge
if has_frontmatter(content):
content_frontmatter = parse_frontmatter(content)
content = remove_frontmatter(content)
# Merge content frontmatter with entity metadata
# (entity_metadata takes precedence for conflicts)
content_frontmatter.update(entity_metadata)
entity_metadata = content_frontmatter
# Remove special fields for ordered frontmatter
for field in ["type", "title", "permalink"]:
frontmatter_metadata.pop(field, None)
entity_metadata.pop(field, None)
# Create Post with ordered fields
# Create Post with fields ordered by insert order
post = Post(
content,
title=schema.title,
type=schema.entity_type,
permalink=schema.permalink,
**frontmatter_metadata,
)
# set the permalink if passed in
if schema.permalink:
post.metadata["permalink"] = schema.permalink
if entity_metadata:
post.metadata.update(entity_metadata)
return post
@@ -5,19 +5,19 @@ providing context from previous interactions to maintain continuity.
"""
from textwrap import dedent
from typing import Optional, Annotated
from typing import Annotated, Optional
from loguru import logger
from pydantic import Field
from basic_memory.mcp.prompts.utils import format_prompt_context, PromptContext, PromptContextItem
from basic_memory.mcp.prompts.utils import PromptContext, PromptContextItem, format_prompt_context
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.build_context import build_context
from basic_memory.mcp.tools.recent_activity import recent_activity
from basic_memory.mcp.tools.search import search_notes
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.memory import GraphContext
from basic_memory.schemas.search import SearchQuery, SearchItemType
from basic_memory.schemas.search import SearchItemType
@mcp.prompt(
@@ -48,7 +48,7 @@ async def continue_conversation(
# If topic provided, search for it
if topic:
search_results = await search_notes(
SearchQuery(text=topic, after_date=timeframe, types=[SearchItemType.ENTITY])
query=topic, after_date=timeframe, entity_types=[SearchItemType.ENTITY]
)
# Build context from results
+2 -2
View File
@@ -12,7 +12,7 @@ from pydantic import Field
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.search import search_notes as search_tool
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.search import SearchQuery, SearchResponse
from basic_memory.schemas.search import SearchResponse
@mcp.prompt(
@@ -40,7 +40,7 @@ async def search_prompt(
"""
logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}")
search_results = await search_tool(SearchQuery(text=query, after_date=timeframe))
search_results = await search_tool(query=query, after_date=timeframe)
return format_search_results(query, search_results, timeframe)
+29 -3
View File
@@ -1,11 +1,37 @@
"""Enhanced FastMCP server instance for Basic Memory."""
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator, Optional
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.utilities.logging import configure_logging
from mcp.server.fastmcp.utilities.logging import configure_logging as mcp_configure_logging
from dataclasses import dataclass
from basic_memory.config import config as project_config
from basic_memory.services.initialization import initialize_app
# mcp console logging
configure_logging(level="INFO")
mcp_configure_logging(level="ERROR")
@dataclass
class AppContext:
watch_task: Optional[asyncio.Task]
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: # pragma: no cover
"""Manage application lifecycle with type-safe context"""
# Initialize on startup
watch_task = await initialize_app(project_config)
try:
yield AppContext(watch_task=watch_task)
finally:
# Cleanup on shutdown
if watch_task:
watch_task.cancel()
# Create the shared server instance
mcp = FastMCP("Basic Memory")
mcp = FastMCP("Basic Memory", log_level="ERROR", lifespan=app_lifespan)
+4 -5
View File
@@ -9,7 +9,6 @@ from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.search import search_notes
from basic_memory.mcp.tools.utils import call_get
from basic_memory.schemas.memory import memory_url_path
from basic_memory.schemas.search import SearchQuery
@mcp.tool(
@@ -63,7 +62,7 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
# Fallback 1: Try title search via API
logger.info(f"Search title for: {identifier}")
title_results = await search_notes(SearchQuery(title=identifier))
title_results = await search_notes(query=identifier, search_type="title")
if title_results and title_results.results:
result = title_results.results[0] # Get the first/best match
@@ -87,7 +86,7 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
# Fallback 2: Text search as a last resort
logger.info(f"Title search failed, trying text search for: {identifier}")
text_results = await search_notes(SearchQuery(text=identifier))
text_results = await search_notes(query=identifier, search_type="text")
# We didn't find a direct match, construct a helpful error message
if not text_results or not text_results.results:
@@ -112,7 +111,7 @@ def format_not_found_message(identifier: str) -> str:
## Search Instead
Try searching for related content:
```
search(query="{identifier}")
search_notes(query="{identifier}")
```
## Recent Activity
@@ -173,7 +172,7 @@ def format_related_results(identifier: str, results) -> str:
## Search For More Results
To see more related content:
```
search(query="{identifier}")
search_notes(query="{identifier}")
```
## Create New Note
+37 -15
View File
@@ -1,6 +1,6 @@
"""Recent activity tool for Basic Memory MCP server."""
from typing import Optional, List
from typing import List, Union
from loguru import logger
@@ -14,7 +14,7 @@ from basic_memory.schemas.search import SearchItemType
@mcp.tool(
description="""Get recent activity from across the knowledge base.
Timeframe supports natural language formats like:
- "2 days ago"
- "last week"
@@ -25,9 +25,9 @@ from basic_memory.schemas.search import SearchItemType
""",
)
async def recent_activity(
type: Optional[List[SearchItemType]] = None,
depth: Optional[int] = 1,
timeframe: Optional[TimeFrame] = "7d",
type: Union[str, List[str]] = "",
depth: int = 1,
timeframe: TimeFrame = "7d",
page: int = 1,
page_size: int = 10,
max_related: int = 10,
@@ -35,11 +35,14 @@ async def recent_activity(
"""Get recent activity across the knowledge base.
Args:
type: Filter by content type(s). Valid options:
- ["entity"] for knowledge entities
- ["relation"] for connections between entities
- ["observation"] for notes and observations
type: Filter by content type(s). Can be a string or list of strings.
Valid options:
- "entity" or ["entity"] for knowledge entities
- "relation" or ["relation"] for connections between entities
- "observation" or ["observation"] for notes and observations
Multiple types can be combined: ["entity", "relation"]
Case-insensitive: "ENTITY" and "entity" are treated the same.
Default is an empty string, which returns all types.
depth: How many relation hops to traverse (1-3 recommended)
timeframe: Time window to search. Supports natural language:
- Relative: "2 days ago", "last week", "yesterday"
@@ -59,14 +62,17 @@ async def recent_activity(
# Get all entities for the last 10 days (default)
recent_activity()
# Get all entities from yesterday
# Get all entities from yesterday (string format)
recent_activity(type="entity", timeframe="yesterday")
# Get all entities from yesterday (list format)
recent_activity(type=["entity"], timeframe="yesterday")
# Get recent relations and observations
recent_activity(type=["relation", "observation"], timeframe="today")
# Look back further with more context
recent_activity(type=["entity"], depth=2, timeframe="2 weeks ago")
recent_activity(type="entity", depth=2, timeframe="2 weeks ago")
Notes:
- Higher depth values (>3) may impact performance with large result sets
@@ -86,11 +92,27 @@ async def recent_activity(
if timeframe:
params["timeframe"] = timeframe # pyright: ignore
# send enum values if we have an enum, else send string value
# Validate and convert type parameter
if type:
params["type"] = [ # pyright: ignore
type.value if isinstance(type, SearchItemType) else type for type in type
]
# Convert single string to list
if isinstance(type, str):
type_list = [type]
else:
type_list = type
# Validate each type against SearchItemType enum
validated_types = []
for t in type_list:
try:
# Try to convert string to enum
if isinstance(t, str):
validated_types.append(SearchItemType(t.lower()))
except ValueError:
valid_types = [t.value for t in SearchItemType]
raise ValueError(f"Invalid type: {t}. Valid types are: {valid_types}")
# Add validated types to params
params["type"] = [t.value for t in validated_types] # pyright: ignore
response = await call_get(
client,
+64 -28
View File
@@ -1,17 +1,27 @@
"""Search tools for Basic Memory MCP server."""
from typing import List, Optional
from loguru import logger
from basic_memory.mcp.async_client import client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_post
from basic_memory.schemas.search import SearchQuery, SearchResponse
from basic_memory.mcp.async_client import client
from basic_memory.schemas.search import SearchItemType, SearchQuery, SearchResponse
@mcp.tool(
description="Search across all content in the knowledge base.",
)
async def search_notes(query: SearchQuery, page: int = 1, page_size: int = 10) -> SearchResponse:
async def search_notes(
query: str,
page: int = 1,
page_size: int = 10,
search_type: str = "text",
types: Optional[List[str]] = None,
entity_types: Optional[List[str]] = None,
after_date: Optional[str] = None,
) -> SearchResponse:
"""Search across all content in the knowledge base.
This tool searches the knowledge base using full-text search, pattern matching,
@@ -19,59 +29,85 @@ async def search_notes(query: SearchQuery, page: int = 1, page_size: int = 10) -
and date.
Args:
query: SearchQuery object with search parameters including:
- text: Full-text search (e.g., "project planning")
Supports boolean operators: AND, OR, NOT and parentheses for grouping
- title: Search only in titles (e.g., "Meeting notes")
- permalink: Exact permalink match (e.g., "docs/meeting-notes")
- permalink_match: Pattern matching for permalinks (e.g., "docs/*-notes")
- types: Optional list of content types to search (e.g., ["entity", "observation"])
- entity_types: Optional list of entity types to filter by (e.g., ["note", "person"])
- after_date: Optional date filter for recent content (e.g., "1 week", "2d")
query: The search query string
page: The page number of results to return (default 1)
page_size: The number of results to return per page (default 10)
search_type: Type of search to perform, one of: "text", "title", "permalink" (default: "text")
types: Optional list of note types to search (e.g., ["note", "person"])
entity_types: Optional list of entity types to filter by (e.g., ["entity", "observation"])
after_date: Optional date filter for recent content (e.g., "1 week", "2d")
Returns:
SearchResponse with results and pagination info
Examples:
# Basic text search
results = await search_notes(SearchQuery(text="project planning"))
results = await search_notes("project planning")
# Boolean AND search (both terms must be present)
results = await search_notes(SearchQuery(text="project AND planning"))
results = await search_notes("project AND planning")
# Boolean OR search (either term can be present)
results = await search_notes(SearchQuery(text="project OR meeting"))
results = await search_notes("project OR meeting")
# Boolean NOT search (exclude terms)
results = await search_notes(SearchQuery(text="project NOT meeting"))
results = await search_notes("project NOT meeting")
# Boolean search with grouping
results = await search_notes(SearchQuery(text="(project OR planning) AND notes"))
results = await search_notes("(project OR planning) AND notes")
# Search with type filter
results = await search_notes(SearchQuery(
text="meeting notes",
results = await search_notes(
query="meeting notes",
types=["entity"],
))
)
# Search with entity type filter, e.g., note vs
results = await search_notes(
query="meeting notes",
types=["entity"],
)
# Search for recent content
results = await search_notes(SearchQuery(
text="bug report",
results = await search_notes(
query="bug report",
after_date="1 week"
))
)
# Pattern matching on permalinks
results = await search_notes(SearchQuery(
permalink_match="docs/meeting-*"
))
results = await search_notes(
query="docs/meeting-*",
search_type="permalink"
)
"""
logger.info(f"Searching for {query}")
# Create a SearchQuery object based on the parameters
search_query = SearchQuery()
# Set the appropriate search field based on search_type
if search_type == "text":
search_query.text = query
elif search_type == "title":
search_query.title = query
elif search_type == "permalink" and "*" in query:
search_query.permalink_match = query
elif search_type == "permalink":
search_query.permalink = query
else:
search_query.text = query # Default to text search
# Add optional filters if provided
if entity_types:
search_query.entity_types = [SearchItemType(t) for t in entity_types]
if types:
search_query.types = types
if after_date:
search_query.after_date = after_date
logger.info(f"Searching for {search_query}")
response = await call_post(
client,
"/search/",
json=query.model_dump(),
json=search_query.model_dump(),
params={"page": page, "page_size": page_size},
)
return SearchResponse.model_validate(response.json())
+3 -1
View File
@@ -88,8 +88,10 @@ async def write_note(
# Format semantic summary based on status code
action = "Created" if response.status_code == 201 else "Updated"
summary = [
f"# {action} {result.file_path} ({result.checksum[:8] if result.checksum else 'unknown'})",
f"# {action} note",
f"file_path: {result.file_path}",
f"permalink: {result.permalink}",
f"checksum: {result.checksum[:8] if result.checksum else 'unknown'}",
]
# Count observations by category
@@ -137,8 +137,6 @@ class Repository[T: Base]:
async def find_one(self, query: Select[tuple[T]]) -> Optional[T]:
"""Execute a query and retrieve a single record."""
logger.debug(f"Finding one {self.Model.__name__} with query: {query}")
# add in load options
query = query.options(*self.get_load_options())
result = await self.execute_query(query)
@@ -270,11 +268,9 @@ class Repository[T: Base]:
"""Execute a query asynchronously."""
query = query.options(*self.get_load_options()) if use_query_options else query
logger.debug(f"Executing query: {query}")
async with db.scoped_session(self.session_maker) as session:
result = await session.execute(query)
logger.debug("Query executed successfully")
return result
def get_load_options(self) -> List[LoaderOption]:
@@ -4,10 +4,10 @@ import json
import time
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional, Any, Dict
from typing import Any, Dict, List, Optional
from loguru import logger
from sqlalchemy import text, Executable, Result
from sqlalchemy import Executable, Result, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from basic_memory import db
@@ -123,9 +123,9 @@ class SearchRepository:
permalink: Optional[str] = None,
permalink_match: Optional[str] = None,
title: Optional[str] = None,
types: Optional[List[SearchItemType]] = None,
types: Optional[List[str]] = None,
after_date: Optional[datetime] = None,
entity_types: Optional[List[str]] = None,
entity_types: Optional[List[SearchItemType]] = None,
limit: int = 10,
offset: int = 0,
) -> List[SearchIndexRow]:
@@ -174,15 +174,15 @@ class SearchRepository:
else:
conditions.append("permalink MATCH :permalink")
# Handle type filter
if types:
type_list = ", ".join(f"'{t.value}'" for t in types)
conditions.append(f"type IN ({type_list})")
# Handle entity type filter
if entity_types:
entity_type_list = ", ".join(f"'{t}'" for t in entity_types)
conditions.append(f"json_extract(metadata, '$.entity_type') IN ({entity_type_list})")
type_list = ", ".join(f"'{t.value}'" for t in entity_types)
conditions.append(f"type IN ({type_list})")
# Handle type filter
if types:
type_list = ", ".join(f"'{t}'" for t in types)
conditions.append(f"json_extract(metadata, '$.entity_type') IN ({type_list})")
# Handle date filter using datetime() for proper comparison
if after_date:
+2 -2
View File
@@ -49,8 +49,8 @@ class SearchQuery(BaseModel):
title: Optional[str] = None # title only search
# Optional filters
types: Optional[List[SearchItemType]] = None # Filter by item type
entity_types: Optional[List[str]] = None # Filter by entity type
types: Optional[List[str]] = None # Filter by type
entity_types: Optional[List[SearchItemType]] = None # Filter by entity type
after_date: Optional[Union[datetime, str]] = None # Time-based filter
@field_validator("after_date")
+1 -1
View File
@@ -81,7 +81,7 @@ class ContextService:
else:
logger.debug(f"Build context for '{types}'")
primary = await self.search_repository.search(
types=types, after_date=since, limit=limit, offset=offset
entity_types=types, after_date=since, limit=limit, offset=offset
)
# Get type_id pairs for traversal
+10 -10
View File
@@ -1,24 +1,24 @@
"""Service for managing entities in the database."""
from pathlib import Path
from typing import Sequence, List, Optional, Tuple, Union
from typing import List, Optional, Sequence, Tuple, Union
import frontmatter
from loguru import logger
from sqlalchemy.exc import IntegrityError
from basic_memory.markdown import EntityMarkdown
from basic_memory.markdown.entity_parser import EntityParser
from basic_memory.markdown.utils import entity_model_from_markdown, schema_to_markdown
from basic_memory.models import Entity as EntityModel, Observation, Relation
from basic_memory.models import Entity as EntityModel
from basic_memory.models import Observation, Relation
from basic_memory.repository import ObservationRepository, RelationRepository
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.schemas import Entity as EntitySchema
from basic_memory.schemas.base import Permalink
from basic_memory.services.exceptions import EntityNotFoundError, EntityCreationError
from basic_memory.services import FileService
from basic_memory.services import BaseService
from basic_memory.services import BaseService, FileService
from basic_memory.services.exceptions import EntityCreationError, EntityNotFoundError
from basic_memory.services.link_resolver import LinkResolver
from basic_memory.markdown.entity_parser import EntityParser
from basic_memory.utils import generate_permalink
@@ -89,7 +89,7 @@ class EntityService(BaseService[EntityModel]):
logger.debug(f"Creating or updating entity: {schema}")
# Try to find existing entity using smart resolution
existing = await self.link_resolver.resolve_link(schema.permalink)
existing = await self.link_resolver.resolve_link(schema.permalink or schema.file_path)
if existing:
logger.debug(f"Found existing entity: {existing.permalink}")
@@ -100,7 +100,7 @@ class EntityService(BaseService[EntityModel]):
async def create_entity(self, schema: EntitySchema) -> EntityModel:
"""Create a new entity and write to filesystem."""
logger.debug(f"Creating entity: {schema.permalink}")
logger.debug(f"Creating entity: {schema.title}")
# Get file path and ensure it's a Path object
file_path = Path(schema.file_path)
@@ -230,7 +230,7 @@ class EntityService(BaseService[EntityModel]):
Creates the entity with null checksum to indicate sync not complete.
Relations will be added in second pass.
"""
logger.debug(f"Creating entity: {markdown.frontmatter.title}")
logger.debug(f"Creating entity: {markdown.frontmatter.title} file_path: {file_path}")
model = entity_model_from_markdown(file_path, markdown)
# Mark as incomplete because we still need to add relations
@@ -315,7 +315,7 @@ class EntityService(BaseService[EntityModel]):
except IntegrityError:
# Unique constraint violation - relation already exists
logger.debug(
f"Skipping duplicate relation {rel.type} from {db_entity.permalink} target: {rel.target}, type: {rel.type}"
f"Skipping duplicate relation {rel.type} from {db_entity.permalink} target: {rel.target}"
)
continue
+1 -1
View File
@@ -60,7 +60,7 @@ class FileService:
Returns:
Raw content string without metadata sections
"""
logger.debug("Reading entity content", entity_id=entity.id, permalink=entity.permalink)
logger.debug(f"Reading entity content, entity_id={entity.id}, permalink={entity.permalink}")
file_path = self.get_entity_path(entity)
markdown = await self.markdown_processor.read_file(file_path)
+143
View File
@@ -0,0 +1,143 @@
"""Shared initialization service for Basic Memory.
This module provides shared initialization functions used by both CLI and API
to ensure consistent application startup across all entry points.
"""
import asyncio
from typing import Optional
from loguru import logger
from basic_memory import db
from basic_memory.config import ProjectConfig, config_manager
from basic_memory.sync import WatchService
# Import this inside functions to avoid circular imports
# from basic_memory.cli.commands.sync import get_sync_service
async def initialize_database(app_config: ProjectConfig) -> None:
"""Run database migrations to ensure schema is up to date.
Args:
app_config: The Basic Memory project configuration
"""
try:
logger.info("Running database migrations...")
await db.run_migrations(app_config)
logger.info("Migrations completed successfully")
except Exception as e:
logger.error(f"Error running migrations: {e}")
# Allow application to continue - it might still work
# depending on what the error was, and will fail with a
# more specific error if the database is actually unusable
async def initialize_file_sync(
app_config: ProjectConfig,
) -> asyncio.Task:
"""Initialize file synchronization services.
Args:
app_config: The Basic Memory project configuration
Returns:
Tuple of (sync_service, watch_service, watch_task) if sync is enabled,
or (None, None, None) if sync is disabled
"""
# Load app configuration
# Import here to avoid circular imports
from basic_memory.cli.commands.sync import get_sync_service
# Initialize sync service
sync_service = await get_sync_service()
# Initialize watch service
watch_service = WatchService(
sync_service=sync_service,
file_service=sync_service.entity_service.file_service,
config=app_config,
quiet=True,
)
# Create the background task for running sync
async def run_background_sync(): # pragma: no cover
# Run initial full sync
await sync_service.sync(app_config.home)
logger.info("Sync completed successfully")
# Start background sync task
logger.info(f"Starting watch service to sync file changes in dir: {app_config.home}")
# Start watching for changes
await watch_service.run()
watch_task = asyncio.create_task(run_background_sync())
logger.info("Watch service started")
return watch_task
async def initialize_app(
app_config: ProjectConfig,
) -> Optional[asyncio.Task]:
"""Initialize the Basic Memory application.
This function handles all initialization steps needed for both API and shor lived CLI commands.
For long running commands like mcp, a
- Running database migrations
- Setting up file synchronization
Args:
app_config: The Basic Memory project configuration
"""
# Initialize database first
await initialize_database(app_config)
basic_memory_config = config_manager.load_config()
logger.info(f"Sync changes enabled: {basic_memory_config.sync_changes}")
logger.info(
f"Update permalinks on move enabled: {basic_memory_config.update_permalinks_on_move}"
)
if not basic_memory_config.sync_changes: # pragma: no cover
logger.info("Sync changes disabled. Skipping watch service.")
return
# Initialize file sync services
return await initialize_file_sync(app_config)
def ensure_initialization(app_config: ProjectConfig) -> None:
"""Ensure initialization runs in a synchronous context.
This is a wrapper for the async initialize_app function that can be
called from synchronous code like CLI entry points.
Args:
app_config: The Basic Memory project configuration
"""
try:
asyncio.run(initialize_app(app_config))
except Exception as e:
logger.error(f"Error during initialization: {e}")
# Continue execution even if initialization fails
# The command might still work, or will fail with a
# more specific error message
def ensure_initialize_database(app_config: ProjectConfig) -> None:
"""Ensure initialization runs in a synchronous context.
This is a wrapper for the async initialize_database function that can be
called from synchronous code like CLI entry points.
Args:
app_config: The Basic Memory project configuration
"""
try:
asyncio.run(initialize_database(app_config))
except Exception as e:
logger.error(f"Error during initialization: {e}")
# Continue execution even if initialization fails
# The command might still work, or will fail with a
# more specific error message
+8 -1
View File
@@ -46,10 +46,17 @@ class LinkResolver:
logger.debug(f"Found title match: {entity.title}")
return entity
# 3. Try file path
found_path = await self.entity_repository.get_by_file_path(clean_text)
if found_path:
logger.debug(f"Found entity with path: {found_path.file_path}")
return found_path
# search if indicated
if use_search and "*" not in clean_text:
# 3. Fall back to search for fuzzy matching on title
results = await self.search_service.search(
query=SearchQuery(title=clean_text, types=[SearchItemType.ENTITY]),
query=SearchQuery(title=clean_text, entity_types=[SearchItemType.ENTITY]),
)
if results:
+3 -23
View File
@@ -181,17 +181,6 @@ class SearchService:
Each type gets its own row in the search index with appropriate metadata.
"""
if entity.permalink is None: # pragma: no cover
logger.error(
"Missing permalink for markdown entity",
entity_id=entity.id,
title=entity.title,
file_path=entity.file_path,
)
raise ValueError(
f"Entity permalink should not be None for markdown entity: {entity.id} ({entity.title})"
)
content_stems = []
content_snippet = ""
title_variants = self._generate_variants(entity.title)
@@ -202,22 +191,13 @@ class SearchService:
content_stems.append(content)
content_snippet = f"{content[:250]}"
content_stems.extend(self._generate_variants(entity.permalink))
if entity.permalink:
content_stems.extend(self._generate_variants(entity.permalink))
content_stems.extend(self._generate_variants(entity.file_path))
entity_content_stems = "\n".join(p for p in content_stems if p and p.strip())
if entity.permalink is None: # pragma: no cover
logger.error(
"Missing permalink for markdown entity",
entity_id=entity.id,
title=entity.title,
file_path=entity.file_path,
)
raise ValueError(
f"Entity permalink should not be None for markdown entity: {entity.id} ({entity.title})"
)
# Index entity
await self.repository.index_item(
SearchIndexRow(
+122 -195
View File
@@ -1,9 +1,8 @@
"""Service for syncing files between filesystem and database."""
import os
from dataclasses import dataclass
from dataclasses import field
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, Optional, Set, Tuple
@@ -11,13 +10,13 @@ from typing import Dict, Optional, Set, Tuple
from loguru import logger
from sqlalchemy.exc import IntegrityError
from basic_memory.config import ProjectConfig
from basic_memory.file_utils import has_frontmatter
from basic_memory.markdown import EntityParser
from basic_memory.models import Entity
from basic_memory.repository import EntityRepository, RelationRepository
from basic_memory.services import EntityService, FileService
from basic_memory.services.search_service import SearchService
import time
from rich.progress import Progress, TextColumn, BarColumn, TaskProgressColumn
@dataclass
@@ -65,6 +64,7 @@ class SyncService:
def __init__(
self,
config: ProjectConfig,
entity_service: EntityService,
entity_parser: EntityParser,
entity_repository: EntityRepository,
@@ -72,6 +72,7 @@ class SyncService:
search_service: SearchService,
file_service: FileService,
):
self.config = config
self.entity_service = entity_service
self.entity_parser = entity_parser
self.entity_repository = entity_repository
@@ -79,145 +80,51 @@ class SyncService:
self.search_service = search_service
self.file_service = file_service
async def sync(self, directory: Path, show_progress: bool = True) -> SyncReport:
async def sync(self, directory: Path) -> SyncReport:
"""Sync all files with database."""
start_time = time.time()
console = None
progress = None # Will be initialized if show_progress is True
logger.info("Sync operation started", directory=str(directory))
logger.info(f"Sync operation started for directory: {directory}")
# initial paths from db to sync
# path -> checksum
if show_progress:
from rich.console import Console
console = Console()
console.print(f"Scanning directory: {directory}")
report = await self.scan(directory)
# Initialize progress tracking if requested
if show_progress and report.total > 0:
progress = Progress(
TextColumn("[bold blue]{task.description}"),
BarColumn(),
TaskProgressColumn(),
console=console,
expand=True,
)
# order of sync matters to resolve relations effectively
logger.info(
"Sync changes detected",
new_files=len(report.new),
modified_files=len(report.modified),
deleted_files=len(report.deleted),
moved_files=len(report.moves),
f"Sync changes detected: new_files={len(report.new)}, modified_files={len(report.modified)}, "
+ f"deleted_files={len(report.deleted)}, moved_files={len(report.moves)}"
)
if show_progress and report.total > 0:
with progress: # pyright: ignore
# Track each category separately
move_task = None
if report.moves: # pragma: no cover
move_task = progress.add_task("[blue]Moving files...", total=len(report.moves)) # pyright: ignore
delete_task = None
if report.deleted: # pragma: no cover
delete_task = progress.add_task( # pyright: ignore
"[red]Deleting files...", total=len(report.deleted)
# sync moves first
for old_path, new_path in report.moves.items():
# in the case where a file has been deleted and replaced by another file
# it will show up in the move and modified lists, so handle it in modified
if new_path in report.modified:
report.modified.remove(new_path)
logger.debug(
f"File marked as moved and modified: old_path={old_path}, new_path={new_path}"
)
else:
await self.handle_move(old_path, new_path)
new_task = None
if report.new:
new_task = progress.add_task( # pyright: ignore
"[green]Adding new files...", total=len(report.new)
)
# deleted next
for path in report.deleted:
await self.handle_delete(path)
modify_task = None
if report.modified: # pragma: no cover
modify_task = progress.add_task( # pyright: ignore
"[yellow]Updating modified files...", total=len(report.modified)
)
# then new and modified
for path in report.new:
await self.sync_file(path, new=True)
# sync moves first
for i, (old_path, new_path) in enumerate(report.moves.items()):
# in the case where a file has been deleted and replaced by another file
# it will show up in the move and modified lists, so handle it in modified
if new_path in report.modified: # pragma: no cover
report.modified.remove(new_path)
logger.debug(
"File marked as moved and modified",
old_path=old_path,
new_path=new_path,
action="processing as modified",
)
else: # pragma: no cover
await self.handle_move(old_path, new_path)
for path in report.modified:
await self.sync_file(path, new=False)
if move_task is not None: # pragma: no cover
progress.update(move_task, advance=1) # pyright: ignore
# deleted next
for i, path in enumerate(report.deleted): # pragma: no cover
await self.handle_delete(path)
if delete_task is not None: # pragma: no cover
progress.update(delete_task, advance=1) # pyright: ignore
# then new and modified
for i, path in enumerate(report.new):
await self.sync_file(path, new=True)
if new_task is not None:
progress.update(new_task, advance=1) # pyright: ignore
for i, path in enumerate(report.modified): # pragma: no cover
await self.sync_file(path, new=False)
if modify_task is not None: # pragma: no cover
progress.update(modify_task, advance=1) # pyright: ignore
# Final step - resolving relations
if report.total > 0:
relation_task = progress.add_task("[cyan]Resolving relations...", total=1) # pyright: ignore
await self.resolve_relations()
progress.update(relation_task, advance=1) # pyright: ignore
else:
# No progress display - proceed with normal sync
# sync moves first
for old_path, new_path in report.moves.items():
# in the case where a file has been deleted and replaced by another file
# it will show up in the move and modified lists, so handle it in modified
if new_path in report.modified:
report.modified.remove(new_path)
logger.debug(
"File marked as moved and modified",
old_path=old_path,
new_path=new_path,
action="processing as modified",
)
else:
await self.handle_move(old_path, new_path)
# deleted next
for path in report.deleted:
await self.handle_delete(path)
# then new and modified
for path in report.new:
await self.sync_file(path, new=True)
for path in report.modified:
await self.sync_file(path, new=False)
await self.resolve_relations()
await self.resolve_relations()
duration_ms = int((time.time() - start_time) * 1000)
logger.info(
"Sync operation completed",
directory=str(directory),
total_changes=report.total,
duration_ms=duration_ms,
f"Sync operation completed: directory={directory}, total_changes={report.total}, duration_ms={duration_ms}"
)
return report
@@ -226,6 +133,7 @@ class SyncService:
"""Scan directory for changes compared to database state."""
db_paths = await self.get_db_file_state()
logger.debug(f"Found {len(db_paths)} db paths")
# Track potentially moved files by checksum
scan_result = await self.scan_directory(directory)
@@ -276,6 +184,7 @@ class SyncService:
:param db_records: the data from the db
"""
db_records = await self.entity_repository.find_all()
logger.info(f"Found {len(db_records)} db records")
return {r.file_path: r.checksum or "" for r in db_records}
async def sync_file(
@@ -292,10 +201,7 @@ class SyncService:
"""
try:
logger.debug(
"Syncing file",
path=path,
is_new=new,
is_markdown=self.file_service.is_markdown(path),
f"Syncing file path={path} is_new={new} is_markdown={self.file_service.is_markdown(path)}"
)
if self.file_service.is_markdown(path):
@@ -307,7 +213,7 @@ class SyncService:
await self.search_service.index_entity(entity)
logger.debug(
"File sync completed", path=path, entity_id=entity.id, checksum=checksum
f"File sync completed, path={path}, entity_id={entity.id}, checksum={checksum[:8]}"
)
return entity, checksum
@@ -326,54 +232,59 @@ class SyncService:
Tuple of (entity, checksum)
"""
# Parse markdown first to get any existing permalink
logger.debug("Parsing markdown file", path=path)
logger.debug(f"Parsing markdown file, path: {path}, new: {new}")
file_path = self.entity_parser.base_path / path
file_content = file_path.read_text(encoding="utf-8")
file_contains_frontmatter = has_frontmatter(file_content)
# 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)
# Resolve permalink - this handles all the cases including conflicts
permalink = await self.entity_service.resolve_permalink(path, markdown=entity_markdown)
# if the file contains frontmatter, resolve a permalink
if file_contains_frontmatter:
# Resolve permalink - this handles all the cases including conflicts
permalink = await self.entity_service.resolve_permalink(path, markdown=entity_markdown)
# If permalink changed, update the file
if permalink != entity_markdown.frontmatter.permalink:
logger.info(
"Updating permalink",
path=path,
old_permalink=entity_markdown.frontmatter.permalink,
new_permalink=permalink,
)
# If permalink changed, update the file
if permalink != entity_markdown.frontmatter.permalink:
logger.info(
f"Updating permalink for path: {path}, old_permalink: {entity_markdown.frontmatter.permalink}, new_permalink: {permalink}"
)
entity_markdown.frontmatter.metadata["permalink"] = permalink
checksum = await self.file_service.update_frontmatter(path, {"permalink": permalink})
else:
checksum = await self.file_service.compute_checksum(path)
entity_markdown.frontmatter.metadata["permalink"] = permalink
await self.file_service.update_frontmatter(path, {"permalink": permalink})
# if the file is new, create an entity
if new:
# Create entity with final permalink
logger.debug("Creating new entity from markdown", path=path, permalink=permalink)
logger.debug(f"Creating new entity from markdown, path={path}")
await self.entity_service.create_entity_from_markdown(Path(path), entity_markdown)
# otherwise we need to update the entity and observations
else:
logger.debug("Updating entity from markdown", path=path, permalink=permalink)
logger.debug(f"Updating entity from markdown, path={path}")
await self.entity_service.update_entity_and_observations(Path(path), entity_markdown)
# Update relations and search index
entity = await self.entity_service.update_entity_relations(path, entity_markdown)
# After updating relations, we need to compute the checksum again
# This is necessary for files with wikilinks to ensure consistent checksums
# after relation processing is complete
final_checksum = await self.file_service.compute_checksum(path)
# set checksum
await self.entity_repository.update(entity.id, {"checksum": checksum})
await self.entity_repository.update(entity.id, {"checksum": final_checksum})
logger.debug(
"Markdown sync completed",
path=path,
entity_id=entity.id,
observation_count=len(entity.observations),
relation_count=len(entity.relations),
f"Markdown sync completed: path={path}, entity_id={entity.id}, "
f"observation_count={len(entity.observations)}, relation_count={len(entity.relations)}, "
f"checksum={final_checksum[:8]}"
)
return entity, checksum
# Return the final checksum to ensure everything is consistent
return entity, final_checksum
async def sync_regular_file(self, path: str, new: bool = True) -> Tuple[Optional[Entity], str]:
"""Sync a non-markdown file with basic tracking.
@@ -414,7 +325,7 @@ class SyncService:
else:
entity = await self.entity_repository.get_by_file_path(path)
if entity is None: # pragma: no cover
logger.error("Entity not found for existing file", path=path)
logger.error(f"Entity not found for existing file, path={path}")
raise ValueError(f"Entity not found for existing file: {path}")
updated = await self.entity_repository.update(
@@ -422,7 +333,7 @@ class SyncService:
)
if updated is None: # pragma: no cover
logger.error("Failed to update entity", entity_id=entity.id, path=path)
logger.error(f"Failed to update entity, entity_id={entity.id}, path={path}")
raise ValueError(f"Failed to update entity with ID {entity.id}")
return updated, checksum
@@ -434,10 +345,7 @@ class SyncService:
entity = await self.entity_repository.get_by_file_path(file_path)
if entity:
logger.info(
"Deleting entity",
file_path=file_path,
entity_id=entity.id,
permalink=entity.permalink,
f"Deleting entity with file_path={file_path}, entity_id={entity.id}, permalink={entity.permalink}"
)
# Delete from db (this cascades to observations/relations)
@@ -451,10 +359,8 @@ class SyncService:
)
logger.debug(
"Cleaning up search index",
entity_id=entity.id,
file_path=file_path,
index_entries=len(permalinks),
f"Cleaning up search index for entity_id={entity.id}, file_path={file_path}, "
f"index_entries={len(permalinks)}"
)
for permalink in permalinks:
@@ -464,28 +370,49 @@ class SyncService:
await self.search_service.delete_by_entity_id(entity.id)
async def handle_move(self, old_path, new_path):
logger.info("Moving entity", old_path=old_path, new_path=new_path)
logger.debug("Moving entity", old_path=old_path, new_path=new_path)
entity = await self.entity_repository.get_by_file_path(old_path)
if entity:
# Update file_path but keep the same permalink for link stability
updated = await self.entity_repository.update(entity.id, {"file_path": new_path})
# Update file_path in all cases
updates = {"file_path": new_path}
# If configured, also update permalink to match new path
if self.config.update_permalinks_on_move:
# generate new permalink value
new_permalink = await self.entity_service.resolve_permalink(new_path)
# write to file and get new checksum
new_checksum = await self.file_service.update_frontmatter(
new_path, {"permalink": new_permalink}
)
updates["permalink"] = new_permalink
updates["checksum"] = new_checksum
logger.info(
f"Updating permalink on move,old_permalink={entity.permalink}"
f"new_permalink={new_permalink}"
f"new_checksum={new_checksum}"
)
updated = await self.entity_repository.update(entity.id, updates)
if updated is None: # pragma: no cover
logger.error(
"Failed to update entity path",
entity_id=entity.id,
old_path=old_path,
new_path=new_path,
"Failed to update entity path"
f"entity_id={entity.id}"
f"old_path={old_path}"
f"new_path={new_path}"
)
raise ValueError(f"Failed to update entity path for ID {entity.id}")
logger.debug(
"Entity path updated",
entity_id=entity.id,
permalink=entity.permalink,
old_path=old_path,
new_path=new_path,
"Entity path updated"
f"entity_id={entity.id} "
f"permalink={entity.permalink} "
f"old_path={old_path} "
f"new_path={new_path} "
)
# update search index
@@ -500,10 +427,10 @@ class SyncService:
for relation in unresolved_relations:
logger.debug(
"Attempting to resolve relation",
relation_id=relation.id,
from_id=relation.from_id,
to_name=relation.to_name,
"Attempting to resolve relation "
f"relation_id={relation.id} "
f"from_id={relation.from_id} "
f"to_name={relation.to_name}"
)
resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name)
@@ -511,12 +438,12 @@ class SyncService:
# ignore reference to self
if resolved_entity and resolved_entity.id != relation.from_id:
logger.debug(
"Resolved forward reference",
relation_id=relation.id,
from_id=relation.from_id,
to_name=relation.to_name,
resolved_id=resolved_entity.id,
resolved_title=resolved_entity.title,
"Resolved forward reference "
f"relation_id={relation.id} "
f"from_id={relation.from_id} "
f"to_name={relation.to_name} "
f"resolved_id={resolved_entity.id} "
f"resolved_title={resolved_entity.title}",
)
try:
await self.relation_repository.update(
@@ -528,10 +455,10 @@ class SyncService:
)
except IntegrityError: # pragma: no cover
logger.debug(
"Ignoring duplicate relation",
relation_id=relation.id,
from_id=relation.from_id,
to_name=relation.to_name,
"Ignoring duplicate relation "
f"relation_id={relation.id} "
f"from_id={relation.from_id} "
f"to_name={relation.to_name}"
)
# update search index
@@ -549,7 +476,7 @@ class SyncService:
"""
start_time = time.time()
logger.debug("Scanning directory", directory=str(directory))
logger.debug(f"Scanning directory {directory}")
result = ScanResult()
for root, dirnames, filenames in os.walk(str(directory)):
@@ -571,10 +498,10 @@ class SyncService:
duration_ms = int((time.time() - start_time) * 1000)
logger.debug(
"Directory scan completed",
directory=str(directory),
files_found=len(result.files),
duration_ms=duration_ms,
f"{directory} scan completed "
f"directory={str(directory)} "
f"files_found={len(result.files)} "
f"duration_ms={duration_ms}"
)
return result
+49 -30
View File
@@ -70,22 +70,30 @@ class WatchServiceState(BaseModel):
class WatchService:
def __init__(self, sync_service: SyncService, file_service: FileService, config: ProjectConfig):
def __init__(
self,
sync_service: SyncService,
file_service: FileService,
config: ProjectConfig,
quiet: bool = False,
):
self.sync_service = sync_service
self.file_service = file_service
self.config = config
self.state = WatchServiceState()
self.status_path = config.home / ".basic-memory" / WATCH_STATUS_JSON
self.status_path.parent.mkdir(parents=True, exist_ok=True)
self.console = Console()
# quiet mode for mcp so it doesn't mess up stdout
self.console = Console(quiet=quiet)
async def run(self): # pragma: no cover
"""Watch for file changes and sync them"""
logger.info(
"Watch service started",
directory=str(self.config.home),
debounce_ms=self.config.sync_delay,
pid=os.getpid(),
f"directory={str(self.config.home)}",
f"debounce_ms={self.config.sync_delay}",
f"pid={os.getpid()}",
)
self.state.running = True
@@ -111,8 +119,8 @@ class WatchService:
finally:
logger.info(
"Watch service stopped",
directory=str(self.config.home),
runtime_seconds=int((datetime.now() - self.state.start_time).total_seconds()),
f"directory={str(self.config.home)}",
f"runtime_seconds={int((datetime.now() - self.state.start_time).total_seconds())}",
)
self.state.running = False
@@ -154,7 +162,7 @@ class WatchService:
start_time = time.time()
logger.info("Processing file changes", change_count=len(changes), directory=str(directory))
logger.info(f"Processing file changes, change_count={len(changes)}, directory={directory}")
# Group changes by type
adds: List[str] = []
@@ -177,9 +185,17 @@ class WatchService:
modifies.append(relative_path)
logger.debug(
"Grouped file changes", added=len(adds), deleted=len(deletes), modified=len(modifies)
f"Grouped file changes, added={len(adds)}, deleted={len(deletes)}, modified={len(modifies)}"
)
# because of our atomic writes on updates, an add may be an existing file
for added_path in adds: # pragma: no cover TODO add test
entity = await self.sync_service.entity_repository.get_by_file_path(added_path)
if entity is not None:
logger.debug(f"Existing file will be processed as modified, path={added_path}")
adds.remove(added_path)
modifies.append(added_path)
# Track processed files to avoid duplicates
processed: Set[str] = set()
@@ -223,15 +239,16 @@ class WatchService:
status="success",
)
self.console.print(f"[blue]→[/blue] {deleted_path}{added_path}")
logger.info(f"move: {deleted_path} -> {added_path}")
processed.add(added_path)
processed.add(deleted_path)
break
except Exception as e: # pragma: no cover
logger.warning(
"Error checking for move",
old_path=deleted_path,
new_path=added_path,
error=str(e),
f"old_path={deleted_path}",
f"new_path={added_path}",
f"error={str(e)}",
)
# Handle remaining changes - group them by type for concise output
@@ -247,6 +264,7 @@ class WatchService:
await self.sync_service.handle_delete(path)
self.state.add_event(path=path, action="deleted", status="success")
self.console.print(f"[red]✕[/red] {path}")
logger.info(f"deleted: {path}")
processed.add(path)
delete_count += 1
@@ -257,28 +275,27 @@ class WatchService:
full_path = directory / path
if not full_path.exists() or full_path.is_dir():
logger.debug(
"Skipping non-existent or directory path", path=path
f"Skipping non-existent or directory path, path={path}"
) # pragma: no cover
processed.add(path) # pragma: no cover
continue # pragma: no cover
logger.debug("Processing new file", path=path)
logger.debug(f"Processing new file, path={path}")
entity, checksum = await self.sync_service.sync_file(path, new=True)
if checksum:
self.state.add_event(
path=path, action="new", status="success", checksum=checksum
)
self.console.print(f"[green]✓[/green] {path}")
logger.debug(
"Added file processed",
path=path,
entity_id=entity.id if entity else None,
checksum=checksum,
logger.info(
"new file processed",
f"path={path}",
f"checksum={checksum}",
)
processed.add(path)
add_count += 1
else: # pragma: no cover
logger.warning("Error syncing new file", path=path) # pragma: no cover
logger.warning(f"Error syncing new file, path={path}") # pragma: no cover
self.console.print(
f"[orange]?[/orange] Error syncing: {path}"
) # pragma: no cover
@@ -296,7 +313,7 @@ class WatchService:
processed.add(path)
continue
logger.debug("Processing modified file", path=path)
logger.debug(f"Processing modified file: path={path}")
entity, checksum = await self.sync_service.sync_file(path, new=False)
self.state.add_event(
path=path, action="modified", status="success", checksum=checksum
@@ -311,17 +328,18 @@ class WatchService:
f"[yellow]...[/yellow] Repeated changes to {path}"
) # pragma: no cover
else:
# New file being modified
# haven't processed this file
self.console.print(f"[yellow]✎[/yellow] {path}")
logger.info(f"modified: {path}")
last_modified_path = path
repeat_count = 0
modify_count += 1
logger.debug(
"Modified file processed",
path=path,
entity_id=entity.id if entity else None,
checksum=checksum,
"Modified file processed, "
f"path={path} "
f"entity_id={entity.id if entity else None} "
f"checksum={checksum}",
)
processed.add(path)
@@ -339,16 +357,17 @@ class WatchService:
if changes:
self.console.print(f"{', '.join(changes)}", style="dim") # pyright: ignore
logger.info(f"changes: {len(changes)}")
duration_ms = int((time.time() - start_time) * 1000)
self.state.last_scan = datetime.now()
self.state.synced_files += len(processed)
logger.info(
"File change processing completed",
processed_files=len(processed),
total_synced_files=self.state.synced_files,
duration_ms=duration_ms,
"File change processing completed, "
f"processed_files={len(processed)}, "
f"total_synced_files={self.state.synced_files}, "
f"duration_ms={duration_ms}"
)
await self.write_status()
+77 -19
View File
@@ -5,11 +5,11 @@ import os
import logging
import re
import sys
import unicodedata
from pathlib import Path
from typing import Optional, Protocol, Union, runtime_checkable, List
from typing import Optional, Protocol, Union, runtime_checkable, List, Any
from loguru import logger
from unidecode import unidecode
@runtime_checkable
@@ -27,23 +27,23 @@ FilePath = Union[Path, str]
logging.getLogger("opentelemetry.sdk.metrics._internal.instrument").setLevel(logging.ERROR)
def generate_permalink(file_path: Union[Path, str, PathLike]) -> str:
"""Generate a stable permalink from a file path.
Args:
file_path: Original file path (str, Path, or PathLike)
def generate_permalink(file_path: Union[Path, str, Any]) -> str:
"""
Generate a permalink from a file path.
Returns:
Normalized permalink that matches validation rules. Converts spaces and underscores
to hyphens for consistency.
to hyphens for consistency. Preserves non-ASCII characters like Chinese.
Examples:
>>> generate_permalink("docs/My Feature.md")
'docs/my-feature'
>>> generate_permalink("specs/API (v2).md")
>>> generate_permalink("specs/API_v2.md")
'specs/api-v2'
>>> generate_permalink("design/unified_model_refactor.md")
'design/unified-model-refactor'
>>> generate_permalink("中文/测试文档.md")
'中文/测试文档'
"""
# Convert Path to string if needed
path_str = str(file_path)
@@ -51,24 +51,74 @@ def generate_permalink(file_path: Union[Path, str, PathLike]) -> str:
# Remove extension
base = os.path.splitext(path_str)[0]
# Transliterate unicode to ascii
ascii_text = unidecode(base)
# Create a transliteration mapping for specific characters
transliteration_map = {
"ø": "o", # Handle Søren -> soren
"å": "a", # Handle Kierkegård -> kierkegard
"ü": "u", # Handle Müller -> muller
"é": "e", # Handle Café -> cafe
"è": "e", # Handle Mère -> mere
"ê": "e", # Handle Fête -> fete
"à": "a", # Handle À la mode -> a la mode
"ç": "c", # Handle Façade -> facade
"ñ": "n", # Handle Niño -> nino
"ö": "o", # Handle Björk -> bjork
"ä": "a", # Handle Häagen -> haagen
# Add more mappings as needed
}
# Process character by character, transliterating Latin characters with diacritics
result = ""
for char in base:
# Direct mapping for known characters
if char.lower() in transliteration_map:
result += transliteration_map[char.lower()]
# General case using Unicode normalization
elif unicodedata.category(char).startswith("L") and ord(char) > 127:
# Decompose the character (e.g., ü -> u + combining diaeresis)
decomposed = unicodedata.normalize("NFD", char)
# If decomposition produced multiple characters and first one is ASCII
if len(decomposed) > 1 and ord(decomposed[0]) < 128:
# Keep only the base character
result += decomposed[0].lower()
else:
# For non-Latin scripts like Chinese, preserve the character
result += char
else:
# Add the character as is
result += char
# Handle special punctuation cases for apostrophes
result = result.replace("'", "")
# Insert dash between camelCase
ascii_text = re.sub(r"([a-z0-9])([A-Z])", r"\1-\2", ascii_text)
# This regex finds boundaries between lowercase and uppercase letters
result = re.sub(r"([a-z0-9])([A-Z])", r"\1-\2", result)
# Convert to lowercase
lower_text = ascii_text.lower()
# Insert dash between Chinese and Latin character boundaries
# This is needed for cases like "中文English" -> "中文-english"
result = re.sub(r"([\u4e00-\u9fff])([a-zA-Z])", r"\1-\2", result)
result = re.sub(r"([a-zA-Z])([\u4e00-\u9fff])", r"\1-\2", result)
# replace underscores with hyphens
# Convert ASCII letters to lowercase, preserve non-ASCII characters
lower_text = "".join(c.lower() if c.isascii() and c.isalpha() else c for c in result)
# Replace underscores with hyphens
text_with_hyphens = lower_text.replace("_", "-")
# Replace remaining invalid chars with hyphens
clean_text = re.sub(r"[^a-z0-9/\-]", "-", text_with_hyphens)
# Replace spaces and unsafe ASCII characters with hyphens, but preserve non-ASCII characters
# Include common Chinese character ranges and other non-ASCII characters
clean_text = re.sub(
r"[^a-z0-9\u4e00-\u9fff\u3000-\u303f\u3400-\u4dbf/\-]", "-", text_with_hyphens
)
# Collapse multiple hyphens
clean_text = re.sub(r"-+", "-", clean_text)
# Remove hyphens between adjacent Chinese characters only
# This handles cases like "你好-世界" -> "你好世界"
clean_text = re.sub(r"([\u4e00-\u9fff])-([\u4e00-\u9fff])", r"\1\2", clean_text)
# Clean each path segment
segments = clean_text.split("/")
clean_segments = [s.strip("-") for s in segments]
@@ -138,15 +188,23 @@ def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
Returns:
A list of tag strings, or an empty list if no tags
Note:
This function strips leading '#' characters from tags to prevent
their accumulation when tags are processed multiple times.
"""
if tags is None:
return []
# Process list of tags
if isinstance(tags, list):
return tags
# First strip whitespace, then strip leading '#' characters to prevent accumulation
return [tag.strip().lstrip("#") for tag in tags if tag and tag.strip()]
# Process comma-separated string of tags
if isinstance(tags, str):
return [tag.strip() for tag in tags.split(",") if tag.strip()]
# Split by comma, strip whitespace, then strip leading '#' characters
return [tag.strip().lstrip("#") for tag in tags.split(",") if tag and tag.strip()]
# For any other type, try to convert to string and parse
try: # pragma: no cover
+3 -3
View File
@@ -251,7 +251,7 @@ async def test_entity_indexing(client: AsyncClient):
# Verify it's searchable
search_response = await client.post(
"/search/", json={"text": "search", "types": [SearchItemType.ENTITY.value]}
"/search/", json={"text": "search", "entity_types": [SearchItemType.ENTITY.value]}
)
assert search_response.status_code == 200
search_result = SearchResponse.model_validate(search_response.json())
@@ -279,7 +279,7 @@ async def test_entity_delete_indexing(client: AsyncClient):
# Verify it's initially searchable
search_response = await client.post(
"/search/", json={"text": "delete", "types": [SearchItemType.ENTITY.value]}
"/search/", json={"text": "delete", "entity_types": [SearchItemType.ENTITY.value]}
)
search_result = SearchResponse.model_validate(search_response.json())
assert len(search_result.results) == 1
@@ -475,7 +475,7 @@ async def test_update_entity_search_index(client: AsyncClient):
# Search should find new content
search_response = await client.post(
"/search/", json={"text": "sphinx marker", "types": [SearchItemType.ENTITY.value]}
"/search/", json={"text": "sphinx marker", "entity_types": [SearchItemType.ENTITY.value]}
)
results = search_response.json()["results"]
assert len(results) == 1
+3 -3
View File
@@ -2,9 +2,9 @@
import json
from datetime import datetime, timezone
from pathlib import Path
import pytest
from pathlib import Path
from basic_memory.schemas import EntityResponse
@@ -346,7 +346,7 @@ async def test_put_resource_new_file(client, test_config, entity_repository, sea
assert full_path.exists()
# Verify file content
file_content = full_path.read_text()
file_content = full_path.read_text(encoding="utf-8")
assert json.loads(file_content) == canvas_data
# Verify entity was created in DB
@@ -420,7 +420,7 @@ async def test_put_resource_update_existing(client, test_config, entity_reposito
assert response.status_code == 200
# Verify file was updated
updated_content = full_path.read_text()
updated_content = full_path.read_text(encoding="utf-8")
assert json.loads(updated_content) == updated_data
# Verify entity was updated
+8 -8
View File
@@ -48,11 +48,11 @@ async def test_search_basic_pagination(client, indexed_entity):
@pytest.mark.asyncio
async def test_search_with_type_filter(client, indexed_entity):
async def test_search_with_entity_type_filter(client, indexed_entity):
"""Test search with type filter."""
# Should find with correct type
response = await client.post(
"/search/", json={"text": "test", "types": [SearchItemType.ENTITY.value]}
"/search/", json={"text": "test", "entity_types": [SearchItemType.ENTITY.value]}
)
assert response.status_code == 200
search_results = SearchResponse.model_validate(response.json())
@@ -60,7 +60,7 @@ async def test_search_with_type_filter(client, indexed_entity):
# Should find with relation type
response = await client.post(
"/search/", json={"text": "test", "types": [SearchItemType.RELATION.value]}
"/search/", json={"text": "test", "entity_types": [SearchItemType.RELATION.value]}
)
assert response.status_code == 200
search_results = SearchResponse.model_validate(response.json())
@@ -68,16 +68,16 @@ async def test_search_with_type_filter(client, indexed_entity):
@pytest.mark.asyncio
async def test_search_with_entity_type_filter(client, indexed_entity):
async def test_search_with_type_filter(client, indexed_entity):
"""Test search with entity type filter."""
# Should find with correct entity type
response = await client.post("/search/", json={"text": "test", "entity_types": ["test"]})
response = await client.post("/search/", json={"text": "test", "types": ["test"]})
assert response.status_code == 200
search_results = SearchResponse.model_validate(response.json())
assert len(search_results.results) == 1
# Should not find with wrong entity type
response = await client.post("/search/", json={"text": "test", "entity_types": ["note"]})
response = await client.post("/search/", json={"text": "test", "types": ["note"]})
assert response.status_code == 200
search_results = SearchResponse.model_validate(response.json())
assert len(search_results.results) == 0
@@ -153,8 +153,8 @@ async def test_multiple_filters(client, indexed_entity):
"/search/",
json={
"text": "test",
"types": [SearchItemType.ENTITY.value],
"entity_types": ["test"],
"entity_types": [SearchItemType.ENTITY.value],
"types": ["test"],
"after_date": datetime(2020, 1, 1, tzinfo=timezone.utc).isoformat(),
},
)
+28
View File
@@ -3,6 +3,8 @@
These tests use real MCP tools with the test environment instead of mocks.
"""
# Import for testing
import io
from datetime import datetime, timedelta
import json
@@ -410,3 +412,29 @@ def test_continue_conversation_no_results(cli_env):
# Check result contains expected content for no results
assert "Continuing conversation on: NonexistentTopic" in result.stdout
assert "The supplied query did not return any information" in result.stdout
@patch("basic_memory.services.initialization.initialize_database")
def test_ensure_migrations_functionality(mock_initialize_database, test_config, monkeypatch):
"""Test the database initialization functionality."""
from basic_memory.services.initialization import ensure_initialization
# Call the function
ensure_initialization(test_config)
# The underlying asyncio.run should call our mocked function
mock_initialize_database.assert_called_once()
@patch("basic_memory.services.initialization.initialize_database")
def test_ensure_migrations_handles_errors(mock_initialize_database, test_config, monkeypatch):
"""Test that initialization handles errors gracefully."""
from basic_memory.services.initialization import ensure_initialization
# Configure mock to raise an exception
mock_initialize_database.side_effect = Exception("Test error")
# Call the function - should not raise exception
ensure_initialization(test_config)
# We're just making sure it doesn't crash by calling it
+8 -7
View File
@@ -1,10 +1,11 @@
"""Tests for import_chatgpt command."""
import json
import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import import_app, app
from basic_memory.cli.app import app, import_app
from basic_memory.cli.commands import import_chatgpt
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
@@ -144,7 +145,7 @@ def sample_conversation_with_hidden():
def sample_chatgpt_json(tmp_path, sample_conversation):
"""Create a sample ChatGPT JSON file."""
json_file = tmp_path / "conversations.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([sample_conversation], f)
return json_file
@@ -167,7 +168,7 @@ async def test_process_chatgpt_json(tmp_path, sample_chatgpt_json):
assert conv_path.exists()
# Check content formatting
content = conv_path.read_text()
content = conv_path.read_text(encoding="utf-8")
assert "# Test Conversation" in content
assert "### User" in content
assert "Hello, this is a test message" in content
@@ -183,14 +184,14 @@ async def test_process_code_blocks(tmp_path, sample_conversation_with_code):
# Create test file
json_file = tmp_path / "code_test.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([sample_conversation_with_code], f)
await import_chatgpt.process_chatgpt_json(json_file, tmp_path, processor)
# Check content
conv_path = tmp_path / "20250111-code-test.md"
content = conv_path.read_text()
content = conv_path.read_text(encoding="utf-8")
assert "```python" in content
assert "def hello():" in content
assert "```" in content
@@ -204,7 +205,7 @@ async def test_hidden_messages(tmp_path, sample_conversation_with_hidden):
# Create test file
json_file = tmp_path / "hidden_test.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([sample_conversation_with_hidden], f)
results = await import_chatgpt.process_chatgpt_json(json_file, tmp_path, processor)
@@ -214,7 +215,7 @@ async def test_hidden_messages(tmp_path, sample_conversation_with_hidden):
# Check content
conv_path = tmp_path / "20250111-hidden-test.md"
content = conv_path.read_text()
content = conv_path.read_text(encoding="utf-8")
assert "Visible message" in content
assert "Hidden message" not in content
@@ -1,6 +1,7 @@
"""Tests for import_claude command (chat conversations)."""
import json
import pytest
from typer.testing import CliRunner
@@ -44,7 +45,7 @@ def sample_conversation():
def sample_conversations_json(tmp_path, sample_conversation):
"""Create a sample conversations.json file."""
json_file = tmp_path / "conversations.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([sample_conversation], f)
return json_file
@@ -65,7 +66,7 @@ async def test_process_chat_json(tmp_path, sample_conversations_json):
# Check conversation file
conv_path = tmp_path / "20250105-test-conversation.md"
assert conv_path.exists()
content = conv_path.read_text()
content = conv_path.read_text(encoding="utf-8")
# Check content formatting
assert "### Human" in content
@@ -156,7 +157,7 @@ def test_import_conversation_with_attachments(tmp_path):
}
json_file = tmp_path / "with_attachments.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([conversation], f)
# Set up environment
@@ -168,7 +169,7 @@ def test_import_conversation_with_attachments(tmp_path):
# Check attachment formatting
conv_path = tmp_path / "conversations/20250105-test-with-attachments.md"
content = conv_path.read_text()
content = conv_path.read_text(encoding="utf-8")
assert "**Attachment: test.txt**" in content
assert "```" in content
assert "Test file content" in content
+5 -4
View File
@@ -1,6 +1,7 @@
"""Tests for import_claude_projects command."""
import json
import pytest
from typer.testing import CliRunner
@@ -43,7 +44,7 @@ def sample_project():
def sample_projects_json(tmp_path, sample_project):
"""Create a sample projects.json file."""
json_file = tmp_path / "projects.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([sample_project], f)
return json_file
@@ -70,14 +71,14 @@ async def test_process_projects_json(tmp_path, sample_projects_json):
# Check document files
doc1 = project_dir / "docs/test-document.md"
assert doc1.exists()
content1 = doc1.read_text()
content1 = doc1.read_text(encoding="utf-8")
assert "# Test Document" in content1
assert "This is test content" in content1
# Check prompt template
prompt = project_dir / "prompt-template.md"
assert prompt.exists()
prompt_content = prompt.read_text()
prompt_content = prompt.read_text(encoding="utf-8")
assert "# Test Prompt" in prompt_content
assert "This is a test prompt" in prompt_content
@@ -160,7 +161,7 @@ def test_import_project_without_prompt(tmp_path):
}
json_file = tmp_path / "no_prompt.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
json.dump([project], f)
# Set up environment
+4 -3
View File
@@ -1,6 +1,7 @@
"""Tests for import_memory_json command."""
import json
import pytest
from typer.testing import CliRunner
@@ -35,7 +36,7 @@ def sample_entities():
def sample_json_file(tmp_path, sample_entities):
"""Create a sample memory.json file."""
json_file = tmp_path / "memory.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
for entity in sample_entities:
f.write(json.dumps(entity) + "\n")
return json_file
@@ -55,7 +56,7 @@ async def test_process_memory_json(tmp_path, sample_json_file):
# Check file was created
entity_file = tmp_path / "test/test_entity.md"
assert entity_file.exists()
content = entity_file.read_text()
content = entity_file.read_text(encoding="utf-8")
assert "Test observation 1" in content
assert "Test observation 2" in content
assert "test_relation [[related_entity]]" in content
@@ -120,7 +121,7 @@ def test_import_json_command_handle_old_format(tmp_path):
]
json_file = tmp_path / "old_format.json"
with open(json_file, "w") as f:
with open(json_file, "w", encoding="utf-8") as f:
for item in old_format:
f.write(json.dumps(item) + "\n")
+5 -3
View File
@@ -1,14 +1,14 @@
"""Common test fixtures."""
from datetime import datetime, timezone
from pathlib import Path
from textwrap import dedent
from typing import AsyncGenerator
from datetime import datetime, timezone
import pytest
import pytest_asyncio
from loguru import logger
from sqlalchemy.ext.asyncio import AsyncSession, AsyncEngine, async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from basic_memory import db
from basic_memory.config import ProjectConfig
@@ -140,6 +140,7 @@ def entity_parser(test_config):
@pytest_asyncio.fixture
async def sync_service(
test_config: ProjectConfig,
entity_service: EntityService,
entity_parser: EntityParser,
entity_repository: EntityRepository,
@@ -149,6 +150,7 @@ async def sync_service(
) -> SyncService:
"""Create sync service for testing."""
return SyncService(
config=test_config,
entity_service=entity_service,
entity_repository=entity_repository,
relation_repository=relation_repository,
@@ -349,5 +351,5 @@ def test_files(test_config) -> dict[str, Path]:
@pytest_asyncio.fixture
async def synced_files(sync_service, test_config, test_files):
# Initial sync - should create forward reference
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
return test_files
+31
View File
@@ -217,6 +217,37 @@ def test_parse_empty_content():
assert len(result.relations) == 0
@pytest.mark.asyncio
async def test_parse_file_with_absolute_path(test_config, entity_parser):
"""Test parsing a file with an absolute path."""
content = dedent("""
---
type: component
permalink: absolute_path_test
---
# Absolute Path Test
A file with an absolute path.
""")
# Create a test file in the project directory
test_file = test_config.home / "absolute_path_test.md"
test_file.write_text(content)
# Get the absolute path to the test file
absolute_path = test_file.resolve()
# Parse the file using the absolute path
entity = await entity_parser.parse_file(absolute_path)
# Verify the file was parsed correctly
assert entity.frontmatter.permalink == "absolute_path_test"
assert "Absolute Path Test" in entity.content
assert entity.created is not None
assert entity.modified is not None
# @pytest.mark.asyncio
# async def test_parse_file_invalid_yaml(test_config, entity_parser):
# """Test parsing file with invalid YAML frontmatter."""
+5 -5
View File
@@ -8,10 +8,10 @@ from pathlib import Path
import pytest
from basic_memory.markdown.markdown_processor import MarkdownProcessor, DirtyFileError
from basic_memory.markdown.markdown_processor import DirtyFileError, MarkdownProcessor
from basic_memory.markdown.schemas import (
EntityMarkdown,
EntityFrontmatter,
EntityMarkdown,
Observation,
Relation,
)
@@ -41,7 +41,7 @@ async def test_write_new_minimal_file(markdown_processor: MarkdownProcessor, tmp
await markdown_processor.write_file(path, markdown)
# Read back and verify
content = path.read_text()
content = path.read_text(encoding="utf-8")
assert "---" in content # Has frontmatter
assert "type: note" in content
assert "permalink: test" in content
@@ -90,7 +90,7 @@ async def test_write_new_file_with_content(markdown_processor: MarkdownProcessor
await markdown_processor.write_file(path, markdown)
# Read back and verify
content = path.read_text()
content = path.read_text(encoding="utf-8")
# Check content preserved exactly
assert "# Custom Title" in content
@@ -169,7 +169,7 @@ async def test_dirty_file_detection(markdown_processor: MarkdownProcessor, tmp_p
checksum = await markdown_processor.write_file(path, initial)
# Modify file directly
path.write_text(path.read_text() + "\nModified!")
path.write_text(path.read_text(encoding="utf-8") + "\nModified!")
# Try to update with old checksum
update = EntityMarkdown(
@@ -5,12 +5,9 @@ from datetime import datetime
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.tools import build_context, recent_activity
from basic_memory.mcp.tools import build_context
from basic_memory.schemas.memory import (
GraphContext,
EntitySummary,
ObservationSummary,
RelationSummary,
)
@@ -83,53 +80,6 @@ invalid_timeframes = [
]
@pytest.mark.asyncio
async def test_recent_activity_timeframe_formats(client, test_graph):
"""Test that recent_activity accepts various timeframe formats."""
# Test each valid timeframe
for timeframe in valid_timeframes:
try:
result = await recent_activity(
type=["entity"], timeframe=timeframe, page=1, page_size=10, max_related=10
)
assert result is not None
except Exception as e:
pytest.fail(f"Failed with valid timeframe '{timeframe}': {str(e)}")
# Test invalid timeframes should raise ValidationError
for timeframe in invalid_timeframes:
with pytest.raises(ToolError):
await recent_activity(timeframe=timeframe)
@pytest.mark.asyncio
async def test_recent_activity_type_filters(client, test_graph):
"""Test that recent_activity correctly filters by types."""
# Test single type
result = await recent_activity(type=["entity"])
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)
# Test multiple types
result = await recent_activity(type=["entity", "observation"])
assert result is not None
assert all(
isinstance(r, EntitySummary) or isinstance(r, ObservationSummary)
for r in result.primary_results
)
# Test all types
result = await recent_activity(type=["entity", "observation", "relation"])
assert result is not None
# Results can be any type
assert all(
isinstance(r, EntitySummary)
or isinstance(r, ObservationSummary)
or isinstance(r, RelationSummary)
for r in result.primary_results
)
@pytest.mark.asyncio
async def test_build_context_timeframe_formats(client, test_graph):
"""Test that build_context accepts various timeframe formats."""
+4 -4
View File
@@ -46,7 +46,7 @@ async def test_create_canvas(app, test_config):
assert file_path.exists()
# Verify content is correct
content = json.loads(file_path.read_text())
content = json.loads(file_path.read_text(encoding="utf-8"))
assert content["nodes"] == nodes
assert content["edges"] == edges
@@ -81,7 +81,7 @@ async def test_create_canvas_with_extension(app, test_config):
assert file_path.exists()
# Verify content
content = json.loads(file_path.read_text())
content = json.loads(file_path.read_text(encoding="utf-8"))
assert content["nodes"] == nodes
@@ -134,7 +134,7 @@ async def test_update_existing_canvas(app, test_config):
assert "Updated: visualizations/update-test.canvas" in result
# Verify content was updated
content = json.loads(file_path.read_text())
content = json.loads(file_path.read_text(encoding="utf-8"))
assert content["nodes"] == updated_nodes
assert content["edges"] == updated_edges
@@ -252,7 +252,7 @@ async def test_create_canvas_complex_content(app, test_config):
assert file_path.exists()
# Verify content is correct with all complex structures
content = json.loads(file_path.read_text())
content = json.loads(file_path.read_text(encoding="utf-8"))
assert len(content["nodes"]) == 4
assert len(content["edges"]) == 2
+10 -5
View File
@@ -51,8 +51,10 @@ async def test_note_unicode_content(app):
assert (
dedent("""
# Created test/Unicode Test.md (272389cd)
# Created note
file_path: test/Unicode Test.md
permalink: test/unicode-test
checksum: 272389cd
""").strip()
in result
)
@@ -201,7 +203,8 @@ async def test_read_note_title_search_fallback(mock_call_get, mock_search):
# Verify title search was used
mock_search.assert_called_once()
assert mock_search.call_args[0][0].title == "Test Note"
assert mock_search.call_args[1]["query"] == "Test Note"
assert mock_search.call_args[1]["search_type"] == "title"
# Verify second lookup was used
assert mock_call_get.call_count == 2
@@ -254,15 +257,17 @@ async def test_read_note_text_search_fallback(mock_call_get, mock_search):
# Verify both search types were used
assert mock_search.call_count == 2
assert mock_search.call_args_list[0][0][0].title == "some query" # Title search
assert mock_search.call_args_list[1][0][0].text == "some query" # Text search
assert mock_search.call_args_list[0][1]["query"] == "some query" # Title search
assert mock_search.call_args_list[0][1]["search_type"] == "title"
assert mock_search.call_args_list[1][1]["query"] == "some query" # Text search
assert mock_search.call_args_list[1][1]["search_type"] == "text"
# Verify result contains helpful information
assert "Note Not Found" in result
assert "Related Note 1" in result
assert "Related Note 2" in result
assert 'read_note("notes/related-note-1")' in result
assert "search(query=" in result
assert "search_notes(query=" in result
assert "write_note(" in result
+110
View File
@@ -0,0 +1,110 @@
"""Tests for discussion context MCP tool."""
import pytest
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.tools import recent_activity
from basic_memory.schemas.memory import (
EntitySummary,
ObservationSummary,
RelationSummary,
)
from basic_memory.schemas.search import SearchItemType
# Test data for different timeframe formats
valid_timeframes = [
"7d", # Standard format
"yesterday", # Natural language
"0d", # Zero duration
]
invalid_timeframes = [
"invalid", # Nonsense string
"tomorrow", # Future date
]
@pytest.mark.asyncio
async def test_recent_activity_timeframe_formats(client, test_graph):
"""Test that recent_activity accepts various timeframe formats."""
# Test each valid timeframe
for timeframe in valid_timeframes:
try:
result = await recent_activity(
type=["entity"], timeframe=timeframe, page=1, page_size=10, max_related=10
)
assert result is not None
except Exception as e:
pytest.fail(f"Failed with valid timeframe '{timeframe}': {str(e)}")
# Test invalid timeframes should raise ValidationError
for timeframe in invalid_timeframes:
with pytest.raises(ToolError):
await recent_activity(timeframe=timeframe)
@pytest.mark.asyncio
async def test_recent_activity_type_filters(client, test_graph):
"""Test that recent_activity correctly filters by types."""
# Test single string type
result = await recent_activity(type=SearchItemType.ENTITY)
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)
# Test single string type
result = await recent_activity(type="entity")
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)
# Test single type
result = await recent_activity(type=["entity"])
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)
# Test multiple types
result = await recent_activity(type=["entity", "observation"])
assert result is not None
assert all(
isinstance(r, EntitySummary) or isinstance(r, ObservationSummary)
for r in result.primary_results
)
# Test multiple types
result = await recent_activity(type=[SearchItemType.ENTITY, SearchItemType.OBSERVATION])
assert result is not None
assert all(
isinstance(r, EntitySummary) or isinstance(r, ObservationSummary)
for r in result.primary_results
)
# Test all types
result = await recent_activity(type=["entity", "observation", "relation"])
assert result is not None
# Results can be any type
assert all(
isinstance(r, EntitySummary)
or isinstance(r, ObservationSummary)
or isinstance(r, RelationSummary)
for r in result.primary_results
)
@pytest.mark.asyncio
async def test_recent_activity_type_invalid(client, test_graph):
"""Test that recent_activity correctly filters by types."""
# Test single invalid string type
with pytest.raises(ValueError) as e:
await recent_activity(type="note")
assert (
str(e.value) == "Invalid type: note. Valid types are: ['entity', 'observation', 'relation']"
)
# Test invalid string array type
with pytest.raises(ValueError) as e:
await recent_activity(type=["note"])
assert (
str(e.value) == "Invalid type: note. Valid types are: ['entity', 'observation', 'relation']"
)
+27
View File
@@ -42,6 +42,33 @@ async def test_read_file_text_file(app, synced_files):
assert response["encoding"] == "utf-8"
@pytest.mark.asyncio
async def test_read_content_file_path(app, synced_files):
"""Test reading a text file.
Should:
- Correctly identify text content
- Return the content as text
- Include correct metadata
"""
# First create a text file via notes
result = await write_note(
title="Text Resource",
folder="test",
content="This is a test text resource",
tags=["test", "resource"],
)
assert result is not None
# Now read it as a resource
response = await read_content("test/Text Resource.md")
assert response["type"] == "text"
assert "This is a test text resource" in response["text"]
assert response["content_type"].startswith("text/")
assert response["encoding"] == "utf-8"
@pytest.mark.asyncio
async def test_read_file_image_file(app, synced_files):
"""Test reading an image file.
+82 -10
View File
@@ -5,11 +5,10 @@ from datetime import datetime, timedelta
from basic_memory.mcp.tools import write_note
from basic_memory.mcp.tools.search import search_notes
from basic_memory.schemas.search import SearchQuery, SearchItemType
@pytest.mark.asyncio
async def test_search_basic(client):
async def test_search_text(client):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
@@ -21,8 +20,67 @@ async def test_search_basic(client):
assert result
# Search for it
query = SearchQuery(text="searchable")
response = await search_notes(query)
response = await search_notes(query="searchable")
# Verify results
assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results)
@pytest.mark.asyncio
async def test_search_title(client):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
title="Test Search Note",
folder="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it
response = await search_notes(query="Search Note", search_type="title")
# Verify results
assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results)
@pytest.mark.asyncio
async def test_search_permalink(client):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
title="Test Search Note",
folder="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it
response = await search_notes(query="test/test-search-note", search_type="permalink")
# Verify results
assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results)
@pytest.mark.asyncio
async def test_search_permalink_match(client):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
title="Test Search Note",
folder="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it
response = await search_notes(query="test/test-search-*", search_type="permalink")
# Verify results
assert len(response.results) > 0
@@ -42,8 +100,7 @@ async def test_search_pagination(client):
assert result
# Search for it
query = SearchQuery(text="searchable")
response = await search_notes(query, page=1, page_size=1)
response = await search_notes(query="searchable", page=1, page_size=1)
# Verify results
assert len(response.results) == 1
@@ -61,8 +118,24 @@ async def test_search_with_type_filter(client):
)
# Search with type filter
query = SearchQuery(text="type", types=[SearchItemType.ENTITY])
response = await search_notes(query)
response = await search_notes(query="type", types=["note"])
# Verify all results are entities
assert all(r.type == "entity" for r in response.results)
@pytest.mark.asyncio
async def test_search_with_entity_type_filter(client):
"""Test search with entity type filter."""
# Create test content
await write_note(
title="Entity Type Test",
folder="test",
content="# Test\nFiltered by type",
)
# Search with entity type filter
response = await search_notes(query="type", entity_types=["entity"])
# Verify all results are entities
assert all(r.type == "entity" for r in response.results)
@@ -80,8 +153,7 @@ async def test_search_with_date_filter(client):
# Search with date filter
one_hour_ago = datetime.now() - timedelta(hours=1)
query = SearchQuery(text="recent", after_date=one_hour_ago)
response = await search_notes(query)
response = await search_notes(query="recent", after_date=one_hour_ago.isoformat())
# Verify we get results within timeframe
assert len(response.results) > 0
@@ -26,8 +26,10 @@ async def test_write_note(app):
assert result
assert (
dedent("""
# Created test/Test Note.md (159f2168)
# Created note
file_path: test/Test Note.md
permalink: test/test-note
checksum: 159f2168
## Tags
- test, documentation
@@ -63,8 +65,10 @@ async def test_write_note_no_tags(app):
assert result
assert (
dedent("""
# Created test/Simple Note.md (9a1ff079)
# Created note
file_path: test/Simple Note.md
permalink: test/simple-note
checksum: 9a1ff079
""").strip()
in result
)
@@ -104,8 +108,10 @@ async def test_write_note_update_existing(app):
assert result # Got a valid permalink
assert (
dedent("""
# Created test/Test Note.md (159f2168)
# Created note
file_path: test/Test Note.md
permalink: test/test-note
checksum: 159f2168
## Tags
- test, documentation
@@ -121,8 +127,10 @@ async def test_write_note_update_existing(app):
)
assert (
dedent("""
# Updated test/Test Note.md (a8eb4d44)
# Updated note
file_path: test/Test Note.md
permalink: test/test-note
checksum: a8eb4d44
## Tags
- test, documentation
@@ -235,8 +243,10 @@ async def test_write_note_verbose(app):
assert (
dedent("""
# Created test/Test Note.md (06873a7a)
# Created note
file_path: test/Test Note.md
permalink: test/test-note
checksum: 06873a7a
## Observations
- note: 1
@@ -302,7 +312,7 @@ async def test_write_note_preserves_custom_metadata(app, test_config):
)
# Verify the update was successful
assert "Updated test/Custom Metadata Note.md" in result
assert ("Updated note\nfile_path: test/Custom Metadata Note.md") in result
# Read the note back and check if custom frontmatter is preserved
content = await read_note("test/custom-metadata-note")
@@ -320,3 +330,50 @@ async def test_write_note_preserves_custom_metadata(app, test_config):
# And tags should be updated
assert "'#test'" in content
assert "'#updated'" in content
@pytest.mark.asyncio
async def test_write_note_preserves_content_frontmatter(app):
"""Test creating a new note."""
await write_note(
title="Test Note",
folder="test",
content=dedent(
"""
---
title: Test Note
type: note
version: 1.0
author: name
---
# Test
This is a test note
"""
),
tags=["test", "documentation"],
)
# Try reading it back via permalink
content = await read_note("test/test-note")
assert (
dedent(
"""
---
title: Test Note
type: note
permalink: test/test-note
version: 1.0
author: name
tags:
- '#test'
- '#documentation'
---
# Test
This is a test note
"""
).strip()
in content
)
+4 -4
View File
@@ -32,12 +32,12 @@ def test_search_filters():
"""Test search result filtering."""
query = SearchQuery(
text="search",
types=[SearchItemType.ENTITY],
entity_types=["component"],
entity_types=[SearchItemType.ENTITY],
types=["component"],
after_date=datetime(2024, 1, 1),
)
assert query.types == [SearchItemType.ENTITY]
assert query.entity_types == ["component"]
assert query.entity_types == [SearchItemType.ENTITY]
assert query.types == ["component"]
assert query.after_date == "2024-01-01T00:00:00"
+34 -2
View File
@@ -6,12 +6,14 @@ from textwrap import dedent
import pytest
import yaml
from basic_memory.config import ProjectConfig
from basic_memory.markdown import EntityParser
from basic_memory.models import Entity as EntityModel
from basic_memory.repository import EntityRepository
from basic_memory.schemas import Entity as EntitySchema
from basic_memory.services import FileService
from basic_memory.services.entity_service import EntityService
from basic_memory.services.exceptions import EntityNotFoundError, EntityCreationError
from basic_memory.services.exceptions import EntityCreationError, EntityNotFoundError
from basic_memory.utils import generate_permalink
@@ -261,7 +263,7 @@ async def test_get_entities_by_permalinks(entity_service: EntityService):
@pytest.mark.asyncio
async def test_oget_entities_empty_input(entity_service: EntityService):
async def test_get_entities_empty_input(entity_service: EntityService):
"""Test opening nodes with empty path ID list."""
found = await entity_service.get_entities_by_permalinks([])
assert len(found) == 0
@@ -557,3 +559,33 @@ async def test_update_with_content(entity_service: EntityService, file_service:
# assert content is in file
assert update_content.strip() == file_content
@pytest.mark.asyncio
async def test_create_with_no_frontmatter(
test_config: ProjectConfig,
entity_parser: EntityParser,
entity_service: EntityService,
file_service: FileService,
):
# contains no frontmatter
content = "# Git Workflow Guide"
file_path = Path("test/Git Workflow Guide.md")
full_path = test_config.home / file_path
await file_service.write_file(Path(full_path), content)
entity_markdown = await entity_parser.parse_file(full_path)
created = await entity_service.create_entity_from_markdown(file_path, entity_markdown)
file_content, _ = await file_service.read_file(created.file_path)
assert str(file_path) == str(created.file_path)
assert created.title == "Git Workflow Guide"
assert created.entity_type == "note"
assert created.permalink is None
# assert file
expected = dedent("""
# Git Workflow Guide
""").strip()
assert expected == file_content
+49
View File
@@ -0,0 +1,49 @@
"""Tests for the initialization service."""
from unittest.mock import patch
import pytest
from basic_memory.services.initialization import (
ensure_initialization,
initialize_app,
initialize_database,
)
@pytest.mark.asyncio
@patch("basic_memory.services.initialization.db.run_migrations")
async def test_initialize_database(mock_run_migrations, test_config):
"""Test initializing the database."""
await initialize_database(test_config)
mock_run_migrations.assert_called_once_with(test_config)
@pytest.mark.asyncio
@patch("basic_memory.services.initialization.db.run_migrations")
async def test_initialize_database_error(mock_run_migrations, test_config):
"""Test handling errors during database initialization."""
mock_run_migrations.side_effect = Exception("Test error")
await initialize_database(test_config)
mock_run_migrations.assert_called_once_with(test_config)
@pytest.mark.asyncio
@patch("basic_memory.services.initialization.initialize_database")
@patch("basic_memory.services.initialization.initialize_file_sync")
async def test_initialize_app(mock_initialize_file_sync, mock_initialize_database, test_config):
"""Test app initialization."""
mock_initialize_file_sync.return_value = "task"
result = await initialize_app(test_config)
mock_initialize_database.assert_called_once_with(test_config)
mock_initialize_file_sync.assert_called_once_with(test_config)
assert result == "task"
@patch("basic_memory.services.initialization.asyncio.run")
def test_ensure_initialization(mock_run, test_config):
"""Test synchronous initialization wrapper."""
ensure_initialization(test_config)
mock_run.assert_called_once()
+13 -6
View File
@@ -75,7 +75,7 @@ async def test_search_permalink_wildcard2(search_service, test_graph):
async def test_search_text(search_service, test_graph):
"""Full-text search"""
results = await search_service.search(
SearchQuery(text="Root Entity", types=[SearchItemType.ENTITY])
SearchQuery(text="Root Entity", entity_types=[SearchItemType.ENTITY])
)
assert len(results) >= 1
assert results[0].permalink == "test/root"
@@ -84,7 +84,9 @@ async def test_search_text(search_service, test_graph):
@pytest.mark.asyncio
async def test_search_title(search_service, test_graph):
"""Title only search"""
results = await search_service.search(SearchQuery(title="Root", types=[SearchItemType.ENTITY]))
results = await search_service.search(
SearchQuery(title="Root", entity_types=[SearchItemType.ENTITY])
)
assert len(results) >= 1
assert results[0].permalink == "test/root"
@@ -140,7 +142,7 @@ async def test_filters(search_service, test_graph):
"""Test search filters."""
# Combined filters
results = await search_service.search(
SearchQuery(text="Deep", types=[SearchItemType.ENTITY], entity_types=["deep"])
SearchQuery(text="Deep", entity_types=[SearchItemType.ENTITY], types=["deep"])
)
assert len(results) == 1
for r in results:
@@ -179,13 +181,18 @@ async def test_search_type(search_service, test_graph):
"""Test search filters."""
# Should find only type
results = await search_service.search(SearchQuery(types=[SearchItemType.ENTITY]))
results = await search_service.search(SearchQuery(types=["test"]))
assert len(results) > 0
for r in results:
assert r.type == SearchItemType.ENTITY
# Should find only types passed in
results = await search_service.search(SearchQuery(types=[SearchItemType.ENTITY]))
@pytest.mark.asyncio
async def test_search_entity_type(search_service, test_graph):
"""Test search filters."""
# Should find only type
results = await search_service.search(SearchQuery(entity_types=[SearchItemType.ENTITY]))
assert len(results) > 0
for r in results:
assert r.type == SearchItemType.ENTITY
+119 -64
View File
@@ -3,8 +3,10 @@
import asyncio
from datetime import datetime, timezone
from pathlib import Path
from textwrap import dedent
import pytest
import pytest_asyncio
from basic_memory.config import ProjectConfig
from basic_memory.models import Entity
@@ -44,7 +46,7 @@ type: knowledge
await create_test_file(project_dir / "source.md", source_content)
# Initial sync - should create forward reference
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify forward reference
source = await entity_service.get_by_permalink("source")
@@ -63,7 +65,7 @@ Target content
await create_test_file(project_dir / "target_doc.md", target_content)
# Sync again - should resolve the reference
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify reference is now resolved
source = await entity_service.get_by_permalink("source")
@@ -116,7 +118,7 @@ A test concept.
await entity_service.repository.add(other)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify results
entities = await entity_service.repository.find_all()
@@ -146,7 +148,7 @@ async def test_sync_hidden_file(
await create_test_file(project_dir / "concept/.hidden.md", "hidden")
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify results
entities = await entity_service.repository.find_all()
@@ -180,7 +182,7 @@ modified: 2024-01-01
await create_test_file(project_dir / "concept/depends_on_future.md", content)
# Sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify entity created but no relations
entity = await sync_service.entity_service.repository.get_by_permalink(
@@ -236,7 +238,7 @@ modified: 2024-01-01
await create_test_file(project_dir / "concept/entity_b.md", content_b)
# Sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify both entities and their relations
entity_a = await sync_service.entity_service.repository.get_by_permalink("concept/entity-a")
@@ -307,7 +309,7 @@ modified: 2024-01-01
await create_test_file(project_dir / "concept/duplicate_relations.md", content)
# Sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify duplicates are handled
entity = await sync_service.entity_service.repository.get_by_permalink(
@@ -349,7 +351,7 @@ modified: 2024-01-01
await create_test_file(project_dir / "concept/invalid_category.md", content)
# Sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify observations
entity = await sync_service.entity_service.repository.get_by_permalink(
@@ -429,7 +431,7 @@ modified: 2024-01-01
await create_test_file(project_dir / f"concept/entity_{name}.md", content)
# Sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify all relations are created correctly regardless of order
entity_a = await sync_service.entity_service.repository.get_by_permalink("concept/entity-a")
@@ -449,7 +451,7 @@ modified: 2024-01-01
@pytest.mark.asyncio
async def test_sync_empty_directories(sync_service: SyncService, test_config: ProjectConfig):
"""Test syncing empty directories."""
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Should not raise exceptions for empty dirs
assert (test_config.home).exists()
@@ -484,7 +486,7 @@ modified: 2024-01-01
doc_path.write_text("Modified during sync")
# Run sync and modification concurrently
await asyncio.gather(sync_service.sync(test_config.home, show_progress=False), modify_file())
await asyncio.gather(sync_service.sync(test_config.home), modify_file())
# Verify final state
doc = await sync_service.entity_service.repository.get_by_permalink("changing")
@@ -492,7 +494,7 @@ modified: 2024-01-01
# if we failed in the middle of a sync, the next one should fix it.
if doc.checksum is None:
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
doc = await sync_service.entity_service.repository.get_by_permalink("changing")
assert doc.checksum is not None
@@ -528,7 +530,7 @@ Testing permalink generation.
await create_test_file(test_config.home / filename, content)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify permalinks
entities = await entity_service.repository.find_all()
@@ -599,7 +601,7 @@ Testing file timestamps
await create_test_file(file_path, file_dates_content)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Check explicit frontmatter dates
explicit_entity = await entity_service.get_by_permalink("explicit-dates")
@@ -639,7 +641,7 @@ Content for move test
await create_test_file(old_path, content)
# Initial sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Move the file
new_path = project_dir / "new" / "moved_file.md"
@@ -647,7 +649,7 @@ Content for move test
old_path.rename(new_path)
# Sync again
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Check search index has updated path
results = await search_service.search(SearchQuery(text="Content for move test"))
@@ -689,7 +691,7 @@ modified: 2024-01-01
await create_test_file(test_config.home / "concept/incomplete.md", content)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify entity was properly synced
updated = await entity_service.get_by_permalink("concept/incomplete")
@@ -718,7 +720,7 @@ Content for move test
await create_test_file(old_path, content)
# Initial sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Move the file
new_path = project_dir / "new" / "moved_file.md"
@@ -726,7 +728,7 @@ Content for move test
old_path.rename(new_path)
# Sync again
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
file_content, _ = await file_service.read_file(new_path)
assert "permalink: old/test-move" in file_content
@@ -745,7 +747,7 @@ Content for move test
await create_test_file(old_path, content)
# Sync new file
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# assert permalink is unique
file_content, _ = await file_service.read_file(old_path)
@@ -763,11 +765,31 @@ async def test_sync_permalink_resolved_on_update(
one_file = project_dir / "one.md"
two_file = project_dir / "two.md"
await create_test_file(one_file)
await create_test_file(two_file)
await create_test_file(
one_file,
content=dedent(
"""
---
permalink: one
---
test content
"""
),
)
await create_test_file(
two_file,
content=dedent(
"""
---
permalink: two
---
test content
"""
),
)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Check permalinks
file_one_content, _ = await file_service.read_file(one_file)
@@ -790,7 +812,7 @@ test content
two_file.write_text(updated_content)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Check permalinks
file_two_content, _ = await file_service.read_file(two_file)
@@ -811,7 +833,7 @@ test content
await create_test_file(new_file, new_content)
# Run another time
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Should have deduplicated permalink
new_file_content, _ = await file_service.read_file(new_file)
@@ -819,7 +841,7 @@ test content
@pytest.mark.asyncio
async def test_sync_duplicate_observations(
async def test_sync_permalink_not_created_if_no_frontmatter(
sync_service: SyncService,
test_config: ProjectConfig,
file_service: FileService,
@@ -827,46 +849,79 @@ async def test_sync_duplicate_observations(
"""Test that sync resolves permalink conflicts on update."""
project_dir = test_config.home
content = """
---
title: a note
type: note
tags: []
---
test content
- [note] one observation
"""
note_file = project_dir / "note.md"
await create_test_file(note_file, content)
file = project_dir / "one.md"
await create_test_file(file)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Check permalinks
file_one_content, _ = await file_service.read_file(note_file)
assert (
"""---
title: a note
type: note
tags: []
permalink: note
---
# Check permalink not created
file_content, _ = await file_service.read_file(file)
assert "permalink:" not in file_content
test content
- [note] one observation
""".strip()
== file_one_content
@pytest_asyncio.fixture
def test_config_update_permamlinks_on_move(tmp_path) -> ProjectConfig:
"""Test configuration using in-memory DB."""
config = ProjectConfig(
project="test-project",
update_permalinks_on_move=True,
)
config.home = tmp_path
(tmp_path / config.home.name).mkdir(parents=True, exist_ok=True)
return config
@pytest.mark.asyncio
async def test_sync_permalink_updated_on_move(
test_config_update_permamlinks_on_move: ProjectConfig,
sync_service: SyncService,
file_service: FileService,
):
"""Test that we update a permalink on a file move if set in config ."""
test_config = test_config_update_permamlinks_on_move
project_dir = test_config.home
sync_service.config = test_config
# Create initial file
content = dedent(
"""
---
type: knowledge
---
# Test Move
Content for move test
"""
)
old_path = project_dir / "old" / "test_move.md"
old_path.parent.mkdir(parents=True)
await create_test_file(old_path, content)
# Initial sync
await sync_service.sync(test_config.home)
# verify permalink
old_content, _ = await file_service.read_file(old_path)
assert "permalink: old/test-move" in old_content
# Move the file
new_path = project_dir / "new" / "moved_file.md"
new_path.parent.mkdir(parents=True)
old_path.rename(new_path)
# Sync again
await sync_service.sync(test_config.home)
file_content, _ = await file_service.read_file(new_path)
assert "permalink: new/moved-file" in file_content
@pytest.mark.asyncio
async def test_sync_non_markdown_files(sync_service, test_config, test_files):
"""Test syncing non-markdown files."""
report = await sync_service.sync(test_config.home, show_progress=False)
report = await sync_service.sync(test_config.home)
assert report.total == 2
# Check files were detected
@@ -889,7 +944,7 @@ async def test_sync_non_markdown_files_modified(
sync_service, test_config, test_files, file_service
):
"""Test syncing non-markdown files."""
report = await sync_service.sync(test_config.home, show_progress=False)
report = await sync_service.sync(test_config.home)
assert report.total == 2
# Check files were detected
@@ -899,7 +954,7 @@ async def test_sync_non_markdown_files_modified(
test_files["pdf"].write_text("New content")
test_files["image"].write_text("New content")
report = await sync_service.sync(test_config.home, show_progress=False)
report = await sync_service.sync(test_config.home)
assert len(report.modified) == 2
pdf_file_content, pdf_checksum = await file_service.read_file(test_files["pdf"].name)
@@ -917,7 +972,7 @@ async def test_sync_non_markdown_files_modified(
@pytest.mark.asyncio
async def test_sync_non_markdown_files_move(sync_service, test_config, test_files):
"""Test syncing non-markdown files updates permalink"""
report = await sync_service.sync(test_config.home, show_progress=False)
report = await sync_service.sync(test_config.home)
assert report.total == 2
# Check files were detected
@@ -925,7 +980,7 @@ async def test_sync_non_markdown_files_move(sync_service, test_config, test_file
assert test_files["image"].name in [f for f in report.new]
test_files["pdf"].rename(test_config.home / "moved_pdf.pdf")
report2 = await sync_service.sync(test_config.home, show_progress=False)
report2 = await sync_service.sync(test_config.home)
assert len(report2.moves) == 1
# Verify entity is updated
@@ -937,7 +992,7 @@ async def test_sync_non_markdown_files_move(sync_service, test_config, test_file
@pytest.mark.asyncio
async def test_sync_non_markdown_files_deleted(sync_service, test_config, test_files):
"""Test syncing non-markdown files updates permalink"""
report = await sync_service.sync(test_config.home, show_progress=False)
report = await sync_service.sync(test_config.home)
assert report.total == 2
# Check files were detected
@@ -945,7 +1000,7 @@ async def test_sync_non_markdown_files_deleted(sync_service, test_config, test_f
assert test_files["image"].name in [f for f in report.new]
test_files["pdf"].unlink()
report2 = await sync_service.sync(test_config.home, show_progress=False)
report2 = await sync_service.sync(test_config.home)
assert len(report2.deleted) == 1
# Verify entity is deleted
@@ -964,14 +1019,14 @@ async def test_sync_non_markdown_files_move_with_delete(
await create_test_file(test_config.home / "other/doc-1.pdf", "content2")
# Initial sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# First move/delete the original file to make way for the move
(test_config.home / "doc.pdf").unlink()
(test_config.home / "other/doc-1.pdf").rename(test_config.home / "doc.pdf")
# Sync again
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify the changes
moved_entity = await sync_service.entity_repository.get_by_file_path("doc.pdf")
@@ -1003,7 +1058,7 @@ tags: []
await create_test_file(note_file, content)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Check permalinks
file_one_content, _ = await file_service.read_file(note_file)
+63
View File
@@ -0,0 +1,63 @@
"""Test for issue #72 - notes with wikilinks staying in modified status."""
from pathlib import Path
import pytest
from basic_memory.sync.sync_service import SyncService
async def create_test_file(path: Path, content: str) -> None:
"""Create a test file with given content."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
@pytest.mark.asyncio
async def test_wikilink_modified_status_issue(sync_service: SyncService, test_config):
"""Test that files with wikilinks don't remain in modified status after sync."""
project_dir = test_config.home
# Create a file with a wikilink
content = """---
title: Test Wikilink
type: note
---
# Test File
This file contains a wikilink to [[another-file]].
"""
test_file_path = project_dir / "test_wikilink.md"
await create_test_file(test_file_path, content)
# Initial sync
report1 = await sync_service.sync(test_config.home)
assert "test_wikilink.md" in report1.new
assert "test_wikilink.md" not in report1.modified
# Sync again without changing the file - should not be modified
report2 = await sync_service.sync(test_config.home)
assert "test_wikilink.md" not in report2.new
assert "test_wikilink.md" not in report2.modified
# Create the target file
target_content = """---
title: Another File
type: note
---
# Another File
This is the target file.
"""
target_file_path = project_dir / "another_file.md"
await create_test_file(target_file_path, target_content)
# Sync again after adding target file
report3 = await sync_service.sync(test_config.home)
assert "another_file.md" in report3.new
assert "test_wikilink.md" not in report3.modified
# Sync one more time - both files should now be stable
report4 = await sync_service.sync(test_config.home)
assert "test_wikilink.md" not in report4.modified
assert "another_file.md" not in report4.modified
+2 -2
View File
@@ -144,14 +144,14 @@ async def test_rapid_atomic_writes(watch_service, test_config):
await asyncio.sleep(0.1)
# Read content to verify
content1 = final_path.read_text()
content1 = final_path.read_text(encoding="utf-8")
assert content1 == "First version"
# Simulate the second atomic write
tmp2_path.rename(final_path)
# Verify content was updated
content2 = final_path.read_text()
content2 = final_path.read_text(encoding="utf-8")
assert content2 == "Second version"
# Create a batch of changes that might arrive in mixed order
+7 -7
View File
@@ -63,7 +63,7 @@ async def test_write_status(watch_service):
await watch_service.write_status()
assert watch_service.status_path.exists()
data = json.loads(watch_service.status_path.read_text())
data = json.loads(watch_service.status_path.read_text(encoding="utf-8"))
assert not data["running"]
assert data["error_count"] == 0
@@ -96,7 +96,7 @@ Test content
# Verify
entity = await watch_service.sync_service.entity_repository.get_by_file_path("new_note.md")
assert entity is not None
assert entity.title == "new_note.md"
assert entity.title == "new_note"
# Check event was recorded
events = [e for e in watch_service.state.recent_events if e.action == "new"]
@@ -125,7 +125,7 @@ Initial content
await create_test_file(test_file, initial_content)
# Initial sync
await watch_service.sync_service.sync(project_dir, show_progress=False)
await watch_service.sync_service.sync(project_dir)
# Modify file
modified_content = """---
@@ -169,7 +169,7 @@ Test content
await create_test_file(test_file, content)
# Initial sync
await watch_service.sync_service.sync(project_dir, show_progress=False)
await watch_service.sync_service.sync(project_dir)
# Delete file
test_file.unlink()
@@ -207,7 +207,7 @@ Test content
await create_test_file(old_path, content)
# Initial sync
await watch_service.sync_service.sync(project_dir, show_progress=False)
await watch_service.sync_service.sync(project_dir)
initial_entity = await watch_service.sync_service.entity_repository.get_by_file_path(
"old/test_move.md"
)
@@ -306,7 +306,7 @@ type: knowledge
Test content for rapid moves
"""
await create_test_file(original_path, content)
await watch_service.sync_service.sync(project_dir, show_progress=False)
await watch_service.sync_service.sync(project_dir)
# Perform rapid moves
temp_path = project_dir / "temp.md"
@@ -393,7 +393,7 @@ This is a test file in a directory
await create_test_file(file_in_dir, content)
# Initial sync to add the file to the database
await watch_service.sync_service.sync(project_dir, show_progress=False)
await watch_service.sync_service.sync(project_dir)
# Rename the directory
new_dir_path = project_dir / "new_dir"
+11 -11
View File
@@ -5,16 +5,16 @@ from pathlib import Path
import pytest
from basic_memory.file_utils import (
compute_checksum,
ensure_directory,
write_file_atomic,
parse_frontmatter,
has_frontmatter,
remove_frontmatter,
FileError,
FileWriteError,
ParseError,
compute_checksum,
ensure_directory,
has_frontmatter,
parse_frontmatter,
remove_frontmatter,
update_frontmatter,
write_file_atomic,
)
@@ -52,7 +52,7 @@ async def test_write_file_atomic(tmp_path: Path):
await write_file_atomic(test_file, content)
assert test_file.exists()
assert test_file.read_text() == content
assert test_file.read_text(encoding="utf-8") == content
# Temp file should be cleaned up
assert not test_file.with_suffix(".tmp").exists()
@@ -185,7 +185,7 @@ async def test_update_frontmatter(tmp_path: Path):
checksum = await update_frontmatter(test_file, updates)
# Verify content
updated = test_file.read_text()
updated = test_file.read_text(encoding="utf-8")
assert "title: Test" in updated
assert "type: note" in updated
assert "Test Content" in updated
@@ -204,13 +204,13 @@ async def test_update_frontmatter(tmp_path: Path):
assert new_checksum != checksum
# Verify content
updated = test_file.read_text()
updated = test_file.read_text(encoding="utf-8")
fm = parse_frontmatter(updated)
assert fm == {"title": "Test", "type": "doc", "tags": ["test"]}
assert "Test Content" in updated
# Test 3: Update with empty dict shouldn't change anything
checksum_before = await compute_checksum(test_file.read_text())
checksum_before = await compute_checksum(test_file.read_text(encoding="utf-8"))
new_checksum = await update_frontmatter(test_file, {})
assert new_checksum == checksum_before
@@ -229,7 +229,7 @@ More content here"""
test_file.write_text(content)
await update_frontmatter(test_file, {"title": "Test"})
updated = test_file.read_text()
updated = test_file.read_text(encoding="utf-8")
assert remove_frontmatter(updated).strip() == content
+50
View File
@@ -0,0 +1,50 @@
"""Tests for parse_tags utility function."""
from typing import List, Union
import pytest
from basic_memory.utils import parse_tags
@pytest.mark.parametrize(
"input_tags,expected",
[
# None input
(None, []),
# List inputs
([], []),
(["tag1", "tag2"], ["tag1", "tag2"]),
(["tag1", "", "tag2"], ["tag1", "tag2"]), # Empty tags are filtered
([" tag1 ", " tag2 "], ["tag1", "tag2"]), # Whitespace is stripped
# String inputs
("", []),
("tag1", ["tag1"]),
("tag1,tag2", ["tag1", "tag2"]),
("tag1, tag2", ["tag1", "tag2"]), # Whitespace after comma is stripped
("tag1,,tag2", ["tag1", "tag2"]), # Empty tags are filtered
# Tags with leading '#' characters - these should be stripped
(["#tag1", "##tag2"], ["tag1", "tag2"]),
("#tag1,##tag2", ["tag1", "tag2"]),
(["tag1", "#tag2", "##tag3"], ["tag1", "tag2", "tag3"]),
# Mixed whitespace and '#' characters
([" #tag1 ", " ##tag2 "], ["tag1", "tag2"]),
(" #tag1 , ##tag2 ", ["tag1", "tag2"]),
],
)
def test_parse_tags(input_tags: Union[List[str], str, None], expected: List[str]) -> None:
"""Test tag parsing with various input formats."""
result = parse_tags(input_tags)
assert result == expected
def test_parse_tags_special_case() -> None:
"""Test parsing from non-string, non-list types."""
# Test with custom object that has __str__ method
class TagObject:
def __str__(self) -> str:
return "tag1,tag2"
result = parse_tags(TagObject()) # pyright: ignore [reportArgumentType]
assert result == ["tag1", "tag2"]
+55 -2
View File
@@ -1,11 +1,13 @@
"""Test permalink formatting during sync."""
import pytest
from pathlib import Path
import pytest
from basic_memory.config import ProjectConfig
from basic_memory.services import EntityService
from basic_memory.sync.sync_service import SyncService
from basic_memory.utils import generate_permalink
async def create_test_file(path: Path, content: str = "test content") -> None:
@@ -57,7 +59,7 @@ Testing permalink generation.
await create_test_file(project_dir / filename, content)
# Run sync
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify permalinks
for filename, expected_permalink in test_cases:
@@ -65,3 +67,54 @@ Testing permalink generation.
assert entity.permalink == expected_permalink, (
f"File {filename} should have permalink {expected_permalink}"
)
@pytest.mark.parametrize(
"input_path, expected",
[
("test/Über File.md", "test/uber-file"),
("docs/résumé.md", "docs/resume"),
("notes/Déjà vu.md", "notes/deja-vu"),
("papers/Jürgen's Findings.md", "papers/jurgens-findings"),
("archive/François Müller.md", "archive/francois-muller"),
("research/Søren Kierkegård.md", "research/soren-kierkegard"),
("articles/El Niño.md", "articles/el-nino"),
],
)
def test_latin_accents_transliteration(input_path, expected):
"""Test that Latin letters with accents are properly transliterated."""
assert generate_permalink(input_path) == expected
@pytest.mark.parametrize(
"input_path, expected",
[
("中文/测试文档.md", "中文/测试文档"),
("notes/北京市.md", "notes/北京市"),
("research/上海简介.md", "research/上海简介"),
("docs/中文 English Mixed.md", "docs/中文-english-mixed"),
("articles/东京Tokyo混合.md", "articles/东京-tokyo-混合"),
("papers/汉字_underscore_test.md", "papers/汉字-underscore-test"),
("projects/中文CamelCase测试.md", "projects/中文-camel-case-测试"),
],
)
def test_chinese_character_preservation(input_path, expected):
"""Test that Chinese characters are preserved in permalinks."""
assert generate_permalink(input_path) == expected
@pytest.mark.parametrize(
"input_path, expected",
[
("mixed/北京Café.md", "mixed/北京-cafe"),
("notes/东京Tōkyō.md", "notes/东京-tokyo"),
("research/München中文.md", "research/munchen-中文"),
("docs/Über测试.md", "docs/uber-测试"),
("complex/北京Beijing上海Shanghai.md", "complex/北京-beijing-上海-shanghai"),
("special/中文!@#$%^&*()_+.md", "special/中文"),
("punctuation/你好,世界!.md", "punctuation/你好世界"),
],
)
def test_mixed_character_sets(input_path, expected):
"""Test handling of mixed character sets and edge cases."""
assert generate_permalink(input_path) == expected
+2 -2
View File
@@ -3,7 +3,7 @@
import pytest
from basic_memory import file_utils
from basic_memory.file_utils import write_file_atomic, compute_checksum
from basic_memory.file_utils import compute_checksum, write_file_atomic
@pytest.mark.asyncio
@@ -130,7 +130,7 @@ permalink: i18n/utf8-document
test_file.write_text(utf8_content, encoding="utf-8")
# Sync the file
await sync_service.sync(test_config.home, show_progress=False)
await sync_service.sync(test_config.home)
# Verify entity was created
entity = await sync_service.entity_service.get_by_permalink("i18n/utf8-document")
Generated
+1 -1
View File
@@ -71,7 +71,7 @@ wheels = [
[[package]]
name = "basic-memory"
version = "0.10.1"
version = "0.12.2"
source = { editable = "." }
dependencies = [
{ name = "aiosqlite" },