Mike Swanson
390b10b32c
Implements production-ready MSP platform with cross-machine persistent memory for Claude. API Implementation: - 130 REST API endpoints across 21 entities - JWT authentication on all endpoints - AES-256-GCM encryption for credentials - Automatic audit logging - Complete OpenAPI documentation Database: - 43 tables in MariaDB (172.16.3.20:3306) - 42 SQLAlchemy models with modern 2.0 syntax - Full Alembic migration system - 99.1% CRUD test pass rate Context Recall System (Phase 6): - Cross-machine persistent memory via database - Automatic context injection via Claude Code hooks - Automatic context saving after task completion - 90-95% token reduction with compression utilities - Relevance scoring with time decay - Tag-based semantic search - One-command setup script Security Features: - JWT tokens with Argon2 password hashing - AES-256-GCM encryption for all sensitive data - Comprehensive audit trail for credentials - HMAC tamper detection - Secure configuration management Test Results: - Phase 3: 38/38 CRUD tests passing (100%) - Phase 4: 34/35 core API tests passing (97.1%) - Phase 5: 62/62 extended API tests passing (100%) - Phase 6: 10/10 compression tests passing (100%) - Overall: 144/145 tests passing (99.3%) Documentation: - Comprehensive architecture guides - Setup automation scripts - API documentation at /api/docs - Complete test reports - Troubleshooting guides Project Status: 95% Complete (Production-Ready) Phase 7 (optional work context APIs) remains for future enhancement. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
368 lines
9.4 KiB
Python
368 lines
9.4 KiB
Python
"""
|
|
ContextSnippet service layer for business logic and database operations.
|
|
|
|
Handles all database operations for context snippets, providing reusable
|
|
knowledge storage and retrieval.
|
|
"""
|
|
|
|
import json
|
|
from typing import List, Optional
|
|
from uuid import UUID
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from api.models.context_snippet import ContextSnippet
|
|
from api.schemas.context_snippet import ContextSnippetCreate, ContextSnippetUpdate
|
|
|
|
|
|
def get_context_snippets(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> tuple[list[ContextSnippet], int]:
|
|
"""
|
|
Retrieve a paginated list of context snippets.
|
|
|
|
Args:
|
|
db: Database session
|
|
skip: Number of records to skip (for pagination)
|
|
limit: Maximum number of records to return
|
|
|
|
Returns:
|
|
tuple: (list of context snippets, total count)
|
|
"""
|
|
# Get total count
|
|
total = db.query(ContextSnippet).count()
|
|
|
|
# Get paginated results, ordered by relevance and usage
|
|
snippets = (
|
|
db.query(ContextSnippet)
|
|
.order_by(ContextSnippet.relevance_score.desc(), ContextSnippet.usage_count.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
return snippets, total
|
|
|
|
|
|
def get_context_snippet_by_id(db: Session, snippet_id: UUID) -> ContextSnippet:
|
|
"""
|
|
Retrieve a single context snippet by its ID.
|
|
|
|
Automatically increments usage_count when snippet is retrieved.
|
|
|
|
Args:
|
|
db: Database session
|
|
snippet_id: UUID of the context snippet to retrieve
|
|
|
|
Returns:
|
|
ContextSnippet: The context snippet object
|
|
|
|
Raises:
|
|
HTTPException: 404 if context snippet not found
|
|
"""
|
|
snippet = db.query(ContextSnippet).filter(ContextSnippet.id == str(snippet_id)).first()
|
|
|
|
if not snippet:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"ContextSnippet with ID {snippet_id} not found"
|
|
)
|
|
|
|
# Increment usage count
|
|
snippet.usage_count += 1
|
|
db.commit()
|
|
db.refresh(snippet)
|
|
|
|
return snippet
|
|
|
|
|
|
def get_context_snippets_by_project(
|
|
db: Session,
|
|
project_id: UUID,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> tuple[list[ContextSnippet], int]:
|
|
"""
|
|
Retrieve context snippets for a specific project.
|
|
|
|
Args:
|
|
db: Database session
|
|
project_id: UUID of the project
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
|
|
Returns:
|
|
tuple: (list of context snippets, total count)
|
|
"""
|
|
# Get total count for project
|
|
total = db.query(ContextSnippet).filter(
|
|
ContextSnippet.project_id == str(project_id)
|
|
).count()
|
|
|
|
# Get paginated results
|
|
snippets = (
|
|
db.query(ContextSnippet)
|
|
.filter(ContextSnippet.project_id == str(project_id))
|
|
.order_by(ContextSnippet.relevance_score.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
return snippets, total
|
|
|
|
|
|
def get_context_snippets_by_client(
|
|
db: Session,
|
|
client_id: UUID,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> tuple[list[ContextSnippet], int]:
|
|
"""
|
|
Retrieve context snippets for a specific client.
|
|
|
|
Args:
|
|
db: Database session
|
|
client_id: UUID of the client
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
|
|
Returns:
|
|
tuple: (list of context snippets, total count)
|
|
"""
|
|
# Get total count for client
|
|
total = db.query(ContextSnippet).filter(
|
|
ContextSnippet.client_id == str(client_id)
|
|
).count()
|
|
|
|
# Get paginated results
|
|
snippets = (
|
|
db.query(ContextSnippet)
|
|
.filter(ContextSnippet.client_id == str(client_id))
|
|
.order_by(ContextSnippet.relevance_score.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
return snippets, total
|
|
|
|
|
|
def get_context_snippets_by_tags(
|
|
db: Session,
|
|
tags: List[str],
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> tuple[list[ContextSnippet], int]:
|
|
"""
|
|
Retrieve context snippets filtered by tags.
|
|
|
|
Args:
|
|
db: Database session
|
|
tags: List of tags to filter by (OR logic - any tag matches)
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
|
|
Returns:
|
|
tuple: (list of context snippets, total count)
|
|
"""
|
|
# Build tag filters
|
|
tag_filters = []
|
|
for tag in tags:
|
|
tag_filters.append(ContextSnippet.tags.contains(f'"{tag}"'))
|
|
|
|
# Get total count
|
|
if tag_filters:
|
|
total = db.query(ContextSnippet).filter(or_(*tag_filters)).count()
|
|
else:
|
|
total = 0
|
|
|
|
# Get paginated results
|
|
if tag_filters:
|
|
snippets = (
|
|
db.query(ContextSnippet)
|
|
.filter(or_(*tag_filters))
|
|
.order_by(ContextSnippet.relevance_score.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
else:
|
|
snippets = []
|
|
|
|
return snippets, total
|
|
|
|
|
|
def get_top_relevant_snippets(
|
|
db: Session,
|
|
limit: int = 10,
|
|
min_relevance_score: float = 7.0
|
|
) -> list[ContextSnippet]:
|
|
"""
|
|
Get the top most relevant context snippets.
|
|
|
|
Args:
|
|
db: Database session
|
|
limit: Maximum number of snippets to return (default 10)
|
|
min_relevance_score: Minimum relevance score threshold (default 7.0)
|
|
|
|
Returns:
|
|
list: Top relevant context snippets
|
|
"""
|
|
snippets = (
|
|
db.query(ContextSnippet)
|
|
.filter(ContextSnippet.relevance_score >= min_relevance_score)
|
|
.order_by(ContextSnippet.relevance_score.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
return snippets
|
|
|
|
|
|
def create_context_snippet(
|
|
db: Session,
|
|
snippet_data: ContextSnippetCreate
|
|
) -> ContextSnippet:
|
|
"""
|
|
Create a new context snippet.
|
|
|
|
Args:
|
|
db: Database session
|
|
snippet_data: Context snippet creation data
|
|
|
|
Returns:
|
|
ContextSnippet: The created context snippet object
|
|
|
|
Raises:
|
|
HTTPException: 500 if database error occurs
|
|
"""
|
|
try:
|
|
# Create new context snippet instance
|
|
db_snippet = ContextSnippet(**snippet_data.model_dump())
|
|
|
|
# Add to database
|
|
db.add(db_snippet)
|
|
db.commit()
|
|
db.refresh(db_snippet)
|
|
|
|
return db_snippet
|
|
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Database error: {str(e)}"
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to create context snippet: {str(e)}"
|
|
)
|
|
|
|
|
|
def update_context_snippet(
|
|
db: Session,
|
|
snippet_id: UUID,
|
|
snippet_data: ContextSnippetUpdate
|
|
) -> ContextSnippet:
|
|
"""
|
|
Update an existing context snippet.
|
|
|
|
Args:
|
|
db: Database session
|
|
snippet_id: UUID of the context snippet to update
|
|
snippet_data: Context snippet update data
|
|
|
|
Returns:
|
|
ContextSnippet: The updated context snippet object
|
|
|
|
Raises:
|
|
HTTPException: 404 if context snippet not found
|
|
HTTPException: 500 if database error occurs
|
|
"""
|
|
# Get existing snippet (without incrementing usage count)
|
|
snippet = db.query(ContextSnippet).filter(ContextSnippet.id == str(snippet_id)).first()
|
|
|
|
if not snippet:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"ContextSnippet with ID {snippet_id} not found"
|
|
)
|
|
|
|
try:
|
|
# Update only provided fields
|
|
update_data = snippet_data.model_dump(exclude_unset=True)
|
|
|
|
# Apply updates
|
|
for field, value in update_data.items():
|
|
setattr(snippet, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(snippet)
|
|
|
|
return snippet
|
|
|
|
except HTTPException:
|
|
db.rollback()
|
|
raise
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Database error: {str(e)}"
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to update context snippet: {str(e)}"
|
|
)
|
|
|
|
|
|
def delete_context_snippet(db: Session, snippet_id: UUID) -> dict:
|
|
"""
|
|
Delete a context snippet by its ID.
|
|
|
|
Args:
|
|
db: Database session
|
|
snippet_id: UUID of the context snippet to delete
|
|
|
|
Returns:
|
|
dict: Success message
|
|
|
|
Raises:
|
|
HTTPException: 404 if context snippet not found
|
|
HTTPException: 500 if database error occurs
|
|
"""
|
|
# Get existing snippet (without incrementing usage count)
|
|
snippet = db.query(ContextSnippet).filter(ContextSnippet.id == str(snippet_id)).first()
|
|
|
|
if not snippet:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"ContextSnippet with ID {snippet_id} not found"
|
|
)
|
|
|
|
try:
|
|
db.delete(snippet)
|
|
db.commit()
|
|
|
|
return {
|
|
"message": "ContextSnippet deleted successfully",
|
|
"snippet_id": str(snippet_id)
|
|
}
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to delete context snippet: {str(e)}"
|
|
)
|