Coverage for src/qdrant_loader/connectors/jira/connector.py: 77%
234 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Jira connector implementation."""
3import asyncio
4import warnings
5from abc import abstractmethod
6from collections.abc import AsyncGenerator, AsyncIterator
7from datetime import datetime
8from urllib.parse import urlparse # noqa: F401 - may be used in URL handling
10import requests
11from requests.auth import HTTPBasicAuth # noqa: F401 - compatibility
13from qdrant_loader.config.types import SourceType
14from qdrant_loader.connectors.base import BaseConnector, ConnectorConfigurationError
15from qdrant_loader.connectors.jira.auth import (
16 auto_detect_deployment_type as _auto_detect_type,
17)
18from qdrant_loader.connectors.jira.auth import setup_authentication as _setup_auth
19from qdrant_loader.connectors.jira.config import (
20 JiraDeploymentType,
21 JiraExtraField,
22 JiraProjectConfig,
23)
24from qdrant_loader.connectors.jira.mappers import (
25 parse_attachment as _parse_attachment_helper,
26)
27from qdrant_loader.connectors.jira.mappers import parse_comment as _parse_comment_helper
28from qdrant_loader.connectors.jira.mappers import parse_issue as _parse_issue_helper
29from qdrant_loader.connectors.jira.mappers import parse_user as _parse_user_helper
30from qdrant_loader.connectors.jira.models import (
31 JiraAttachment,
32 JiraComment,
33 JiraIssue,
34 JiraUser,
35)
36from qdrant_loader.connectors.shared.attachments import AttachmentReader
37from qdrant_loader.connectors.shared.attachments.metadata import (
38 jira_attachment_to_metadata,
39)
40from qdrant_loader.connectors.shared.http import (
41 RateLimiter,
42)
43from qdrant_loader.connectors.shared.http import (
44 request_with_policy as _http_request_with_policy,
45)
46from qdrant_loader.core.attachment_downloader import (
47 AttachmentDownloader,
48 AttachmentMetadata,
49)
50from qdrant_loader.core.document import Document
51from qdrant_loader.core.file_conversion import (
52 FileConversionConfig,
53 FileConverter,
54 FileDetector,
55)
56from qdrant_loader.utils.logging import LoggingConfig
58logger = LoggingConfig.get_logger(__name__)
61class BaseJiraConnector(BaseConnector):
62 """Base class for all Jira connectors."""
64 def __init__(self, config: JiraProjectConfig, checkpoint_cursor: str | None = None):
65 """Initialize the Jira connector.
67 Args:
68 config: The Jira configuration.
69 checkpoint_cursor: Optional pagination cursor to resume from (WS-2 feature).
71 Raises:
72 ValueError: If required authentication parameters are not set.
73 """
74 super().__init__(config)
75 self.config = config
76 self.base_url = str(config.base_url).rstrip("/")
78 # Initialize session
79 self.session = requests.Session()
81 # Set up authentication based on deployment type
82 self._setup_authentication()
84 self._last_sync: datetime | None = None
85 self._rate_limiter = RateLimiter.per_minute(self.config.requests_per_minute)
86 self._initialized = False
88 # Checkpoint support (WS-2 feature)
89 self._checkpoint_cursor = checkpoint_cursor
91 # Initialize file conversion components if enabled
92 self.file_converter: FileConverter | None = None
93 self.file_detector: FileDetector | None = None
94 self.attachment_reader: AttachmentReader | None = None
96 if config.enable_file_conversion:
97 self.file_detector = FileDetector()
98 # FileConverter will be initialized when file_conversion_config is set
100 if config.download_attachments:
101 self.attachment_reader = AttachmentReader(
102 session=self.session,
103 downloader=AttachmentDownloader(session=self.session),
104 )
106 def _setup_authentication(self):
107 """Set up authentication based on deployment type."""
108 _setup_auth(self.session, self.config)
110 def _auto_detect_deployment_type(self) -> JiraDeploymentType:
111 """Auto-detect the Jira deployment type based on the base URL.
113 Returns:
114 JiraDeploymentType: Detected deployment type
115 """
116 return _auto_detect_type(str(self.base_url))
118 def set_file_conversion_config(self, config: FileConversionConfig) -> None:
119 """Set the file conversion configuration.
121 Args:
122 config: File conversion configuration
123 """
124 if self.config.enable_file_conversion and self.file_detector:
125 self.file_converter = FileConverter(config)
126 if self.config.download_attachments:
127 # Clean up any existing attachment reader to avoid resource leaks
128 old_reader = self.attachment_reader
129 if old_reader is not None:
130 try:
131 close_callable = None
132 if hasattr(old_reader, "aclose"):
133 close_callable = old_reader.aclose
134 elif hasattr(old_reader, "close"):
135 close_callable = old_reader.close
136 elif hasattr(old_reader, "cleanup"):
137 close_callable = old_reader.cleanup
139 if close_callable is not None:
140 result = close_callable()
141 if asyncio.iscoroutine(result):
142 try:
143 # Try to schedule/await coroutine cleanup safely
144 try:
145 loop = asyncio.get_running_loop()
146 except RuntimeError:
147 loop = None
148 if loop and not loop.is_closed():
149 loop.create_task(result)
150 else:
151 asyncio.run(result)
152 except Exception:
153 # Ignore cleanup errors to not block reconfiguration
154 pass
155 except Exception:
156 # Ignore cleanup errors to avoid masking the config update
157 pass
159 # Drop reference before creating a new reader
160 self.attachment_reader = None
162 # Reinitialize reader with new downloader config
163 self.attachment_reader = AttachmentReader(
164 session=self.session,
165 downloader=AttachmentDownloader(
166 session=self.session,
167 file_conversion_config=config,
168 enable_file_conversion=True,
169 max_attachment_size=config.max_file_size,
170 ),
171 )
173 async def _validate_connection(self) -> None:
174 """Validate connectivity, auth, and project access before use.
176 Raises:
177 ConnectorConfigurationError: for invalid URL, bad credentials,
178 missing permissions, or unknown project key.
179 """
180 # ── Step 1: reachability + authentication (/myself endpoint) ──────────
181 try:
182 await self._make_request("GET", "myself")
183 except requests.exceptions.Timeout as exc:
184 raise ConnectorConfigurationError(
185 f"Connection to Jira at '{self.base_url}' timed out. "
186 "Verify network connectivity and try again."
187 ) from exc
188 except requests.exceptions.ConnectionError as exc:
189 raise ConnectorConfigurationError(
190 f"Cannot connect to Jira at '{self.base_url}'. "
191 "Verify that base_url is correct and the server is reachable."
192 ) from exc
193 except requests.exceptions.HTTPError as exc:
194 status = exc.response.status_code if exc.response is not None else None
195 if status == 401:
196 raise ConnectorConfigurationError(
197 f"Authentication failed for Jira at '{self.base_url}' (HTTP 401). "
198 "Check that token and email are valid."
199 ) from exc
200 if status == 403:
201 raise ConnectorConfigurationError(
202 f"Access denied to Jira at '{self.base_url}' (HTTP 403). "
203 "The account does not have sufficient permissions."
204 ) from exc
205 raise ConnectorConfigurationError(
206 f"Validation request to Jira at '{self.base_url}' failed "
207 f"with HTTP {status}: {exc}"
208 ) from exc
209 except requests.exceptions.RequestException as exc:
210 raise ConnectorConfigurationError(
211 f"Validation request to Jira at '{self.base_url}' failed: {exc}"
212 ) from exc
214 # ── Step 2: project key exists and is accessible ───────────────────────
215 try:
216 await self._make_request("GET", f"project/{self.config.project_key}")
217 except requests.exceptions.Timeout as exc:
218 raise ConnectorConfigurationError(
219 f"Connection to Jira at '{self.base_url}' timed out while validating "
220 f"project '{self.config.project_key}'."
221 ) from exc
222 except requests.exceptions.ConnectionError as exc:
223 raise ConnectorConfigurationError(
224 f"Connection to Jira at '{self.base_url}' was lost while validating "
225 f"project '{self.config.project_key}' (between validation steps). "
226 "Verify network connectivity and Jira availability."
227 ) from exc
228 except requests.exceptions.HTTPError as exc:
229 status = exc.response.status_code if exc.response is not None else None
230 if status == 404:
231 raise ConnectorConfigurationError(
232 f"Project '{self.config.project_key}' not found in Jira (HTTP 404). "
233 "Check that project_key is correct."
234 ) from exc
235 if status == 403:
236 raise ConnectorConfigurationError(
237 f"No permission to access project '{self.config.project_key}' "
238 f"in Jira (HTTP 403)."
239 ) from exc
240 raise ConnectorConfigurationError(
241 f"Validation request for project '{self.config.project_key}' at "
242 f"'{self.base_url}' failed with HTTP {status}: {exc}"
243 ) from exc
244 except requests.exceptions.RequestException as exc:
245 raise ConnectorConfigurationError(
246 f"Validation request for project '{self.config.project_key}' at "
247 f"'{self.base_url}' failed: {exc}"
248 ) from exc
250 @staticmethod
251 def _escape_jql_literal(value: str) -> str:
252 """Escape special characters in JQL string literals.
254 Escapes backslashes and double quotes to prevent JQL injection
255 and query breaking when config values contain these characters.
257 Args:
258 value: The string value to escape
260 Returns:
261 str: The escaped string safe for inclusion in JQL quoted literals
262 """
263 # Replace backslash first to avoid double-escaping
264 value = value.replace("\\", "\\\\")
265 # Then escape double quotes
266 value = value.replace('"', '\\"')
267 return value
269 def _build_jql_filter(self, updated_after: datetime | None = None) -> str:
270 """Build JQL filter query with project key, issue types, and statuses.
272 Args:
273 updated_after: Optional datetime to filter issues updated after this time
275 Returns:
276 str: JQL filter query
277 """
278 escaped_project_key = self._escape_jql_literal(self.config.project_key)
279 jql = f'project = "{escaped_project_key}"'
281 # Add issue type filter if configured
282 if self.config.issue_types:
283 escaped_types = [
284 self._escape_jql_literal(t) for t in self.config.issue_types
285 ]
286 types_str = ", ".join(f'"{t}"' for t in escaped_types)
287 jql += f" AND type IN ({types_str})"
288 logger.debug(f"Applied JIRA issue type filter: {self.config.issue_types}")
290 # Add status filter if configured
291 if self.config.include_statuses:
292 escaped_statuses = [
293 self._escape_jql_literal(s) for s in self.config.include_statuses
294 ]
295 statuses_str = ", ".join(f'"{s}"' for s in escaped_statuses)
296 jql += f" AND status IN ({statuses_str})"
297 logger.debug(f"Applied JIRA status filter: {self.config.include_statuses}")
299 # Add updated_after filter if provided
300 if updated_after:
301 jql += f" AND updated >= '{updated_after.strftime('%Y-%m-%d %H:%M')}'"
303 # Pagination (and offset-based checkpoint resume, see
304 # JiraDataCenterConnector.get_issues) relies on a stable row order
305 # across requests; without an explicit ORDER BY, Jira may reorder
306 # results between pages (e.g. new issues created mid-scan), causing
307 # resume to skip or duplicate issues.
308 jql += " ORDER BY key ASC"
310 return jql
312 async def __aenter__(self):
313 """Async context manager entry."""
314 if not self._initialized:
315 await self._validate_connection()
316 self._initialized = True
317 return self
319 async def __aexit__(self, exc_type, exc_val, _exc_tb):
320 """Async context manager exit."""
321 try:
322 self.session.close()
323 finally:
324 self._initialized = False
326 @abstractmethod
327 def _get_api_url(self, endpoint: str) -> str:
328 """Construct the full API URL for an endpoint."""
329 ...
331 async def _make_request(self, method: str, endpoint: str, **kwargs) -> dict:
332 """Make an authenticated request to the Jira API.
334 Args:
335 method: HTTP method
336 endpoint: API endpoint path
337 **kwargs: Additional request parameters
339 Returns:
340 dict: Response data
342 Raises:
343 requests.exceptions.RequestException: If the request fails
344 """
345 url = self._get_api_url(endpoint)
347 if "timeout" not in kwargs:
348 kwargs["timeout"] = 60
350 try:
351 logger.debug(
352 "Making JIRA API request",
353 method=method,
354 endpoint=endpoint,
355 url=url,
356 timeout=kwargs.get("timeout"),
357 )
359 if not self.session.headers.get("Authorization"):
360 kwargs["auth"] = self.session.auth
362 response = await _http_request_with_policy(
363 self.session,
364 method,
365 url,
366 rate_limiter=self._rate_limiter,
367 retries=3,
368 backoff_factor=0.5,
369 status_forcelist=(429, 500, 502, 503, 504),
370 overall_timeout=90.0,
371 **kwargs,
372 )
374 response.raise_for_status()
376 logger.debug(
377 "JIRA API request completed successfully",
378 method=method,
379 endpoint=endpoint,
380 status_code=response.status_code,
381 response_size=(
382 len(response.content) if hasattr(response, "content") else 0
383 ),
384 )
386 return response.json()
388 except TimeoutError:
389 logger.error(
390 "JIRA API request timed out",
391 method=method,
392 url=url,
393 timeout=kwargs.get("timeout"),
394 )
395 raise requests.exceptions.Timeout(
396 f"Request to {url} timed out after {kwargs.get('timeout')} seconds"
397 )
399 except requests.exceptions.RequestException as e:
400 logger.error(
401 "Failed to make request to JIRA API",
402 method=method,
403 url=url,
404 error=str(e),
405 error_type=type(e).__name__,
406 )
407 logger.error(
408 "Request details",
409 deployment_type=self.config.deployment_type,
410 has_auth_header=bool(self.session.headers.get("Authorization")),
411 has_session_auth=bool(self.session.auth),
412 )
413 raise
415 @abstractmethod
416 async def get_issues(
417 self, updated_after: datetime | None = None
418 ) -> AsyncGenerator[JiraIssue, None]:
419 """Get all issues from Jira."""
420 ...
422 def _parse_issue(
423 self, raw_issue: dict, extra_fields: list[JiraExtraField] | None = None
424 ) -> JiraIssue:
425 """Parse a raw issue from the Jira response into a JiraIssue object."""
426 return _parse_issue_helper(raw_issue, extra_fields)
428 def _parse_user(
429 self, raw_user: dict | None, required: bool = False
430 ) -> JiraUser | None:
431 """Parse a raw user from the Jira response into a JiraUser object."""
432 return _parse_user_helper(raw_user, required)
434 def _parse_attachment(self, raw_attachment: dict) -> JiraAttachment:
435 """Parse a raw attachment from the Jira response into a JiraAttachment object."""
436 return _parse_attachment_helper(raw_attachment)
438 def _parse_comment(self, raw_comment: dict) -> JiraComment:
439 """Parse a raw comment from the Jira response into a JiraComment object."""
440 return _parse_comment_helper(raw_comment)
442 def _get_issue_attachments(self, issue: JiraIssue) -> list[AttachmentMetadata]:
443 """Convert JIRA issue attachments to AttachmentMetadata objects.
445 Args:
446 issue: JIRA issue with attachments
448 Returns:
449 List of attachment metadata objects
450 """
451 if not self.config.download_attachments or not issue.attachments:
452 return []
454 attachment_metadata = [
455 jira_attachment_to_metadata(att, parent_id=issue.id)
456 for att in issue.attachments
457 ]
459 return attachment_metadata
461 async def fetch_by_id(self, entity_id: str) -> Document | None:
462 """Fetch a single Jira issue by key or numeric ID."""
463 try:
464 raw_issue = await self._make_request("GET", f"issue/{entity_id}")
465 issue = self._parse_issue(raw_issue, self.config.extra_fields)
466 documents = await self._issues_to_documents(
467 [issue], include_attachments=False
468 )
469 except Exception as exc:
470 logger.error(
471 "Failed to fetch/parse Jira issue by id",
472 entity_id=entity_id,
473 error=str(exc),
474 )
475 return None
477 return documents[0] if documents else None
479 async def list_entity_ids(self) -> AsyncIterator[str]:
480 """Stream all issue keys in the configured project."""
481 async for issue in self.get_issues():
482 yield issue.key
484 async def _issues_to_documents(
485 self, issues: list[JiraIssue], include_attachments: bool = True
486 ) -> list[Document]:
487 """Convert Jira issues to Document objects.
489 Set ``include_attachments=False`` for single-entity fetches where the
490 parent document must remain available even if attachment materialization
491 fails.
493 .. deprecated:: WS-1
494 Use :meth:`_stream_issues_to_documents` for streaming.
495 """
496 documents: list[Document] = []
497 async for document in self._stream_issues_to_documents(
498 issues, include_attachments=include_attachments
499 ):
500 documents.append(document)
501 return documents
503 async def _stream_issues_to_documents(
504 self, issues: list[JiraIssue], include_attachments: bool = True
505 ) -> AsyncGenerator[Document, None]:
506 """Stream Jira issues as Document objects, including attachments.
508 Yields documents one at a time, with attachments yielded immediately after
509 their parent issue document.
510 """
511 for issue in issues:
512 content_parts = [issue.summary]
513 if issue.description:
514 content_parts.append(issue.description)
516 for comment in issue.comments:
517 content_parts.append(
518 f"\nComment by {comment.author.display_name} on {comment.created.strftime('%Y-%m-%d %H:%M')}:"
519 )
520 content_parts.append(comment.body)
522 content = "\n\n".join(content_parts)
523 metadata = {
524 "project": self.config.project_key,
525 "issue_type": issue.issue_type,
526 "status": issue.status,
527 "key": issue.key,
528 "priority": issue.priority,
529 "labels": issue.labels,
530 "reporter": issue.reporter.display_name if issue.reporter else None,
531 "assignee": issue.assignee.display_name if issue.assignee else None,
532 "created": issue.created.isoformat(),
533 "updated": issue.updated.isoformat(),
534 "parent_key": issue.parent_key,
535 "subtasks": issue.subtasks,
536 "linked_issues": issue.linked_issues,
537 "linked_issue_details": [
538 {
539 "key": link.key,
540 "link_type": link.link_type,
541 "direction": link.direction,
542 "relation": link.relation,
543 }
544 for link in issue.linked_issue_details
545 ],
546 "comments": [
547 {
548 "id": comment.id,
549 "body": comment.body,
550 "created": comment.created.isoformat(),
551 "updated": (
552 comment.updated.isoformat() if comment.updated else None
553 ),
554 "author": (
555 comment.author.display_name if comment.author else None
556 ),
557 }
558 for comment in issue.comments
559 ],
560 "attachments": (
561 [
562 {
563 "id": att.id,
564 "filename": att.filename,
565 "size": att.size,
566 "mime_type": att.mime_type,
567 "created": att.created.isoformat(),
568 "author": (att.author.display_name if att.author else None),
569 }
570 for att in issue.attachments
571 ]
572 if issue.attachments
573 else []
574 ),
575 }
576 if self.config.extra_fields:
577 for field in self.config.extra_fields:
578 metadata[field.name] = getattr(issue, field.name)
579 # Propagate checkpoint info into document metadata if present
580 cp = getattr(issue, "ingestion_checkpoint", None)
581 if cp:
582 metadata["__ingestion_checkpoint"] = cp
583 base_url = str(self.config.base_url).rstrip("/")
584 document = Document(
585 id=issue.id,
586 content=content,
587 content_type="text",
588 source=self.config.source,
589 source_type=SourceType.JIRA,
590 created_at=issue.created,
591 url=f"{base_url}/browse/{issue.key}",
592 title=issue.summary,
593 updated_at=issue.updated,
594 is_deleted=False,
595 metadata=metadata,
596 )
597 logger.debug(
598 "Jira document created",
599 document_id=document.id,
600 source_type=document.source_type,
601 source=document.source,
602 title=document.title,
603 )
604 yield document
606 if (
607 include_attachments
608 and self.config.download_attachments
609 and self.attachment_reader
610 ):
611 attachment_metadata = self._get_issue_attachments(issue)
612 if attachment_metadata:
613 logger.info(
614 "Processing attachments for JIRA issue",
615 issue_key=issue.key,
616 attachment_count=len(attachment_metadata),
617 )
619 attachment_documents = (
620 await self.attachment_reader.fetch_and_process(
621 attachment_metadata, document
622 )
623 )
624 for attachment_document in attachment_documents:
625 yield attachment_document
627 logger.debug(
628 "Processed attachments for JIRA issue",
629 issue_key=issue.key,
630 processed_count=len(attachment_documents),
631 )
633 async def stream_documents(
634 self, since: datetime | None = None
635 ) -> AsyncGenerator[Document, None]:
636 """Stream documents from Jira (WS-1 connector contract)."""
637 effective_since = since if since is not None else self.config.updated_after
638 async for issue in self.get_issues(updated_after=effective_since):
639 async for document in self._stream_issues_to_documents([issue]):
640 yield document
642 async def get_documents(self) -> list[Document]:
643 """Fetch and process documents from Jira (DEPRECATED - use stream_documents)."""
644 warnings.warn(
645 "BaseJiraConnector.get_documents is deprecated. Implement stream_documents() "
646 "or use connector.stream_documents() to avoid materializing the full "
647 "document list in memory.",
648 DeprecationWarning,
649 stacklevel=2,
650 )
651 documents = []
652 async for document in self.stream_documents():
653 documents.append(document)
654 return documents