Smart Memory System Development Guide

This document provides detailed guidance for developers on interface definitions, Mock implementations, and decoupled development.

📋 Table of Contents

Interface Definition and Implementation
Mock Implementation and Decoupled Development
Dependency Injection Best Practices
Development Environment Configuration

🔧 Interface Definition and Implementation

1. Define Abstract Interfaces

Use Python's Abstract Base Classes (ABC) to define clear interfaces:

from abc import ABC, abstractmethod
from typing import Dict, List, Optional

class UserRepository(ABC):
    """User storage interface"""
    
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        """Find user by ID"""
        pass
    
    @abstractmethod
    def save(self, user: Dict) -> Dict:
        """Save user"""
        pass
    
    @abstractmethod
    def find_by_email(self, email: str) -> Optional[Dict]:
        """Find user by email"""
        pass

class NotificationService(ABC):
    """Notification service interface"""
    
    @abstractmethod
    async def send_notification(self, user_id: int, message: str) -> bool:
        """Send notification"""
        pass

2. Implement Concrete Classes

Use dependency injection decorators to mark implementation classes:

from core.di.decorators import repository, service

@repository("mysql_user_repo")
class MySQLUserRepository(UserRepository):
    """MySQL user storage implementation"""
    
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        # Actual database query logic
        return {"id": user_id, "name": f"User {user_id}", "source": "mysql"}
    
    def save(self, user: Dict) -> Dict:
        # Actual save logic
        return {**user, "id": 123, "created_at": "2024-01-01"}
    
    def find_by_email(self, email: str) -> Optional[Dict]:
        # Actual query logic
        return {"id": 456, "email": email, "source": "mysql"}

@repository("redis_user_repo") 
class RedisUserRepository(UserRepository):
    """Redis user storage implementation (cache layer)"""
    
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        # Redis cache query logic
        return {"id": user_id, "name": f"Cached User {user_id}", "source": "redis"}
    
    def save(self, user: Dict) -> Dict:
        # Redis cache save logic
        return {**user, "cached": True}
    
    def find_by_email(self, email: str) -> Optional[Dict]:
        # Redis cache query logic
        return None  # Not in cache

@service("email_notification")
class EmailNotificationService(NotificationService):
    """Email notification service implementation"""
    
    async def send_notification(self, user_id: int, message: str) -> bool:
        # Actual email sending logic
        print(f"📧 Sending email to user {user_id}: {message}")
        return True

3. Set Primary Implementation

When there are multiple implementations, use primary=True to mark the default implementation:

@repository("primary_user_repo", primary=True)
class PrimaryUserRepository(UserRepository):
    """Primary user storage implementation"""
    
    def __init__(self):
        # Can combine multiple implementations
        self.mysql_repo = MySQLUserRepository()
        self.redis_repo = RedisUserRepository()
    
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        # Check cache first, then database
        user = self.redis_repo.find_by_id(user_id)
        if user:
            return user
        return self.mysql_repo.find_by_id(user_id)
    
    def save(self, user: Dict) -> Dict:
        # Save to database and update cache
        saved_user = self.mysql_repo.save(user)
        self.redis_repo.save(saved_user)
        return saved_user
    
    def find_by_email(self, email: str) -> Optional[Dict]:
        return self.mysql_repo.find_by_email(email)

🧪 Mock Implementation and Decoupled Development

1. Define Mock Implementation

Use the @mock_impl decorator to define Mock implementations:

from core.di.decorators import mock_impl

@mock_impl("mock_user_repo")
class MockUserRepository(UserRepository):
    """Mock user storage implementation"""
    
    def __init__(self):
        # Simulated data in memory
        self.users = {
            1: {"id": 1, "name": "Mock User 1", "email": "user1@mock.com"},
            2: {"id": 2, "name": "Mock User 2", "email": "user2@mock.com"}
        }
        self.next_id = 3
    
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        return self.users.get(user_id)
    
    def save(self, user: Dict) -> Dict:
        if "id" not in user:
            user["id"] = self.next_id
            self.next_id += 1
        self.users[user["id"]] = user
        return user
    
    def find_by_email(self, email: str) -> Optional[Dict]:
        for user in self.users.values():
            if user.get("email") == email:
                return user
        return None

@mock_impl("mock_notification")
class MockNotificationService(NotificationService):
    """Mock notification service implementation"""
    
    def __init__(self):
        self.sent_messages = []  # Record sent messages for test verification
    
    async def send_notification(self, user_id: int, message: str) -> bool:
        # Simulate sending, actually just record
        notification = {
            "user_id": user_id,
            "message": message,
            "timestamp": "2024-01-01T10:00:00"
        }
        self.sent_messages.append(notification)
        print(f"🧪 Mock notification sent: {notification}")
        return True
    
    def get_sent_messages(self) -> List[Dict]:
        """Get sent messages (for testing)"""
        return self.sent_messages.copy()

2. Mock Mode Toggle Control

2.1 Environment Variable Control

Configure in the environment variable file (.env):

# Development environment configuration
MOCK_MODE=true

# Production environment configuration
# MOCK_MODE=false

2.2 Dynamic Switching in Code

import os
from core.di.utils import enable_mock_mode, disable_mock_mode, get_bean_by_type

def setup_mock_mode():
    """Set Mock mode based on environment variable"""
    if os.getenv("MOCK_MODE", "false").lower() == "true":
        enable_mock_mode()
        print("🧪 Mock mode enabled")
    else:
        disable_mock_mode()
        print("🔧 Using real implementation")

# Call during application startup
def initialize_app():
    setup_mock_mode()
    
    # Now the implementation will automatically switch based on Mock mode
    user_service = get_bean_by_type(UserService)
    return user_service

2.3 Automatic Control at Startup

The application will automatically check the MOCK_MODE environment variable at startup:

# Implementation in run.py
if os.getenv("MOCK_MODE") and os.getenv("MOCK_MODE").lower() == "true":
    enable_mock_mode()
    logger.info("🚀 Mock mode enabled")
else:
    logger.info("🚀 Mock mode disabled")

3. Conditional Mock Implementation

Use different Mock implementations based on different conditions:

@mock_impl("mock_fast_notification")
class FastMockNotificationService(NotificationService):
    """Fast Mock notification (for testing)"""
    
    async def send_notification(self, user_id: int, message: str) -> bool:
        print(f"⚡ Fast Mock notification: User {user_id} - {message}")
        return True

@mock_impl("mock_slow_notification") 
class SlowMockNotificationService(NotificationService):
    """Slow Mock notification (for performance testing)"""
    
    async def send_notification(self, user_id: int, message: str) -> bool:
        import asyncio
        await asyncio.sleep(0.1)  # Simulate network delay
        print(f"🐌 Slow Mock notification: User {user_id} - {message}")
        return True

# Choose Mock implementation based on test type
def setup_test_environment(test_type: str):
    enable_mock_mode()
    
    if test_type == "performance":
        # Use slow Mock for performance testing
        from core.di.utils import register_bean
        slow_mock = SlowMockNotificationService()
        register_bean(NotificationService, slow_mock, "mock_notification")
    else:
        # Use fast Mock for normal testing
        fast_mock = FastMockNotificationService()
        register_bean(NotificationService, fast_mock, "mock_notification")

🏗️ Dependency Injection Best Practices

1. Interface Design Principles

Single Responsibility: Each interface should be responsible for one clear responsibility
Interface Segregation: Clients should not depend on interfaces they don't need
Dependency Inversion: High-level modules should not depend on low-level modules; both should depend on abstractions

# Good design: Clear responsibilities
class UserRepository(ABC):
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        pass

class UserValidator(ABC):
    @abstractmethod
    def validate(self, user: Dict) -> bool:
        pass

# Avoid this design: Mixed responsibilities
class UserService(ABC):  # Not recommended: Mixes storage and validation responsibilities
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[Dict]:
        pass
    
    @abstractmethod
    def validate(self, user: Dict) -> bool:
        pass

2. Decorator Usage Guidelines

# Import from specific decorator modules
from core.di.decorators import repository, service, component, mock_impl, factory

# Data access layer
@repository("user_repository")
class UserRepositoryImpl(UserRepository):
    pass

# Business service layer
@service("user_service")
class UserService:
    pass

# General components
@component("config_manager")
class ConfigManager:
    pass

# Mock implementation
@mock_impl("mock_external_api")
class MockExternalApiClient(ExternalApiClient):
    pass

# Factory method
@factory(DatabaseConnection, "db_connection")
def create_database_connection() -> DatabaseConnection:
    config = load_config()
    return DatabaseConnection(config.db_url)

3. Circular Dependency Handling

Use lazy injection to avoid circular dependencies:

from core.di.decorators import service
from core.di.utils import lazy_inject

@service("order_service")
class OrderService:
    def __init__(self):
        # Get dependencies lazily to avoid circular dependencies
        self.user_service_lazy = lazy_inject(UserService)
        self.payment_service_lazy = lazy_inject(PaymentService)
    
    def create_order(self, order_data: Dict) -> Dict:
        user_service = self.user_service_lazy()  # Get only when called
        payment_service = self.payment_service_lazy()
        
        # Business logic
        user = user_service.get_user(order_data["user_id"])
        payment_result = payment_service.process_payment(order_data["amount"])
        
        return {"order_id": 123, "status": "created"}

⚙️ Development Environment Configuration

Note: Before starting development, run make dev-setup to set up the development environment (sync dependencies + install code check hooks).

1. Environment Variable Configuration

Create .env file:

# Development environment configuration
ENVIRONMENT=development
DEBUG=true

# Mock mode configuration
MOCK_MODE=true

# Logging configuration
LOG_LEVEL=DEBUG
LOG_FORMAT=detailed

# External service configuration (use test addresses in development)
EXTERNAL_API_URL=https://api-test.example.com
DATABASE_URL=postgresql://dev:password@localhost:5432/memsys_dev
REDIS_URL=redis://localhost:6379/0

2. Development Script Template

For development scripts that need to be run, add environment initialization at the beginning of the script:

#!/usr/bin/env python3
"""
Development script template - Data processing/testing scripts, etc.
"""
import os

# ============= Development Environment Initialization (Must be at top) =============
# 1. Set environment variables and Python path
from common_utils.load_env import setup_environment
setup_environment(load_env_file_name=".env", check_env_var="MONGODB_HOST")

# 2. Enable Mock mode (enabled by default in development environment)
from core.di.utils import enable_mock_mode
if os.getenv("MOCK_MODE", "true").lower() == "true":
    enable_mock_mode()
    print("🧪 Development script: Mock mode enabled")

# 3. Initialize dependency injection
from application_startup import setup_all
setup_all()
# ================================================

# Now you can normally import and use project modules
from core.di.utils import get_bean_by_type
from core.observation.logger import get_logger

logger = get_logger(__name__)

def main():
    """Script main logic"""
    logger.info("🚀 Development script execution started")
    
    # Example: Use dependency injection to get services
    # user_service = get_bean_by_type(UserService)
    # result = user_service.process_data()
    
    # Your script logic...
    
    logger.info("✅ Development script execution completed")

if __name__ == "__main__":
    main()

Actual Usage Example

#!/usr/bin/env python3
"""
User data migration script
"""
import os

# ============= Development Environment Initialization =============
from common_utils.load_env import setup_environment
setup_environment(load_env_file_name=".env")

from core.di.utils import enable_mock_mode
if os.getenv("MOCK_MODE", "true").lower() == "true":
    enable_mock_mode()
    print("🧪 Data migration script: Using Mock data")

from application_startup import setup_all
setup_all()
# =======================================

from core.di.utils import get_bean_by_type
from core.observation.logger import get_logger

logger = get_logger(__name__)

def migrate_user_data():
    """Migrate user data"""
    logger.info("Starting user data migration...")
    
    # Get service (will automatically use Mock implementation if Mock mode is enabled)
    user_service = get_bean_by_type(UserService)
    
    # Process data migration
    users = user_service.get_all_users()
    for user in users:
        # Migration logic...
        logger.info(f"Migrating user: {user['name']}")
    
    logger.info("User data migration completed")

if __name__ == "__main__":
    migrate_user_data()

3. Development Startup Methods

Run Development Scripts

# Enter src directory
cd src

# Run data processing script
python your_dev_script.py

# Run migration script
python migrate_data.py

# Run test script
python test_service.py

Start Development Service

# Start web service (automatically loads .env file)
python run.py

# Or set environment variable and start
export MOCK_MODE=true
python run.py

VS Code Debug Configuration

Add development configuration to VS Code's launch.json:

{
    "name": "Development Mode Launch",
    "type": "debugpy",
    "request": "launch",
    "env": {
        "PYTHONPATH": "${workspaceFolder}/src"
    },
    "envFile": "${workspaceFolder}/.env",
    "cwd": "${workspaceFolder}/src",
    "python": "${workspaceFolder}/.venv/bin/python",
    "program": "dev_run.py",
    "console": "integratedTerminal",
    "justMyCode": false
}

4. Mock Mode Verification

After starting the application, you can confirm Mock mode status through logs:

# Mock mode status will be displayed when starting the application
python run.py

# Output example:
# 🚀 Mock mode enabled  (when MOCK_MODE=true)
# 🚀 Mock mode disabled  (when MOCK_MODE=false or not set)

📝 Practical Development Example

Complete Development Workflow Example

from core.di.decorators import service, mock_impl

# 1. Define interface
class PaymentProcessor(ABC):
    @abstractmethod
    async def process_payment(self, amount: float, payment_method: str) -> Dict:
        pass

# 2. Implement real service
@service("stripe_payment")
class StripePaymentProcessor(PaymentProcessor):
    async def process_payment(self, amount: float, payment_method: str) -> Dict:
        # Real Stripe API call
        return {"transaction_id": "stripe_123", "status": "success"}

# 3. Implement Mock service
@mock_impl("mock_payment")
class MockPaymentProcessor(PaymentProcessor):
    async def process_payment(self, amount: float, payment_method: str) -> Dict:
        # Mock implementation for development and testing
        return {"transaction_id": "mock_123", "status": "success"}

# 4. Business service uses interface
@service("order_service")
class OrderService:
    def __init__(self, payment_processor: PaymentProcessor):
        self.payment_processor = payment_processor
    
    async def place_order(self, order_data: Dict) -> Dict:
        # Process payment
        payment_result = await self.payment_processor.process_payment(
            order_data["amount"], 
            order_data["payment_method"]
        )
        
        if payment_result["status"] == "success":
            return {"order_id": 456, "status": "confirmed"}
        else:
            return {"error": "Payment failed"}

# 5. Use during development
def development_workflow():
    from core.di.utils import enable_mock_mode, get_bean_by_type
    
    # Enable Mock mode for development
    enable_mock_mode()
    
    # Get service (automatically uses Mock implementation)
    order_service = get_bean_by_type(OrderService)
    
    # Test business logic without real payment
    order_data = {
        "amount": 99.99,
        "payment_method": "credit_card"
    }
    
    result = await order_service.place_order(order_data)
    print(f"Order result: {result}")

# 6. Use in production environment
def production_workflow():
    from core.di.utils import disable_mock_mode, get_bean_by_type
    
    # Disable Mock mode to use real service
    disable_mock_mode()
    
    # Get service (automatically uses real implementation)
    order_service = get_bean_by_type(OrderService)
    
    # Real business processing
    result = await order_service.place_order(order_data)
    print(f"Real order result: {result}")

This approach allows developers to:

Parallel Development: Frontend and backend can develop simultaneously, with backend using Mock data
Fast Testing: No need to set up complete external service environments
Decoupled Development: Each module can be developed and tested independently
Flexible Switching: Switch between Mock and real implementations through simple configuration

This architecture greatly improves development efficiency and code quality while maintaining system testability and maintainability.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Smart Memory System Development Guide

📋 Table of Contents

🔧 Interface Definition and Implementation

1. Define Abstract Interfaces

2. Implement Concrete Classes

3. Set Primary Implementation

🧪 Mock Implementation and Decoupled Development

1. Define Mock Implementation

2. Mock Mode Toggle Control

2.1 Environment Variable Control

2.2 Dynamic Switching in Code

2.3 Automatic Control at Startup

3. Conditional Mock Implementation

🏗️ Dependency Injection Best Practices

1. Interface Design Principles

2. Decorator Usage Guidelines

3. Circular Dependency Handling

⚙️ Development Environment Configuration

1. Environment Variable Configuration

2. Development Script Template

Actual Usage Example

3. Development Startup Methods

Run Development Scripts

Start Development Service

VS Code Debug Configuration

4. Mock Mode Verification

📝 Practical Development Example

Complete Development Workflow Example

FilesExpand file tree

development_guide.md

Latest commit

History

development_guide.md

File metadata and controls

Smart Memory System Development Guide

📋 Table of Contents

🔧 Interface Definition and Implementation

1. Define Abstract Interfaces

2. Implement Concrete Classes

3. Set Primary Implementation

🧪 Mock Implementation and Decoupled Development

1. Define Mock Implementation

2. Mock Mode Toggle Control

2.1 Environment Variable Control

2.2 Dynamic Switching in Code

2.3 Automatic Control at Startup

3. Conditional Mock Implementation

🏗️ Dependency Injection Best Practices

1. Interface Design Principles

2. Decorator Usage Guidelines

3. Circular Dependency Handling

⚙️ Development Environment Configuration

1. Environment Variable Configuration

2. Development Script Template

Actual Usage Example

3. Development Startup Methods

Run Development Scripts

Start Development Service

VS Code Debug Configuration

4. Mock Mode Verification

📝 Practical Development Example

Complete Development Workflow Example