services: add missing type annotations

bugwarrior/services/azuredevops.py

@@ -185,8 +185,10 @@ class AzureDevopsService(Service[AzureDevopsIssue]):
     ISSUE_CLASS = AzureDevopsIssue
     CONFIG_SCHEMA = AzureDevopsConfig
 
-    def __init__(self, *args: Any, **kw: Any) -> None:
-        super().__init__(*args, **kw)
+    def __init__(
+        self, config: AzureDevopsConfig, main_config: config.MainSectionConfig
+    ) -> None:
+        super().__init__(config, main_config)
         self.client = AzureDevopsClient(
             pat=self.get_secret('PAT'),
             project=self.config.project,

bugwarrior/services/bitbucket.py

@@ -79,16 +79,18 @@ def get_default_description(self) -> str:
         )
 
 
-class BitbucketService(Service[BitbucketIssue], Client):
+class BitbucketService(Service[BitbucketIssue]):
     API_VERSION = 1.0
     ISSUE_CLASS = BitbucketIssue
     CONFIG_SCHEMA = BitbucketConfig
 
     BASE_API2 = 'https://api.bitbucket.org/2.0'
     BASE_URL = 'https://bitbucket.org/'
 
-    def __init__(self, *args: Any, **kw: Any) -> None:
-        super().__init__(*args, **kw)
+    def __init__(
+        self, config: BitbucketConfig, main_config: config.MainSectionConfig
+    ) -> None:
+        super().__init__(config, main_config)
 
         oauth = (self.config.key, self.get_secret('secret', self.config.key))
         refresh_token = self.main_config.data.get('bitbucket_refresh_token')
@@ -135,7 +137,7 @@ def filter_repos(self, repo_tag: str) -> bool:
 
     def get_data(self, url: str) -> dict[str, Any]:
         """Perform a request to the fully qualified url and return json."""
-        return self.json_response(requests.get(url, **self.requests_kwargs))
+        return Client.json_response(requests.get(url, **self.requests_kwargs))
 
     def get_collection(self, url: str) -> Iterator[Any]:
         """Pages through an object collection from the bitbucket API.

bugwarrior/services/bts.py

@@ -1,12 +1,15 @@
+from collections.abc import Iterable, Iterator
 import logging
 import typing
+from typing import Any
 
 import debianbts
 import pydantic
 from pydantic import model_validator
 import requests
 
 from bugwarrior import config
+from bugwarrior.config import Priority
 from bugwarrior.services import Client, Issue, Service
 
 log = logging.getLogger(__name__)
@@ -30,13 +33,13 @@ class BTSConfig(config.ServiceConfig):
     also_unassigned: config.UnsupportedOption[bool] = False
 
     @model_validator(mode='after')
-    def require_email_or_packages(self):
+    def require_email_or_packages(self) -> "BTSConfig":
         if not self.email and not self.packages:
             raise ValueError('section requires one of:\n    email\n    packages')
         return self
 
     @model_validator(mode='after')
-    def udd_needs_email(self):
+    def udd_needs_email(self) -> "BTSConfig":
         if self.udd and not self.email:
             raise ValueError("no 'email' but UDD search was requested")
         return self
@@ -62,7 +65,7 @@ class BTSIssue(Issue):
     }
     UNIQUE_KEY = (URL,)
 
-    PRIORITY_MAP = {
+    PRIORITY_MAP: dict[str, Priority] = {
         'wishlist': 'L',
         'minor': 'L',
         'normal': 'M',
@@ -72,7 +75,7 @@ class BTSIssue(Issue):
         'critical': 'H',
     }
 
-    def to_taskwarrior(self):
+    def to_taskwarrior(self) -> dict[str, Any]:
         return {
             'priority': self.get_priority(),
             'annotations': self.extra.get('annotations', []),
@@ -85,30 +88,30 @@ def to_taskwarrior(self):
             self.STATUS: self.record['status'],
         }
 
-    def get_default_description(self):
+    def get_default_description(self) -> str:
         return self.build_default_description(
             title=self.record['subject'],
             url=self.record['url'],
             number=self.record['number'],
             cls='issue',
         )
 
-    def get_priority(self):
+    def get_priority(self) -> config.Priority:
         return self.PRIORITY_MAP.get(
             self.record.get('severity', ''), self.config.default_priority
         )
 
 
-class BTSService(Service, Client):
+class BTSService(Service[BTSIssue]):
     API_VERSION = 1.0
     ISSUE_CLASS = BTSIssue
     CONFIG_SCHEMA = BTSConfig
 
     @staticmethod
-    def get_keyring_service(config):
+    def get_keyring_service(config: BTSConfig) -> str:
         return 'bts://'
 
-    def _record_for_bug(self, bug):
+    def _record_for_bug(self, bug: debianbts.Bugreport) -> dict[str, Any]:
         return {
             'number': bug.bug_num,
             'url': 'https://bugs.debian.org/' + str(bug.bug_num),
@@ -120,17 +123,17 @@ def _record_for_bug(self, bug):
             'status': bug.pending,
         }
 
-    def _get_udd_bugs(self):
+    def _get_udd_bugs(self) -> Iterable[dict[str, Any]]:
         request_params = {'format': 'json', 'dmd': 1, 'email1': self.config.email}
         if self.config.udd_ignore_sponsor:
             request_params['nosponsor1'] = "on"
         resp = requests.get(UDD_BUGS_SEARCH, request_params)
-        return self.json_response(resp)
+        return Client.json_response(resp)
 
-    def annotations(self, issue):
+    def annotations(self, issue: dict[str, Any]) -> list[str]:
         return self.build_annotations([], issue['url'])
 
-    def issues(self):
+    def issues(self) -> Iterator[BTSIssue]:
         # Initialise empty list of bug numbers
         collected_bugs = []
 

bugwarrior/services/bz.py

@@ -1,8 +1,9 @@
+from collections.abc import Iterator
 import datetime
 import logging
 import time
 import typing
-from typing import Annotated
+from typing import Annotated, Any
 import urllib.parse
 import xmlrpc.client
 
@@ -17,7 +18,7 @@
 log = logging.getLogger(__name__)
 
 
-def validate_url(value: str):
+def validate_url(value: str) -> str:
     if not urllib.parse.urlparse(value).scheme:
         value = f'https://{value}'
         log.warning(
@@ -87,7 +88,7 @@ class BugzillaIssue(Issue):
         'urgent': 'H',
     }
 
-    def to_taskwarrior(self):
+    def to_taskwarrior(self) -> dict[str, Any]:
         task = {
             'project': self.record['component'],
             'priority': self.get_priority(),
@@ -107,7 +108,7 @@ def to_taskwarrior(self):
 
         return task
 
-    def get_default_description(self):
+    def get_default_description(self) -> str:
         return self.build_default_description(
             title=self.record['summary'],
             url=self.extra['url'],
@@ -116,7 +117,7 @@ def get_default_description(self):
         )
 
 
-class BugzillaService(Service):
+class BugzillaService(Service[BugzillaIssue]):
     API_VERSION = 1.0
     ISSUE_CLASS = BugzillaIssue
     CONFIG_SCHEMA = BugzillaConfig
@@ -133,8 +134,10 @@ class BugzillaService(Service):
         'assigned_to',
     ]
 
-    def __init__(self, *args, **kw):
-        super().__init__(*args, **kw)
+    def __init__(
+        self, config: BugzillaConfig, main_config: config.MainSectionConfig
+    ) -> None:
+        super().__init__(config, main_config)
         log.debug(" filtering on statuses: %r", self.config.open_statuses)
 
         force_rest_kwargs = {}
@@ -156,13 +159,13 @@ def __init__(self, *args, **kw):
                 self.bz.login(self.config.username, password)
 
     @staticmethod
-    def get_keyring_service(config):
+    def get_keyring_service(config: BugzillaConfig) -> str:
         return f"bugzilla://{config.username}@{config.base_uri}"
 
-    def get_owner(self, issue):
+    def get_owner(self, issue: dict[str, Any]) -> str:
         return issue['assigned_to']
 
-    def include(self, issue):
+    def include(self, issue: dict[str, Any]) -> bool:
         """Return true if the issue in question should be included"""
         if self.config.only_if_assigned:
             owner = self.get_owner(issue)
@@ -175,7 +178,7 @@ def include(self, issue):
 
         return True
 
-    def annotations(self, tag, issue):
+    def annotations(self, tag: str, issue: dict[str, Any]) -> list[str]:
         base_url = "%s/show_bug.cgi?id=" % self.config.base_uri
         long_url = base_url + str(issue['id'])
         url = long_url
@@ -192,20 +195,21 @@ def annotations(self, tag, issue):
             # version of bugzilla itself.  :(
             comments = issue.get('longdescs', [])
 
-            def _parse_author(obj):
+            def _parse_author(obj: dict[str, Any] | str) -> str:
                 if isinstance(obj, dict):
                     return obj['login_name'].split('@')[0]
                 else:
                     return obj
 
-            def _parse_body(obj):
+            def _parse_body(obj: dict[str, Any]) -> str | None:
                 return obj.get('text', obj.get('body'))
 
             return self.build_annotations(
-                ((_parse_author(c['author']), _parse_body(c)) for c in comments), url
+                ((_parse_author(c['author']), _parse_body(c) or "") for c in comments),
+                url,
             )
 
-    def issues(self):
+    def issues(self) -> Iterator[BugzillaIssue]:
         email = self.config.username
         # TODO -- doing something with blockedby would be nice.
 
@@ -286,7 +290,7 @@ def issues(self):
             issue_obj.extra.update(extra)
             yield issue_obj
 
-    def _get_assigned_date(self, issue):
+    def _get_assigned_date(self, issue: dict[str, Any]) -> str | None:
         bug = self.bz.getbug(issue['id'])
         history = bug.get_history_raw()['bugs'][0]['history']
 
@@ -297,7 +301,7 @@ def _get_assigned_date(self, issue):
                     return _ensure_datetime(h['when']).isoformat()
 
 
-def _get_bug_attr(bug, attr):
+def _get_bug_attr(bug: Any, attr: str) -> Any:
     """Default longdescs/flags case to [] since they may not be present."""
     if attr in ("longdescs", "flags"):
         return getattr(bug, attr, [])

bugwarrior/services/clickup.py

@@ -1,7 +1,8 @@
+from collections.abc import Iterator
 import datetime
 import logging
 import typing
-from typing import Generator, Optional
+from typing import Any, Generator, Optional
 
 import requests
 
@@ -20,11 +21,11 @@ class ClickupConfig(config.ServiceConfig):
 class ClickupClient(Client):
     """Abstraction of Clickup API v2"""
 
-    def __init__(self, token):
+    def __init__(self, token: str) -> None:
         self.token = token
 
     @staticmethod
-    def _get_url_for_tasks(team_id: int, page: int = 0):
+    def _get_url_for_tasks(team_id: int, page: int = 0) -> str:
         base_url = "https://api.clickup.com/api/v2/"
         query = f"include_closed=false&page={page}"
         return f"{base_url}team/{team_id}/task?{query}"
@@ -78,7 +79,7 @@ class ClickupIssue(Issue):
 
     PRIORITY_MAP = {"urgent": "H", "high": "M", "normal": "L", "low": ""}
 
-    def to_taskwarrior(self):
+    def to_taskwarrior(self) -> dict[str, Any]:
         if not self.record["project"]["hidden"]:
             project = self.record["project"]["name"]
         else:
@@ -102,7 +103,7 @@ def to_taskwarrior(self):
             self.NAME: self.record["name"],
         }
 
-    def get_default_description(self):
+    def get_default_description(self) -> str:
         return self.build_default_description(
             title=self.record["name"], url=self.record["url"]
         )
@@ -118,17 +119,19 @@ def parse_timestamp(
         return datetime.datetime.fromtimestamp(seconds_unix, tz=datetime.timezone.utc)
 
 
-class ClickupService(Service):
+class ClickupService(Service[ClickupIssue]):
     API_VERSION = 1.0
     ISSUE_CLASS = ClickupIssue
     CONFIG_SCHEMA = ClickupConfig
 
-    def __init__(self, *args, **kw):
-        super().__init__(*args, **kw)
+    def __init__(
+        self, config: ClickupConfig, main_config: config.MainSectionConfig
+    ) -> None:
+        super().__init__(config, main_config)
         self.client = ClickupClient(token=self.get_secret('token'))
 
     @staticmethod
-    def get_keyring_service(config):
+    def get_keyring_service(config: ClickupConfig) -> str:
         return "clickup://"
 
     def is_assigned(self, issue: dict) -> bool:
@@ -144,7 +147,7 @@ def is_assigned(self, issue: dict) -> bool:
 
         return False
 
-    def issues(self):
+    def issues(self) -> Iterator[ClickupIssue]:
         for task in self.client.get_tasks_for_team(self.config.team_id):
             if self.is_assigned(task):
                 yield self.get_issue_for_record(task)