implement-repository-pattern（实施存储库模式）

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