GitHub Issue: #253 Date: 2025-01-09 Status: ✅ Utility Ready - Migration In Progress
We've created a standardized utility (lazy_singleton.py) that eliminates 10-15 lines of duplicate lazy initialization code across 10+ API files.
Impact: Standardized lazy singleton initialization with idempotency, error handling, and consistent logging.
All lazy initialization getters in API files follow this pattern:
Example from autobot-backend/api/chat.py:66-79:
def get_chat_history_manager(request):
"""Lazy-initialize chat history manager on app state"""
manager = getattr(request.app.state, "chat_history_manager", None)
if manager is None:
try:
from src.chat_history_manager import ChatHistoryManager
manager = ChatHistoryManager()
request.app.state.chat_history_manager = manager
logger.info("✅ Lazy-initialized chat_history_manager")
except Exception as e:
logger.error(f"Failed to lazy-initialize chat_history_manager: {e}")
return managerExample from autobot-backend/api/chat.py:92-105:
def get_llm_service(request):
"""Lazy-initialize LLM service on app state"""
llm_service = getattr(request.app.state, "llm_service", None)
if llm_service is None:
try:
from src.llm_service import LLMService
llm_service = LLMService()
request.app.state.llm_service = llm_service
logger.info("✅ Lazy-initialized llm_service")
except Exception as e:
logger.error(f"Failed to lazy-initialize llm_service: {e}")
return llm_serviceLines per implementation: ~14 lines Total across 10+ files: ~140-210 lines of duplicate code
Common pattern elements:
getattr(request.app.state, "attribute_name", None)- Check if existsif instance is None:- Idempotency checktry/except- Error handling- Import statement inside function (lazy import)
- Instance creation
setattr(request.app.state, "attribute_name", instance)- Store on app state- Success/failure logging
✅ Idempotency: Safe to call multiple times, returns same instance ✅ Error Handling: Standardized error logging and graceful failure ✅ Lazy Import Support: Import inside getter function still supported ✅ Consistent Logging: Automatic success/failure messages ✅ Thread-Safe: Uses getattr/setattr (atomic in Python) ✅ Flexible: Works with classes, factory functions, async factories ✅ Validation Support: Optional validation checks on retrieval
BEFORE (autobot-backend/api/chat.py:66-79 - 14 lines):
def get_chat_history_manager(request):
"""Lazy-initialize chat history manager on app state"""
manager = getattr(request.app.state, "chat_history_manager", None)
if manager is None:
try:
from src.chat_history_manager import ChatHistoryManager
manager = ChatHistoryManager()
request.app.state.chat_history_manager = manager
logger.info("✅ Lazy-initialized chat_history_manager")
except Exception as e:
logger.error(f"Failed to lazy-initialize chat_history_manager: {e}")
return managerAFTER (3 lines - 11 lines saved):
def get_chat_history_manager(request):
from src.chat_history_manager import ChatHistoryManager
return lazy_init_singleton(request.app.state, "chat_history_manager", ChatHistoryManager)BEFORE (18 lines):
def get_llm_service(request):
"""Lazy-initialize LLM service with custom config"""
llm_service = getattr(request.app.state, "llm_service", None)
if llm_service is None:
try:
from src.llm_service import LLMService
llm_service = LLMService(
model="gpt-4",
temperature=0.7,
max_tokens=2000
)
request.app.state.llm_service = llm_service
logger.info("✅ Lazy-initialized llm_service")
except Exception as e:
logger.error(f"Failed to lazy-initialize llm_service: {e}")
return llm_serviceAFTER (7 lines - 11 lines saved):
def get_llm_service(request):
from src.llm_service import LLMService
return lazy_init_singleton(
request.app.state,
"llm_service",
LLMService,
model="gpt-4",
temperature=0.7,
max_tokens=2000
)BEFORE (17 lines):
async def get_async_service(request):
"""Lazy-initialize async service"""
service = getattr(request.app.state, "async_service", None)
if service is None:
try:
from src.async_service import AsyncService
service = await AsyncService.create()
request.app.state.async_service = service
logger.info("✅ Lazy-initialized async_service")
except Exception as e:
logger.error(f"Failed to lazy-initialize async_service: {e}")
return serviceAFTER (6 lines - 11 lines saved):
async def get_async_service(request):
from src.async_service import AsyncService
async def factory():
return await AsyncService.create()
return await lazy_init_singleton_async(
request.app.state, "async_service", factory
)BEFORE (22 lines):
def get_complex_service(request):
"""Lazy-initialize service with complex setup"""
service = getattr(request.app.state, "complex_service", None)
if service is None:
try:
from src.complex_service import ComplexService
from src.config import get_config
service = ComplexService()
config = get_config()
service.configure(config)
service.initialize()
request.app.state.complex_service = service
logger.info("✅ Lazy-initialized complex_service")
except Exception as e:
logger.error(f"Failed to lazy-initialize complex_service: {e}")
return serviceAFTER (11 lines - 11 lines saved):
def get_complex_service(request):
from src.complex_service import ComplexService
from src.config import get_config
def factory():
service = ComplexService()
config = get_config()
service.configure(config)
service.initialize()
return service
return lazy_init_singleton(request.app.state, "complex_service", factory)BEFORE (14 lines):
def get_knowledge_base(request):
"""Lazy-initialize knowledge base"""
kb = getattr(request.app.state, "knowledge_base", None)
if kb is None:
try:
from src.knowledge_base import KnowledgeBase
kb = KnowledgeBase()
request.app.state.knowledge_base = kb
logger.info("✅ Lazy-initialized knowledge_base")
except Exception as e:
logger.error(f"Failed to lazy-initialize knowledge_base: {e}")
return kbAFTER (4 lines - 10 lines saved):
@singleton_getter("knowledge_base", KnowledgeBase)
def get_knowledge_base(request):
from src.knowledge_base import KnowledgeBase
return KnowledgeBaseSignature:
def lazy_init_singleton(
storage: Any,
attribute_name: str,
factory: Callable[..., T],
*args,
**kwargs,
) -> Optional[T]Use Cases:
- Basic lazy initialization
- Services requiring initialization arguments
- Factory functions with complex setup
Example:
from src.utils.lazy_singleton import lazy_init_singleton
def get_my_service(request):
from src.my_service import MyService
return lazy_init_singleton(request.app.state, "my_service", MyService)Signature:
async def lazy_init_singleton_async(
storage: Any,
attribute_name: str,
factory: Callable[..., T],
*args,
**kwargs,
) -> Optional[T]Use Cases:
- Services with async initialization
- Async factory functions
Example:
from src.utils.lazy_singleton import lazy_init_singleton_async
async def get_async_service(request):
from src.async_service import AsyncService
return await lazy_init_singleton_async(
request.app.state, "async_service", AsyncService
)Signature:
def lazy_init_singleton_with_check(
storage: Any,
attribute_name: str,
factory: Callable[..., T],
validator: Optional[Callable[[T], bool]] = None,
*args,
**kwargs,
) -> Optional[T]Use Cases:
- Services requiring health checks
- Validation before returning instance
Example:
from src.utils.lazy_singleton import lazy_init_singleton_with_check
def get_validated_service(request):
from src.validated_service import ValidatedService
def validator(service):
return service.is_healthy()
return lazy_init_singleton_with_check(
request.app.state,
"validated_service",
ValidatedService,
validator
)Signature:
def singleton_getter(attribute_name: str, factory: Callable)Use Cases:
- Clean decorator-based approach
- Simplifies getter functions
Example:
from src.utils.lazy_singleton import singleton_getter
@singleton_getter("my_service", MyService)
def get_my_service(request):
from src.my_service import MyService
return MyServiceSignature:
def global_lazy_singleton(
attribute_name: str, factory: Callable[..., T], *args, **kwargs
) -> Optional[T]Use Cases:
- Module-level singletons (use sparingly)
- CLI tools and scripts
Example:
from src.utils.lazy_singleton import global_lazy_singleton
# In module scope
def get_global_service():
from src.global_service import GlobalService
return global_lazy_singleton("global_service", GlobalService)For each file you migrate:
- Identify pattern: Find functions with
getattr(request.app.state, ..., None) - Add import:
from src.utils.lazy_singleton import lazy_init_singleton - Simplify function: Replace 14 lines with 3 lines
- Keep lazy import: Import statement inside function is still fine
- Pass arguments: Forward any initialization arguments to factory
- Handle async: Use
lazy_init_singleton_async()if needed - Remove boilerplate: Delete try/except, getattr, setattr, logging
- Test: Verify lazy initialization works correctly
| File | Function | Lines Saved |
|---|---|---|
autobot-backend/api/chat.py:66 |
get_chat_history_manager() |
~11 lines |
autobot-backend/api/chat.py:92 |
get_llm_service() |
~11 lines |
autobot-backend/api/chat.py:120 |
get_knowledge_base() |
~11 lines |
autobot-backend/api/chat.py:145 |
get_rag_service() |
~11 lines |
autobot-backend/api/intelligence.py:45 |
get_intelligent_agent() |
~11 lines |
autobot-backend/api/tools.py:38 |
get_tool_manager() |
~11 lines |
autobot-backend/api/memory.py- Memory-related gettersautobot-backend/api/workflow.py- Workflow gettersautobot-backend/api/agents.py- Agent management getters- 5+ more API files with similar patterns
Total Estimated Savings: 100-150 lines across 10+ files
| Benefit | Before | After |
|---|---|---|
| Lines per Function | 14-22 lines | 3-11 lines |
| Total Savings | N/A | 100-150 lines |
| Idempotency | ✅ Automatic | |
| Error Handling | ✅ Standardized | |
| Logging | ✅ Automatic | |
| Code Duplication | ❌ High | ✅ Eliminated |
| Maintenance | ✅ 1 place | |
| Testing | ✅ Centralized (25 tests) |
Problem: Need to keep lazy imports inside function
Solution: Keep import inside getter function - this is supported
# ✅ CORRECT - Lazy import preserved
def get_service(request):
from src.service import Service # Still lazy
return lazy_init_singleton(request.app.state, "service", Service)Problem: Service requires multi-step setup
Solution: Use factory function
def get_service(request):
from src.service import Service
def factory():
service = Service()
service.configure(config)
service.initialize()
return service
return lazy_init_singleton(request.app.state, "service", factory)Problem: Service creation is async
Solution: Use lazy_init_singleton_async()
async def get_service(request):
from src.service import AsyncService
return await lazy_init_singleton_async(
request.app.state, "service", AsyncService
)Problem: Want to check if service is healthy before returning
Solution: Use lazy_init_singleton_with_check()
def get_service(request):
from src.service import Service
def validator(service):
return service.is_healthy()
return lazy_init_singleton_with_check(
request.app.state, "service", Service, validator
)The utility has 25 comprehensive tests in tests/test_lazy_singleton.py:
- ✅ Basic lazy initialization
- ✅ Idempotency (multiple calls)
- ✅ With initialization arguments
- ✅ With factory functions
- ✅ Error handling
- ✅ Async initialization
- ✅ Validation support
- ✅ Decorator pattern
- ✅ Global singletons
All tests passing - utility is production-ready.
After migration, verify:
# Test in actual API endpoint
@app.get("/test-lazy-init")
def test_endpoint(request: Request):
service = get_my_service(request)
assert service is not None
return {"status": "ok"}If issues arise:
- Revert specific file: Change getter back to manual pattern
- Keep utility: Other files can still use it
- No breaking changes: Utility is additive, doesn't affect non-migrated files
Use Case: Need singleton outside of FastAPI request context
from src.utils.lazy_singleton import SingletonStorage, lazy_init_singleton
# Create storage
storage = SingletonStorage()
# Use like request.app.state
service = lazy_init_singleton(storage, "service", MyService)Use Case: CLI tool needs global singleton
from src.utils.lazy_singleton import global_lazy_singleton
def get_cli_service():
from src.cli_service import CLIService
return global_lazy_singleton("cli_service", CLIService)request.app.state in web applications.
- getattr/setattr are atomic in Python (CPython GIL)
- Race condition: Rare edge case where two threads create instance simultaneously
- Impact: Minimal - one instance wins, other is garbage collected
- No data corruption: Both instances are valid
- Negligible: Stores single reference on app.state
- Lifecycle: Lives for lifetime of application
- Cleanup: Automatic when application shuts down
Questions? See:
autobot-backend/utils/lazy_singleton.py- Utility implementationtests/test_lazy_singleton.py- Comprehensive tests (25 tests)CLAUDE.md- Initialization patterns section
Issues? Report with:
- File being migrated
- Error messages
- Expected vs actual behavior
- Review this guide - Understand patterns and utilities
- Start with Priority 1 files - API files with highest impact
- Test after each migration - Verify endpoints work correctly
- Track progress - Update migration checklist
- Document any issues - Help improve guide for others
Migration Status: Ready to begin - utility tested and functional ✅