import json
try:
import requests
from requests.exceptions import RequestException
from requests.packages.urllib3.exceptions import InsecureRequestWarning
except ImportError: # pragma: no cover
pass
from . import exceptions, utils
from .models import (
Epics,
History,
IssueAttachments,
IssueAttributes,
Issues,
IssueStatuses,
IssueTypes,
Milestones,
Points,
Priorities,
Projects,
Roles,
Severities,
SwimLanes,
TaskAttachments,
TaskAttributes,
Tasks,
TaskStatuses,
Users,
UserStories,
UserStoryAttachments,
UserStoryAttributes,
UserStoryStatuses,
Webhooks,
WikiLinks,
WikiPages,
)
from .requestmaker import RequestMaker
class SearchResult:
count = 0
tasks = []
issues = []
user_stories = []
wikipages = []
epics = []
[docs]
class TaigaAPI:
"""
TaigaAPI class
:param host: the host of your Taiga.io instance
:param token: the token you may provide
:param token_type: the token type
:param tls_verify: verify server certificate
:param auth_type: authentication type identifier
"""
def __init__(
self, host="https://api.taiga.io", token=None, token_type="Bearer", tls_verify=True, auth_type="normal"
):
self.host = host
self.token = token
self.token_refresh = None
self.token_type = token_type
self.tls_verify = tls_verify
self.auth_type = auth_type
if not self.tls_verify:
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
if token:
self.raw_request = RequestMaker("/api/v1", self.host, self.token, self.token_type, self.tls_verify)
self._init_resources()
def _init_resources(self):
self.projects = Projects(self.raw_request)
self.user_stories = UserStories(self.raw_request)
self.user_story_attachments = UserStoryAttachments(self.raw_request)
self.users = Users(self.raw_request)
self.swimlanes = SwimLanes(self.raw_request)
self.issues = Issues(self.raw_request)
self.issue_attachments = IssueAttachments(self.raw_request)
self.tasks = Tasks(self.raw_request)
self.task_attachments = TaskAttachments(self.raw_request)
self.milestones = Milestones(self.raw_request)
self.severities = Severities(self.raw_request)
self.roles = Roles(self.raw_request)
self.points = Points(self.raw_request)
self.issue_statuses = IssueStatuses(self.raw_request)
self.issue_types = IssueTypes(self.raw_request)
self.issue_attributes = IssueAttributes(self.raw_request)
self.task_attributes = TaskAttributes(self.raw_request)
self.user_story_attributes = UserStoryAttributes(self.raw_request)
self.task_statuses = TaskStatuses(self.raw_request)
self.priorities = Priorities(self.raw_request)
self.user_story_statuses = UserStoryStatuses(self.raw_request)
self.wikipages = WikiPages(self.raw_request)
self.wikilinks = WikiLinks(self.raw_request)
self.history = History(self.raw_request)
self.webhooks = Webhooks(self.raw_request)
self.epics = Epics(self.raw_request)
[docs]
def me(self):
"""
Get a :class:`taiga.models.models.User` representing me
"""
return self.users.get("me")
[docs]
def search(self, project, text=""):
"""
Search in your Taiga.io instance
:param project: the project id
:param text: the query of your search
"""
result = self.raw_request.get("search", query={"project": project, "text": text})
result = result.json()
search_result = SearchResult()
search_result.count = result["count"]
search_result.tasks = self.tasks.parse_list(result["tasks"])
search_result.issues = self.issues.parse_list(result["issues"])
search_result.user_stories = self.user_stories.parse_list(result["userstories"])
search_result.wikipages = self.wikipages.parse_list(result["wikipages"])
search_result.epics = self.epics.parse_list(result["epics"])
return search_result
[docs]
def auth(self, username, password):
"""
Authenticate you
:param username: your username
:param password: your password
"""
headers = {"Content-type": "application/json"}
payload = {"type": self.auth_type, "username": username, "password": password}
try:
full_url = utils.urljoin(self.host, "/api/v1/auth")
response = requests.post(full_url, data=json.dumps(payload), headers=headers, verify=self.tls_verify)
except RequestException:
raise exceptions.TaigaRestException(full_url, 400, "NETWORK ERROR", "POST")
if response.status_code != 200:
raise exceptions.TaigaRestException(full_url, response.status_code, response.text, "POST")
self.token = response.json()["auth_token"]
self.token_refresh = response.json()["refresh"]
self.raw_request = RequestMaker("/api/v1", self.host, self.token, "Bearer", self.tls_verify)
self._init_resources()
[docs]
def auth_app(self, app_id, app_secret, auth_code, state=""):
"""
Authenticate an app
:param app_id: the app id
:param app_secret: the app secret
:param auth_code: the app auth code
"""
headers = {"Content-type": "application/json"}
payload = {"application": app_id, "auth_code": auth_code, "state": state}
try:
full_url = utils.urljoin(self.host, "/api/v1/application-tokens/validate")
response = requests.post(full_url, data=json.dumps(payload), headers=headers, verify=self.tls_verify)
except RequestException:
raise exceptions.TaigaRestException(full_url, 400, "NETWORK ERROR", "POST")
if response.status_code != 200:
raise exceptions.TaigaRestException(full_url, response.status_code, response.text, "POST")
cyphered_token = response.json().get("cyphered_token", "")
if cyphered_token:
from jwkest.jwe import JWE
from jwkest.jwk import SYMKey
sym_key = SYMKey(key=app_secret, alg="A128KW")
data, success = JWE().decrypt(cyphered_token, keys=[sym_key]), True
if isinstance(data, tuple):
data, success = data
try:
self.token = json.loads(data.decode("utf-8")).get("token", None)
except ValueError: # pragma: no cover
self.token = None
if not success:
self.token = None
else:
self.token = None
if self.token is None:
raise exceptions.TaigaRestException(full_url, 400, "INVALID TOKEN", "POST")
self.raw_request = RequestMaker("/api/v1", self.host, self.token, "Application", self.tls_verify)
self._init_resources()
[docs]
def refresh_token(self, token_refresh=""):
"""
Refresh auth token.
Passing a token_refresh will use passed token, otherwise it will try to use self.token_refresh.
:param token_refresh: the refresh token to be used to refresh api token
"""
if not token_refresh:
if self.token_refresh:
token_refresh = self.token_refresh
else:
raise ValueError("Refresh token not set")
headers = {"Content-type": "application/json"}
payload = {"refresh": token_refresh}
try:
full_url = utils.urljoin(self.host, "/api/v1/auth/refresh")
response = requests.post(full_url, data=json.dumps(payload), headers=headers, verify=self.tls_verify)
except RequestException:
raise exceptions.TaigaRestException(full_url, 400, "NETWORK ERROR", "POST")
if response.status_code != 200:
raise exceptions.TaigaRestException(full_url, response.status_code, response.text, "POST")
self.token = response.json()["auth_token"]
self.token_refresh = response.json()["refresh"]
self.raw_request = RequestMaker("/api/v1", self.host, self.token, "Bearer", self.tls_verify)
self._init_resources()