so4t_api_v3.py

# Standard Python libraries
import json
import time

# Third-party libraries
import requests

# Local libraries
import so4t_request_validate


class V3Client(object):

    def __init__(self, url, token):

        print("Initializing API v3 client...")

        if not url: # check if URL is provided; if not, exit
            print("Missing required argument. Please provide a URL.")
            raise SystemExit

        if not token: # check if API token is provided; if not, exit
            print("Missing required argument. Please provide an API token.")
            raise SystemExit
        else:
            self.token = token
            self.headers = {
                'Authorization': f'Bearer {self.token}',
                'User-Agent': 'so4t_api_user_report/1.0 (http://your-app-url.com; your-contact@email.com)'
            }

        if "stackoverflowteams.com" in url: # Stack Internal (Business) or Basic
            self.team_slug = url.split("https://stackoverflowteams.com/c/")[1]
            self.api_url = f"https://api.stackoverflowteams.com/v3/teams/{self.team_slug}"
        else: # Stack Internal (Enterprise)
            self.api_url = url + "/api/v3"

        self.ssl_verify = self.test_connection() # test the API connection

    
    def test_connection(self):

        endpoint = "/tags"
        endpoint_url = self.api_url + endpoint
        ssl_verify = True

        print("Testing API v3 connection...")
        try:
            response = requests.get(endpoint_url, headers=self.headers)
        except requests.exceptions.SSLError:
            print("SSL error. Trying again without SSL verification...")
            response = requests.get(endpoint_url, headers=self.headers, verify=False)
            ssl_verify = False
        
        if response.status_code == 200:
            print("API connection successful")
            return ssl_verify
        else:
            print("Unable to connect to API. Please check your URL and API token.")
            print(f"Status code: {response.status_code}")
            print(f"Response from server: {response.text}")
            raise SystemExit


    def get_all_questions(self):
            
            method = "get"
            endpoint = "/questions"
            params = {
                'page': 1,
                'pagesize': 100,
            }
            questions = self.send_api_call(method, endpoint, params)
    
            return questions


    def get_all_tags(self):

        method = "get"
        endpoint = "/tags"
        params = {
            'page': 1,
            'pagesize': 100,
        }
        tags = self.send_api_call(method, endpoint, params)

        return tags


    def get_tag_smes(self, tag_id):

        method = "get"
        endpoint = f"/tags/{tag_id}/subject-matter-experts"
        smes = self.send_api_call(method, endpoint)

        return smes


    def get_user(self, user_id):

        method = "get"
        endpoint = f"/users/{user_id}"
        user = self.send_api_call(method, endpoint)

        return user
    

    def get_all_users(self):
            
            method = "get"
            endpoint = "/users"
            params = {
                'page': 1,
                'pagesize': 100,
            }
            users = self.send_api_call(method, endpoint, params)
    
            return users


    def send_api_call(self, method, endpoint, params={}):

        get_response = getattr(requests, method, None) # get the method from the requests library
        endpoint_url = self.api_url + endpoint

        data = []
        while True:
            try:
                if method == 'get':
                    response = get_response(endpoint_url, headers=self.headers, params=params,
                                            verify=self.ssl_verify,
                                            timeout=so4t_request_validate.timeout)
                else:
                    response = get_response(endpoint_url, headers=self.headers, json=params,
                                            verify=self.ssl_verify,
                                            timeout=so4t_request_validate.timeout)
            except Exception as ex:
                so4t_request_validate.handle_except(ex)
                continue

            if response.status_code not in [200, 201, 204]:
                print(f"API call to {endpoint_url} failed with status code {response.status_code}")
                print(f"Response from server: {response.text}")
                raise SystemExit

            # Respect v3 throttle headers to avoid being blocked
            burst_left = response.headers.get('x-burst-throttle-calls-left')
            if burst_left is not None and int(burst_left) < 5:
                burst_wait = int(response.headers.get('x-burst-throttle-seconds-until-full', 2))
                print(f"Approaching burst throttle limit ({burst_left} calls left). "
                      f"Waiting {burst_wait} seconds...")
                time.sleep(burst_wait)

            bucket_left = response.headers.get('x-token-bucket-calls-left')
            if bucket_left is not None and int(bucket_left) < 100:
                bucket_wait = int(response.headers.get('x-token-bucket-seconds-until-next-refill', 60))
                print(f"Token bucket running low ({bucket_left} tokens left). "
                      f"Waiting {bucket_wait} seconds for refill...")
                time.sleep(bucket_wait)

            try:
                json_data = response.json()
            except json.decoder.JSONDecodeError: # some API calls do not return JSON data
                print(f"API request successfully sent to {endpoint_url}")
                return

            if type(params) == dict and params.get('page'): # check request for pagination
                print(f"Received page {params['page']} from {endpoint_url}")
                data += json_data['items']
                if params['page'] == json_data['totalPages']:
                    break
                params['page'] += 1
            else:
                print(f"API request successfully sent to {endpoint_url}")
                data = json_data
                break

        return data