fix test coverage and type checks

Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
phernandez
2025-03-24 22:45:56 -05:00
parent e716946b44
commit b667bca5a2
12 changed files with 134 additions and 130 deletions
+4 -3
View File
@@ -98,15 +98,16 @@ def set_default_project(
try:
# Set the default project
config_manager.set_default_project(name)
# Also activate it for the current session by setting the environment variable
os.environ["BASIC_MEMORY_PROJECT"] = name
# Reload configuration to apply the change
from importlib import reload
from basic_memory import config as config_module
reload(config_module)
console.print(f"[green]Project '{name}' set as default and activated[/green]")
except ValueError as e: # pragma: no cover
console.print(f"[red]Error: {e}[/red]")
+13 -10
View File
@@ -4,21 +4,22 @@ Uses markdown-it with plugins to parse structured data from markdown content.
"""
from dataclasses import dataclass, field
from pathlib import Path
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
import dateparser
from markdown_it import MarkdownIt
import dateparser
import frontmatter
from markdown_it import MarkdownIt
from basic_memory.markdown.plugins import observation_plugin, relation_plugin
from basic_memory.markdown.schemas import (
EntityMarkdown,
EntityFrontmatter,
EntityMarkdown,
Observation,
Relation,
)
from basic_memory.utils import parse_tags
md = MarkdownIt().use(observation_plugin).use(relation_plugin)
@@ -56,11 +57,11 @@ def parse(content: str) -> EntityContent:
)
def parse_tags(tags: Any) -> list[str]:
"""Parse tags into list of strings."""
if isinstance(tags, (list, tuple)):
return [str(t).strip() for t in tags if str(t).strip()]
return [t.strip() for t in tags.split(",") if t.strip()]
# def parse_tags(tags: Any) -> list[str]:
# """Parse tags into list of strings."""
# if isinstance(tags, (list, tuple)):
# return [str(t).strip() for t in tags if str(t).strip()]
# return [t.strip() for t in tags.split(",") if t.strip()]
class EntityParser:
@@ -101,7 +102,9 @@ class EntityParser:
metadata = post.metadata
metadata["title"] = post.metadata.get("title", absolute_path.name)
metadata["type"] = post.metadata.get("type", "note")
metadata["tags"] = parse_tags(post.metadata.get("tags", []))
tags = parse_tags(post.metadata.get("tags", [])) # pyright: ignore
if tags:
metadata["tags"] = tags
# frontmatter
entity_frontmatter = EntityFrontmatter(
+1 -1
View File
@@ -42,7 +42,7 @@ class EntityFrontmatter(BaseModel):
@property
def tags(self) -> List[str]:
return self.metadata.get("tags") if self.metadata else [] # pyright: ignore
return self.metadata.get("tags") if self.metadata else None # pyright: ignore
@property
def title(self) -> str:
-27
View File
@@ -5,7 +5,6 @@ to the Basic Memory API, with improved error handling and logging.
"""
import typing
from typing import Union, List
from httpx import Response, URL, AsyncClient, HTTPStatusError
from httpx._client import UseClientDefault, USE_CLIENT_DEFAULT
@@ -24,32 +23,6 @@ from loguru import logger
from mcp.server.fastmcp.exceptions import ToolError
def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
"""Parse tags from various input formats into a consistent list.
Args:
tags: Can be a list of strings, a comma-separated string, or None
Returns:
A list of tag strings, or an empty list if no tags
"""
if tags is None:
return []
if isinstance(tags, list):
return tags
if isinstance(tags, str):
return [tag.strip() for tag in tags.split(",") if tag.strip()]
# For any other type, try to convert to string and parse
try:
return parse_tags(str(tags))
except (ValueError, TypeError):
logger.warning(f"Couldn't parse tags from input of type {type(tags)}: {tags}")
return []
def get_error_message(status_code: int, url: URL | str, method: str) -> str:
"""Get a friendly error message based on the HTTP status code.
+4 -3
View File
@@ -6,9 +6,10 @@ 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_put, parse_tags
from basic_memory.mcp.tools.utils import call_put
from basic_memory.schemas import EntityResponse
from basic_memory.schemas.base import Entity
from basic_memory.utils import parse_tags
# Define TagType as a Union that can accept either a string or a list of strings or None
TagType = Union[List[str], str, None]
@@ -21,7 +22,7 @@ async def write_note(
title: str,
content: str,
folder: str,
tags = None, # Remove type hint completely to avoid schema issues
tags=None, # Remove type hint completely to avoid schema issues
) -> str:
"""Write a markdown note to the knowledge base.
@@ -64,7 +65,7 @@ async def write_note(
# Process tags using the helper function
tag_list = parse_tags(tags)
# Create the entity request
metadata = {"tags": [f"#{tag}" for tag in tag_list]} if tag_list else None
entity = Entity(
+4 -7
View File
@@ -146,15 +146,12 @@ class EntityService(BaseService[EntityModel]):
# Create post with new content from schema
post = await schema_to_markdown(schema)
# Merge new metadata with existing metadata
existing_markdown.frontmatter.metadata.update(post.metadata)
# Create a new post with merged metadata
merged_post = frontmatter.Post(
post.content,
**existing_markdown.frontmatter.metadata
)
merged_post = frontmatter.Post(post.content, **existing_markdown.frontmatter.metadata)
# write file
final_content = frontmatter.dumps(merged_post, sort_keys=False)
@@ -322,4 +319,4 @@ class EntityService(BaseService[EntityModel]):
)
continue
return await self.repository.get_by_file_path(path)
return await self.repository.get_by_file_path(path)
+1 -1
View File
@@ -351,4 +351,4 @@ class WatchService:
duration_ms=duration_ms,
)
await self.write_status()
await self.write_status()
+27 -1
View File
@@ -6,7 +6,7 @@ import logging
import re
import sys
from pathlib import Path
from typing import Optional, Protocol, Union, runtime_checkable
from typing import Optional, Protocol, Union, runtime_checkable, List
from loguru import logger
from unidecode import unidecode
@@ -128,3 +128,29 @@ def setup_logging(
# Set log levels for noisy loggers
for logger_name, level in noisy_loggers.items():
logging.getLogger(logger_name).setLevel(level)
def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
"""Parse tags from various input formats into a consistent list.
Args:
tags: Can be a list of strings, a comma-separated string, or None
Returns:
A list of tag strings, or an empty list if no tags
"""
if tags is None:
return []
if isinstance(tags, list):
return tags
if isinstance(tags, str):
return [tag.strip() for tag in tags.split(",") if tag.strip()]
# For any other type, try to convert to string and parse
try: # pragma: no cover
return parse_tags(str(tags))
except (ValueError, TypeError): # pragma: no cover
logger.warning(f"Couldn't parse tags from input of type {type(tags)}: {tags}")
return []