feat: Add bulk insert with ON CONFLICT handling for relations

Add add_all_ignore_duplicates() method to RelationRepository for bulk
inserting relations with ON CONFLICT DO NOTHING. This handles cases
where the same [[wiki link]] appears multiple times in a document,
silently ignoring duplicates based on the (from_id, to_name, relation_type)
unique constraint.

Works with both SQLite and PostgreSQL dialects.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: phernandez

src/basic_memory/repository/relation_repository.py

@@ -4,6 +4,8 @@
 
 import logfire
 from sqlalchemy import and_, delete, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlalchemy.orm import selectinload, aliased
 from sqlalchemy.orm.interfaces import LoaderOption
@@ -92,5 +94,56 @@ async def find_unresolved_relations_for_entity(self, entity_id: int) -> Sequence
         result = await self.execute_query(query)
         return result.scalars().all()
 
+    @logfire.instrument()
+    async def add_all_ignore_duplicates(self, relations: List[Relation]) -> int:
+        """Bulk insert relations, ignoring duplicates.
+
+        Uses ON CONFLICT DO NOTHING to skip relations that would violate the
+        unique constraint on (from_id, to_name, relation_type). This is useful
+        for bulk operations where the same link may appear multiple times in
+        a document.
+
+        Works with both SQLite and PostgreSQL dialects.
+
+        Args:
+            relations: List of Relation objects to insert
+
+        Returns:
+            Number of relations actually inserted (excludes duplicates)
+        """
+        if not relations:
+            return 0
+
+        # Convert Relation objects to dicts for insert
+        values = [
+            {
+                "from_id": r.from_id,
+                "to_id": r.to_id,
+                "to_name": r.to_name,
+                "relation_type": r.relation_type,
+                "context": r.context,
+            }
+            for r in relations
+        ]
+
+        async with db.scoped_session(self.session_maker) as session:
+            # Check dialect to use appropriate insert
+            dialect_name = session.bind.dialect.name if session.bind else "sqlite"
+
+            if dialect_name == "postgresql":
+                stmt = pg_insert(Relation).values(values)
+                stmt = stmt.on_conflict_do_nothing(
+                    index_elements=["from_id", "to_name", "relation_type"]
+                )
+            else:
+                # SQLite
+                stmt = sqlite_insert(Relation).values(values)
+                stmt = stmt.on_conflict_do_nothing(
+                    index_elements=["from_id", "to_name", "relation_type"]
+                )
+
+            result = await session.execute(stmt)
+            return result.rowcount if result.rowcount else 0
+
     def get_load_options(self) -> List[LoaderOption]:
         return [selectinload(Relation.from_entity), selectinload(Relation.to_entity)]

tests/repository/test_relation_repository.py

@@ -349,3 +349,156 @@ async def test_delete_nonexistent_relation(relation_repository):
     """Test deleting a relation that doesn't exist."""
     result = await relation_repository.delete_by_fields(relation_type="nonexistent")
     assert result is False
+
+
+# -------------------------------------------------------------------------
+# Tests for add_all_ignore_duplicates
+# -------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_add_all_ignore_duplicates_basic(
+    relation_repository: RelationRepository, sample_entity: Entity, related_entity: Entity
+):
+    """Test bulk inserting relations with ON CONFLICT DO NOTHING."""
+    relations = [
+        Relation(
+            from_id=sample_entity.id,
+            to_id=related_entity.id,
+            to_name=related_entity.title,
+            relation_type="links_to",
+        ),
+        Relation(
+            from_id=sample_entity.id,
+            to_id=related_entity.id,
+            to_name=related_entity.title,
+            relation_type="references",
+        ),
+    ]
+
+    inserted = await relation_repository.add_all_ignore_duplicates(relations)
+
+    # Both should be inserted
+    assert inserted == 2
+
+    # Verify they exist
+    found = await relation_repository.find_by_entities(sample_entity.id, related_entity.id)
+    assert len(found) == 2
+    relation_types = {r.relation_type for r in found}
+    assert relation_types == {"links_to", "references"}
+
+
+@pytest.mark.asyncio
+async def test_add_all_ignore_duplicates_skips_duplicates(
+    relation_repository: RelationRepository, sample_entity: Entity, related_entity: Entity
+):
+    """Test that duplicate relations are silently ignored."""
+    # Same relation appearing multiple times (common when same [[link]] appears twice in doc)
+    relations = [
+        Relation(
+            from_id=sample_entity.id,
+            to_id=None,  # Unresolved
+            to_name="Some Target",
+            relation_type="links_to",
+        ),
+        Relation(
+            from_id=sample_entity.id,
+            to_id=None,
+            to_name="Some Target",  # Duplicate!
+            relation_type="links_to",
+        ),
+        Relation(
+            from_id=sample_entity.id,
+            to_id=None,
+            to_name="Some Target",  # Triple duplicate!
+            relation_type="links_to",
+        ),
+    ]
+
+    inserted = await relation_repository.add_all_ignore_duplicates(relations)
+
+    # Only 1 should be inserted (duplicates ignored)
+    assert inserted == 1
+
+    # Verify only one exists
+    all_relations = await relation_repository.find_all()
+    matching = [r for r in all_relations if r.to_name == "Some Target"]
+    assert len(matching) == 1
+
+
+@pytest.mark.asyncio
+async def test_add_all_ignore_duplicates_empty_list(relation_repository: RelationRepository):
+    """Test with empty list returns 0."""
+    inserted = await relation_repository.add_all_ignore_duplicates([])
+    assert inserted == 0
+
+
+@pytest.mark.asyncio
+async def test_add_all_ignore_duplicates_mixed(
+    relation_repository: RelationRepository, sample_entity: Entity, related_entity: Entity
+):
+    """Test with mix of new and duplicate relations."""
+    # First, insert one relation
+    first_relation = Relation(
+        from_id=sample_entity.id,
+        to_id=None,
+        to_name="Existing Target",
+        relation_type="links_to",
+    )
+    await relation_repository.add_all_ignore_duplicates([first_relation])
+
+    # Now try to insert a mix of new and duplicate
+    relations = [
+        Relation(
+            from_id=sample_entity.id,
+            to_id=None,
+            to_name="Existing Target",  # Duplicate of first_relation
+            relation_type="links_to",
+        ),
+        Relation(
+            from_id=sample_entity.id,
+            to_id=None,
+            to_name="New Target 1",  # New
+            relation_type="links_to",
+        ),
+        Relation(
+            from_id=sample_entity.id,
+            to_id=None,
+            to_name="New Target 2",  # New
+            relation_type="references",
+        ),
+    ]
+
+    inserted = await relation_repository.add_all_ignore_duplicates(relations)
+
+    # Only 2 new ones should be inserted
+    assert inserted == 2
+
+    # Verify total count
+    all_relations = await relation_repository.find_all()
+    from_sample = [r for r in all_relations if r.from_id == sample_entity.id]
+    assert len(from_sample) == 3  # 1 existing + 2 new
+
+
+@pytest.mark.asyncio
+async def test_add_all_ignore_duplicates_with_context(
+    relation_repository: RelationRepository, sample_entity: Entity, related_entity: Entity
+):
+    """Test that context field is properly inserted."""
+    relations = [
+        Relation(
+            from_id=sample_entity.id,
+            to_id=related_entity.id,
+            to_name=related_entity.title,
+            relation_type="links_to",
+            context="some context here",
+        ),
+    ]
+
+    inserted = await relation_repository.add_all_ignore_duplicates(relations)
+    assert inserted == 1
+
+    # Verify context was saved
+    found = await relation_repository.find_by_entities(sample_entity.id, related_entity.id)
+    assert len(found) == 1
+    assert found[0].context == "some context here"