chore: apply lint and formatting fixes for 0.14.4 release (#290)

Signed-off-by: Joe P <joe@basicmemory.com>
Co-authored-by: Claude <noreply@anthropic.com>

src/basic_memory/alembic/versions/a1b2c3d4e5f6_fix_project_foreign_keys.py

@@ -20,14 +20,14 @@ depends_on: Union[str, Sequence[str], None] = None
 
 def upgrade() -> None:
     """Re-establish foreign key constraints that were lost during project table recreation.
-    
+
     The migration 647e7a75e2cd recreated the project table but did not re-establish
     the foreign key constraint from entity.project_id to project.id, causing
     foreign key constraint failures when trying to delete projects with related entities.
     """
     # SQLite doesn't allow adding foreign key constraints to existing tables easily
     # We need to be careful and handle the case where the constraint might already exist
-    
+
     with op.batch_alter_table("entity", schema=None) as batch_op:
         # Try to drop existing foreign key constraint (may not exist)
         try:
@@ -35,19 +35,15 @@ def upgrade() -> None:
         except Exception:
             # Constraint may not exist, which is fine - we'll create it next
             pass
-        
+
         # Add the foreign key constraint with CASCADE DELETE
         # This ensures that when a project is deleted, all related entities are also deleted
         batch_op.create_foreign_key(
-            "fk_entity_project_id", 
-            "project", 
-            ["project_id"], 
-            ["id"],
-            ondelete="CASCADE"
+            "fk_entity_project_id", "project", ["project_id"], ["id"], ondelete="CASCADE"
         )
 
 
 def downgrade() -> None:
     """Remove the foreign key constraint."""
     with op.batch_alter_table("entity", schema=None) as batch_op:
-        batch_op.drop_constraint("fk_entity_project_id", type_="foreignkey")
+        batch_op.drop_constraint("fk_entity_project_id", type_="foreignkey")

src/basic_memory/file_utils.py

@@ -240,40 +240,37 @@ async def update_frontmatter(path: FilePath, updates: Dict[str, Any]) -> str:
 def dump_frontmatter(post: frontmatter.Post) -> str:
     """
     Serialize frontmatter.Post to markdown with Obsidian-compatible YAML format.
-    
+
     This function ensures that tags are formatted as YAML lists instead of JSON arrays:
-    
+
     Good (Obsidian compatible):
     ---
     tags:
     - system
-    - overview  
+    - overview
     - reference
     ---
-    
+
     Bad (current behavior):
     ---
     tags: ["system", "overview", "reference"]
     ---
-    
+
     Args:
         post: frontmatter.Post object to serialize
-        
+
     Returns:
         String containing markdown with properly formatted YAML frontmatter
-    """    
+    """
     if not post.metadata:
         # No frontmatter, just return content
         return post.content
-        
+
     # Serialize YAML with block style for lists
     yaml_str = yaml.dump(
-        post.metadata,
-        sort_keys=False,
-        allow_unicode=True,
-        default_flow_style=False
+        post.metadata, sort_keys=False, allow_unicode=True, default_flow_style=False
     )
-    
+
     # Construct the final markdown with frontmatter
     if post.content:
         return f"---\n{yaml_str}---\n\n{post.content}"
@@ -297,4 +294,3 @@ def sanitize_for_filename(text: str, replacement: str = "-") -> str:
     text = re.sub(f"{re.escape(replacement)}+", replacement, text)
 
     return text.strip(replacement)
-

src/basic_memory/markdown/plugins.py

@@ -9,6 +9,7 @@ from markdown_it.token import Token
 def is_observation(token: Token) -> bool:
     """Check if token looks like our observation format."""
     import re
+
     if token.type != "inline":  # pragma: no cover
         return False
     # Use token.tag which contains the actual content for test tokens, fallback to content
@@ -18,15 +19,15 @@ def is_observation(token: Token) -> bool:
     # if it's a markdown_task, return false
     if content.startswith("[ ]") or content.startswith("[x]") or content.startswith("[-]"):
         return False
-    
+
     # Exclude markdown links: [text](url)
     if re.match(r"^\[.*?\]\(.*?\)$", content):
         return False
-    
+
     # Exclude wiki links: [[text]]
     if re.match(r"^\[\[.*?\]\]$", content):
         return False
-    
+
     # Check for proper observation format: [category] content
     match = re.match(r"^\[([^\[\]()]+)\]\s+(.+)", content)
     has_tags = "#" in content
@@ -36,9 +37,10 @@ def is_observation(token: Token) -> bool:
 def parse_observation(token: Token) -> Dict[str, Any]:
     """Extract observation parts from token."""
     import re
+
     # Use token.tag which contains the actual content for test tokens, fallback to content
     content = (token.tag or token.content).strip()
-    
+
     # Parse [category] with regex
     match = re.match(r"^\[([^\[\]()]+)\]\s+(.+)", content)
     category = None
@@ -50,7 +52,7 @@ def parse_observation(token: Token) -> Dict[str, Any]:
         empty_match = re.match(r"^\[\]\s+(.+)", content)
         if empty_match:
             content = empty_match.group(1).strip()
-    
+
     # Parse (context)
     context = None
     if content.endswith(")"):
@@ -58,7 +60,7 @@ def parse_observation(token: Token) -> Dict[str, Any]:
         if start != -1:
             context = content[start + 1 : -1].strip()
             content = content[:start].strip()
-    
+
     # Extract tags and keep original content
     tags = []
     parts = content.split()
@@ -69,7 +71,7 @@ def parse_observation(token: Token) -> Dict[str, Any]:
                 tags.extend(subtags)
             else:
                 tags.append(part[1:])
-    
+
     return {
         "category": category,
         "content": content,

src/basic_memory/mcp/tools/build_context.py

@@ -17,6 +17,7 @@ from basic_memory.schemas.memory import (
 
 type StringOrInt = str | int
 
+
 @mcp.tool(
     description="""Build context from a memory:// URI to continue conversations naturally.
     
@@ -81,15 +82,16 @@ async def build_context(
         build_context("memory://specs/search", project="work-project")
     """
     logger.info(f"Building context from {url}")
-    
+
     # Convert string depth to integer if needed
     if isinstance(depth, str):
         try:
             depth = int(depth)
         except ValueError:
             from mcp.server.fastmcp.exceptions import ToolError
+
             raise ToolError(f"Invalid depth parameter: '{depth}' is not a valid integer")
-    
+
     # URL is already validated and normalized by MemoryUrl type annotation
 
     # Get the active project first to check project-specific sync status

src/basic_memory/mcp/tools/project_management.py

@@ -223,7 +223,8 @@ async def set_default_project(project_name: str, ctx: Context | None = None) ->
 
     # Call API to set default project using URL encoding for special characters
     from urllib.parse import quote
-    encoded_name = quote(project_name, safe='')
+
+    encoded_name = quote(project_name, safe="")
     response = await call_put(client, f"/projects/{encoded_name}/default")
     status_response = ProjectStatusResponse.model_validate(response.json())
 
@@ -337,7 +338,7 @@ async def delete_project(project_name: str, ctx: Context | None = None) -> str:
         if p.name.lower() == project_name.lower():
             target_project = p
             break
-    
+
     if not target_project:
         available_projects = [p.name for p in project_list.projects]
         raise ValueError(
@@ -346,7 +347,8 @@ async def delete_project(project_name: str, ctx: Context | None = None) -> str:
 
     # Call API to delete project using URL encoding for special characters
     from urllib.parse import quote
-    encoded_name = quote(target_project.name, safe='')
+
+    encoded_name = quote(target_project.name, safe="")
     response = await call_delete(client, f"/projects/{encoded_name}")
     status_response = ProjectStatusResponse.model_validate(response.json())
 

src/basic_memory/mcp/tools/read_note.py

@@ -60,8 +60,10 @@ async def read_note(
     # We need to check both the raw identifier and the processed path
     processed_path = memory_url_path(identifier)
     project_path = active_project.home
-    
-    if not validate_project_path(identifier, project_path) or not validate_project_path(processed_path, project_path):
+
+    if not validate_project_path(identifier, project_path) or not validate_project_path(
+        processed_path, project_path
+    ):
         logger.warning(
             "Attempted path traversal attack blocked",
             identifier=identifier,

src/basic_memory/models/knowledge.py

@@ -74,8 +74,14 @@ class Entity(Base):
     checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True)
 
     # Metadata and tracking
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now().astimezone())
-    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now().astimezone(), onupdate=lambda: datetime.now().astimezone())
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now().astimezone()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now().astimezone(),
+        onupdate=lambda: datetime.now().astimezone(),
+    )
 
     # Relationships
     project = relationship("Project", back_populates="entities")
@@ -104,15 +110,15 @@ class Entity(Base):
     def is_markdown(self):
         """Check if the entity is a markdown file."""
         return self.content_type == "text/markdown"
-    
+
     def __getattribute__(self, name):
         """Override attribute access to ensure datetime fields are timezone-aware."""
         value = super().__getattribute__(name)
-        
+
         # Ensure datetime fields are timezone-aware
-        if name in ('created_at', 'updated_at') and isinstance(value, datetime):
+        if name in ("created_at", "updated_at") and isinstance(value, datetime):
             return ensure_timezone_aware(value)
-        
+
         return value
 
     def __repr__(self) -> str:

src/basic_memory/models/project.py

@@ -52,9 +52,13 @@ class Project(Base):
     is_default: Mapped[Optional[bool]] = mapped_column(Boolean, default=None, nullable=True)
 
     # Timestamps
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(UTC)
+    )
     updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)
+        DateTime(timezone=True),
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
     )
 
     # Define relationships to entities, observations, and relations

src/basic_memory/repository/search_repository.py

@@ -62,7 +62,7 @@ class SearchIndexRow:
 
         # Normalize path separators to handle both Windows (\) and Unix (/) paths
         normalized_path = Path(self.file_path).as_posix()
-        
+
         # Split the path by slashes
         parts = normalized_path.split("/")
 
@@ -527,7 +527,9 @@ class SearchRepository:
         async with db.scoped_session(self.session_maker) as session:
             # Delete existing record if any
             await session.execute(
-                text("DELETE FROM search_index WHERE permalink = :permalink AND project_id = :project_id"),
+                text(
+                    "DELETE FROM search_index WHERE permalink = :permalink AND project_id = :project_id"
+                ),
                 {"permalink": search_index_row.permalink, "project_id": self.project_id},
             )
 

src/basic_memory/schemas/base.py

@@ -71,7 +71,7 @@ def parse_timeframe(timeframe: str) -> datetime:
         parsed = parse(timeframe)
         if not parsed:
             raise ValueError(f"Could not parse timeframe: {timeframe}")
-        
+
         # If the parsed datetime is naive, make it timezone-aware in local system timezone
         if parsed.tzinfo is None:
             return parsed.astimezone()

src/basic_memory/schemas/memory.py

@@ -117,7 +117,7 @@ def memory_url_path(url: memory_url) -> str:  # pyright: ignore
 
 class EntitySummary(BaseModel):
     """Simplified entity representation."""
-    
+
     model_config = ConfigDict(json_encoders={datetime: lambda dt: dt.isoformat()})
 
     type: Literal["entity"] = "entity"
@@ -130,7 +130,7 @@ class EntitySummary(BaseModel):
 
 class RelationSummary(BaseModel):
     """Simplified relation representation."""
-    
+
     model_config = ConfigDict(json_encoders={datetime: lambda dt: dt.isoformat()})
 
     type: Literal["relation"] = "relation"
@@ -145,7 +145,7 @@ class RelationSummary(BaseModel):
 
 class ObservationSummary(BaseModel):
     """Simplified observation representation."""
-    
+
     model_config = ConfigDict(json_encoders={datetime: lambda dt: dt.isoformat()})
 
     type: Literal["observation"] = "observation"
@@ -159,7 +159,7 @@ class ObservationSummary(BaseModel):
 
 class MemoryMetadata(BaseModel):
     """Simplified response metadata."""
-    
+
     model_config = ConfigDict(json_encoders={datetime: lambda dt: dt.isoformat()})
 
     uri: Optional[str] = None
@@ -178,8 +178,8 @@ class ContextResult(BaseModel):
     """Context result containing a primary item with its observations and related items."""
 
     primary_result: Annotated[
-        Union[EntitySummary, RelationSummary, ObservationSummary], 
-        Field(discriminator="type", description="Primary item")
+        Union[EntitySummary, RelationSummary, ObservationSummary],
+        Field(discriminator="type", description="Primary item"),
     ]
 
     observations: Sequence[ObservationSummary] = Field(
@@ -188,8 +188,7 @@ class ContextResult(BaseModel):
 
     related_results: Sequence[
         Annotated[
-            Union[EntitySummary, RelationSummary, ObservationSummary], 
-            Field(discriminator="type")
+            Union[EntitySummary, RelationSummary, ObservationSummary], Field(discriminator="type")
         ]
     ] = Field(description="Related items", default_factory=list)
 

src/basic_memory/services/context_service.py

@@ -246,7 +246,11 @@ class ContextService:
         values = ", ".join([f"('{t}', {i})" for t, i in type_id_pairs])
 
         # Parameters for bindings - include project_id for security filtering
-        params = {"max_depth": max_depth, "max_results": max_results, "project_id": self.search_repository.project_id}
+        params = {
+            "max_depth": max_depth,
+            "max_results": max_results,
+            "project_id": self.search_repository.project_id,
+        }
 
         # Build date and timeframe filters conditionally based on since parameter
         if since:
@@ -258,7 +262,7 @@ class ContextService:
             date_filter = ""
             relation_date_filter = ""
             timeframe_condition = ""
-        
+
         # Add project filtering for security - ensure all entities and relations belong to the same project
         project_filter = "AND e.project_id = :project_id"
         relation_project_filter = "AND e_from.project_id = :project_id"

src/basic_memory/services/entity_service.py

@@ -9,7 +9,12 @@ from loguru import logger
 from sqlalchemy.exc import IntegrityError
 
 from basic_memory.config import ProjectConfig, BasicMemoryConfig
-from basic_memory.file_utils import has_frontmatter, parse_frontmatter, remove_frontmatter, dump_frontmatter
+from basic_memory.file_utils import (
+    has_frontmatter,
+    parse_frontmatter,
+    remove_frontmatter,
+    dump_frontmatter,
+)
 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

src/basic_memory/sync/watch_service.py

@@ -288,9 +288,13 @@ class WatchService:
                 full_path = directory / path
                 if full_path.exists() and full_path.is_file():
                     # File still exists despite DELETE event - treat as modification
-                    logger.debug("File exists despite DELETE event, treating as modification", path=path)
+                    logger.debug(
+                        "File exists despite DELETE event, treating as modification", path=path
+                    )
                     entity, checksum = await sync_service.sync_file(path, new=False)
-                    self.state.add_event(path=path, action="modified", status="success", checksum=checksum)
+                    self.state.add_event(
+                        path=path, action="modified", status="success", checksum=checksum
+                    )
                     self.console.print(f"[yellow]✎[/yellow] {path} (atomic write)")
                     logger.info(f"atomic write detected: {path}")
                     processed.add(path)
@@ -302,10 +306,12 @@ class WatchService:
                     entity = await sync_service.entity_repository.get_by_file_path(path)
                     if entity is None:
                         # No entity means this was likely a directory - skip it
-                        logger.debug(f"Skipping deleted path with no entity (likely directory), path={path}")
+                        logger.debug(
+                            f"Skipping deleted path with no entity (likely directory), path={path}"
+                        )
                         processed.add(path)
                         continue
-                    
+
                     # File truly deleted
                     logger.debug("Processing deleted file", path=path)
                     await sync_service.handle_delete(path)

src/basic_memory/utils.py

@@ -223,7 +223,8 @@ def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
     if isinstance(tags, str):
         # Check if it's a JSON array string (common issue from AI assistants)
         import json
-        if tags.strip().startswith('[') and tags.strip().endswith(']'):
+
+        if tags.strip().startswith("[") and tags.strip().endswith("]"):
             try:
                 # Try to parse as JSON array
                 parsed_json = json.loads(tags)
@@ -233,7 +234,7 @@ def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
             except json.JSONDecodeError:
                 # Not valid JSON, fall through to comma-separated parsing
                 pass
-        
+
         # Split by comma, strip whitespace, then strip leading '#' characters
         return [tag.strip().lstrip("#") for tag in tags.split(",") if tag and tag.strip()]