OAuth and token support

bitbucket/aclient.py

@@ -23,12 +23,21 @@ class Client(BaseClient):
     """
 
     async def __aenter__(self):
-        self._session = httpx.AsyncClient(
-            auth=(
-                self.user,
-                self.password,
+        if self.use_password:
+            self._session = httpx.AsyncClient(
+                auth=(
+                    self.user,
+                    self.password,
+                )
+            )
+        elif self.use_token:
+            headers = {
+                "Accept": "application/json",
+                "Authorization": f"Bearer {self.token}"
+            }
+            self._session = httpx.AsyncClient(
+                headers=headers
             )
-        )
 
         user_data = await self.get_user()
 

bitbucket/base.py

@@ -1,4 +1,5 @@
 import typing
+import requests
 
 from .exceptions import (
     InvalidIDError,
@@ -11,11 +12,28 @@
 
 class BaseClient(object):
     BASE_URL = "https://api.bitbucket.org/"
+    TOKEN_URL = 'https://bitbucket.org/site/oauth2/access_token'
 
-    def __init__(self, user: str, password: str, owner: typing.Union[str, None] = None):
+    def __init__(self, user: str=None, password: str=None,token: str=None,client_id: str=None, client_secret: str=None, owner: typing.Union[str, None] = None):
         self.user = user
         self.password = password
         self.username = owner
+        self.use_password = False
+        self.use_token = False
+        if user and password:
+            self.use_password = True
+        self.token = token
+        if token:
+            self.use_token = True
+        elif client_id and client_secret:
+            token_req_payload = {'grant_type': 'client_credentials'}
+            response = requests.post(self.TOKEN_URL, data=token_req_payload, allow_redirects=False, auth=(client_id, client_secret))
+            response = self.parse(response)
+            self.token = response['access_token']
+            self.use_token = True
+
+        if not (self.use_password or self.token):
+            raise NotAuthenticatedError("Insufficient credentials")
 
     def parse(self, response) -> typing.Union[typing.Dict[str, typing.Any], None]:
         """

bitbucket/client.py

@@ -2,10 +2,10 @@
 import requests
 
 from .base import BaseClient
-
+from .exceptions import NotAuthenticatedError
 
 class Client(BaseClient):
-    def __init__(self, user, password, owner=None):
+    def __init__(self, user=None, password=None, token=None, client_id=None, client_secret=None, owner=None):
         """Initial session with user/password, and setup repository owner
 
         Args:
@@ -14,7 +14,7 @@ def __init__(self, user, password, owner=None):
         Returns:
 
         """
-        super().__init__(user, password, owner)
+        super().__init__(user, password,token,client_id,client_secret, owner)
 
         # for shared repo, set baseURL to owner
         if owner is None:
@@ -424,34 +424,74 @@ def delete_webhook(self, repository_slug, webhook_uid, params=None):
             params=params,
         )
 
-    def _get(self, endpoint: str, params=None):
-        response = requests.get(
-            endpoint if endpoint.startswith("http") else self.BASE_URL + endpoint,
-            params=params,
-            auth=(self.user, self.password),
-        )
+    def _get(self, endpoint, params=None):
+        if self.use_password:
+            response = requests.get(self.BASE_URL + endpoint, params=params, auth=(self.user, self.password))
+        elif self.use_token:
+            headers = {
+                "Accept": "application/json",
+                "Authorization": f"Bearer {self.token}"
+            }
+            response = requests.request(
+                "GET",
+                self.BASE_URL + endpoint,
+                params=params,
+                headers=headers
+            )
+        else:
+            raise NotAuthenticatedError("Insufficient credentials")
         return self.parse(response)
 
     def _post(self, endpoint, params=None, data=None):
-        response = requests.post(
-            self.BASE_URL + endpoint,
-            params=params,
-            json=data,
-            auth=(self.user, self.password),
-        )
+        if self.use_password:
+            response = requests.post(self.BASE_URL + endpoint, params=params, json=data, auth=(self.user, self.password))
+        elif self.use_token:
+            headers = {
+                "Accept": "application/json",
+                "Authorization": f"Bearer {self.token}"
+            }
+            response = requests.request(
+                "POST",
+                self.BASE_URL + endpoint,
+                params=params, json=data,
+                headers=headers
+            )
+        else:
+            raise NotAuthenticatedError("Insufficient credentials")
         return self.parse(response)
 
     def _put(self, endpoint, params=None, data=None):
-        response = requests.put(
-            self.BASE_URL + endpoint,
-            params=params,
-            json=data,
-            auth=(self.user, self.password),
-        )
+        if self.use_password:
+            response = requests.put(self.BASE_URL + endpoint, params=params, json=data, auth=(self.user, self.password))
+        elif self.use_token:
+            headers = {
+                "Accept": "application/json",
+                "Authorization": f"Bearer {self.token}"
+            }
+            response = requests.request(
+                "PUT",
+                self.BASE_URL + endpoint,
+                params=params, json=data,
+                headers=headers
+            )
+        else:
+            raise NotAuthenticatedError("Insufficient credentials")
         return self.parse(response)
 
     def _delete(self, endpoint, params=None):
-        response = requests.delete(
-            self.BASE_URL + endpoint, params=params, auth=(self.user, self.password)
-        )
+        if self.use_password:
+            response = requests.delete(self.BASE_URL + endpoint, params=params, auth=(self.user, self.password))
+        elif self.use_token:
+            headers = {
+                "Accept": "application/json",
+                "Authorization": f"Bearer {self.token}"
+            }
+            response = requests.request(
+                "DELETE",
+                self.BASE_URL + endpoint,
+                params=params,
+                headers=headers
+            )
+        else:
+            raise NotAuthenticatedError("Insufficient credentials")
         return self.parse(response)