Works with Python files in domain/repositories/ and infrastructure/ directories.

Implement Repository Pattern

Table of Contents

Core Sections

• Purpose
  • Core responsibility: Create repositories with Protocol/Implementation separation
• Quick Start
  • Fastest path: Create repository from user request to working implementation
• Instructions
  • Step 1: Create Domain Protocol (Interface)
  • Step 2: Create Infrastructure Implementation
  • Step 3: Add Cypher Queries
  • Step 4: Register in Container
  • Step 5: Create Tests
• Examples
  • Example 1: Simple CRUD Repository
  • Example 2: Repository with Complex Domain Model
  • Example 3: Repository with Pagination

Patterns & Best Practices

• Common Patterns
  • Pattern 1: Query Parameter Validation
  • Pattern 2: ServiceResult Propagation
  • Pattern 3: Resource Lifecycle
• Red Flags - STOP
  • Critical issues to watch for in repository implementation
• Success Checklist
  • Complete validation before marking repository done

Supporting Resources

• Requirements
  • Dependencies, project structure, and setup requirements
• See Also
  • templates/protocol-template.py - Repository protocol skeleton
  • templates/implementation-template.py - Neo4j implementation skeleton
  • templates/test-template.py - Test suite skeleton
  • references/pattern-guide.md - Complete pattern catalog
  • references/troubleshooting.md - Common issues and solutions
  • scripts/analyze_queries.py - Analyze Cypher queries in repository implementations
  • scripts/generate_repository.py - Generate repository pattern files with domain protocol and implementation
  • scripts/validate_repository_patterns.py - Validate repository pattern compliance across the codebase

Purpose

Create repositories following Clean Architecture principles with Protocol (domain layer) and Implementation (infrastructure layer) separation. Ensures proper dependency inversion, ServiceResult return types, and resource lifecycle management.

When to Use

Use this skill when:

• Adding new data access layer - Creating persistence for domain models
• Creating database interaction - Implementing queries and commands against data stores
• Implementing persistence - Storing and retrieving domain entities
• Need to store/retrieve domain models - Data access abstraction required

Trigger phrases:

• "Create a repository for X"
• "Implement data access for Y"
• "Add persistence layer for Z"
• "Store/retrieve domain model X"

Quick Start

User: "Create a repository for storing search history"

What happens:

1. Create Protocol interface in

   domain/repositories/search_history_repository.py

2. Create Neo4j implementation in

   infrastructure/neo4j/search_history_repository.py

3. Implement ManagedResource for lifecycle
4. Use ServiceResult for all operations
5. Add required Cypher queries

Result: ✅ Repository with Protocol + Implementation ready for dependency injection

Instructions

Step 1: Create Domain Protocol (Interface)

Location:

src/project_watch_mcp/domain/repositories/{name}_repository.py

Pattern:

from abc import ABC, abstractmethod
from project_watch_mcp.domain.common import ServiceResult

class {Name}Repository(ABC):
    """Port for {purpose} storage and retrieval.

    This interface defines the contract for {operations}.
    Concrete implementations will be provided in the infrastructure layer.
    """

    @abstractmethod
    async def {operation}(self, param: Type) -> ServiceResult[ReturnType]:
        """Brief description of operation.

        Args:
            param: Description

        Returns:
            ServiceResult[ReturnType]: Success with data or Failure on errors
        """
        pass

Key Requirements:

• Inherit from

  ABC

• Use

  @abstractmethod

  decorator
• Return

  ServiceResult[T]

  for all operations
• Document expected behavior in docstrings
• No implementation details (pure interface)

Step 2: Create Infrastructure Implementation

Location:

src/project_watch_mcp/infrastructure/neo4j/{name}_repository.py

Pattern:

from neo4j import AsyncDriver, RoutingControl
from project_watch_mcp.config.settings import Settings
from project_watch_mcp.domain.common import ServiceResult
from project_watch_mcp.domain.repositories.{name}_repository import {Name}Repository
from project_watch_mcp.domain.services.resource_manager import ManagedResource

class Neo4j{Name}Repository({Name}Repository, ManagedResource):
    """Neo4j adapter implementing {Name}Repository interface."""

    def __init__(self, driver: AsyncDriver, settings: Settings):
        if not driver:
            raise ValueError("Neo4j driver is required")
        if not settings:
            raise ValueError("Settings is required")

        self.driver = driver
        self.settings = settings
        self.database = settings.neo4j.database_name

    async def _execute_with_retry(
        self,
        query: str,
        parameters: dict[str, Any] | None = None,
        routing: RoutingControl = RoutingControl.WRITE,
    ) -> ServiceResult[list[dict]]:
        """Execute query with parameter validation and retry logic."""
        # Validate before executing
        validation_result = validate_and_build_query(query, parameters, strict=True)
        if validation_result.is_failure:
            return ServiceResult.fail(f"Validation failed: {validation_result.error}")

        validated_query = validation_result.data

        try:
            records, _, _ = await self.driver.execute_query(
                cast(LiteralString, validated_query.query),
                validated_query.parameters,
                database_=self.database,
                routing_=routing,
            )
            return ServiceResult.ok([dict(record) for record in records])
        except Neo4jError as e:
            return ServiceResult.fail(f"Database error: {str(e)}")

    async def close(self) -> None:
        """Close and cleanup resources (ManagedResource protocol)."""
        # Repository-specific cleanup if needed
        pass

Key Requirements:

• Implement Protocol interface
• Inherit from

  ManagedResource

• Required constructor params:

  driver: AsyncDriver, settings: Settings

• Validate parameters in constructor (

  if not driver: raise ValueError

  )
• Use

  _execute_with_retry()

  for all database operations
• Implement

  close()

  for resource cleanup
• All operations return

  ServiceResult[T]

Step 3: Add Cypher Queries

Location:

src/project_watch_mcp/infrastructure/neo4j/queries.py

Pattern:

class CypherQueries:
    # Existing queries...

    # {Name}Repository Queries
    GET_{ENTITY} = """
    MATCH (e:{Label} {project_name: $project_name, id: $id})
    RETURN e
    """

    SAVE_{ENTITY} = """
    MERGE (e:{Label} {project_name: $project_name, id: $id})
    SET e += $properties
    SET e.updated_at = datetime()
    RETURN e
    """

Key Requirements:

• Group queries by repository
• Use parameterized queries (prevent injection)
• Use MERGE for upsert operations
• Include timestamp management
• Document query purpose

See: references/query-patterns.md for common patterns

Step 4: Register in Container

Location:

src/project_watch_mcp/infrastructure/container.py

Pattern:

async def {name}_repository(self) -> {Name}Repository:
    """Provide {Name}Repository implementation."""
    driver = await self.neo4j_driver()
    settings = await self.settings()
    return Neo4j{Name}Repository(driver, settings)

Key Requirements:

• Return type is Protocol (not implementation)
• Inject dependencies (driver, settings)
• Use async/await for resource initialization
• Follow naming convention:

  {name}_repository()

Step 5: Create Tests

Unit Tests:

tests/unit/infrastructure/neo4j/test_{name}_repository.py

Integration Tests:

tests/integration/infrastructure/neo4j/test_{name}_repository.py

Pattern:

@pytest.mark.asyncio
async def test_save_{entity}_success(mock_driver, settings):
    """Test successful {entity} save operation."""
    # Arrange
    repo = Neo4j{Name}Repository(mock_driver, settings)
    mock_driver.execute_query.return_value = (
        [{"e": {"id": "test", "name": "Test"}}],
        None,
        None,
    )

    # Act
    result = await repo.save_{entity}(entity_data)

    # Assert
    assert result.is_success
    assert result.data is not None

Key Requirements:

• Test both success and failure cases
• Mock driver.execute_query for unit tests
• Test parameter validation
• Test ServiceResult.ok() and ServiceResult.fail() paths
• Integration tests use real Neo4j instance

Examples

Example 1: Simple CRUD Repository

Protocol:

class SearchHistoryRepository(ABC):
    @abstractmethod
    async def save_query(self, query: str, user_id: str) -> ServiceResult[None]:
        pass

    @abstractmethod
    async def get_recent_queries(self, user_id: str, limit: int) -> ServiceResult[list[str]]:
        pass

Implementation:

class Neo4jSearchHistoryRepository(SearchHistoryRepository, ManagedResource):
    async def save_query(self, query: str, user_id: str) -> ServiceResult[None]:
        cypher = """
        CREATE (q:SearchQuery {query: $query, user_id: $user_id, timestamp: datetime()})
        """
        result = await self._execute_with_retry(cypher, {"query": query, "user_id": user_id})
        return ServiceResult.ok(None) if result.is_success else result

Example 2: Repository with Complex Domain Model

For advanced patterns, see references/pattern-guide.md:

• Converting Neo4j records to domain models
• Handling nested relationships
• Batch operations
• Transaction management

Example 3: Repository with Pagination

For pagination patterns, see references/pattern-guide.md:

• Cursor-based pagination
• Offset-based pagination
• Performance considerations

Requirements

Dependencies:

• neo4j>=5.0.0

  - Async driver

• project_watch_mcp.domain.common

  - ServiceResult

• project_watch_mcp.domain.services.resource_manager

  - ManagedResource

• project_watch_mcp.config.settings

  - Settings injection

Project Structure:

src/project_watch_mcp/
├── domain/
│   └── repositories/
│       └── {name}_repository.py     # Protocol (ABC)
├── infrastructure/
│   └── neo4j/
│       ├── {name}_repository.py     # Implementation
│       └── queries.py               # Cypher queries
└── tests/
    ├── unit/infrastructure/neo4j/
    │   └── test_{name}_repository.py
    └── integration/infrastructure/neo4j/
        └── test_{name}_repository.py

Common Patterns

Pattern 1: Query Parameter Validation

Always validate query parameters before execution:

validation_result = validate_and_build_query(query, parameters, strict=True)
if validation_result.is_failure:
    return ServiceResult.fail(f"Validation failed: {validation_result.error}")

Pattern 2: ServiceResult Propagation

Chain ServiceResult operations:

result = await self._execute_with_retry(query, params)
if result.is_failure:
    return ServiceResult.fail(f"Failed to save: {result.error}")

# Transform data and return success
return ServiceResult.ok(transformed_data)

Pattern 3: Resource Lifecycle

Implement ManagedResource for proper cleanup:

async def close(self) -> None:
    """Cleanup repository-specific resources."""
    # Driver is managed externally by container
    # Only cleanup repository-specific resources here
    logger.debug(f"{self.__class__.__name__} cleanup complete")

See: references/pattern-guide.md for complete pattern catalog

Red Flags - STOP

If you see any of these, investigate immediately:

1. ❌ Protocol in infrastructure layer → Must be in domain
2. ❌ Return

   None

   on error → Use

   ServiceResult.fail()

3. ❌ Optional

   settings

   parameter → Must be required
4. ❌ Direct driver usage → Use

   _execute_with_retry()

5. ❌ Missing parameter validation → Validate in constructor
6. ❌ Raw Cypher in methods → Use

   CypherQueries

   class
7. ❌ Synchronous methods → All methods must be

   async

8. ❌ Missing

   ManagedResource

   → Required for lifecycle
9. ❌ Return domain models from Neo4j layer → Convert in repository
10. ❌ Missing tests → Must have unit + integration tests

Success Checklist

Before marking repository complete:

• Protocol exists in

  domain/repositories/

• Implementation exists in

  infrastructure/neo4j/

• Constructor validates

  driver

  and

  settings

• All methods return

  ServiceResult[T]

• Implements

  ManagedResource

  with

  close()

• Uses

  _execute_with_retry()

  for all operations
• Queries defined in

  CypherQueries

• Registered in container
• Unit tests passing (mocked driver)
• Integration tests passing (real Neo4j)
• Quality gates pass (

  ./scripts/check_all.sh

  )
• Documentation updated (if new pattern)

See Also

• templates/protocol-template.py - Repository protocol skeleton
• templates/implementation-template.py - Neo4j implementation skeleton
• templates/test-template.py - Test suite skeleton
• references/pattern-guide.md - Complete pattern catalog
• references/troubleshooting.md - Common issues and solutions

Related Skills:

• implement-dependency-injection

  - For container registration

• validate-layer-boundaries

  - For architecture compliance

• run-quality-gates

  - For validation before commit

Last Updated: 2025-10-18