Coverage for src/qdrant_loader/core/project_manager.py: 78%
174 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""
2Project Manager for multi-project support.
4This module provides the core project management functionality including:
5- Project discovery from configuration
6- Project validation and metadata management
7- Project context injection and propagation
8- Project lifecycle management
9"""
11import hashlib
12from datetime import UTC, datetime
13from inspect import isawaitable
15from sqlalchemy import select
16from sqlalchemy.exc import IntegrityError
17from sqlalchemy.ext.asyncio import AsyncSession
19from qdrant_loader.config.models import ProjectConfig, ProjectsConfig
20from qdrant_loader.core.state.models import Project, ProjectSource
21from qdrant_loader.utils.logging import LoggingConfig
23logger = LoggingConfig.get_logger(__name__)
26class ProjectContext:
27 """Context information for a specific project."""
29 def __init__(
30 self,
31 project_id: str,
32 display_name: str,
33 description: str | None = None,
34 collection_name: str | None = None,
35 config: ProjectConfig | None = None,
36 ):
37 self.project_id = project_id
38 self.display_name = display_name
39 self.description = description
40 self.collection_name = collection_name
41 self.config = config
42 self.created_at = datetime.now(UTC)
44 def to_metadata(self) -> dict[str, str]:
45 """Convert project context to metadata dictionary for document injection."""
46 metadata = {
47 "project_id": self.project_id,
48 "project_name": self.display_name,
49 }
50 if self.description:
51 metadata["project_description"] = self.description
52 if self.collection_name:
53 metadata["collection_name"] = self.collection_name
54 return metadata
56 def __repr__(self) -> str:
57 return f"ProjectContext(id='{self.project_id}', name='{self.display_name}')"
60class ProjectManager:
61 """Manages projects for multi-project support."""
63 def __init__(self, projects_config: ProjectsConfig, global_collection_name: str):
64 """Initialize the project manager with configuration."""
65 self.projects_config = projects_config
66 self.global_collection_name = global_collection_name
67 self.logger = LoggingConfig.get_logger(__name__)
68 self._project_contexts: dict[str, ProjectContext] = {}
69 self._initialized = False
71 async def initialize(self, session: AsyncSession) -> None:
72 """Initialize the project manager and discover projects."""
73 if self._initialized:
74 return
76 self.logger.info("Initializing Project Manager")
78 # Discover and validate projects from configuration
79 await self._discover_projects(session)
81 self._initialized = True
82 self.logger.info(
83 f"Project Manager initialized with {len(self._project_contexts)} projects"
84 )
86 async def _discover_projects(self, session: AsyncSession) -> None:
87 """Discover projects from configuration and create project contexts."""
88 self.logger.debug(
89 "Discovering projects from configuration",
90 project_count=len(self.projects_config.projects),
91 )
93 for project_id, project_config in self.projects_config.projects.items():
95 # Validate project configuration
96 await self._validate_project_config(project_id, project_config)
98 # Determine collection name using the project's method
99 collection_name = project_config.get_effective_collection_name(
100 self.global_collection_name
101 )
103 # Create project context
104 context = ProjectContext(
105 project_id=project_id,
106 display_name=project_config.display_name,
107 description=project_config.description,
108 collection_name=collection_name,
109 config=project_config,
110 )
112 self._project_contexts[project_id] = context
114 # Ensure project exists in database
115 await self._ensure_project_in_database(session, context, project_config)
117 self.logger.info(
118 f"Discovered project: {project_id} ({project_config.display_name})"
119 )
121 async def _validate_project_config(
122 self, project_id: str, config: ProjectConfig
123 ) -> None:
124 """Validate a project configuration."""
125 self.logger.debug(f"Validating project configuration for: {project_id}")
127 # Check required fields
128 if not config.display_name:
129 raise ValueError(f"Project '{project_id}' missing required display_name")
131 # Validate sources exist - check if any source type has configurations
132 has_sources = any(
133 [
134 bool(config.sources.git),
135 bool(config.sources.confluence),
136 bool(config.sources.jira),
137 bool(config.sources.localfile),
138 bool(config.sources.publicdocs),
139 ]
140 )
142 if not has_sources:
143 self.logger.warning(f"Project '{project_id}' has no configured sources")
145 # Additional validation can be added here
146 self.logger.debug(f"Project configuration valid for: {project_id}")
148 async def _ensure_project_in_database(
149 self, session: AsyncSession, context: ProjectContext, config: ProjectConfig
150 ) -> None:
151 """Ensure project exists in database with current configuration."""
152 self.logger.debug(f"Ensuring project exists in database: {context.project_id}")
154 # Check if project exists
155 result = await session.execute(select(Project).filter_by(id=context.project_id))
156 project = result.scalar_one_or_none()
158 # Calculate configuration hash for change detection
159 config_hash = self._calculate_config_hash(config)
161 now = datetime.now(UTC)
163 if project is not None:
164 # Update existing project if configuration changed
165 current_config_hash = getattr(project, "config_hash", None)
166 if current_config_hash != config_hash:
167 self.logger.info(
168 f"Updating project configuration: {context.project_id}"
169 )
170 # Use setattr for SQLAlchemy model attribute assignment
171 project.display_name = context.display_name # type: ignore
172 project.description = context.description # type: ignore
173 project.collection_name = context.collection_name # type: ignore
174 project.config_hash = config_hash # type: ignore
175 project.updated_at = now # type: ignore
176 else:
177 # Create new project
178 self.logger.info(f"Creating new project: {context.project_id}")
179 project = Project(
180 id=context.project_id,
181 display_name=context.display_name,
182 description=context.description,
183 collection_name=context.collection_name,
184 config_hash=config_hash,
185 created_at=now,
186 updated_at=now,
187 )
188 # SQLAlchemy AsyncSession.add is sync; tests may mock it as async; handle both
189 try:
190 result = session.add(project)
191 if isawaitable(result): # type: ignore[arg-type]
192 await result # pragma: no cover - only for certain mocks
193 except Exception:
194 # Best-effort add; proceed to commit
195 pass
197 try:
198 # Update project sources
199 await self._update_project_sources(session, context.project_id, config)
200 await session.commit()
201 except IntegrityError as e:
202 # The unique-constraint error can be triggered by autoflush during
203 # session.execute() inside _update_project_sources, before commit.
204 await session.rollback()
205 err = str(e)
206 if "projects.collection_name" in err or "uix_project_collection" in err:
207 raise ValueError(
208 "State DB schema is outdated and still enforces unique projects.collection_name. "
209 "Reset state DB once and retry: `qdrant-loader init --workspace . --force`."
210 ) from e
211 raise
212 except Exception:
213 await session.rollback()
214 raise
216 async def _update_project_sources(
217 self, session: AsyncSession, project_id: str, config: ProjectConfig
218 ) -> None:
219 """Update project sources in database."""
220 self.logger.debug(f"Updating project sources for: {project_id}")
222 # Get existing sources
223 result = await session.execute(
224 select(ProjectSource).filter_by(project_id=project_id)
225 )
226 existing_sources_list = result.scalars().all()
227 existing_sources = {
228 (source.source_type, source.source_name): source
229 for source in existing_sources_list
230 }
232 # Track current sources from configuration
233 current_sources = set()
234 now = datetime.now(UTC)
236 # Process each source type from SourcesConfig
237 source_types = {
238 "git": config.sources.git,
239 "confluence": config.sources.confluence,
240 "jira": config.sources.jira,
241 "localfile": config.sources.localfile,
242 "publicdocs": config.sources.publicdocs,
243 }
245 for source_type, sources in source_types.items():
246 if not sources:
247 continue
249 for source_name, source_config in sources.items():
250 current_sources.add((source_type, source_name))
252 # Calculate source configuration hash
253 source_config_hash = self._calculate_source_config_hash(source_config)
255 source_key = (source_type, source_name)
256 if source_key in existing_sources:
257 # Update existing source if configuration changed
258 source = existing_sources[source_key]
259 current_source_config_hash = getattr(source, "config_hash", None)
260 if current_source_config_hash != source_config_hash:
261 self.logger.debug(
262 f"Updating source configuration: {source_type}:{source_name}"
263 )
264 source.config_hash = source_config_hash # type: ignore
265 source.updated_at = now # type: ignore
267 # The source config changed (e.g. JQL/project_key/path
268 # filters), so any saved checkpoint cursor was computed
269 # against the old query and is no longer meaningful.
270 # Drop it so the next run starts fresh instead of
271 # silently resuming mid-page against a different query.
272 if current_source_config_hash is not None:
273 from qdrant_loader.core.state.checkpoint_manager import (
274 CheckpointManager,
275 )
277 await CheckpointManager(session).clear_checkpoint(
278 project_id, source_type, source_name
279 )
280 else:
281 # Create new source
282 self.logger.debug(
283 f"Creating new source: {source_type}:{source_name}"
284 )
285 source = ProjectSource(
286 project_id=project_id,
287 source_type=source_type,
288 source_name=source_name,
289 config_hash=source_config_hash,
290 created_at=now,
291 updated_at=now,
292 )
293 try:
294 result = session.add(source)
295 if isawaitable(result): # type: ignore[arg-type]
296 await result # pragma: no cover - only for certain mocks
297 except Exception:
298 pass
300 # Remove sources that are no longer in configuration
301 for source_key, source in existing_sources.items():
302 if source_key not in current_sources:
303 source_type, source_name = source_key
304 self.logger.info(
305 f"Removing obsolete source: {source_type}:{source_name}"
306 )
307 await session.delete(source)
309 def _calculate_config_hash(self, config: ProjectConfig) -> str:
310 """Calculate hash of project configuration for change detection."""
311 # Create a stable representation of the configuration
312 config_data = {
313 "display_name": config.display_name,
314 "description": config.description,
315 "sources": {
316 "git": {
317 name: self._source_config_to_dict(cfg)
318 for name, cfg in config.sources.git.items()
319 },
320 "confluence": {
321 name: self._source_config_to_dict(cfg)
322 for name, cfg in config.sources.confluence.items()
323 },
324 "jira": {
325 name: self._source_config_to_dict(cfg)
326 for name, cfg in config.sources.jira.items()
327 },
328 "localfile": {
329 name: self._source_config_to_dict(cfg)
330 for name, cfg in config.sources.localfile.items()
331 },
332 "publicdocs": {
333 name: self._source_config_to_dict(cfg)
334 for name, cfg in config.sources.publicdocs.items()
335 },
336 },
337 }
339 # Convert to stable string representation and hash
340 config_str = str(sorted(config_data.items()))
341 return hashlib.sha256(config_str.encode()).hexdigest()[:16]
343 def _calculate_source_config_hash(self, source_config) -> str:
344 """Calculate hash of source configuration for change detection."""
345 config_dict = self._source_config_to_dict(source_config)
346 config_str = str(sorted(config_dict.items()))
347 return hashlib.sha256(config_str.encode()).hexdigest()[:16]
349 def _source_config_to_dict(self, source_config) -> dict:
350 """Convert source configuration to dictionary for hashing."""
351 if hasattr(source_config, "model_dump"):
352 # Pydantic model
353 return source_config.model_dump()
354 elif hasattr(source_config, "__dict__"):
355 # Regular object
356 return {
357 k: v for k, v in source_config.__dict__.items() if not k.startswith("_")
358 }
359 else:
360 # Fallback to string representation
361 return {"config": str(source_config)}
363 def get_project_context(self, project_id: str) -> ProjectContext | None:
364 """Get project context by ID."""
365 return self._project_contexts.get(project_id)
367 def get_all_project_contexts(self) -> dict[str, ProjectContext]:
368 """Get all project contexts."""
369 return self._project_contexts.copy()
371 def list_project_ids(self) -> list[str]:
372 """Get list of all project IDs."""
373 return list(self._project_contexts.keys())
375 def get_project_collection_name(self, project_id: str) -> str | None:
376 """Get the collection name for a specific project."""
377 context = self._project_contexts.get(project_id)
378 return context.collection_name if context else None
380 def inject_project_metadata(
381 self, project_id: str, metadata: dict[str, str]
382 ) -> dict[str, str]:
383 """Inject project metadata into document metadata."""
384 context = self._project_contexts.get(project_id)
385 if not context:
386 self.logger.warning(f"Project context not found for ID: {project_id}")
387 return metadata
389 # Create new metadata dict with project information
390 enhanced_metadata = metadata.copy()
391 enhanced_metadata.update(context.to_metadata())
393 return enhanced_metadata
395 def validate_project_exists(self, project_id: str) -> bool:
396 """Validate that a project exists."""
397 return project_id in self._project_contexts
399 async def get_project_stats(
400 self, session: AsyncSession, project_id: str
401 ) -> dict | None:
402 """Get statistics for a specific project."""
403 if not self.validate_project_exists(project_id):
404 return None
406 context = self._project_contexts[project_id]
408 # Get project from database with related data
409 result = await session.execute(select(Project).filter_by(id=project_id))
410 project = result.scalar_one_or_none()
412 if not project:
413 return None
415 # Calculate statistics
416 stats = {
417 "project_id": project_id,
418 "display_name": context.display_name,
419 "description": context.description,
420 "collection_name": context.collection_name,
421 "created_at": project.created_at,
422 "updated_at": project.updated_at,
423 "source_count": len(project.sources),
424 "document_count": len(project.document_states),
425 "ingestion_count": len(project.ingestion_histories),
426 }
428 return stats
430 def __repr__(self) -> str:
431 return f"ProjectManager(projects={len(self._project_contexts)})"