Coverage for src/qdrant_loader/connectors/jira/config.py: 96%
134 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Configuration for Jira connector."""
3import os
4from datetime import datetime, timedelta
5from enum import StrEnum
6from typing import Self
8from pydantic import (
9 BaseModel,
10 ConfigDict,
11 Field,
12 HttpUrl,
13 field_validator,
14 model_validator,
15)
17from qdrant_loader.config.source_config import SourceConfig
20class JiraDeploymentType(StrEnum):
21 """Jira deployment types."""
23 CLOUD = "cloud"
24 DATACENTER = "datacenter"
27class JiraFieldType(StrEnum):
28 SIMPLE = "simple"
29 OBJECT = "object" # extract attribute from single object
30 ARRAY = "array" # plain list
31 ARRAY_OBJECT = "array_object" # extract attribute from list of objects
34RESERVED_NAMES = {
35 "id",
36 "key",
37 "summary",
38 "description",
39 "issue_type",
40 "status",
41 "priority",
42 "project_key",
43 "created",
44 "updated",
45 "reporter",
46 "assignee",
47 "labels",
48 "attachments",
49 "comments",
50 "parent_key",
51 "subtasks",
52 "linked_issues",
53 "linked_issue_details",
54}
57class JiraExtraField(BaseModel):
58 param_name: str = Field(
59 ...,
60 min_length=1,
61 description="The Jira API parameter name (e.g., 'customfield_11406', 'priority')",
62 )
63 name: str = Field(
64 ...,
65 min_length=1,
66 description="Target attribute name on JiraIssue (e.g., 'sah_project')",
67 )
68 field_type: JiraFieldType = Field(
69 default=JiraFieldType.SIMPLE,
70 description=(
71 "Extraction strategy: "
72 "'simple' = direct value, "
73 "'object' = extract attribute from object (requires attr_name), "
74 "'array' = plain list, "
75 "'array_object' = extract attribute from list of objects (requires attr_name)"
76 ),
77 )
78 attr_name: str | None = Field(
79 default=None,
80 description="Attribute to extract from object(s) (e.g., 'name', 'value')",
81 )
83 @field_validator("param_name", "name", "attr_name", mode="before")
84 @classmethod
85 def normalize_strings(cls, v: str | None) -> str | None:
86 if v is None:
87 return None
88 s = v.strip()
89 if s == "":
90 raise ValueError("Field value cannot be empty or whitespace")
91 return s
93 @model_validator(mode="after")
94 def validate_attr_requirement(self) -> "JiraExtraField":
95 if self.field_type in {JiraFieldType.OBJECT, JiraFieldType.ARRAY_OBJECT}:
96 if not self.attr_name:
97 raise ValueError(
98 f"'attr_name' is required for field_type='{self.field_type}'"
99 )
100 elif self.attr_name is not None:
101 raise ValueError(
102 "'attr_name' is only allowed for field_type='object' or 'array_object'"
103 )
104 return self
106 @model_validator(mode="after")
107 def validate_reserved_name(self) -> "JiraExtraField":
108 if self.name in RESERVED_NAMES:
109 raise ValueError(
110 f"'name' cannot be one of reserved attributes: {sorted(RESERVED_NAMES)}"
111 )
112 return self
115class JiraProjectConfig(SourceConfig):
116 """Configuration for a Jira project."""
118 # Authentication
119 token: str | None = Field(
120 default=None, description="Jira API token or Personal Access Token"
121 )
122 email: str | None = Field(
123 default=None, description="Email associated with the API token (Cloud only)"
124 )
125 base_url: HttpUrl = Field(
126 ...,
127 description="Base URL of the Jira instance (e.g., 'https://your-domain.atlassian.net')",
128 )
130 # Project configuration
131 project_key: str = Field(
132 ..., description="Project key to process (e.g., 'PROJ')", min_length=1
133 )
135 # Deployment type
136 deployment_type: JiraDeploymentType = Field(
137 default=JiraDeploymentType.CLOUD,
138 description="Jira deployment type (cloud, datacenter, or server)",
139 )
141 # Rate limiting
142 requests_per_minute: int = Field(
143 default=60, description="Maximum number of requests per minute", ge=1, le=1000
144 )
146 # Pagination
147 page_size: int = Field(
148 default=100,
149 description="Number of items per page for paginated requests",
150 ge=1,
151 le=100,
152 )
154 # Attachment handling
155 download_attachments: bool = Field(
156 default=False, description="Whether to download and process issue attachments"
157 )
159 # Additional configuration
160 issue_types: list[str] = Field(
161 default=[],
162 description="Optional list of issue types to process (e.g., ['Bug', 'Story']). If empty, all types are processed.",
163 )
164 include_statuses: list[str] = Field(
165 default=[],
166 description="Optional list of statuses to include (e.g., ['Open', 'In Progress']). If empty, all statuses are included.",
167 )
169 # Issue filtering
170 updated_after: datetime | None = Field(
171 default=None,
172 description="Only fetch issues updated after this datetime. Supports ISO 8601 format (e.g., '2026-05-04T12:00:00') or relative format (e.g., '-2 days', '-48h', '-1w'). Set to None to fetch all issues.",
173 )
174 extra_fields: list[JiraExtraField] | None = Field(
175 default=None,
176 description="Optional list of extra Jira fields to retrieve with their extraction type.",
177 )
179 model_config = ConfigDict(validate_default=True, arbitrary_types_allowed=True)
181 @field_validator("deployment_type", mode="before")
182 @classmethod
183 def auto_detect_deployment_type(
184 cls, v: str | JiraDeploymentType
185 ) -> JiraDeploymentType:
186 """Auto-detect deployment type if not specified."""
187 if isinstance(v, str):
188 return JiraDeploymentType(v.lower())
189 return v
191 @field_validator("token", mode="after")
192 @classmethod
193 def load_token_from_env(cls, v: str | None) -> str | None:
194 """Load token from environment variable if not provided."""
195 return v or os.getenv("JIRA_TOKEN")
197 @field_validator("email", mode="after")
198 @classmethod
199 def load_email_from_env(cls, v: str | None) -> str | None:
200 """Load email from environment variable if not provided."""
201 return v or os.getenv("JIRA_EMAIL")
203 @model_validator(mode="after")
204 def validate_no_placeholders(self) -> Self:
205 """Fail immediately if any required field still contains an un-substituted ${VAR} placeholder."""
206 import re
208 _placeholder = re.compile(r"\$\{[^}]+\}")
210 fields_to_check: dict[str, str | None] = {
211 "project_key": self.project_key,
212 "base_url": str(self.base_url) if self.base_url else None,
213 "token": self.token,
214 "email": self.email,
215 }
217 bad: list[str] = []
218 for field_name, value in fields_to_check.items():
219 if value and _placeholder.search(value):
220 # Extract the variable name for a helpful hint
221 var = _placeholder.search(value).group(0) # type: ignore[union-attr]
222 bad.append(f" - {field_name}: {var} (env var not set)")
224 if bad:
225 raise ValueError(
226 "Jira source config contains un-substituted environment variables.\n"
227 "Set the following variables in your .env file or shell before running:\n"
228 + "\n".join(bad)
229 )
231 return self
233 @model_validator(mode="after")
234 def validate_auth_config(self) -> Self:
235 """Validate authentication configuration based on deployment type."""
236 if self.deployment_type == JiraDeploymentType.CLOUD:
237 # Cloud requires email and token
238 if not self.email:
239 raise ValueError("Email is required for Jira Cloud deployment")
240 if not self.token:
241 raise ValueError("API token is required for Jira Cloud deployment")
242 else:
243 # Data Center/Server requires Personal Access Token
244 if not self.token:
245 raise ValueError(
246 "Personal Access Token is required for Jira Data Center/Server deployment"
247 )
249 return self
251 @field_validator("issue_types", "include_statuses")
252 @classmethod
253 def validate_list_items(cls, v: list[str]) -> list[str]:
254 """Validate that list items are not empty strings."""
255 if any(not item.strip() for item in v):
256 raise ValueError("List items cannot be empty strings")
257 return [item.strip() for item in v]
259 @field_validator("extra_fields")
260 @classmethod
261 def validate_extra_fields_unique(
262 cls, v: list[JiraExtraField] | None
263 ) -> list[JiraExtraField] | None:
264 """Validate that extra field param_names and names are unique."""
265 if v is None:
266 return v
267 param_names = [f.param_name for f in v if f.param_name is not None]
268 if len(param_names) != len(set(param_names)):
269 raise ValueError("Extra field 'param_name' values must be unique")
270 names = [f.name for f in v if f.name is not None]
271 if len(names) != len(set(names)):
272 raise ValueError("Extra field 'name' values must be unique")
273 return v
275 @field_validator("updated_after", mode="before")
276 @classmethod
277 def parse_updated_after(cls, v: str | datetime | None) -> datetime | None:
278 """Parse updated_after field supporting relative date strings.
280 Supports formats like:
281 - ISO 8601: "2026-05-04T12:00:00"
282 - Relative: "-2 days", "-48h", "-2d", "-1w"
283 - None: fetch all issues
284 """
285 if v is None or isinstance(v, datetime):
286 return v
288 if isinstance(v, str):
289 import re
291 # Try to parse as ISO 8601 datetime first
292 try:
293 return datetime.fromisoformat(v)
294 except ValueError:
295 pass
297 # Parse relative dates like "-2 days", "-48h", "-2d", "-1w"
298 match = re.match(
299 r"^-(\d+)\s*(days?|hours?|h|d|w|weeks?)$", v.strip(), re.IGNORECASE
300 )
301 if match:
302 amount = int(match.group(1))
303 unit = match.group(2).lower()
305 if unit in ("day", "days", "d"):
306 return datetime.now() - timedelta(days=amount)
307 elif unit in ("hour", "hours", "h"):
308 return datetime.now() - timedelta(hours=amount)
309 elif unit in ("week", "weeks", "w"):
310 return datetime.now() - timedelta(weeks=amount)
312 raise ValueError(
313 f"Invalid updated_after format: '{v}'. "
314 "Use ISO 8601 (e.g., '2026-05-04T12:00:00') or relative format (e.g., '-2 days', '-48h', '-1w')"
315 )
317 raise ValueError(f"updated_after must be a datetime or string, got {type(v)}")