"""
Service API
-----------
"""
import abc
from collections.abc import Iterable, Iterator
import datetime
import logging
import math
import os
import re
from typing import Any, Generic, Optional, TypeVar
import zoneinfo
from dateutil.parser import parse as parse_date
import dogpile.cache
from jinja2 import Template
import requests
from bugwarrior.config import schema, secrets
log = logging.getLogger(__name__)
DOGPILE_CACHE_PATH = os.path.expanduser(
''.join([os.getenv('XDG_CACHE_HOME', '~/.cache'), '/dagd-py3.dbm'])
)
if not os.path.isdir(os.path.dirname(DOGPILE_CACHE_PATH)):
os.makedirs(os.path.dirname(DOGPILE_CACHE_PATH))
CACHE_REGION = dogpile.cache.make_region().configure(
"dogpile.cache.dbm", arguments=dict(filename=DOGPILE_CACHE_PATH)
)
# MAJOR versions signal a breakage in backwards compatibility between services
# and previous releases of bugwarrior. That is, services implementing the new
# spec will cause breakages with older bugwarrior releases.
# MINOR versions signal extensions of the spec which enhance future releases of
# bugwarrior without breaking past releases.
LATEST_API_VERSION = 1.0
class URLShortener:
_instance = None
def __new__(cls, *args: Any, **kwargs: Any) -> "URLShortener":
if not cls._instance:
cls._instance = super().__new__(cls, *args, **kwargs)
return cls._instance
@CACHE_REGION.cache_on_arguments()
def shorten(self, url: str) -> str:
if not url:
return ''
base = 'https://da.gd/s'
return requests.get(base, params=dict(url=url)).text.strip()
def get_processed_url(main_config: schema.MainSectionConfig, url: str) -> str:
"""Returns a URL with conditional processing.
If the following config key are set:
- [general]shorten
returns a shortened URL; otherwise returns the URL unaltered.
"""
if main_config.shorten:
return URLShortener().shorten(url)
return url
[docs]
class Issue(abc.ABC):
"""Base class for translating from foreign records to taskwarrior tasks.
The upper case attributes and abstract methods need to be defined by
service implementations, while the lower case attributes and concrete
methods are provided by the base class.
"""
#: Set to a dictionary mapping UDA short names with type and long name.
#:
#: Example::
#:
#: {
#: 'project_id': {
#: 'type': 'string',
#: 'label': 'Project ID',
#: },
#: 'ticket_number': {
#: 'type': 'number',
#: 'label': 'Ticket Number',
#: },
#: }
#:
#: Note: For best results, dictionary keys should be unique!
UDAS: dict
#: Should be a tuple of field names (can be UDA names) that are usable for
#: uniquely identifying an issue in the foreign system.
UNIQUE_KEY: tuple[str, ...]
#: Should be a dictionary of value-to-level mappings between the foreign
#: system and the string values 'H', 'M' or 'L'.
PRIORITY_MAP: dict
def __init__(
self,
foreign_record: dict[str, Any],
config: schema.ServiceConfig,
main_config: schema.MainSectionConfig,
extra: dict[str, Any],
) -> None:
#: Data retrieved from the external service.
self.record = foreign_record
#: An object whose attributes are this service's configuration values.
self.config: schema.ServiceConfig = config
#: An object whose attributes are the
#: :ref:`common_configuration:Main Section` configuration values.
self.main_config: schema.MainSectionConfig = main_config
#: Data computed by the :class:`Service` class.
self.extra = extra
[docs]
@abc.abstractmethod
def to_taskwarrior(self) -> dict[str, Any]:
"""Transform a foreign record into a taskwarrior dictionary."""
raise NotImplementedError()
[docs]
@abc.abstractmethod
def get_default_description(self) -> str:
"""Return a default description for this task.
You should probably use :meth:`build_default_description` to achieve
this.
"""
raise NotImplementedError()
[docs]
def get_priority(self) -> schema.Priority:
"""Return the priority of this issue, falling back to ``default_priority`` configuration."""
return self.PRIORITY_MAP.get(
self.record.get('priority'), self.config.default_priority
)
[docs]
def parse_date(
self, date: str | None, timezone: str = 'deprecated'
) -> datetime.datetime | None:
"""Parse a date string into a datetime object.
If the parsed date does not have a timezone, the UTC timezone is added.
:param `date`: A time string parseable by `dateutil.parser.parse`
"""
if timezone != 'deprecated':
log.warning(
"Deprecation Warning: Issue.parse_date's timezone parameter is deprecated and will "
"be removed in a future API version."
)
if not date:
return None
_date = parse_date(date)
if not _date.tzinfo:
_date = _date.replace(
tzinfo=datetime.timezone.utc
if timezone == 'deprecated'
else zoneinfo.ZoneInfo(timezone)
)
return _date.replace(microsecond=0)
[docs]
def build_default_description(
self, title: str = '', url: str = '', number: str | int = '', cls: str = "issue"
) -> str:
"""Return a default description, respecting configuration options.
:param `title`: Short description of the task.
:param `url`: URL to the task on the service.
:param `number`: Number associated with the task on the service.
:param `cls`: The abbreviated type of task this is. Preferred options
are ('issue', 'pull_request', 'merge_request', 'todo', 'task',
'subtask').
"""
cls_markup = {
'issue': 'Is',
'pull_request': 'PR',
'merge_request': 'MR',
'todo': '',
'task': '',
'subtask': 'Subtask #',
}
url_separator = ' .. '
url = (
get_processed_url(self.main_config, url)
if self.main_config.inline_links
else ''
)
desc_len = self.main_config.description_length
return "(bw)%s#%s - %s%s%s" % (
cls_markup.get(cls, cls.title()),
number,
title[:desc_len] if desc_len else title,
url_separator if url else '',
url,
)
T_Issue = TypeVar("T_Issue", bound="Issue")
[docs]
class Service(abc.ABC, Generic[T_Issue]):
"""Base class for fetching issues from the service.
The upper case attributes and abstract methods need to be defined by
service implementations, while the lower case attributes and concrete
methods are provided by the base class.
"""
#: Which version of the API does this service implement?
API_VERSION: float
#: Which class should this service instantiate for holding these issues?
ISSUE_CLASS: type[T_Issue]
#: Which class defines this service's configuration options?
CONFIG_SCHEMA: type[schema.ServiceConfig]
def __init__(
self, config: schema.ServiceConfig, main_config: schema.MainSectionConfig
) -> None:
over_version = math.floor(LATEST_API_VERSION) + 1
if self.API_VERSION >= over_version:
raise ValueError(
f"Incompatible Service: {config.service} implements api "
f"version {self.API_VERSION} but this version of bugwarrior "
f"only supports versions less than {over_version}."
)
#: An object whose attributes are this service's configuration values.
self.config = config
#: An object whose attributes are the
#: :ref:`common_configuration:Main Section` configuration values.
self.main_config = main_config
log.info("Working on [%s]", self.config.target)
[docs]
def get_secret(self, key: str, login: str = 'nousername') -> str:
"""Get a secret value, potentially from an :ref:`oracle <Secret Management>`.
The secret key need not be a *password*, per se.
:param `key`: Name of the configuration field of the given secret.
:param `login`: Username associated with the password in a keyring, if
applicable.
"""
password = getattr(self.config, key)
keyring_service = self.get_keyring_service(self.config)
if not password or password.startswith("@oracle:"):
password = secrets.get_service_password(
keyring_service, login, oracle=password
)
return password
[docs]
def get_issue_for_record(
self, record: dict[str, Any], extra: dict[str, Any] | None = None
) -> T_Issue:
"""Instantiate and return an issue for the given record.
:param `record`: Foreign record.
:param `extra`: Computed data which is not directly from the service.
"""
extra = extra if extra is not None else {}
return self.ISSUE_CLASS(record, self.config, self.main_config, extra=extra)
[docs]
def build_annotations(
self, annotations: Iterable[tuple[str, str]], url: Optional[str] = None
) -> list[str]:
"""Format annotations, respecting configuration values.
:param `annotations`: Comments from service.
:param `url`: Url to prepend to the annotations.
"""
final = []
if url and self.main_config.annotation_links:
final.append(get_processed_url(self.main_config, url))
if self.main_config.annotation_comments:
for author, message in annotations:
message = message.strip()
if not message or not author:
continue
if not self.main_config.annotation_newlines:
message = message.replace('\n', '').replace('\r', '')
annotation_length = self.main_config.annotation_length
if annotation_length:
message = '%s%s' % (
message[:annotation_length],
'...' if len(message) > annotation_length else '',
)
final.append('@%s - %s' % (author, message))
return final
[docs]
@abc.abstractmethod
def issues(self) -> Iterator[T_Issue]:
"""A generator yielding Issue instances representing issues from a remote service.
Each item in the list should be a dict that looks something like this:
.. code-block:: python
{
"description": "Some description of the issue",
"project": "some_project",
"priority": "H",
"annotations": [
"This is an annotation",
"This is another annotation",
]
}
The description can be 'anything' but must be consistent and unique for
issues you're pulling from a remote service. You can and should use
the ``.description(...)`` method to help format your descriptions.
The project should be a string and may be anything you like.
The priority should be one of "H", "M", or "L".
"""
raise NotImplementedError()
[docs]
@staticmethod
@abc.abstractmethod
def get_keyring_service(config: schema.ServiceConfig) -> str:
"""Return the keyring name for this service."""
raise NotImplementedError
[docs]
class Client:
"""Base class for making requests to service API's.
This class is not strictly necessary but encourages a well-structured
service in which the details of making and parsing http requests is
compartmentalized.
"""
[docs]
@staticmethod
def json_response(response: requests.Response) -> Any:
"""Return json if response is OK."""
# If we didn't get good results, just bail.
if response.status_code != 200:
raise OSError(
f"Non-200 status code {response.status_code}; {response.url}; {response.text}"
)
return response.json()
# NOTE: __all__ determines the stable, public API.
__all__ = [Client.__name__, Issue.__name__, Service.__name__]