style: apply ruff formatting fixes

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/alembic/versions/j3d4e5f6g7h8_rename_entity_type_to_note_type.py

@@ -22,10 +22,7 @@ def table_exists(connection, table_name: str) -> bool:
     """Check if a table exists (idempotent migration support)."""
     if connection.dialect.name == "postgresql":
         result = connection.execute(
-            text(
-                "SELECT 1 FROM information_schema.tables "
-                "WHERE table_name = :table_name"
-            ),
+            text("SELECT 1 FROM information_schema.tables WHERE table_name = :table_name"),
             {"table_name": table_name},
         )
         return result.fetchone() is not None

src/basic_memory/mcp/prompts/continue_conversation.py

@@ -125,7 +125,7 @@ def _format_continuation_results(result: SearchResponse, topic: str) -> str:
         lines.append(f"### {title}")
         if permalink:
             lines.append(f"permalink: {permalink}")
-            lines.append(f"Read with: `read_note(\"{permalink}\")`")
+            lines.append(f'Read with: `read_note("{permalink}")`')
         if item.content:
             content = item.content[:300] + "..." if len(item.content) > 300 else item.content
             lines.append(f"\n{content}")

src/basic_memory/mcp/tools/search.py

@@ -466,7 +466,12 @@ async def search_notes(
             effective_query = (query or "").strip()
             if effective_query:
                 valid_search_types = {
-                    "text", "title", "permalink", "vector", "semantic", "hybrid",
+                    "text",
+                    "title",
+                    "permalink",
+                    "vector",
+                    "semantic",
+                    "hybrid",
                 }
                 if effective_search_type == "text":
                     search_query.text = effective_query

src/basic_memory/repository/postgres_search_repository.py

@@ -689,9 +689,7 @@ async def search(
             for idx, note_type in enumerate(note_types):
                 param_name = f"note_type_{idx}"
                 params[param_name] = json.dumps({"note_type": note_type})
-                type_conditions.append(
-                    f"search_index.metadata @> CAST(:{param_name} AS jsonb)"
-                )
+                type_conditions.append(f"search_index.metadata @> CAST(:{param_name} AS jsonb)")
             conditions.append(f"({' OR '.join(type_conditions)})")
 
         # Handle date filter

src/basic_memory/services/project_service.py

@@ -977,16 +977,13 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
                 total_indexed_entities=total_indexed_entities,
                 vector_tables_exist=False,
                 reindex_recommended=True,
-                reindex_reason=(
-                    "Vector tables not initialized — run: bm reindex --embeddings"
-                ),
+                reindex_reason=("Vector tables not initialized — run: bm reindex --embeddings"),
             )
 
         # --- Count queries (tables exist) ---
         si_result = await self.repository.execute_query(
             text(
-                "SELECT COUNT(DISTINCT entity_id) FROM search_index "
-                "WHERE project_id = :project_id"
+                "SELECT COUNT(DISTINCT entity_id) FROM search_index WHERE project_id = :project_id"
             ),
             {"project_id": project_id},
         )
@@ -1040,9 +1037,7 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
                 "WHERE c.project_id = :project_id AND e.rowid IS NULL"
             )
 
-        orphan_result = await self.repository.execute_query(
-            orphan_sql, {"project_id": project_id}
-        )
+        orphan_result = await self.repository.execute_query(orphan_sql, {"project_id": project_id})
         orphaned_chunks = orphan_result.scalar() or 0
 
         # --- Reindex recommendation logic (priority order) ---
@@ -1051,9 +1046,7 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
 
         if total_indexed_entities > 0 and total_chunks == 0:
             reindex_recommended = True
-            reindex_reason = (
-                "Embeddings have never been built — run: bm reindex --embeddings"
-            )
+            reindex_reason = "Embeddings have never been built — run: bm reindex --embeddings"
         elif orphaned_chunks > 0:
             reindex_recommended = True
             reindex_reason = (
@@ -1063,9 +1056,7 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
         elif total_indexed_entities > total_entities_with_chunks:
             missing = total_indexed_entities - total_entities_with_chunks
             reindex_recommended = True
-            reindex_reason = (
-                f"{missing} entities missing embeddings — run: bm reindex --embeddings"
-            )
+            reindex_reason = f"{missing} entities missing embeddings — run: bm reindex --embeddings"
 
         return EmbeddingStatus(
             semantic_search_enabled=True,

test-int/cli/test_cli_tool_edit_note_integration.py

@@ -262,7 +262,14 @@ def test_edit_note_json_format_contract(app, app_config, test_project, config_ma
 
     assert result.exit_code == 0, result.output
     data = json.loads(result.stdout)
-    assert set(data.keys()) == {"title", "permalink", "file_path", "operation", "checksum", "fileCreated"}
+    assert set(data.keys()) == {
+        "title",
+        "permalink",
+        "file_path",
+        "operation",
+        "checksum",
+        "fileCreated",
+    }
     assert data["operation"] == "append"
     assert data["fileCreated"] is False
     assert data["title"] == "Edit JSON Note"

tests/api/v2/test_knowledge_router.py

@@ -836,9 +836,7 @@ async def test_delete_directory_v2_nested_structure(client: AsyncClient, v2_proj
 
 
 @pytest.mark.asyncio
-async def test_entity_response_includes_user_tracking_fields(
-    client: AsyncClient, v2_project_url
-):
+async def test_entity_response_includes_user_tracking_fields(client: AsyncClient, v2_project_url):
     """EntityResponseV2 includes created_by and last_updated_by fields (null for local)."""
     entity_data = {
         "title": "UserTrackingTest",

tests/repository/test_hybrid_fusion.py

@@ -248,8 +248,11 @@ async def test_dual_source_result_keeps_vector_matched_chunk():
     ]
     vector_results = [
         FakeRow(
-            id=1, score=0.9, title="both",
-            content_snippet=content, matched_chunk_text=vector_chunk,
+            id=1,
+            score=0.9,
+            title="both",
+            content_snippet=content,
+            matched_chunk_text=vector_chunk,
         ),
     ]
 

tests/repository/test_postgres_search_repository.py

@@ -580,8 +580,8 @@ async def test_postgres_note_types_sql_injection_returns_empty(session_maker, te
     repo = PostgresSearchRepository(session_maker, project_id=test_project.id)
 
     malicious_payloads = [
-        'note"}}\' OR \'1\'=\'1',
-        "note\"; DROP TABLE search_index;--",
+        "note\"}}' OR '1'='1",
+        'note"; DROP TABLE search_index;--',
         'note"}} UNION SELECT * FROM entity--',
     ]
     for payload in malicious_payloads:

tests/services/test_project_service_embedding_status.py

@@ -40,7 +40,11 @@ async def test_embedding_status_vector_tables_missing(
     """When vector tables don't exist, recommend reindex."""
     # Drop the chunks table created by the fixture to simulate missing vector tables
     # Postgres requires CASCADE (due to index dependencies); SQLite doesn't support it
-    drop_sql = "DROP TABLE IF EXISTS search_vector_chunks CASCADE" if _is_postgres() else "DROP TABLE IF EXISTS search_vector_chunks"
+    drop_sql = (
+        "DROP TABLE IF EXISTS search_vector_chunks CASCADE"
+        if _is_postgres()
+        else "DROP TABLE IF EXISTS search_vector_chunks"
+    )
     await project_service.repository.execute_query(text(drop_sql), {})
 
     with patch.object(
@@ -112,9 +116,7 @@ async def test_embedding_status_orphaned_chunks(
     # SQLite queries join on rowid which aliases INTEGER PRIMARY KEY.
     await project_service.repository.execute_query(
         text(
-            "CREATE TABLE IF NOT EXISTS search_vector_embeddings ("
-            "  chunk_id INTEGER PRIMARY KEY"
-            ")"
+            "CREATE TABLE IF NOT EXISTS search_vector_embeddings (  chunk_id INTEGER PRIMARY KEY)"
         ),
         {},
     )
@@ -141,14 +143,10 @@ async def test_embedding_status_orphaned_chunks(
 
 
 @pytest.mark.asyncio
-async def test_embedding_status_healthy(
-    project_service: ProjectService, test_graph, test_project
-):
+async def test_embedding_status_healthy(project_service: ProjectService, test_graph, test_project):
     """When all entities have embeddings, no reindex recommended."""
     # Clear any leftover data from prior tests
-    await project_service.repository.execute_query(
-        text("DELETE FROM search_vector_chunks"), {}
-    )
+    await project_service.repository.execute_query(text("DELETE FROM search_vector_chunks"), {})
 
     # Drop any existing virtual table (may have been created by search_service init)
     # and recreate as a simple regular table for testing the join logic.
@@ -158,19 +156,13 @@ async def test_embedding_status_healthy(
         text("DROP TABLE IF EXISTS search_vector_embeddings"), {}
     )
     await project_service.repository.execute_query(
-        text(
-            "CREATE TABLE search_vector_embeddings ("
-            "  chunk_id INTEGER PRIMARY KEY"
-            ")"
-        ),
+        text("CREATE TABLE search_vector_embeddings (  chunk_id INTEGER PRIMARY KEY)"),
         {},
     )
 
     # Insert a chunk + matching embedding for every search_index entity
     entity_result = await project_service.repository.execute_query(
-        text(
-            "SELECT DISTINCT entity_id FROM search_index WHERE project_id = :project_id"
-        ),
+        text("SELECT DISTINCT entity_id FROM search_index WHERE project_id = :project_id"),
         {"project_id": test_project.id},
     )
     entity_ids = [row[0] for row in entity_result.fetchall()]