mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d8cb7313db |
@@ -38,6 +38,7 @@ from basic_memory.schemas.v2 import (
|
||||
MoveEntityRequestV2,
|
||||
MoveDirectoryRequestV2,
|
||||
DeleteDirectoryRequestV2,
|
||||
OrphanEntitiesResponse,
|
||||
)
|
||||
from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
|
||||
|
||||
@@ -110,6 +111,43 @@ async def get_graph(
|
||||
return GraphResponse(nodes=nodes, edges=edges)
|
||||
|
||||
|
||||
## Orphan entities endpoint
|
||||
|
||||
|
||||
@router.get("/orphans", response_model=OrphanEntitiesResponse)
|
||||
async def get_orphan_entities(
|
||||
project_id: ProjectExternalIdPathDep,
|
||||
entity_repository: EntityRepositoryV2ExternalDep,
|
||||
) -> OrphanEntitiesResponse:
|
||||
"""Return entities that have no relations in the knowledge graph.
|
||||
|
||||
Orphan entities have no incoming or outgoing connections — they are isolated
|
||||
nodes that may indicate newly created notes not yet linked to the graph,
|
||||
or entities whose relations have been removed.
|
||||
"""
|
||||
with logfire.span(
|
||||
"api.request.knowledge.get_orphans",
|
||||
entrypoint="api",
|
||||
domain="knowledge",
|
||||
action="get_orphans",
|
||||
):
|
||||
logger.info("API v2 request: get_orphan_entities")
|
||||
|
||||
entities = await entity_repository.find_without_relations()
|
||||
nodes = [
|
||||
GraphNode(
|
||||
external_id=entity.external_id,
|
||||
title=entity.title,
|
||||
note_type=entity.note_type,
|
||||
file_path=entity.file_path,
|
||||
)
|
||||
for entity in entities
|
||||
]
|
||||
|
||||
logger.info(f"API v2 response: {len(nodes)} orphan entities")
|
||||
return OrphanEntitiesResponse(entities=nodes, total=len(nodes))
|
||||
|
||||
|
||||
## Resolution endpoint
|
||||
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ from . import status, db, doctor, import_memory_json, mcp, import_claude_convers
|
||||
from . import (
|
||||
import_claude_projects,
|
||||
import_chatgpt,
|
||||
orphans,
|
||||
tool,
|
||||
project,
|
||||
format,
|
||||
@@ -20,6 +21,7 @@ __all__ = [
|
||||
"import_claude_conversations",
|
||||
"import_claude_projects",
|
||||
"import_chatgpt",
|
||||
"orphans",
|
||||
"tool",
|
||||
"project",
|
||||
"format",
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Orphans command - show entities with no relations in the knowledge graph."""
|
||||
|
||||
import json
|
||||
from typing import Annotated, Optional
|
||||
|
||||
import typer
|
||||
from loguru import logger
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
from basic_memory.cli.app import app
|
||||
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
|
||||
from basic_memory.config import ConfigManager
|
||||
from basic_memory.mcp.async_client import get_client
|
||||
from basic_memory.mcp.clients.knowledge import KnowledgeClient
|
||||
from basic_memory.mcp.project_context import get_active_project
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
async def run_orphans(project: Optional[str] = None) -> tuple[str, list[dict]]:
|
||||
"""Fetch entities that have no relations in the knowledge graph."""
|
||||
project = project or ConfigManager().default_project
|
||||
|
||||
async with get_client(project_name=project) as client:
|
||||
project_item = await get_active_project(client, project, None)
|
||||
entities = await KnowledgeClient(client, project_item.external_id).get_orphans()
|
||||
return project_item.name, entities
|
||||
|
||||
|
||||
@app.command()
|
||||
def orphans(
|
||||
project: Annotated[
|
||||
Optional[str],
|
||||
typer.Option(help="The project name."),
|
||||
] = None,
|
||||
json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
|
||||
local: bool = typer.Option(
|
||||
False, "--local", help="Force local API routing (ignore cloud mode)"
|
||||
),
|
||||
cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
|
||||
):
|
||||
"""Show entities that have no relations in the knowledge graph.
|
||||
|
||||
Orphan entities have no incoming or outgoing connections. These may indicate
|
||||
newly created notes not yet linked to other entities, or notes that have had
|
||||
their relations removed.
|
||||
|
||||
Use --json for machine-readable output.
|
||||
Use --local to force local routing when cloud mode is enabled.
|
||||
Use --cloud to force cloud routing when cloud mode is disabled.
|
||||
"""
|
||||
from basic_memory.cli.commands.command_utils import run_with_cleanup
|
||||
|
||||
try:
|
||||
validate_routing_flags(local, cloud)
|
||||
if not local and not cloud:
|
||||
local = True
|
||||
with force_routing(local=local, cloud=cloud):
|
||||
project_name, entities = run_with_cleanup(run_orphans(project))
|
||||
|
||||
if json_output:
|
||||
print(json.dumps(entities, indent=2, default=str))
|
||||
else:
|
||||
if not entities:
|
||||
console.print(
|
||||
f"[green]No orphan entities in project '{project_name}'[/green]"
|
||||
)
|
||||
return
|
||||
|
||||
table = Table(
|
||||
title=f"{project_name}: Entities Without Relations ({len(entities)} total)"
|
||||
)
|
||||
table.add_column("Title", style="cyan")
|
||||
table.add_column("File Path", style="yellow")
|
||||
table.add_column("Type", style="green")
|
||||
|
||||
for entity in entities:
|
||||
table.add_row(
|
||||
entity.get("title", ""),
|
||||
entity.get("file_path", ""),
|
||||
entity.get("note_type") or "",
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
|
||||
except (ValueError, typer.Exit):
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching orphan entities: {e}")
|
||||
if json_output:
|
||||
print(json.dumps({"error": str(e)}, indent=2))
|
||||
else:
|
||||
console.print(f"[red]Error: {e}[/red]")
|
||||
raise typer.Exit(code=1) # pragma: no cover
|
||||
@@ -24,6 +24,7 @@ if not _version_only_invocation(sys.argv[1:]):
|
||||
import_claude_projects,
|
||||
import_memory_json,
|
||||
mcp,
|
||||
orphans,
|
||||
project,
|
||||
schema,
|
||||
status,
|
||||
|
||||
@@ -275,6 +275,31 @@ class KnowledgeClient:
|
||||
)
|
||||
return DirectoryDeleteResult.model_validate(response.json())
|
||||
|
||||
# --- Orphan detection ---
|
||||
|
||||
async def get_orphans(self) -> list[dict]:
|
||||
"""Get entities that have no incoming or outgoing relations.
|
||||
|
||||
Returns:
|
||||
List of entity dicts with external_id, title, note_type, file_path
|
||||
|
||||
Raises:
|
||||
ToolError: If the request fails
|
||||
"""
|
||||
with logfire.span(
|
||||
"mcp.client.knowledge.get_orphans",
|
||||
client_name="knowledge",
|
||||
operation="get_orphans",
|
||||
):
|
||||
response = await call_get(
|
||||
self.http_client,
|
||||
f"{self._base_path}/orphans",
|
||||
client_name="knowledge",
|
||||
operation="get_orphans",
|
||||
path_template="/v2/projects/{project_id}/knowledge/orphans",
|
||||
)
|
||||
return response.json()["entities"]
|
||||
|
||||
# --- Resolution ---
|
||||
|
||||
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
|
||||
|
||||
@@ -5,7 +5,7 @@ from typing import List, Optional, Sequence, Union, Any
|
||||
|
||||
|
||||
from loguru import logger
|
||||
from sqlalchemy import select, func
|
||||
from sqlalchemy import exists, func, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
from sqlalchemy.orm import selectinload
|
||||
@@ -436,6 +436,32 @@ class EntityRepository(Repository[Entity]):
|
||||
result = await self.execute_query(query, use_query_options=False)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def find_without_relations(self) -> Sequence[Entity]:
|
||||
"""Find entities that have no incoming or outgoing relations.
|
||||
|
||||
An orphan entity has no entries as from_id in any relation and no resolved
|
||||
entries as to_id. These are isolated nodes in the knowledge graph.
|
||||
|
||||
Returns:
|
||||
Sequence of entities with no outgoing or incoming relations
|
||||
"""
|
||||
# Trigger: entity appears as a source in any relation (outgoing link)
|
||||
# Why: even unresolved outgoing links mean the entity references something
|
||||
# Outcome: entities with any outgoing relation are excluded
|
||||
has_outgoing = exists().where(Relation.from_id == Entity.id)
|
||||
|
||||
# Trigger: entity appears as a resolved target in any relation (incoming link)
|
||||
# Why: to_id is null for unresolved links; only resolved links form graph edges
|
||||
# Outcome: entities referenced by resolved links are excluded
|
||||
has_incoming = exists().where(
|
||||
Relation.to_id == Entity.id,
|
||||
Relation.to_id.is_not(None),
|
||||
)
|
||||
|
||||
query = self.select().where(~has_outgoing).where(~has_incoming)
|
||||
result = await self.execute_query(query, use_query_options=False)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_distinct_directories(self) -> List[str]:
|
||||
"""Extract unique directory paths from file_path column.
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ from basic_memory.schemas.v2.graph import (
|
||||
GraphEdge,
|
||||
GraphNode,
|
||||
GraphResponse,
|
||||
OrphanEntitiesResponse,
|
||||
)
|
||||
from basic_memory.schemas.v2.resource import (
|
||||
CreateResourceRequest,
|
||||
@@ -33,6 +34,7 @@ __all__ = [
|
||||
"GraphEdge",
|
||||
"GraphNode",
|
||||
"GraphResponse",
|
||||
"OrphanEntitiesResponse",
|
||||
"CreateResourceRequest",
|
||||
"UpdateResourceRequest",
|
||||
"ResourceResponse",
|
||||
|
||||
@@ -29,3 +29,12 @@ class GraphResponse(BaseModel):
|
||||
edges: list[GraphEdge] = Field(
|
||||
default_factory=list, description="All resolved relations as edges"
|
||||
)
|
||||
|
||||
|
||||
class OrphanEntitiesResponse(BaseModel):
|
||||
"""Entities that have no incoming or outgoing relations in the knowledge graph."""
|
||||
|
||||
entities: list[GraphNode] = Field(
|
||||
default_factory=list, description="Entities with no relations"
|
||||
)
|
||||
total: int = Field(..., description="Total count of orphan entities")
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Tests for the /knowledge/orphans API endpoint."""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_orphan_entities_empty_project(client: AsyncClient, v2_project_url):
|
||||
"""An empty project returns an empty orphans list."""
|
||||
response = await client.get(f"{v2_project_url}/knowledge/orphans")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert data["entities"] == []
|
||||
assert data["total"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_orphan_entities_returns_unlinked_entities(
|
||||
client: AsyncClient, v2_project_url
|
||||
):
|
||||
"""Entities with no relations appear in the orphans endpoint."""
|
||||
r1 = await client.post(
|
||||
f"{v2_project_url}/knowledge/entities",
|
||||
json={"title": "Orphan One", "directory": "orphan", "content": "No links here"},
|
||||
)
|
||||
assert r1.status_code == 200
|
||||
|
||||
r2 = await client.post(
|
||||
f"{v2_project_url}/knowledge/entities",
|
||||
json={"title": "Orphan Two", "directory": "orphan", "content": "Also no links"},
|
||||
)
|
||||
assert r2.status_code == 200
|
||||
|
||||
response = await client.get(f"{v2_project_url}/knowledge/orphans")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert "entities" in data
|
||||
assert "total" in data
|
||||
titles = {e["title"] for e in data["entities"]}
|
||||
assert "Orphan One" in titles
|
||||
assert "Orphan Two" in titles
|
||||
assert data["total"] == len(data["entities"])
|
||||
assert data["total"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_orphan_entities_excludes_entity_with_outgoing_relation(
|
||||
client: AsyncClient, v2_project_url
|
||||
):
|
||||
"""An entity with an outgoing wiki-link relation is excluded from orphans."""
|
||||
# Source entity references another via wikilink in content
|
||||
r_source = await client.post(
|
||||
f"{v2_project_url}/knowledge/entities",
|
||||
json={
|
||||
"title": "Source Note",
|
||||
"directory": "linked",
|
||||
"content": "- links_to [[Target Note]]",
|
||||
},
|
||||
)
|
||||
assert r_source.status_code == 200
|
||||
|
||||
# Target entity (no outgoing links)
|
||||
await client.post(
|
||||
f"{v2_project_url}/knowledge/entities",
|
||||
json={"title": "Target Note", "directory": "linked", "content": "Referenced entity"},
|
||||
)
|
||||
|
||||
# Unlinked entity - should appear in orphans
|
||||
await client.post(
|
||||
f"{v2_project_url}/knowledge/entities",
|
||||
json={"title": "Standalone Note", "directory": "linked", "content": "No links at all"},
|
||||
)
|
||||
|
||||
response = await client.get(f"{v2_project_url}/knowledge/orphans")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
titles = {e["title"] for e in data["entities"]}
|
||||
|
||||
# Source has outgoing relation — not an orphan
|
||||
assert "Source Note" not in titles
|
||||
# Standalone has no links — is an orphan
|
||||
assert "Standalone Note" in titles
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_orphan_entities_response_shape(client: AsyncClient, v2_project_url):
|
||||
"""Each entity in the response has the expected fields."""
|
||||
await client.post(
|
||||
f"{v2_project_url}/knowledge/entities",
|
||||
json={"title": "Shape Test", "directory": "shape", "content": "Testing response shape"},
|
||||
)
|
||||
|
||||
response = await client.get(f"{v2_project_url}/knowledge/orphans")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert data["total"] >= 1
|
||||
|
||||
entity = next(e for e in data["entities"] if e["title"] == "Shape Test")
|
||||
assert "external_id" in entity
|
||||
assert "title" in entity
|
||||
assert "file_path" in entity
|
||||
assert "note_type" in entity
|
||||
assert entity["title"] == "Shape Test"
|
||||
assert entity["file_path"].endswith(".md")
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Tests for the 'basic-memory orphans' CLI command."""
|
||||
|
||||
import json
|
||||
from contextlib import asynccontextmanager
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from basic_memory.cli.main import app as cli_app
|
||||
|
||||
# Register the orphans command on the shared app.
|
||||
import basic_memory.cli.commands.orphans as orphans_cmd # noqa: F401
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
_MOCK_PROJECT_ITEM = MagicMock()
|
||||
_MOCK_PROJECT_ITEM.name = "test-project"
|
||||
_MOCK_PROJECT_ITEM.external_id = "11111111-1111-1111-1111-111111111111"
|
||||
|
||||
_ORPHAN_ENTITIES = [
|
||||
{
|
||||
"external_id": "aaaa-1111",
|
||||
"title": "Isolated Note",
|
||||
"file_path": "notes/isolated.md",
|
||||
"note_type": "note",
|
||||
},
|
||||
{
|
||||
"external_id": "bbbb-2222",
|
||||
"title": "Dangling Spec",
|
||||
"file_path": "specs/dangling.md",
|
||||
"note_type": "spec",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def _mock_config_manager():
|
||||
mock_cm = MagicMock()
|
||||
mock_cm.default_project = "test-project"
|
||||
return mock_cm
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _fake_get_client(project_name=None):
|
||||
yield MagicMock()
|
||||
|
||||
|
||||
@patch("basic_memory.cli.commands.orphans.ConfigManager")
|
||||
@patch("basic_memory.cli.commands.orphans.get_active_project", new_callable=AsyncMock)
|
||||
@patch("basic_memory.cli.commands.orphans.get_client")
|
||||
@patch("basic_memory.cli.commands.orphans.KnowledgeClient")
|
||||
def test_orphans_json_output(mock_knowledge_cls, mock_get_client, mock_get_active, mock_config_cls):
|
||||
"""bm orphans --json outputs a JSON array of entity objects."""
|
||||
mock_config_cls.return_value = _mock_config_manager()
|
||||
mock_get_active.return_value = _MOCK_PROJECT_ITEM
|
||||
mock_get_client.side_effect = _fake_get_client
|
||||
|
||||
mock_knowledge_instance = AsyncMock()
|
||||
mock_knowledge_instance.get_orphans.return_value = _ORPHAN_ENTITIES
|
||||
mock_knowledge_cls.return_value = mock_knowledge_instance
|
||||
|
||||
result = runner.invoke(cli_app, ["orphans", "--json"])
|
||||
|
||||
assert result.exit_code == 0, f"CLI failed: {result.output}"
|
||||
start = result.output.index("[")
|
||||
data = json.loads(result.output[start:])
|
||||
assert len(data) == 2
|
||||
titles = {e["title"] for e in data}
|
||||
assert "Isolated Note" in titles
|
||||
assert "Dangling Spec" in titles
|
||||
|
||||
|
||||
@patch("basic_memory.cli.commands.orphans.ConfigManager")
|
||||
@patch("basic_memory.cli.commands.orphans.get_active_project", new_callable=AsyncMock)
|
||||
@patch("basic_memory.cli.commands.orphans.get_client")
|
||||
@patch("basic_memory.cli.commands.orphans.KnowledgeClient")
|
||||
def test_orphans_table_output(mock_knowledge_cls, mock_get_client, mock_get_active, mock_config_cls):
|
||||
"""bm orphans (default) renders a Rich table with titles and paths."""
|
||||
mock_config_cls.return_value = _mock_config_manager()
|
||||
mock_get_active.return_value = _MOCK_PROJECT_ITEM
|
||||
mock_get_client.side_effect = _fake_get_client
|
||||
|
||||
mock_knowledge_instance = AsyncMock()
|
||||
mock_knowledge_instance.get_orphans.return_value = _ORPHAN_ENTITIES
|
||||
mock_knowledge_cls.return_value = mock_knowledge_instance
|
||||
|
||||
result = runner.invoke(cli_app, ["orphans"])
|
||||
|
||||
assert result.exit_code == 0, f"CLI failed: {result.output}"
|
||||
assert "Isolated Note" in result.output
|
||||
assert "Dangling Spec" in result.output
|
||||
assert "notes/isolated.md" in result.output
|
||||
|
||||
|
||||
@patch("basic_memory.cli.commands.orphans.ConfigManager")
|
||||
@patch("basic_memory.cli.commands.orphans.get_active_project", new_callable=AsyncMock)
|
||||
@patch("basic_memory.cli.commands.orphans.get_client")
|
||||
@patch("basic_memory.cli.commands.orphans.KnowledgeClient")
|
||||
def test_orphans_no_results(mock_knowledge_cls, mock_get_client, mock_get_active, mock_config_cls):
|
||||
"""bm orphans prints a success message when no orphans are found."""
|
||||
mock_config_cls.return_value = _mock_config_manager()
|
||||
mock_get_active.return_value = _MOCK_PROJECT_ITEM
|
||||
mock_get_client.side_effect = _fake_get_client
|
||||
|
||||
mock_knowledge_instance = AsyncMock()
|
||||
mock_knowledge_instance.get_orphans.return_value = []
|
||||
mock_knowledge_cls.return_value = mock_knowledge_instance
|
||||
|
||||
result = runner.invoke(cli_app, ["orphans"])
|
||||
|
||||
assert result.exit_code == 0, f"CLI failed: {result.output}"
|
||||
assert "No orphan entities" in result.output
|
||||
@@ -1114,3 +1114,108 @@ async def test_get_file_path_to_permalink_map(entity_repository: EntityRepositor
|
||||
assert len(mapping) == 2
|
||||
assert mapping["test/entity1.md"] == "test/entity1"
|
||||
assert mapping["test/entity2.md"] == "test/entity2"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_without_relations_returns_isolated_entities(
|
||||
entity_repository: EntityRepository, session_maker, test_project: Project
|
||||
):
|
||||
"""Entities with no outgoing or incoming relations are returned as orphans."""
|
||||
async with db.scoped_session(session_maker) as session:
|
||||
orphan = Entity(
|
||||
project_id=test_project.id,
|
||||
title="Orphan",
|
||||
note_type="test",
|
||||
permalink="orphan/orphan",
|
||||
file_path="orphan/orphan.md",
|
||||
content_type="text/markdown",
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
source = Entity(
|
||||
project_id=test_project.id,
|
||||
title="Source",
|
||||
note_type="test",
|
||||
permalink="source/source",
|
||||
file_path="source/source.md",
|
||||
content_type="text/markdown",
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
target = Entity(
|
||||
project_id=test_project.id,
|
||||
title="Target",
|
||||
note_type="test",
|
||||
permalink="target/target",
|
||||
file_path="target/target.md",
|
||||
content_type="text/markdown",
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
session.add_all([orphan, source, target])
|
||||
await session.flush()
|
||||
|
||||
relation = Relation(
|
||||
project_id=test_project.id,
|
||||
from_id=source.id,
|
||||
to_id=target.id,
|
||||
to_name=target.title,
|
||||
relation_type="links_to",
|
||||
)
|
||||
session.add(relation)
|
||||
|
||||
result = await entity_repository.find_without_relations()
|
||||
titles = {e.title for e in result}
|
||||
|
||||
assert "Orphan" in titles
|
||||
assert "Source" not in titles
|
||||
assert "Target" not in titles
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_without_relations_empty_project(entity_repository: EntityRepository):
|
||||
"""An empty project returns no orphans."""
|
||||
result = await entity_repository.find_without_relations()
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_without_relations_all_connected(
|
||||
entity_repository: EntityRepository, session_maker, test_project: Project
|
||||
):
|
||||
"""When all entities are connected, no orphans are returned."""
|
||||
async with db.scoped_session(session_maker) as session:
|
||||
a = Entity(
|
||||
project_id=test_project.id,
|
||||
title="A",
|
||||
note_type="test",
|
||||
permalink="conn/a",
|
||||
file_path="conn/a.md",
|
||||
content_type="text/markdown",
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
b = Entity(
|
||||
project_id=test_project.id,
|
||||
title="B",
|
||||
note_type="test",
|
||||
permalink="conn/b",
|
||||
file_path="conn/b.md",
|
||||
content_type="text/markdown",
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
session.add_all([a, b])
|
||||
await session.flush()
|
||||
session.add(
|
||||
Relation(
|
||||
project_id=test_project.id,
|
||||
from_id=a.id,
|
||||
to_id=b.id,
|
||||
to_name=b.title,
|
||||
relation_type="connects",
|
||||
)
|
||||
)
|
||||
|
||||
result = await entity_repository.find_without_relations()
|
||||
assert result == []
|
||||
|
||||
Reference in New Issue
Block a user