multi-agent-system / app /services /architecture_patterns.py
firepenguindisopanda
Refactor test documentation and improve consistency in comments
e382248
Raw
History Blame Contribute Delete
4.67 kB
"""Pattern library service - wraps ArchitectureMemory with consistent error handling."""
import logging
from typing import Any
from fastapi import HTTPException
from app.core.architecture_memory import get_architecture_memory, ArchitectureMemory
logger = logging.getLogger("specs-before-code-api.services.architecture_patterns")
class PatternService:
"""Facade over ArchitectureMemory (Redis-backed pattern library)."""
def __init__(self) -> None:
self._memory: ArchitectureMemory = get_architecture_memory()
@property
def is_configured(self) -> bool:
return self._memory.is_configured
async def list_patterns(
self,
tag: str | None = None,
search: str | None = None,
) -> dict[str, Any]:
if not self._memory.is_configured:
return {"patterns": [], "message": "Redis not configured"}
try:
if tag:
patterns = await self._memory.get_patterns_by_tag(tag)
elif search:
patterns = await self._memory.search_patterns(search)
else:
patterns = await self._memory.get_all_patterns()
return {
"patterns": [
{
"id": p.id,
"name": p.name,
"description": p.description,
"use_cases": p.use_cases,
"tags": p.tags,
"components": p.components,
"pros": p.pros,
"cons": p.cons,
"tech_stack": p.tech_stack,
"estimated_cost": p.estimated_cost,
"estimated_setup_time": p.estimated_setup_time,
}
for p in patterns
]
}
except Exception as e:
logger.error("Failed to list patterns: %s", e)
return {"error": str(e)}
async def get_pattern(self, pattern_id: str) -> dict[str, Any]:
if not self._memory.is_configured:
return {"error": "Redis not configured"}
try:
pattern = await self._memory.get_pattern(pattern_id)
if not pattern:
return {"error": "Pattern not found"}
return {
"id": pattern.id,
"name": pattern.name,
"description": pattern.description,
"use_cases": pattern.use_cases,
"tags": pattern.tags,
"components": pattern.components,
"pros": pattern.pros,
"cons": pattern.cons,
"tech_stack": pattern.tech_stack,
"estimated_cost": pattern.estimated_cost,
"estimated_setup_time": pattern.estimated_setup_time,
}
except Exception as e:
logger.error("Failed to get pattern %s: %s", pattern_id, e)
return {"error": str(e)}
async def seed_patterns(self) -> dict[str, Any]:
if not self._memory.is_configured:
return {"error": "Redis not configured"}
try:
seeded = await self._memory.seed_default_patterns()
return {"message": f"Seeded {seeded} default patterns", "seeded_count": seeded}
except Exception as e:
logger.error("Failed to seed patterns: %s", e)
return {"error": str(e)}
async def import_to_session(self, pattern_id: str) -> dict[str, Any] | None:
"""Fetch a pattern; returns the raw pattern dict or raises HTTPException."""
if not self._memory.is_configured:
raise HTTPException(status_code=503, detail="Pattern library not available")
pattern = await self._memory.get_pattern(pattern_id)
if not pattern:
raise HTTPException(status_code=404, detail="Pattern not found")
return {
"id": pattern.id,
"name": pattern.name,
"description": pattern.description,
"use_cases": pattern.use_cases,
"tags": pattern.tags,
"components": pattern.components,
"pros": pattern.pros,
"cons": pattern.cons,
"tech_stack": pattern.tech_stack,
"estimated_cost": pattern.estimated_cost,
"estimated_setup_time": pattern.estimated_setup_time,
}
async def get_pattern_raw(self, pattern_id: str) -> Any | None:
if not self._memory.is_configured:
raise HTTPException(status_code=503, detail="Pattern library not available")
return await self._memory.get_pattern(pattern_id)