feat(library-desk): implement Phase 1 service clients and infrastructure

Implements comprehensive service client layer for Library Desk API to support
Librarian AI agent with multi-tenant knowledge management across Neo4j, Qdrant,
Wiki.js, SearXNG, and Ollama.

## Service Clients (src/clients/)
- Neo4j async client with connection pooling and user-scoped labels
- Qdrant vector store with collection-per-user multi-tenancy
- Wiki.js GraphQL API client for page/dossier management
- SearXNG client for web search integration
- Ollama client for text embeddings (nomic-embed-text)

## Core Infrastructure (src/core/)
- Multi-tenancy helpers for user namespace management
  - Wiki.js: path-based namespaces (/users/{user})
  - Neo4j: user-specific labels (User_{User}_Document)
  - Qdrant: collection per user (library_desk_{user})
- Dependency injection with FastAPI Depends and @lru_cache singletons
- Lifecycle management (startup/shutdown) for all service connections

## Background Jobs (src/jobs/)
- Redis-based job manager for long-running operations
- Job status tracking with 24-hour TTL
- Support for queued, processing, completed, failed states

## Configuration
- Updated config.py with Redis DB 4 for library-desk jobs
- Updated docker-compose.yml: REDIS_DB from 2 to 4
- Added pytest and pytest-asyncio to requirements.txt

## Testing
- Unit tests: 25/25 passed (multi-tenancy helpers)
- Integration tests: 12/12 passed (all services verified)
  - Neo4j connection and CRUD operations
  - Qdrant vector operations with 768-dim embeddings
  - Wiki.js GraphQL queries
  - SearXNG web search
  - Job Manager with Redis
  - Dependency injection lifecycle
- pytest.ini configuration with asyncio support

## Health Monitoring
- Real-time service health checks via /health endpoint
- Connection status for all 5 external services
- Graceful degradation for partial service availability

## Architecture
- Follows async/await pattern throughout
- Connection pooling for Neo4j (singleton driver)
- HTTP client lifecycle management (httpx)
- Multi-tenancy enforced at client layer
- Default user: jpmschweitzer

Files changed: 26 files
- 5 new service clients (~1500 lines)
- 2 core modules (~500 lines)
- 1 job manager (~350 lines)
- 3 test files with 37 test cases
- Updated main.py with lifecycle hooks

All services tested and operational. Ready for Phase 2 (routers/services).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-08 19:09:54 +01:00
co-authored by Claude Opus 4.5
parent e886a2f9ba
commit 1a41e5bb80
26 changed files with 4165 additions and 12 deletions
@@ -0,0 +1,413 @@
"""
Neo4j async client for Library Desk.
Provides async Neo4j operations with:
- Connection pooling via AsyncGraphDatabase
- Session management with context managers
- Multi-tenancy support via user labels
- Automatic retry on transient failures
"""
from neo4j import AsyncGraphDatabase, AsyncDriver, AsyncSession
from typing import Optional, List, Dict, Any
import logging
from src.core.multi_tenancy import get_neo4j_user_label
logger = logging.getLogger(__name__)
class Neo4jClient:
"""
Async Neo4j client with connection pooling.
Features:
- Singleton driver instance (expensive to create)
- Session-per-request pattern (lightweight)
- Automatic transaction retry
- Multi-tenancy via user-specific labels
"""
def __init__(self, uri: str, user: str, password: str):
"""
Initialize Neo4j client.
Args:
uri: Neo4j Bolt URI (e.g., "bolt://neo4j:7687")
user: Neo4j username
password: Neo4j password
"""
self.uri = uri
self._driver: Optional[AsyncDriver] = None
self._auth = (user, password)
async def connect(self):
"""
Initialize connection pool.
Should be called once at app startup.
Driver handles connection pooling internally.
"""
if not self._driver:
self._driver = AsyncGraphDatabase.driver(
self.uri,
auth=self._auth,
max_connection_pool_size=50,
connection_timeout=30.0,
max_transaction_retry_time=30.0
)
# Verify connectivity
await self._driver.verify_connectivity()
logger.info(f"Connected to Neo4j at {self.uri}")
async def close(self):
"""
Close connection pool.
Should be called once at app shutdown.
"""
if self._driver:
await self._driver.close()
self._driver = None
logger.info("Closed Neo4j connection")
async def execute_query(
self,
cypher: str,
parameters: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
Execute Cypher query and return results.
Args:
cypher: Cypher query string
parameters: Query parameters
Returns:
List of result records as dictionaries
Raises:
Exception: If driver not initialized or query fails
"""
if not self._driver:
await self.connect()
async with self._driver.session() as session:
result = await session.run(cypher, parameters or {})
records = await result.data()
return records
async def execute_write(
self,
cypher: str,
parameters: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
Execute write transaction with automatic retry.
Args:
cypher: Cypher query string
parameters: Query parameters
Returns:
List of result records as dictionaries
"""
if not self._driver:
await self.connect()
async def write_tx(tx):
result = await tx.run(cypher, parameters or {})
return await result.data()
async with self._driver.session() as session:
return await session.execute_write(write_tx)
# Multi-tenancy helpers
def get_user_label(self, user: str) -> str:
"""
Get Neo4j label for user's documents.
Args:
user: User identifier
Returns:
Neo4j label string (e.g., "User_Jpmschweitzer_Document")
"""
return get_neo4j_user_label(user)
# Document node operations
async def create_document_node(
self,
user: str,
doc_id: str,
properties: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""
Create document node with user label.
Node structure:
(doc:Document:User_{user}_Document {
id: "doc_123",
source: "github",
repository: "anthropic-cookbook",
path: "skills/citation/guide.md",
title: "Citation Guide",
created_at: timestamp(),
updated_at: timestamp(),
content_hash: "sha256:..."
})
Args:
user: User identifier
doc_id: Unique document ID
properties: Document properties
Returns:
Created node properties or None on failure
"""
user_label = self.get_user_label(user)
# Ensure required properties
properties["id"] = doc_id
if "created_at" not in properties:
properties["created_at"] = "timestamp()"
cypher = f"""
CREATE (doc:Document:{user_label})
SET doc = $properties
SET doc.created_at = timestamp()
SET doc.updated_at = timestamp()
RETURN doc
"""
try:
result = await self.execute_write(cypher, {"properties": properties})
return result[0]["doc"] if result else None
except Exception as e:
logger.error(f"Failed to create document node: {e}", exc_info=True)
return None
async def get_document_node(
self,
user: str,
doc_id: str
) -> Optional[Dict[str, Any]]:
"""
Get document node by ID.
Args:
user: User identifier
doc_id: Document ID
Returns:
Document node properties or None if not found
"""
user_label = self.get_user_label(user)
cypher = f"""
MATCH (doc:Document:{user_label} {{id: $doc_id}})
RETURN doc
"""
try:
result = await self.execute_query(cypher, {"doc_id": doc_id})
return result[0]["doc"] if result else None
except Exception as e:
logger.error(f"Failed to get document node: {e}", exc_info=True)
return None
async def delete_document_node(
self,
user: str,
doc_id: str
) -> bool:
"""
Delete document node and all its relationships.
Args:
user: User identifier
doc_id: Document ID
Returns:
True if deleted, False otherwise
"""
user_label = self.get_user_label(user)
cypher = f"""
MATCH (doc:Document:{user_label} {{id: $doc_id}})
DETACH DELETE doc
RETURN count(doc) as deleted
"""
try:
result = await self.execute_write(cypher, {"doc_id": doc_id})
return result[0]["deleted"] > 0 if result else False
except Exception as e:
logger.error(f"Failed to delete document node: {e}", exc_info=True)
return False
async def find_similar_documents(
self,
user: str,
doc_ids: List[str],
max_depth: int = 2,
limit: int = 20
) -> List[Dict[str, Any]]:
"""
Find documents similar to given docs via graph traversal.
Uses: Shared concepts, shared entities, citation links.
Args:
user: User identifier
doc_ids: List of source document IDs
max_depth: Maximum traversal depth
limit: Maximum results to return
Returns:
List of similar documents with connection strength
"""
user_label = self.get_user_label(user)
cypher = f"""
MATCH (source:Document:{user_label})
WHERE source.id IN $doc_ids
MATCH (source)-[*1..{max_depth}]-(related:Document:{user_label})
WHERE related.id <> source.id AND NOT related.id IN $doc_ids
WITH related, count(*) as connection_strength
ORDER BY connection_strength DESC
LIMIT $limit
RETURN related, connection_strength
"""
try:
result = await self.execute_query(
cypher,
{"doc_ids": doc_ids, "limit": limit}
)
return result
except Exception as e:
logger.error(f"Failed to find similar documents: {e}", exc_info=True)
return []
async def list_user_documents(
self,
user: str,
limit: int = 100,
offset: int = 0
) -> List[Dict[str, Any]]:
"""
List all documents for a user.
Args:
user: User identifier
limit: Maximum results to return
offset: Number of results to skip
Returns:
List of document nodes
"""
user_label = self.get_user_label(user)
cypher = f"""
MATCH (doc:Document:{user_label})
RETURN doc
ORDER BY doc.created_at DESC
SKIP $offset
LIMIT $limit
"""
try:
result = await self.execute_query(
cypher,
{"offset": offset, "limit": limit}
)
return [r["doc"] for r in result]
except Exception as e:
logger.error(f"Failed to list documents: {e}", exc_info=True)
return []
# Concept/entity operations
async def create_concept_node(
self,
concept_name: str,
concept_type: str,
properties: Optional[Dict[str, Any]] = None
) -> Optional[Dict[str, Any]]:
"""
Create or update concept node.
Args:
concept_name: Concept name
concept_type: Concept type (Technique, Tool, Pattern, etc.)
properties: Additional properties
Returns:
Concept node properties
"""
cypher = """
MERGE (concept:Concept {name: $name})
ON CREATE SET
concept.type = $type,
concept.first_seen = timestamp(),
concept.mention_count = 1
ON MATCH SET
concept.mention_count = concept.mention_count + 1
SET concept += $properties
RETURN concept
"""
try:
result = await self.execute_write(
cypher,
{
"name": concept_name,
"type": concept_type,
"properties": properties or {}
}
)
return result[0]["concept"] if result else None
except Exception as e:
logger.error(f"Failed to create concept node: {e}", exc_info=True)
return None
async def link_document_to_concept(
self,
user: str,
doc_id: str,
concept_name: str
) -> bool:
"""
Create MENTIONS relationship between document and concept.
Args:
user: User identifier
doc_id: Document ID
concept_name: Concept name
Returns:
True if link created
"""
user_label = self.get_user_label(user)
cypher = f"""
MATCH (doc:Document:{user_label} {{id: $doc_id}})
MATCH (concept:Concept {{name: $concept_name}})
MERGE (doc)-[r:MENTIONS]->(concept)
ON CREATE SET r.count = 1
ON MATCH SET r.count = r.count + 1
RETURN r
"""
try:
result = await self.execute_write(
cypher,
{"doc_id": doc_id, "concept_name": concept_name}
)
return len(result) > 0
except Exception as e:
logger.error(f"Failed to link document to concept: {e}", exc_info=True)
return False
@@ -0,0 +1,366 @@
"""
Ollama client for embeddings generation.
Provides async embedding operations via Ollama API:
- Text embedding generation
- Batch embedding support
- Model management
"""
import httpx
from typing import List, Dict, Any, Optional
import logging
logger = logging.getLogger(__name__)
class OllamaClient:
"""
Ollama API client for embeddings.
Documentation: https://github.com/ollama/ollama/blob/main/docs/api.md
Default model: nomic-embed-text (768-dimensional embeddings)
"""
def __init__(self, base_url: str, model: str = "nomic-embed-text"):
"""
Initialize Ollama client.
Args:
base_url: Ollama server URL (e.g., "http://ollama:11434")
model: Embedding model name (default: "nomic-embed-text")
"""
self.base_url = base_url.rstrip("/")
self.model = model
self.embeddings_url = f"{self.base_url}/api/embeddings"
self.generate_url = f"{self.base_url}/api/generate"
self.tags_url = f"{self.base_url}/api/tags"
self.client = httpx.AsyncClient(timeout=120.0) # Embeddings can be slow
logger.info(f"Initialized Ollama client: {base_url}, model: {model}")
async def close(self):
"""Close HTTP client"""
await self.client.aclose()
async def embed(self, text: str) -> Optional[List[float]]:
"""
Generate embedding for single text.
Args:
text: Text to embed
Returns:
Embedding vector (768-dimensional for nomic-embed-text) or None on failure
Example:
>>> embedding = await client.embed("Hello world")
>>> len(embedding)
768
"""
try:
payload = {
"model": self.model,
"prompt": text
}
response = await self.client.post(
self.embeddings_url,
json=payload
)
response.raise_for_status()
data = response.json()
embedding = data.get("embedding")
if not embedding:
logger.error(f"No embedding in response: {data}")
return None
return embedding
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
return None
except Exception as e:
logger.error(f"Embedding failed: {e}", exc_info=True)
return None
async def embed_batch(
self,
texts: List[str],
show_progress: bool = False
) -> List[Optional[List[float]]]:
"""
Generate embeddings for multiple texts.
Args:
texts: List of texts to embed
show_progress: Log progress for large batches
Returns:
List of embedding vectors (same order as input)
None entries for texts that failed to embed
Example:
>>> texts = ["Hello", "World", "Test"]
>>> embeddings = await client.embed_batch(texts)
>>> len(embeddings)
3
"""
embeddings = []
for i, text in enumerate(texts):
if show_progress and i % 10 == 0:
logger.info(f"Embedding progress: {i}/{len(texts)}")
embedding = await self.embed(text)
embeddings.append(embedding)
if show_progress:
logger.info(f"Embedding complete: {len(embeddings)}/{len(texts)}")
return embeddings
async def embed_batch_filtered(
self,
texts: List[str],
show_progress: bool = False
) -> List[List[float]]:
"""
Generate embeddings for multiple texts, filtering out failures.
Args:
texts: List of texts to embed
show_progress: Log progress for large batches
Returns:
List of successful embedding vectors (may be shorter than input)
Example:
>>> texts = ["Hello", "World", "Test"]
>>> embeddings = await client.embed_batch_filtered(texts)
>>> all(e is not None for e in embeddings)
True
"""
all_embeddings = await self.embed_batch(texts, show_progress)
return [e for e in all_embeddings if e is not None]
async def embed_documents(
self,
documents: List[Dict[str, Any]],
content_field: str = "content",
show_progress: bool = False
) -> List[Dict[str, Any]]:
"""
Embed documents with metadata preservation.
Args:
documents: List of document dictionaries
content_field: Field name containing text to embed
show_progress: Log progress for large batches
Returns:
List of documents with added "embedding" field
Example:
>>> docs = [
... {"content": "Hello world", "id": 1},
... {"content": "Test doc", "id": 2}
... ]
>>> embedded = await client.embed_documents(docs)
>>> "embedding" in embedded[0]
True
"""
texts = [doc.get(content_field, "") for doc in documents]
embeddings = await self.embed_batch(texts, show_progress)
results = []
for doc, embedding in zip(documents, embeddings):
doc_copy = doc.copy()
doc_copy["embedding"] = embedding
results.append(doc_copy)
return results
async def get_embedding_dimension(self) -> Optional[int]:
"""
Get embedding dimension for current model.
Returns:
Embedding dimension (e.g., 768 for nomic-embed-text) or None on failure
Example:
>>> dim = await client.get_embedding_dimension()
>>> dim
768
"""
# Generate a test embedding to determine dimension
test_embedding = await self.embed("test")
if test_embedding:
return len(test_embedding)
return None
async def list_models(self) -> List[Dict[str, Any]]:
"""
List available Ollama models.
Returns:
List of model information dictionaries
Example:
>>> models = await client.list_models()
>>> any(m["name"] == "nomic-embed-text" for m in models)
True
"""
try:
response = await self.client.get(self.tags_url)
response.raise_for_status()
data = response.json()
return data.get("models", [])
except Exception as e:
logger.error(f"Failed to list models: {e}")
return []
async def check_model_available(self, model_name: Optional[str] = None) -> bool:
"""
Check if a model is available.
Args:
model_name: Model name to check (defaults to self.model)
Returns:
True if model is available, False otherwise
"""
check_model = model_name or self.model
models = await self.list_models()
return any(m.get("name") == check_model for m in models)
async def generate_text(
self,
prompt: str,
model: Optional[str] = None,
stream: bool = False
) -> Optional[str]:
"""
Generate text completion (for non-embedding use cases).
Args:
prompt: Input prompt
model: Model name (defaults to self.model)
stream: Enable streaming response
Returns:
Generated text or None on failure
Note: This is primarily for debugging/testing. Use specialized
LLM services for production text generation.
"""
try:
payload = {
"model": model or self.model,
"prompt": prompt,
"stream": stream
}
response = await self.client.post(
self.generate_url,
json=payload
)
response.raise_for_status()
if stream:
# For streaming, return first chunk
# Full streaming implementation would need async generator
return response.text
else:
data = response.json()
return data.get("response")
except Exception as e:
logger.error(f"Text generation failed: {e}")
return None
async def health_check(self) -> bool:
"""
Check if Ollama server is reachable and model is available.
Returns:
True if healthy, False otherwise
"""
try:
# Check server is up
response = await self.client.get(self.tags_url, timeout=5.0)
response.raise_for_status()
# Check model is available
models_available = await self.check_model_available()
if not models_available:
logger.warning(f"Model '{self.model}' not found in Ollama")
return False
return True
except Exception as e:
logger.error(f"Health check failed: {e}")
return False
def estimate_tokens(self, text: str) -> int:
"""
Rough estimate of token count for text.
Uses simple heuristic: ~4 characters per token.
Args:
text: Input text
Returns:
Estimated token count
"""
return len(text) // 4
def chunk_text_for_embedding(
self,
text: str,
max_tokens: int = 512,
overlap: int = 50
) -> List[str]:
"""
Chunk text into segments suitable for embedding.
Args:
text: Input text
max_tokens: Maximum tokens per chunk
overlap: Token overlap between chunks
Returns:
List of text chunks
Example:
>>> chunks = client.chunk_text_for_embedding(long_text, max_tokens=512)
>>> all(client.estimate_tokens(c) <= 512 for c in chunks)
True
"""
# Convert tokens to approximate character count
max_chars = max_tokens * 4
overlap_chars = overlap * 4
if len(text) <= max_chars:
return [text]
chunks = []
start = 0
while start < len(text):
end = start + max_chars
chunk = text[start:end]
# Try to break at sentence boundary
if end < len(text):
last_period = chunk.rfind(". ")
if last_period > max_chars * 0.5: # Only break if > 50% through chunk
end = start + last_period + 1
chunk = text[start:end]
chunks.append(chunk.strip())
start = end - overlap_chars
return chunks
@@ -0,0 +1,412 @@
"""
Qdrant client wrapper for Library Desk.
Provides async vector operations with:
- Collection-per-user multi-tenancy
- Document chunk storage with embeddings
- Semantic search
- Similarity queries
"""
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchValue
)
from typing import List, Dict, Any, Optional
import uuid
import logging
from src.core.multi_tenancy import get_qdrant_collection_name
logger = logging.getLogger(__name__)
class QdrantClientWrapper:
"""
Qdrant client wrapper with multi-tenancy support.
Pattern: Collection per user (from qdrant_memory.py)
Each user has isolated vector collection for their documents.
"""
def __init__(self, url: str, embedding_dim: int = 768):
"""
Initialize Qdrant client.
Args:
url: Qdrant server URL (e.g., "http://qdrant:6333")
embedding_dim: Vector embedding dimension (default 768 for nomic-embed-text)
"""
self.client = QdrantClient(url=url)
self.embedding_dim = embedding_dim
logger.info(f"Initialized Qdrant client: {url}")
def get_collection_name(self, user: str) -> str:
"""
Get collection name for user.
Args:
user: User identifier
Returns:
Collection name (e.g., "library_desk_jpmschweitzer")
"""
return get_qdrant_collection_name(user)
async def ensure_collection(self, user: str):
"""
Create user's collection if it doesn't exist.
Args:
user: User identifier
"""
collection_name = self.get_collection_name(user)
try:
collections = self.client.get_collections()
existing = [c.name for c in collections.collections]
if collection_name not in existing:
logger.info(f"Creating Qdrant collection: {collection_name}")
self.client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=self.embedding_dim,
distance=Distance.COSINE
)
)
logger.info(f"Created collection: {collection_name}")
except Exception as e:
logger.error(f"Error ensuring collection: {e}", exc_info=True)
raise
async def upsert_document_chunks(
self,
user: str,
doc_id: str,
chunks: List[Dict[str, Any]],
embeddings: List[List[float]]
) -> int:
"""
Upsert document chunks with embeddings.
Point structure:
{
id: "doc_123_chunk_0",
vector: [...],
payload: {
doc_id: "doc_123",
chunk_index: 0,
content: "text content",
metadata: {...}
}
}
Args:
user: User identifier
doc_id: Document ID
chunks: List of chunk dictionaries with content
embeddings: List of embedding vectors
Returns:
Number of chunks upserted
Raises:
ValueError: If chunks and embeddings length mismatch
"""
if len(chunks) != len(embeddings):
raise ValueError(
f"Chunks ({len(chunks)}) and embeddings ({len(embeddings)}) length mismatch"
)
collection_name = self.get_collection_name(user)
await self.ensure_collection(user)
points = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
# Generate deterministic point ID
point_id = str(uuid.uuid5(
uuid.NAMESPACE_DNS,
f"{doc_id}_chunk_{i}"
))
points.append(PointStruct(
id=point_id,
vector=embedding,
payload={
"doc_id": doc_id,
"chunk_index": i,
"content": chunk.get("content", ""),
"metadata": chunk.get("metadata", {})
}
))
try:
self.client.upsert(
collection_name=collection_name,
points=points
)
logger.info(f"Upserted {len(points)} chunks for {doc_id}")
return len(points)
except Exception as e:
logger.error(f"Failed to upsert chunks: {e}", exc_info=True)
raise
async def search(
self,
user: str,
query_vector: List[float],
limit: int = 10,
score_threshold: float = 0.7,
filter_dict: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
Semantic search in user's collection.
Args:
user: User identifier
query_vector: Query embedding vector
limit: Maximum results to return
score_threshold: Minimum similarity score (0.0-1.0)
filter_dict: Optional payload filters
Returns:
List of matches with scores and payloads
Example:
results = await client.search(
user="jpmschweitzer",
query_vector=[0.1, 0.2, ...],
limit=5,
filter_dict={"doc_id": "doc_123"}
)
"""
collection_name = self.get_collection_name(user)
# Build filter if provided
query_filter = None
if filter_dict:
conditions = []
for key, value in filter_dict.items():
conditions.append(
FieldCondition(key=key, match=MatchValue(value=value))
)
query_filter = Filter(must=conditions)
try:
results = self.client.search(
collection_name=collection_name,
query_vector=query_vector,
limit=limit,
score_threshold=score_threshold,
query_filter=query_filter,
with_payload=True
)
return [
{
"id": str(result.id),
"score": result.score,
"doc_id": result.payload["doc_id"],
"chunk_index": result.payload["chunk_index"],
"content": result.payload["content"],
"metadata": result.payload.get("metadata", {})
}
for result in results
]
except Exception as e:
logger.error(f"Search failed: {e}", exc_info=True)
return []
async def delete_document(self, user: str, doc_id: str) -> bool:
"""
Delete all chunks for a document.
Args:
user: User identifier
doc_id: Document ID
Returns:
True if deleted successfully
"""
collection_name = self.get_collection_name(user)
try:
self.client.delete(
collection_name=collection_name,
points_selector=Filter(
must=[
FieldCondition(
key="doc_id",
match=MatchValue(value=doc_id)
)
]
)
)
logger.info(f"Deleted chunks for {doc_id}")
return True
except Exception as e:
logger.error(f"Failed to delete document: {e}", exc_info=True)
return False
async def get_document_chunks(
self,
user: str,
doc_id: str
) -> List[Dict[str, Any]]:
"""
Get all chunks for a document.
Args:
user: User identifier
doc_id: Document ID
Returns:
List of chunks with content and metadata
"""
collection_name = self.get_collection_name(user)
try:
# Scroll through points with doc_id filter
points, _ = self.client.scroll(
collection_name=collection_name,
scroll_filter=Filter(
must=[
FieldCondition(
key="doc_id",
match=MatchValue(value=doc_id)
)
]
),
limit=1000,
with_payload=True,
with_vectors=False
)
return [
{
"id": str(point.id),
"chunk_index": point.payload["chunk_index"],
"content": point.payload["content"],
"metadata": point.payload.get("metadata", {})
}
for point in points
]
except Exception as e:
logger.error(f"Failed to get document chunks: {e}", exc_info=True)
return []
async def count_documents(self, user: str) -> int:
"""
Count total number of unique documents in user's collection.
Args:
user: User identifier
Returns:
Number of unique documents
"""
collection_name = self.get_collection_name(user)
try:
# Get collection info
collection_info = self.client.get_collection(collection_name)
# This gives total points, not unique docs
# For unique docs, would need to aggregate by doc_id
return collection_info.points_count
except Exception as e:
logger.error(f"Failed to count documents: {e}", exc_info=True)
return 0
async def delete_collection(self, user: str) -> bool:
"""
Delete user's entire collection.
Warning: This removes all data for the user!
Args:
user: User identifier
Returns:
True if deleted successfully
"""
collection_name = self.get_collection_name(user)
try:
self.client.delete_collection(collection_name)
logger.warning(f"Deleted collection: {collection_name}")
return True
except Exception as e:
logger.error(f"Failed to delete collection: {e}", exc_info=True)
return False
async def find_similar_chunks(
self,
user: str,
doc_id: str,
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Find chunks similar to those in a given document.
Strategy: Get all chunks from doc, use their vectors to find similar chunks.
Args:
user: User identifier
doc_id: Source document ID
limit: Maximum results per chunk
Returns:
List of similar chunks from other documents
"""
collection_name = self.get_collection_name(user)
try:
# Get source document chunks with vectors
source_points, _ = self.client.scroll(
collection_name=collection_name,
scroll_filter=Filter(
must=[
FieldCondition(
key="doc_id",
match=MatchValue(value=doc_id)
)
]
),
limit=10, # Sample first 10 chunks
with_payload=True,
with_vectors=True
)
if not source_points:
return []
# Search using first chunk's vector
# (could aggregate multiple chunks for better results)
first_vector = source_points[0].vector
results = self.client.search(
collection_name=collection_name,
query_vector=first_vector,
limit=limit * 2, # Get more to filter out same doc
with_payload=True
)
# Filter out chunks from same document
similar = [
{
"id": str(r.id),
"score": r.score,
"doc_id": r.payload["doc_id"],
"content": r.payload["content"]
}
for r in results
if r.payload["doc_id"] != doc_id
]
return similar[:limit]
except Exception as e:
logger.error(f"Failed to find similar chunks: {e}", exc_info=True)
return []
@@ -0,0 +1,332 @@
"""
SearXNG search client for Library Desk.
Provides async web search via SearXNG instance:
- General web search
- Category-specific search
- Language filtering
- Result formatting
"""
import httpx
from typing import Optional, List, Dict, Any
import logging
logger = logging.getLogger(__name__)
class SearXNGClient:
"""
SearXNG search API client.
Documentation: https://docs.searxng.org/dev/search_api.html
API Format: JSON (?format=json)
"""
def __init__(self, base_url: str):
"""
Initialize SearXNG client.
Args:
base_url: SearXNG base URL (e.g., "http://searxng:8080")
"""
self.base_url = base_url.rstrip("/")
self.search_url = f"{self.base_url}/search"
self.client = httpx.AsyncClient(timeout=30.0)
logger.info(f"Initialized SearXNG client: {base_url}")
async def close(self):
"""Close HTTP client"""
await self.client.aclose()
async def search(
self,
query: str,
categories: Optional[List[str]] = None,
language: str = "en",
time_range: Optional[str] = None,
safesearch: int = 0,
pageno: int = 1,
limit: int = 10
) -> Dict[str, Any]:
"""
Search via SearXNG.
Args:
query: Search query string
categories: List of categories (e.g., ["general", "images", "news"])
language: Language code (e.g., "en", "fr", "auto")
time_range: Time range filter ("day", "week", "month", "year", None for all)
safesearch: Safe search level (0=off, 1=moderate, 2=strict)
pageno: Page number (starts at 1)
limit: Maximum results to return
Returns:
Search results dictionary with:
- query: Original query
- results: List of result items
- number_of_results: Total results found
- suggestions: Query suggestions
Raises:
Exception: If search fails
"""
params = {
"q": query,
"format": "json",
"language": language,
"safesearch": safesearch,
"pageno": pageno
}
# Add optional parameters
if categories:
params["categories"] = ",".join(categories)
if time_range:
params["time_range"] = time_range
try:
response = await self.client.get(self.search_url, params=params)
response.raise_for_status()
data = response.json()
# Limit results if requested
results = data.get("results", [])
if limit:
results = results[:limit]
return {
"query": data.get("query", query),
"results": results,
"number_of_results": data.get("number_of_results", 0),
"suggestions": data.get("suggestions", []),
"answers": data.get("answers", []),
"infoboxes": data.get("infoboxes", [])
}
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
raise Exception(f"SearXNG search failed: {e.response.status_code}")
except Exception as e:
logger.error(f"Search failed: {e}", exc_info=True)
raise
async def search_general(
self,
query: str,
language: str = "en",
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Simplified general web search.
Args:
query: Search query
language: Language code
limit: Maximum results
Returns:
List of search results with:
- title: Page title
- url: Page URL
- content: Page description/snippet
- engine: Search engine that provided result
- score: Relevance score (if available)
"""
try:
result = await self.search(
query=query,
categories=["general"],
language=language,
limit=limit
)
return result.get("results", [])
except Exception as e:
logger.error(f"General search failed: {e}")
return []
async def search_with_context(
self,
query: str,
context: Optional[str] = None,
language: str = "en",
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Search with additional context appended to query.
Useful for RAG queries where you want to enhance search with context.
Args:
query: Primary search query
context: Additional context to append
language: Language code
limit: Maximum results
Returns:
List of search results
"""
enhanced_query = f"{query} {context}" if context else query
return await self.search_general(enhanced_query, language, limit)
async def search_documentation(
self,
query: str,
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Search for technical documentation.
Enhances query with "documentation" and filters for technical content.
Args:
query: Search query (e.g., "FastAPI")
limit: Maximum results
Returns:
List of documentation search results
"""
enhanced_query = f"{query} documentation"
return await self.search_general(enhanced_query, limit=limit)
async def search_recent(
self,
query: str,
time_range: str = "month",
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Search for recent content only.
Args:
query: Search query
time_range: Time range ("day", "week", "month", "year")
limit: Maximum results
Returns:
List of recent search results
"""
try:
result = await self.search(
query=query,
categories=["general"],
time_range=time_range,
limit=limit
)
return result.get("results", [])
except Exception as e:
logger.error(f"Recent search failed: {e}")
return []
async def get_suggestions(self, query: str) -> List[str]:
"""
Get search query suggestions.
Args:
query: Partial query
Returns:
List of suggested queries
"""
try:
result = await self.search(query=query, limit=1)
return result.get("suggestions", [])
except Exception as e:
logger.error(f"Failed to get suggestions: {e}")
return []
async def search_images(
self,
query: str,
safesearch: int = 1,
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Search for images.
Args:
query: Search query
safesearch: Safe search level (0=off, 1=moderate, 2=strict)
limit: Maximum results
Returns:
List of image results with:
- title: Image title
- url: Image URL
- thumbnail_src: Thumbnail URL
- img_src: Full image URL
- content: Description
"""
try:
result = await self.search(
query=query,
categories=["images"],
safesearch=safesearch,
limit=limit
)
return result.get("results", [])
except Exception as e:
logger.error(f"Image search failed: {e}")
return []
async def search_news(
self,
query: str,
time_range: str = "week",
limit: int = 10
) -> List[Dict[str, Any]]:
"""
Search for news articles.
Args:
query: Search query
time_range: Time range ("day", "week", "month")
limit: Maximum results
Returns:
List of news results
"""
try:
result = await self.search(
query=query,
categories=["news"],
time_range=time_range,
limit=limit
)
return result.get("results", [])
except Exception as e:
logger.error(f"News search failed: {e}")
return []
async def format_results_for_rag(
self,
results: List[Dict[str, Any]],
max_snippet_length: int = 500
) -> List[Dict[str, str]]:
"""
Format search results for RAG context.
Extracts relevant fields and truncates content.
Args:
results: Raw search results
max_snippet_length: Maximum length of content snippet
Returns:
Formatted results with title, url, snippet
"""
formatted = []
for result in results:
content = result.get("content", "")
if len(content) > max_snippet_length:
content = content[:max_snippet_length] + "..."
formatted.append({
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": content,
"engine": result.get("engine", ""),
"score": result.get("score", 0.0)
})
return formatted
@@ -0,0 +1,502 @@
"""
Wiki.js GraphQL client for Library Desk.
Provides async Wiki.js operations via GraphQL API:
- Page CRUD (create, read, update, delete)
- Search and listing
- Tag management
- Multi-tenancy via path namespaces
"""
import httpx
from typing import Optional, List, Dict, Any
import logging
logger = logging.getLogger(__name__)
class WikiJSClient:
"""
Wiki.js GraphQL API client.
Documentation: https://docs.requarks.io/dev/api
Authentication: Bearer token in Authorization header
"""
def __init__(self, base_url: str, api_key: str):
"""
Initialize Wiki.js client.
Args:
base_url: Wiki.js base URL (e.g., "http://wiki:3000")
api_key: Wiki.js API key (from Admin → API Access)
"""
self.base_url = base_url.rstrip("/")
self.graphql_url = f"{self.base_url}/graphql"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.client = httpx.AsyncClient(timeout=30.0)
logger.info(f"Initialized Wiki.js client: {base_url}")
async def close(self):
"""Close HTTP client"""
await self.client.aclose()
async def _execute_query(
self,
query: str,
variables: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Execute GraphQL query.
Args:
query: GraphQL query string
variables: Query variables
Returns:
Query result data
Raises:
Exception: If query fails or returns errors
"""
payload = {
"query": query,
"variables": variables or {}
}
try:
response = await self.client.post(
self.graphql_url,
headers=self.headers,
json=payload
)
response.raise_for_status()
result = response.json()
if "errors" in result:
logger.error(f"GraphQL errors: {result['errors']}")
raise Exception(f"GraphQL errors: {result['errors']}")
return result.get("data", {})
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
raise
except Exception as e:
logger.error(f"GraphQL query failed: {e}", exc_info=True)
raise
async def list_pages(
self,
path_prefix: str = "",
tags: Optional[List[str]] = None,
limit: int = 50
) -> List[Dict[str, Any]]:
"""
List pages with optional filtering.
Multi-tenancy: Use path_prefix to filter by user namespace.
Args:
path_prefix: Filter by path prefix (e.g., "/users/jpmschweitzer")
tags: Filter by tags (e.g., ["projects"])
limit: Maximum results
Returns:
List of page objects
"""
query = """
query ListPages($limit: Int, $orderBy: PageOrderBy) {
pages {
list(limit: $limit, orderBy: $orderBy) {
id
path
title
description
tags
createdAt
updatedAt
isPublished
}
}
}
"""
variables = {
"limit": limit,
"orderBy": "TITLE"
}
data = await self._execute_query(query, variables)
pages = data.get("pages", {}).get("list", [])
# Filter by path prefix (client-side if API doesn't support)
if path_prefix:
pages = [p for p in pages if p["path"].startswith(path_prefix)]
# Filter by tags (dossiers)
if tags:
pages = [
p for p in pages
if any(tag in (p.get("tags") or []) for tag in tags)
]
return pages
async def get_page(self, page_id: int) -> Optional[Dict[str, Any]]:
"""
Get single page by ID.
Args:
page_id: Page ID
Returns:
Page object or None if not found
"""
query = """
query GetPage($id: Int!) {
pages {
single(id: $id) {
id
path
title
description
content
tags
createdAt
updatedAt
isPublished
editor
}
}
}
"""
try:
data = await self._execute_query(query, {"id": page_id})
return data.get("pages", {}).get("single")
except Exception as e:
logger.error(f"Failed to get page {page_id}: {e}")
return None
async def create_page(
self,
path: str,
title: str,
content: str,
description: str = "",
tags: Optional[List[str]] = None,
is_published: bool = True,
editor: str = "markdown"
) -> Dict[str, Any]:
"""
Create new page.
Multi-tenancy: Ensure path starts with user namespace.
Args:
path: Page path (e.g., "/users/jpmschweitzer/projects/library-desk")
title: Page title
content: Page content (markdown)
description: Short description
tags: List of tags (for dossier organization)
is_published: Whether page is published
editor: Editor type (markdown, wysiwyg, etc.)
Returns:
Created page object
Raises:
Exception: If creation fails
"""
mutation = """
mutation CreatePage(
$content: String!,
$description: String!,
$editor: String!,
$isPublished: Boolean!,
$locale: String!,
$path: String!,
$tags: [String]!,
$title: String!
) {
pages {
create(
content: $content,
description: $description,
editor: $editor,
isPublished: $isPublished,
locale: $locale,
path: $path,
tags: $tags,
title: $title
) {
responseResult {
succeeded
errorCode
message
}
page {
id
path
title
}
}
}
}
"""
variables = {
"path": path,
"title": title,
"content": content,
"description": description,
"tags": tags or [],
"isPublished": is_published,
"editor": editor,
"locale": "en"
}
data = await self._execute_query(mutation, variables)
result = data.get("pages", {}).get("create", {})
if not result.get("responseResult", {}).get("succeeded"):
error = result.get("responseResult", {})
raise Exception(f"Failed to create page: {error}")
logger.info(f"Created page: {path}")
return result.get("page")
async def update_page(
self,
page_id: int,
content: Optional[str] = None,
title: Optional[str] = None,
description: Optional[str] = None,
tags: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Update existing page.
Args:
page_id: Page ID
content: New content (optional)
title: New title (optional)
description: New description (optional)
tags: New tags (optional)
Returns:
Updated page object
Raises:
Exception: If update fails
"""
mutation = """
mutation UpdatePage(
$id: Int!,
$content: String,
$title: String,
$description: String,
$tags: [String]
) {
pages {
update(
id: $id,
content: $content,
title: $title,
description: $description,
tags: $tags
) {
responseResult {
succeeded
errorCode
message
}
page {
id
updatedAt
}
}
}
}
"""
variables = {"id": page_id}
if content is not None:
variables["content"] = content
if title is not None:
variables["title"] = title
if description is not None:
variables["description"] = description
if tags is not None:
variables["tags"] = tags
data = await self._execute_query(mutation, variables)
result = data.get("pages", {}).get("update", {})
if not result.get("responseResult", {}).get("succeeded"):
error = result.get("responseResult", {})
raise Exception(f"Failed to update page: {error}")
logger.info(f"Updated page: {page_id}")
return result.get("page")
async def delete_page(self, page_id: int):
"""
Delete page.
Args:
page_id: Page ID
Raises:
Exception: If deletion fails
"""
mutation = """
mutation DeletePage($id: Int!) {
pages {
delete(id: $id) {
responseResult {
succeeded
errorCode
message
}
}
}
}
"""
data = await self._execute_query(mutation, {"id": page_id})
result = data.get("pages", {}).get("delete", {})
if not result.get("responseResult", {}).get("succeeded"):
error = result.get("responseResult", {})
raise Exception(f"Failed to delete page: {error}")
logger.info(f"Deleted page: {page_id}")
async def search_pages(
self,
query: str,
path_prefix: Optional[str] = None
) -> List[Dict[str, Any]]:
"""
Search pages by content.
Args:
query: Search query
path_prefix: Optional path prefix filter
Returns:
List of matching pages
"""
gql_query = """
query SearchPages($query: String!) {
pages {
search(query: $query) {
results {
id
path
title
description
}
}
}
}
"""
data = await self._execute_query(gql_query, {"query": query})
results = data.get("pages", {}).get("search", {}).get("results", [])
# Filter by path prefix if provided
if path_prefix:
results = [r for r in results if r["path"].startswith(path_prefix)]
return results
async def get_page_tree(self, path: str = "/") -> List[Dict[str, Any]]:
"""
Get page tree structure.
Args:
path: Root path
Returns:
Tree structure of pages
"""
query = """
query GetPageTree($parent: Int, $mode: String!) {
pages {
tree(parent: $parent, mode: $mode) {
id
path
title
isFolder
pageId
}
}
}
"""
try:
data = await self._execute_query(
query,
{"parent": 0, "mode": "all"}
)
return data.get("pages", {}).get("tree", [])
except Exception as e:
logger.error(f"Failed to get page tree: {e}")
return []
async def move_page(
self,
page_id: int,
new_path: str,
locale: str = "en"
) -> bool:
"""
Move/rename page.
Args:
page_id: Page ID
new_path: New page path
locale: Page locale
Returns:
True if successful
"""
mutation = """
mutation MovePage($id: Int!, $destinationPath: String!, $destinationLocale: String!) {
pages {
move(id: $id, destinationPath: $destinationPath, destinationLocale: $destinationLocale) {
responseResult {
succeeded
errorCode
message
}
}
}
}
"""
try:
data = await self._execute_query(
mutation,
{
"id": page_id,
"destinationPath": new_path,
"destinationLocale": locale
}
)
result = data.get("pages", {}).get("move", {})
success = result.get("responseResult", {}).get("succeeded", False)
if success:
logger.info(f"Moved page {page_id} to {new_path}")
return success
except Exception as e:
logger.error(f"Failed to move page: {e}")
return False
+2 -2
View File
@@ -41,10 +41,10 @@ class Settings(BaseSettings):
ollama_url: str = Field(default="http://ollama:11434", description="Ollama URL")
ollama_model: str = Field(default="nomic-embed-text", description="Ollama embedding model")
# Redis Configuration
# Redis Configuration (for job tracking - separate DB from wiki)
redis_host: str = Field(default="redis-shared", description="Redis host")
redis_port: int = Field(default=6379, description="Redis port")
redis_db: int = Field(default=2, description="Redis database number")
redis_db: int = Field(default=4, description="Redis database number (4 for library-desk jobs)")
# Application
app_name: str = Field(default="Library Desk", description="Application name")
@@ -0,0 +1,286 @@
"""
Dependency injection for Library Desk.
Provides FastAPI dependencies for service clients with:
- Singleton pattern via @lru_cache
- Lazy initialization
- Proper lifecycle management
- Type aliases for clean endpoint signatures
"""
from functools import lru_cache
from typing import Annotated
from fastapi import Depends
import logging
from src.config import Settings, get_settings
from src.clients.neo4j_client import Neo4jClient
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.wikijs_client import WikiJSClient
from src.clients.searxng_client import SearXNGClient
from src.clients.ollama_client import OllamaClient
logger = logging.getLogger(__name__)
# Settings dependency
SettingsDep = Annotated[Settings, Depends(get_settings)]
# Client factory functions with @lru_cache for singletons
@lru_cache
def get_neo4j_client() -> Neo4jClient:
"""
Get Neo4j client singleton.
Returns:
Initialized Neo4j client (not yet connected)
Note: Call client.connect() during app startup
"""
settings = get_settings()
client = Neo4jClient(
uri=settings.neo4j_uri,
user=settings.neo4j_user,
password=settings.neo4j_password
)
logger.debug("Created Neo4j client instance")
return client
@lru_cache
def get_qdrant_client() -> QdrantClientWrapper:
"""
Get Qdrant client singleton.
Returns:
Initialized Qdrant client
Note: Collections are created lazily per-user
"""
settings = get_settings()
client = QdrantClientWrapper(
url=settings.qdrant_url,
embedding_dim=768 # nomic-embed-text default
)
logger.debug("Created Qdrant client instance")
return client
@lru_cache
def get_wikijs_client() -> WikiJSClient:
"""
Get Wiki.js client singleton.
Returns:
Initialized Wiki.js GraphQL client
"""
settings = get_settings()
client = WikiJSClient(
base_url=settings.wikijs_url,
api_key=settings.wikijs_api_key
)
logger.debug("Created Wiki.js client instance")
return client
@lru_cache
def get_searxng_client() -> SearXNGClient:
"""
Get SearXNG client singleton.
Returns:
Initialized SearXNG search client
"""
settings = get_settings()
client = SearXNGClient(base_url=settings.searxng_url)
logger.debug("Created SearXNG client instance")
return client
@lru_cache
def get_ollama_client() -> OllamaClient:
"""
Get Ollama client singleton.
Returns:
Initialized Ollama embeddings client
"""
settings = get_settings()
client = OllamaClient(
base_url=settings.ollama_url,
model=settings.ollama_model
)
logger.debug("Created Ollama client instance")
return client
# Type aliases for FastAPI endpoint dependencies
# Usage: def my_endpoint(neo4j: Neo4jDep):
Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)]
QdrantDep = Annotated[QdrantClientWrapper, Depends(get_qdrant_client)]
WikiJSDep = Annotated[WikiJSClient, Depends(get_wikijs_client)]
SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)]
OllamaDep = Annotated[OllamaClient, Depends(get_ollama_client)]
# Lifecycle management functions
async def startup_clients():
"""
Initialize all service clients at application startup.
Should be called in FastAPI lifespan or startup event.
Performs:
- Neo4j connection pool initialization
- Neo4j connectivity verification
- Ollama model availability check
"""
logger.info("Starting up service clients...")
# Initialize Neo4j connection pool
neo4j = get_neo4j_client()
try:
await neo4j.connect()
logger.info("✓ Neo4j connected")
except Exception as e:
logger.error(f"✗ Neo4j connection failed: {e}")
# Don't fail startup - allow degraded operation
pass
# Check Ollama availability
ollama = get_ollama_client()
try:
is_healthy = await ollama.health_check()
if is_healthy:
logger.info(f"✓ Ollama ready (model: {ollama.model})")
else:
logger.warning(f"✗ Ollama model '{ollama.model}' not available")
except Exception as e:
logger.error(f"✗ Ollama health check failed: {e}")
pass
# Qdrant, Wiki.js, SearXNG are lazy-initialized
logger.info("Service clients startup complete")
async def shutdown_clients():
"""
Cleanup all service clients at application shutdown.
Should be called in FastAPI lifespan or shutdown event.
Performs:
- Close Neo4j connection pool
- Close HTTP clients
"""
logger.info("Shutting down service clients...")
# Close Neo4j driver
neo4j = get_neo4j_client()
try:
await neo4j.close()
logger.info("✓ Neo4j closed")
except Exception as e:
logger.error(f"Error closing Neo4j: {e}")
# Close HTTP clients
clients_to_close = [
("Wiki.js", get_wikijs_client()),
("SearXNG", get_searxng_client()),
("Ollama", get_ollama_client())
]
for name, client in clients_to_close:
try:
await client.close()
logger.info(f"✓ {name} client closed")
except Exception as e:
logger.error(f"Error closing {name} client: {e}")
logger.info("Service clients shutdown complete")
async def check_service_health() -> dict:
"""
Check health of all service clients.
Returns:
Dictionary with health status of each service:
{
"neo4j": bool,
"qdrant": bool,
"wikijs": bool,
"searxng": bool,
"ollama": bool
}
Usage:
>>> health = await check_service_health()
>>> health["neo4j"]
True
"""
health = {}
# Neo4j
try:
neo4j = get_neo4j_client()
# Simple query to check connectivity
await neo4j.execute_query("RETURN 1 as test", {})
health["neo4j"] = True
except Exception as e:
logger.error(f"Neo4j health check failed: {e}")
health["neo4j"] = False
# Qdrant
try:
qdrant = get_qdrant_client()
# Check if we can list collections
collections = qdrant.client.get_collections()
health["qdrant"] = True
except Exception as e:
logger.error(f"Qdrant health check failed: {e}")
health["qdrant"] = False
# Wiki.js
try:
wikijs = get_wikijs_client()
# Try a simple query (list pages with limit 1)
await wikijs.list_pages(limit=1)
health["wikijs"] = True
except Exception as e:
logger.error(f"Wiki.js health check failed: {e}")
health["wikijs"] = False
# SearXNG
try:
searxng = get_searxng_client()
# Try a simple search
await searxng.search_general("test", limit=1)
health["searxng"] = True
except Exception as e:
logger.error(f"SearXNG health check failed: {e}")
health["searxng"] = False
# Ollama
try:
ollama = get_ollama_client()
is_healthy = await ollama.health_check()
health["ollama"] = is_healthy
except Exception as e:
logger.error(f"Ollama health check failed: {e}")
health["ollama"] = False
return health
# Utility: Get default user from settings or multi_tenancy
def get_default_user() -> str:
"""
Get default user for operations.
Returns:
Default user identifier
"""
from src.core.multi_tenancy import DEFAULT_USER
return DEFAULT_USER
@@ -0,0 +1,182 @@
"""
Multi-tenancy helpers for Library Desk.
Provides utilities for user namespace management across:
- Wiki.js (path-based namespaces)
- Neo4j (user-specific labels)
- Qdrant (collection per user)
"""
import re
# Default user for all operations
DEFAULT_USER = "jpmschweitzer"
def sanitize_user_id(user_id: str) -> str:
"""
Sanitize user ID for use in collection names, labels, and paths.
Converts special characters to underscores and ensures alphanumeric safety.
Args:
user_id: Raw user identifier (email, username, etc.)
Returns:
Sanitized user ID safe for use in identifiers
Examples:
>>> sanitize_user_id("john@example.com")
'john_at_example_com'
>>> sanitize_user_id("user.name")
'user_name'
>>> sanitize_user_id("User Name")
'user_name'
"""
sanitized = user_id.lower()
# Convert @ to _at_
sanitized = sanitized.replace("@", "_at_")
# Convert dots to underscores
sanitized = sanitized.replace(".", "_")
# Replace any non-alphanumeric characters with underscores
sanitized = re.sub(r'[^a-z0-9_]', '_', sanitized)
# Remove consecutive underscores
sanitized = re.sub(r'_+', '_', sanitized)
# Remove leading/trailing underscores
sanitized = sanitized.strip('_')
return sanitized
def get_qdrant_collection_name(user_id: str) -> str:
"""
Get Qdrant collection name for user.
Pattern: library_desk_{sanitized_user_id}
Args:
user_id: User identifier
Returns:
Qdrant collection name
Examples:
>>> get_qdrant_collection_name("jpmschweitzer")
'library_desk_jpmschweitzer'
>>> get_qdrant_collection_name("john@example.com")
'library_desk_john_at_example_com'
"""
sanitized = sanitize_user_id(user_id)
return f"library_desk_{sanitized}"
def get_wikijs_namespace(user_id: str) -> str:
"""
Get Wiki.js namespace (path prefix) for user.
Pattern: /users/{sanitized_user_id}
All wiki pages for a user will be under this namespace.
Args:
user_id: User identifier
Returns:
Wiki.js path prefix
Examples:
>>> get_wikijs_namespace("jpmschweitzer")
'/users/jpmschweitzer'
>>> get_wikijs_namespace("john@example.com")
'/users/john_at_example_com'
"""
sanitized = sanitize_user_id(user_id)
return f"/users/{sanitized}"
def get_neo4j_user_label(user_id: str) -> str:
"""
Get Neo4j label for user's documents.
Pattern: User_{Sanitized}_Document
Uses title case for Neo4j label convention.
Args:
user_id: User identifier
Returns:
Neo4j label for user's document nodes
Examples:
>>> get_neo4j_user_label("jpmschweitzer")
'User_Jpmschweitzer_Document'
>>> get_neo4j_user_label("john@example.com")
'User_John_At_Example_Com_Document'
"""
sanitized = sanitize_user_id(user_id)
# Title case each segment for Neo4j label convention
parts = sanitized.split('_')
titled = '_'.join(part.capitalize() for part in parts if part)
return f"User_{titled}_Document"
def validate_user_id(user_id: str) -> bool:
"""
Validate that a user ID is acceptable.
Checks:
- Not empty
- Not too long (max 100 chars)
- Contains some alphanumeric characters
Args:
user_id: User identifier to validate
Returns:
True if valid, False otherwise
Examples:
>>> validate_user_id("jpmschweitzer")
True
>>> validate_user_id("")
False
>>> validate_user_id("a" * 101)
False
"""
if not user_id or len(user_id) > 100:
return False
# Must contain at least one alphanumeric character
if not re.search(r'[a-zA-Z0-9]', user_id):
return False
return True
def is_path_in_user_namespace(path: str, user_id: str) -> bool:
"""
Check if a Wiki.js path belongs to user's namespace.
Args:
path: Wiki.js page path
user_id: User identifier
Returns:
True if path is in user's namespace
Examples:
>>> is_path_in_user_namespace("/users/jpmschweitzer/projects", "jpmschweitzer")
True
>>> is_path_in_user_namespace("/users/other/projects", "jpmschweitzer")
False
>>> is_path_in_user_namespace("/public/docs", "jpmschweitzer")
False
"""
namespace = get_wikijs_namespace(user_id)
return path.startswith(namespace)
@@ -0,0 +1,426 @@
"""
Redis-based job tracking for Library Desk.
Provides background job management with:
- Job creation and status tracking
- Progress updates
- Result storage
- Auto-expiration after 24 hours
- User-scoped job queries
"""
import redis.asyncio as redis
import json
import uuid
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class JobStatus(str, Enum):
"""Job status enumeration."""
QUEUED = "queued"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class JobType(str, Enum):
"""Job type enumeration."""
DOCUMENT_INGESTION = "document_ingestion"
BATCH_INGESTION = "batch_ingestion"
GRAPH_UPDATE = "graph_update"
VECTOR_UPDATE = "vector_update"
EMBEDDINGS_GENERATION = "embeddings_generation"
class JobManager:
"""
Redis-based job tracking manager.
Key structure:
- library:job:{job_id} - Job data (JSON, TTL 24h)
- library:user_jobs:{user} - Set of job IDs for user
- library:active_jobs - Set of active (non-completed) job IDs
Job data format:
{
"job_id": "uuid",
"job_type": "document_ingestion",
"status": "processing",
"user": "jpmschweitzer",
"parameters": {...},
"result": {...},
"error": "error message",
"progress": 50,
"created_at": "2024-01-15T10:30:00Z",
"started_at": "2024-01-15T10:30:05Z",
"completed_at": "2024-01-15T10:35:00Z"
}
"""
def __init__(self, redis_url: str):
"""
Initialize job manager.
Args:
redis_url: Redis connection URL (e.g., "redis://redis-shared:6379/4")
"""
self.redis_url = redis_url
self._redis: Optional[redis.Redis] = None
self.ttl = 86400 # 24 hours
logger.info(f"Initialized JobManager: {redis_url}")
async def connect(self):
"""Initialize Redis connection."""
if not self._redis:
self._redis = await redis.from_url(
self.redis_url,
encoding="utf-8",
decode_responses=True
)
logger.info("Connected to Redis for job tracking")
async def close(self):
"""Close Redis connection."""
if self._redis:
await self._redis.close()
self._redis = None
logger.info("Closed Redis connection")
def _job_key(self, job_id: str) -> str:
"""Get Redis key for job."""
return f"library:job:{job_id}"
def _user_jobs_key(self, user: str) -> str:
"""Get Redis key for user's jobs."""
return f"library:user_jobs:{user}"
def _active_jobs_key(self) -> str:
"""Get Redis key for active jobs."""
return "library:active_jobs"
async def create_job(
self,
job_type: JobType,
user: str,
parameters: Dict[str, Any]
) -> str:
"""
Create new job.
Args:
job_type: Type of job
user: User identifier
parameters: Job parameters
Returns:
Job ID (UUID)
Example:
>>> job_id = await manager.create_job(
... JobType.DOCUMENT_INGESTION,
... "jpmschweitzer",
... {"doc_url": "https://example.com/doc.pdf"}
... )
"""
if not self._redis:
await self.connect()
job_id = str(uuid.uuid4())
job_data = {
"job_id": job_id,
"job_type": job_type.value,
"status": JobStatus.QUEUED.value,
"user": user,
"parameters": parameters,
"result": None,
"error": None,
"progress": 0,
"created_at": datetime.now(timezone.utc).isoformat()
}
# Store job data
await self._redis.setex(
self._job_key(job_id),
self.ttl,
json.dumps(job_data)
)
# Add to user's job set
await self._redis.sadd(self._user_jobs_key(user), job_id)
await self._redis.expire(self._user_jobs_key(user), self.ttl)
# Add to active jobs set
await self._redis.sadd(self._active_jobs_key(), job_id)
logger.info(f"Created job {job_id}: {job_type.value} for user {user}")
return job_id
async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
"""
Get job by ID.
Args:
job_id: Job ID
Returns:
Job data dictionary or None if not found
Example:
>>> job = await manager.get_job(job_id)
>>> job["status"]
'processing'
"""
if not self._redis:
await self.connect()
job_json = await self._redis.get(self._job_key(job_id))
if not job_json:
return None
return json.loads(job_json)
async def update_job_status(
self,
job_id: str,
status: JobStatus,
progress: Optional[int] = None,
result: Optional[Dict[str, Any]] = None,
error: Optional[str] = None
):
"""
Update job status.
Args:
job_id: Job ID
status: New status
progress: Progress percentage (0-100)
result: Result data (for completed jobs)
error: Error message (for failed jobs)
Example:
>>> await manager.update_job_status(
... job_id,
... JobStatus.PROCESSING,
... progress=50
... )
"""
if not self._redis:
await self.connect()
job = await self.get_job(job_id)
if not job:
logger.error(f"Job {job_id} not found")
return
# Update fields
job["status"] = status.value
if progress is not None:
job["progress"] = progress
if result is not None:
job["result"] = result
if error is not None:
job["error"] = error
# Update timestamps
now = datetime.now(timezone.utc).isoformat()
if status == JobStatus.PROCESSING and not job.get("started_at"):
job["started_at"] = now
if status in (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED):
job["completed_at"] = now
# Save updated job
await self._redis.setex(
self._job_key(job_id),
self.ttl,
json.dumps(job)
)
# Remove from active jobs if completed
if status in (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED):
await self._redis.srem(self._active_jobs_key(), job_id)
logger.info(f"Updated job {job_id}: status={status.value}, progress={progress}")
async def get_user_jobs(
self,
user: str,
limit: int = 50
) -> List[Dict[str, Any]]:
"""
Get all jobs for a user.
Args:
user: User identifier
limit: Maximum jobs to return
Returns:
List of job dictionaries (most recent first)
Example:
>>> jobs = await manager.get_user_jobs("jpmschweitzer", limit=10)
>>> len(jobs)
10
"""
if not self._redis:
await self.connect()
# Get job IDs for user
job_ids = await self._redis.smembers(self._user_jobs_key(user))
if not job_ids:
return []
# Fetch job data
jobs = []
for job_id in job_ids:
job = await self.get_job(job_id)
if job:
jobs.append(job)
# Sort by created_at descending
jobs.sort(key=lambda j: j.get("created_at", ""), reverse=True)
return jobs[:limit]
async def get_active_jobs(self, limit: int = 100) -> List[Dict[str, Any]]:
"""
Get all active (non-completed) jobs.
Args:
limit: Maximum jobs to return
Returns:
List of active job dictionaries
Example:
>>> active = await manager.get_active_jobs()
>>> all(j["status"] in ["queued", "processing"] for j in active)
True
"""
if not self._redis:
await self.connect()
job_ids = await self._redis.smembers(self._active_jobs_key())
if not job_ids:
return []
jobs = []
for job_id in job_ids:
job = await self.get_job(job_id)
if job:
jobs.append(job)
return jobs[:limit]
async def cancel_job(self, job_id: str):
"""
Cancel a job.
Args:
job_id: Job ID
Note: This only marks the job as cancelled. The actual job
worker must check status and stop processing.
"""
await self.update_job_status(
job_id,
JobStatus.CANCELLED
)
logger.info(f"Cancelled job {job_id}")
async def delete_job(self, job_id: str, user: str):
"""
Delete a job.
Args:
job_id: Job ID
user: User identifier (for authorization)
Returns:
True if deleted, False if not found or unauthorized
"""
if not self._redis:
await self.connect()
job = await self.get_job(job_id)
if not job:
return False
# Check user authorization
if job.get("user") != user:
logger.warning(f"User {user} attempted to delete job {job_id} owned by {job.get('user')}")
return False
# Delete job data
await self._redis.delete(self._job_key(job_id))
# Remove from user's jobs
await self._redis.srem(self._user_jobs_key(user), job_id)
# Remove from active jobs
await self._redis.srem(self._active_jobs_key(), job_id)
logger.info(f"Deleted job {job_id}")
return True
async def cleanup_expired_jobs(self):
"""
Clean up expired jobs from sets.
Redis will auto-expire job data, but set memberships need manual cleanup.
Should be called periodically (e.g., hourly).
"""
if not self._redis:
await self.connect()
# Cleanup active jobs set
job_ids = await self._redis.smembers(self._active_jobs_key())
for job_id in job_ids:
exists = await self._redis.exists(self._job_key(job_id))
if not exists:
await self._redis.srem(self._active_jobs_key(), job_id)
logger.info("Cleaned up expired jobs")
async def get_job_stats(self, user: Optional[str] = None) -> Dict[str, int]:
"""
Get job statistics.
Args:
user: Optional user filter
Returns:
Statistics dictionary:
{
"total": 100,
"queued": 5,
"processing": 10,
"completed": 80,
"failed": 5
}
"""
if user:
jobs = await self.get_user_jobs(user, limit=1000)
else:
jobs = await self.get_active_jobs(limit=1000)
stats = {
"total": len(jobs),
"queued": 0,
"processing": 0,
"completed": 0,
"failed": 0,
"cancelled": 0
}
for job in jobs:
status = job.get("status", "")
if status in stats:
stats[status] += 1
return stats
+186 -9
View File
@@ -69,9 +69,7 @@ class HealthResponse(BaseModel):
status: str
app_name: str
version: str
neo4j: str
qdrant: str
wiki: str
services: Dict[str, Any]
class StatsResponse(BaseModel):
@@ -98,13 +96,42 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse:
Health check endpoint.
Returns status of all connected services.
"""
from src.core.dependencies import check_service_health
# Check service connectivity
service_health = await check_service_health()
# Overall status is healthy if at least Neo4j and Qdrant are up
all_healthy = service_health.get("neo4j", False) and service_health.get("qdrant", False)
overall_status = "healthy" if all_healthy else "degraded"
return HealthResponse(
status="healthy",
status=overall_status,
app_name=settings.app_name,
version=settings.app_version,
neo4j=settings.neo4j_uri,
qdrant=settings.qdrant_url,
wiki=settings.wikijs_url
services={
"neo4j": {
"url": settings.neo4j_uri,
"healthy": service_health.get("neo4j", False)
},
"qdrant": {
"url": settings.qdrant_url,
"healthy": service_health.get("qdrant", False)
},
"wikijs": {
"url": settings.wikijs_url,
"healthy": service_health.get("wikijs", False)
},
"searxng": {
"url": settings.searxng_url,
"healthy": service_health.get("searxng", False)
},
"ollama": {
"url": settings.ollama_url,
"model": settings.ollama_model,
"healthy": service_health.get("ollama", False)
}
}
)
@@ -128,6 +155,114 @@ async def stats(
)
# Ingestion endpoints (for Scheduler integration)
@app.post("/ingest/document", tags=["Ingestion"])
async def ingest_document(
document: Dict[str, Any],
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Ingest a single document for indexing.
Used by The Scheduler to add mirrored documentation to the knowledge base.
Expected fields:
- source: str (e.g., "github", "gitea")
- repository: str (e.g., "anthropic-cookbook")
- path: str (file path)
- content: str (document content)
- metadata: dict (commit, author, tags, etc.)
TODO: Implement document ingestion pipeline:
1. Chunk content
2. Generate embeddings (Ollama)
3. Extract entities (NLP)
4. Index in Qdrant
5. Create graph nodes/relationships in Neo4j
"""
return {
"message": "Document ingestion not yet implemented",
"document_id": f"doc_{document.get('path', 'unknown')}",
"status": "stub"
}
@app.post("/ingest/batch", tags=["Ingestion"])
async def batch_ingest(
batch: Dict[str, Any],
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Ingest multiple documents in a batch.
More efficient than individual ingestion for large syncs.
TODO: Implement batch processing with task queue
"""
document_count = len(batch.get("documents", []))
return {
"message": "Batch ingestion not yet implemented",
"batch_id": "batch_stub",
"total_documents": document_count,
"status": "stub"
}
@app.post("/ingest/check-updates", tags=["Ingestion"])
async def check_updates(
documents: Dict[str, Any],
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Check which documents need updating based on content hashes.
Used by Scheduler to determine what changed since last sync.
TODO: Implement update detection:
1. Query existing documents by path
2. Compare content hashes
3. Return list of updates needed
"""
return {
"message": "Update checking not yet implemented",
"updates_needed": [],
"up_to_date": [],
"new_documents": []
}
@app.get("/ingest/status/{document_id}", tags=["Ingestion"])
async def get_ingestion_status(
document_id: str,
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Get processing status for a document.
TODO: Implement status tracking
"""
return {
"message": "Status tracking not yet implemented",
"document_id": document_id,
"status": "unknown"
}
@app.get("/ingest/repo-status/{repository}", tags=["Ingestion"])
async def get_repo_status(
repository: str,
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Get indexing status for an entire repository.
TODO: Implement repository-level statistics
"""
return {
"message": "Repository status not yet implemented",
"repository": repository,
"total_documents": 0,
"indexed_documents": 0
}
# Query endpoints (stubs for future implementation)
@app.post("/query/hybrid", tags=["Query"])
async def hybrid_query(
@@ -180,20 +315,62 @@ async def graph_query(
}
# Deduplication endpoints
@app.post("/deduplicate/check", tags=["Deduplication"])
async def check_duplicates(
request: Dict[str, Any],
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Check for duplicate or highly similar documents.
Uses vector similarity and graph analysis.
Expected fields:
- document_id: str
- similarity_threshold: float (default 0.85)
TODO: Implement deduplication:
1. Get document embedding from Qdrant
2. Find similar vectors above threshold
3. Check graph relationships
4. Return candidates with similarity scores
"""
document_id = request.get("document_id")
threshold = request.get("similarity_threshold", 0.85)
return {
"message": "Deduplication not yet implemented",
"document_id": document_id,
"threshold": threshold,
"duplicates": [],
"suggestions": None
}
# Application lifecycle
@app.on_event("startup")
async def startup_event():
"""Initialize connections and resources on startup."""
from src.core.dependencies import startup_clients
settings = get_settings()
logger.info(f"Starting {settings.app_name} v{settings.app_version}")
logger.info(f"Neo4j: {settings.neo4j_uri}")
logger.info(f"Qdrant: {settings.qdrant_url}")
logger.info(f"Wiki.js: {settings.wikijs_url}")
# TODO: Initialize database connections
logger.info(f"SearXNG: {settings.searxng_url}")
logger.info(f"Ollama: {settings.ollama_url}")
# Initialize all service clients
await startup_clients()
@app.on_event("shutdown")
async def shutdown_event():
"""Clean up resources on shutdown."""
from src.core.dependencies import shutdown_clients
logger.info("Shutting down Library Desk API")
# TODO: Close database connections
# Close all service clients
await shutdown_clients()