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

Implement Repository Pattern

Core Sections

Purpose
- Core responsibility: Create repositories with Protocol/Implementation separation
Quick Start
- Fastest path: Create repository from user request to working implementation
Instructions
Examples

Patterns & Best Practices

Common Patterns
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:

Create Protocol interface in

domain/repositories/search_history_repository.py

Create Neo4j implementation in

infrastructure/neo4j/search_history_repository.py

Implement ManagedResource for lifecycle
Use ServiceResult for all operations
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:

python

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:

python

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:

python

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:

python

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:

python

@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:

python

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:

python

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:

python

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:

python

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:

python

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:

❌ Protocol in infrastructure layer → Must be in domain
❌ Return
```
None
```
on error → Use
```
ServiceResult.fail()
```
❌ Optional
```
settings
```
parameter → Must be required
❌ Direct driver usage → Use
```
_execute_with_retry()
```
❌ Missing parameter validation → Validate in constructor
❌ Raw Cypher in methods → Use
```
CypherQueries
```
class
❌ Synchronous methods → All methods must be
```
async
```
❌ Missing
```
ManagedResource
```
→ Required for lifecycle
❌ Return domain models from Neo4j layer → Convert in repository
❌ 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)

implement-repository-pattern

NPX Install

Tags

SKILL.md Content

Implement Repository Pattern

Table of Contents

Core Sections

Patterns & Best Practices

Supporting Resources

Purpose

When to Use

Quick Start

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

Requirements

Common Patterns

Pattern 1: Query Parameter Validation

Pattern 2: ServiceResult Propagation

Pattern 3: Resource Lifecycle

Red Flags - STOP

Success Checklist

See Also