bitbucket.py
188 lines
| 6.0 KiB
| text/x-python
|
PythonLexer
r0 | # -*- coding: utf-8 -*- | |||
r112 | # Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors | |||
r0 | # | |||
r112 | # Licensed under the Apache License, Version 2.0 (the "License"); | |||
# you may not use this file except in compliance with the License. | ||||
# You may obtain a copy of the License at | ||||
r0 | # | |||
r112 | # http://www.apache.org/licenses/LICENSE-2.0 | |||
r0 | # | |||
r112 | # Unless required by applicable law or agreed to in writing, software | |||
# distributed under the License is distributed on an "AS IS" BASIS, | ||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||||
# See the License for the specific language governing permissions and | ||||
# limitations under the License. | ||||
r0 | ||||
import requests | ||||
from requests_oauthlib import OAuth1 | ||||
r153 | from appenlight.models.integrations import IntegrationBase, IntegrationException | |||
r0 | ||||
_ = str | ||||
class NotFoundException(Exception): | ||||
pass | ||||
class BitbucketIntegration(IntegrationBase): | ||||
r153 | __mapper_args__ = {"polymorphic_identity": "bitbucket"} | |||
r0 | front_visible = True | |||
as_alert_channel = False | ||||
supports_report_alerting = False | ||||
action_notification = True | ||||
r153 | integration_action = "Add issue to Bitbucket" | |||
r0 | ||||
@classmethod | ||||
def create_client(cls, request, user_name=None, repo_name=None): | ||||
""" | ||||
Creates REST client that can authenticate to specific repo | ||||
uses auth tokens for current request user | ||||
""" | ||||
config = request.registry.settings | ||||
token = None | ||||
secret = None | ||||
for identity in request.user.external_identities: | ||||
r153 | if identity.provider_name == "bitbucket": | |||
r0 | token = identity.access_token | |||
secret = identity.token_secret | ||||
break | ||||
if not token: | ||||
r153 | raise IntegrationException("No valid auth token present for this service") | |||
client = BitbucketClient( | ||||
token, | ||||
secret, | ||||
user_name, | ||||
repo_name, | ||||
config["authomatic.pr.bitbucket.key"], | ||||
config["authomatic.pr.bitbucket.secret"], | ||||
) | ||||
r0 | return client | |||
class BitbucketClient(object): | ||||
r153 | api_url = "https://bitbucket.org/api/1.0" | |||
repo_type = "bitbucket" | ||||
def __init__( | ||||
self, | ||||
token, | ||||
secret, | ||||
owner, | ||||
repo_name, | ||||
bitbucket_consumer_key, | ||||
bitbucket_consumer_secret, | ||||
): | ||||
r0 | self.access_token = token | |||
self.token_secret = secret | ||||
self.owner = owner | ||||
self.repo_name = repo_name | ||||
self.bitbucket_consumer_key = bitbucket_consumer_key | ||||
self.bitbucket_consumer_secret = bitbucket_consumer_secret | ||||
possible_keys = { | ||||
r153 | "status": [ | |||
"new", | ||||
"open", | ||||
"resolved", | ||||
"on hold", | ||||
"invalid", | ||||
"duplicate", | ||||
"wontfix", | ||||
], | ||||
"priority": ["trivial", "minor", "major", "critical", "blocker"], | ||||
"kind": ["bug", "enhancement", "proposal", "task"], | ||||
r0 | } | |||
def get_statuses(self): | ||||
"""Gets list of possible item statuses""" | ||||
r153 | return self.possible_keys["status"] | |||
r0 | ||||
def get_priorities(self): | ||||
"""Gets list of possible item statuses""" | ||||
r153 | return self.possible_keys["priority"] | |||
r0 | ||||
r153 | def make_request(self, url, method="get", data=None, headers=None): | |||
r0 | """ | |||
Performs HTTP request to bitbucket | ||||
""" | ||||
r153 | auth = OAuth1( | |||
self.bitbucket_consumer_key, | ||||
self.bitbucket_consumer_secret, | ||||
self.access_token, | ||||
self.token_secret, | ||||
) | ||||
r0 | try: | |||
r153 | resp = getattr(requests, method)(url, data=data, auth=auth, timeout=10) | |||
r0 | except Exception as e: | |||
raise IntegrationException( | ||||
r153 | _("Error communicating with Bitbucket: %s") % (e,) | |||
) | ||||
r0 | if resp.status_code == 401: | |||
r153 | raise IntegrationException(_("You are not authorized to access this repo")) | |||
r0 | elif resp.status_code == 404: | |||
r153 | raise IntegrationException(_("User or repo name are incorrect")) | |||
r0 | elif resp.status_code not in [200, 201]: | |||
raise IntegrationException( | ||||
r153 | _("Bitbucket response_code: %s") % resp.status_code | |||
) | ||||
r0 | try: | |||
return resp.json() | ||||
except Exception as e: | ||||
raise IntegrationException( | ||||
r153 | _("Error decoding response from Bitbucket: %s") % (e,) | |||
) | ||||
r0 | ||||
def get_assignees(self): | ||||
"""Gets list of possible assignees""" | ||||
r153 | url = "%(api_url)s/privileges/%(owner)s/%(repo_name)s" % { | |||
"api_url": self.api_url, | ||||
"owner": self.owner, | ||||
"repo_name": self.repo_name, | ||||
} | ||||
r0 | ||||
data = self.make_request(url) | ||||
r153 | results = [{"user": self.owner, "name": "(Repo owner)"}] | |||
r0 | if data: | |||
for entry in data: | ||||
r153 | results.append( | |||
{ | ||||
"user": entry["user"]["username"], | ||||
"name": entry["user"].get("display_name"), | ||||
} | ||||
) | ||||
r0 | return results | |||
def create_issue(self, form_data): | ||||
""" | ||||
Sends creates a new issue in tracker using REST call | ||||
""" | ||||
r153 | url = "%(api_url)s/repositories/%(owner)s/%(repo_name)s/issues/" % { | |||
"api_url": self.api_url, | ||||
"owner": self.owner, | ||||
"repo_name": self.repo_name, | ||||
} | ||||
r0 | ||||
payload = { | ||||
r153 | "title": form_data["title"], | |||
"content": form_data["content"], | ||||
"kind": form_data["kind"], | ||||
"priority": form_data["priority"], | ||||
"responsible": form_data["responsible"], | ||||
r0 | } | |||
r153 | data = self.make_request(url, "post", payload) | |||
r0 | f_args = { | |||
"owner": self.owner, | ||||
"repo_name": self.repo_name, | ||||
r153 | "issue_id": data["local_id"], | |||
r0 | } | |||
r153 | web_url = ( | |||
"https://bitbucket.org/%(owner)s/%(repo_name)s" | ||||
"/issue/%(issue_id)s/issue-title" % f_args | ||||
) | ||||
r0 | to_return = { | |||
r153 | "id": data["local_id"], | |||
"resource_url": data["resource_uri"], | ||||
"web_url": web_url, | ||||
r0 | } | |||
return to_return | ||||