PK!0e11api_buddy/__init__.pyfrom .utils import VERSION __version__ = VERSION PK!!22api_buddy/cli.pyfrom colorama import init as init_colorama from .config.help import HELP from .config.options import load_options from .config.preferences import load_prefs from .config.variables import interpolate_variables from .network.request import send_request from .network.response import print_response from .network.session import get_session from .utils import VERSION, PREFS_FILE from .utils.exceptions import APIBuddyException, exit_with_exception def run() -> None: init_colorama() try: opts = load_options(HELP) if opts['--version']: print(VERSION) return if opts['--help']: print(HELP) return prefs = load_prefs(PREFS_FILE) interpolated_opts = interpolate_variables(opts, prefs) sesh = get_session(interpolated_opts, prefs, PREFS_FILE) resp = send_request(sesh, prefs, interpolated_opts, PREFS_FILE) print_response(resp, prefs) except APIBuddyException as err: exit_with_exception(err) return if __name__ == '__main__': run() PK!api_buddy/config/__init__.pyPK!Mfro api_buddy/config/help.pyfrom colorama import Fore, Style BRIGHT_GREEN = f'{Style.BRIGHT}{Fore.GREEN}' BRIGHT_NORMAL = f'{Style.RESET_ALL}{Style.BRIGHT}' API_CLI = f'{BRIGHT_GREEN}api{BRIGHT_NORMAL}' BACKSLASH = f'{Fore.MAGENTA}\\' EXAMPLE_PREFS = """ api_url: https://api.url.com auth_type: oauth2 oauth2: client_id: your_client_id client_secret: your_client_secret scopes: - one_scope - another_scope redirec_uri: http://loclahost:8080 state: something token_path: /token authorize_path: /authorize authorize_params: - select_profile=true auth_test_status: 401 api_version: 2 verify_ssl: false timeout: 100 headers: Cookie: flavor=chocolate-chip; milk=please; Origin: your-face verboseness: request: true response: true indent: 4 theme: paraiso-dark variables: user_id: ab12c3d email: me@email.com """ HELP = f"""\nExplore OAuth2 APIs from your console with API Buddy It's as easy as: {API_CLI} get some-endpoint{Style.RESET_ALL} HTTP Method defaults to get: {API_CLI} this-endpoint{Style.RESET_ALL} You can add query params in key=val format: {API_CLI} get {BACKSLASH} {BRIGHT_NORMAL}my/favorite/endpoint {BACKSLASH} {BRIGHT_NORMAL}first_name=cosmo {BACKSLASH} {BRIGHT_NORMAL}last_name=kramer{Style.RESET_ALL} You can also add request body data in JSON format: {API_CLI} post {BACKSLASH} {BRIGHT_NORMAL}some-endpoint {BACKSLASH} {Fore.RED}'{{"id": 1, "field": "value"}}'{Style.RESET_ALL} Note the single-quotes, you can also expand this accross multiple lines: {API_CLI} post {BACKSLASH} {BRIGHT_NORMAL}some-endpoint {BACKSLASH} {Fore.RED}'{{ "id": 1, "field": "value" }}'{Style.RESET_ALL} Variables can be interpolated within your endpoint, as part of values in your query params, or anywhere in your request body data, as long as they're defined by name in your preferences: {API_CLI} post {BACKSLASH} {BRIGHT_NORMAL}users/#{{user_id}} {BACKSLASH} {BRIGHT_NORMAL}name=#{{name}} {BACKSLASH} {Fore.RED}'{{ "occupation"="#{{occupation}}" }}'{Style.RESET_ALL} Your preferences live in {Fore.MAGENTA}~/.api-buddy.yaml{Style.RESET_ALL} They can look something like this: {Style.BRIGHT}{EXAMPLE_PREFS}{Style.RESET_ALL} Check out GitHub for more info {Fore.BLUE}{Style.BRIGHT}https://github.com/fonsecapeter/api-buddy{Style.RESET_ALL} Arguments: http_method (optional, default: get) One of [get, post, patch, put, delete] endpoint The relative path to an API endpoint params (optional) A list of key=val query params data (optional) A JSON string of request body data, for all methods but 'get' Usage: api help api (-h | --help) api (-v | --version) api get [ ...] api post [ ...] [] api patch [ ...] [] api put [ ...] [] api delete [ ...] [] api [ ...] Options: -h, --help Show this help message -v, --version Show installed version """ PK!#zapi_buddy/config/options.pyfrom docopt import docopt from ..utils.typing import Options from ..validation.options import validate_options def load_options(doc: str) -> Options: opts = docopt(doc) return validate_options(opts) PK![ssapi_buddy/config/preferences.pyimport yaml from os import path from copy import deepcopy from typing import Any from ..utils.exceptions import PrefsException from ..utils.http import unpack_query_params from ..utils.typing import Preferences from ..validation.preferences import ( DEFAULT_PREFS, NESTED_DEFAULT_PREFS, validate_preferences, ) EXAMPLE_OAUTH2_PREFS = { 'client_id': 'your_client_id', 'client_secret': 'your_client_secret', 'scopes': ['one_scope', 'another_scope'], } EXAMPLE_PREFS = { 'api_url': 'https://ron-swanson-quotes.herokuapp.com', } def _remove_defaults(prefs: Preferences) -> Preferences: """Remove defaults if they haven't been changed""" filtered_prefs = deepcopy(prefs) for key, default_val in DEFAULT_PREFS.items(): if key in NESTED_DEFAULT_PREFS: continue if filtered_prefs[key] == default_val: # type: ignore del filtered_prefs[key] # type: ignore for nested_name, nested_defaults in NESTED_DEFAULT_PREFS.items(): nested_prefs = filtered_prefs[nested_name] # type: ignore for nested_key, default_nested_val in nested_defaults.items(): if nested_prefs[nested_key] == default_nested_val: del filtered_prefs[nested_name][nested_key] # type: ignore return filtered_prefs def _convert_types(prefs: Preferences) -> Preferences: """Convert any types that are changed in validation for saving""" converted_prefs = deepcopy(prefs) auth_prefs = converted_prefs.get('oauth2') if auth_prefs: auth_params = auth_prefs.get('authorize_params') if auth_params: converted_prefs['oauth2']['authorize_params'] = ( # type: ignore unpack_query_params(auth_params) ) return converted_prefs def _extract_yaml_from_file(file_name: str) -> Any: """Load contents of yaml file Retuns: - None if file doesn't exist - The python-native data if it does Raises: PrefsException if: - file contents are not valid yaml - user preferences are None """ if not path.isfile(file_name): return None with open(file_name, 'r') as prefs_file: try: user_prefs = yaml.load(prefs_file) except yaml.YAMLError: raise PrefsException( title=f'There was a problem reading the file', message=( 'Please make sure it\'s valid yaml: ' 'http://www.yaml.org/start.html' ), ) if user_prefs is None: raise PrefsException( title='It looks like your file is empty', message=( f'Make sure you have something in there\n' f'For example:\n\n{yaml.dump(EXAMPLE_PREFS)}' ) ) return user_prefs def load_prefs( file_name: str, ) -> Preferences: """Load preferences from a yaml file Notes: - Expands ~ - Creates a preferences file if it doesn't exist - Merges with defaults """ expanded_file_name = path.expanduser(file_name) raw_prefs = _extract_yaml_from_file(expanded_file_name) if raw_prefs is None: prefs = validate_preferences(EXAMPLE_PREFS) save_prefs(prefs, expanded_file_name) else: prefs = validate_preferences(raw_prefs) return prefs def save_prefs( preferences: Preferences, file_name: str, ) -> None: """Save preferences as a yaml file Notes: - Expands ~ - Ignores defaults if they haven't changed """ expanded_file_name = path.expanduser(file_name) minimal_prefs = _remove_defaults(preferences) converted_prefs = _convert_types(minimal_prefs) with open(expanded_file_name, 'w') as prefs_file: yaml.dump(converted_prefs, prefs_file) PK!B)api_buddy/config/themes.pyfrom pygments.style import Style from pygments.token import ( Comment, Error, Keyword, Literal, Name, Number, Operator, Punctuation, String, Text, ) SHELLECTRIC = 'shellectric' class Shellectric(Style): # type: ignore default_style = "" styles = { Text: '#ansifuchsia', Comment: '#ansilightgray', Keyword: '#ansigreen', Keyword.Constant: '#ansiturquoise', # constant Keyword.Namespace: '#ansilightgray', Keyword.Pseudo: '#ansilightgray', Name: '#ansibrown', Name.Builtin.Pseudo: '#ansilightgray', # self Name.Function: '#ansiturquoise', Name.Class: '#ansiturquoise', Name.Exception: '#ansiturquoise', Name.Decorator: '#ansiturquoise', String: '#ansired', String.Other: '#ansibrown', Literal: '#ansired', Number: '#ansiblue', Operator: '#ansilightgray', Error: '#ansiyellow', Punctuation: '#ansilightgray', } PK!5SSapi_buddy/config/variables.pyimport json from copy import deepcopy from typing import Any, Callable, Dict, List, Union from ..utils.typing import Options, Preferences def _interpolate(thing: str, name: str, val: str) -> str: return thing.replace(f'#{{{name}}}', val) def _interpolate_these(name: str, val: str) -> Callable[[str], str]: def wrapper(thing: str) -> str: return _interpolate(thing, name, val) return wrapper def _interpolate_params( params: Dict[str, Union[str, List[str]]], name: str, val: str, ) -> Dict[str, Union[str, List[str]]]: interpolated_params: Dict[str, Union[str, List[str]]] = {} for query_name, query_val in params.items(): if isinstance(query_val, list): interpolated_params[query_name] = list(map( _interpolate_these(name, val), query_val, )) else: # is str interpolated_params[query_name] = _interpolate( query_val, name, val, ) return interpolated_params def _interpolate_data( data: Any, name: str, val: str, ) -> Any: # there's probably a faster way but eh this is safe / simple json_data = json.dumps(data) interpolated_json_data = _interpolate(json_data, name, val) return json.loads(interpolated_json_data) def interpolate_variables( opts: Options, prefs: Preferences ) -> Options: """Replace any instances of variables with their values""" interpolated_opts = deepcopy(opts) for name, val in prefs['variables'].items(): interpolated_opts[''] = _interpolate( interpolated_opts[''], name, val, ) interpolated_opts[''] = _interpolate_params( interpolated_opts[''], name, val, ) interpolated_opts[''] = _interpolate_data( interpolated_opts[''], name, val, ) return interpolated_opts PK!api_buddy/network/__init__.pyPK!"api_buddy/network/auth/__init__.pyPK!uF api_buddy/network/auth/oauth2.pyimport webbrowser from colorama import Fore, Style from os import environ from typing import Optional from urllib.parse import urljoin from requests_oauthlib import OAuth2Session from api_buddy.utils.typing import Options, Preferences, QueryParams from api_buddy.config.preferences import save_prefs APPLICATION_JSON = 'application/json' HEADERS = { 'Accept': APPLICATION_JSON, 'Content-Type': APPLICATION_JSON, } def _get_authorization_response_url() -> str: return input( # pragma: no cover f'{Fore.GREEN}Enter the full url{Fore.BLACK}{Style.BRIGHT}:' f'{Style.RESET_ALL} ' ) def _authenticate( sesh: OAuth2Session, client_secret: str, api_url: str, redirect_uri: str, state: Optional[str], token_path: str, authorize_path: str, authorize_params: QueryParams, ) -> str: """Perform OAuth2 Flow and get a new token Note: Implicitly updates the OAuth2Session """ authorization_url, state = sesh.authorization_url( urljoin(api_url, authorize_path), state=state, kwargs=authorize_params, ) print( f'Opening browser to visit:\n\n' f'{Fore.BLUE}{Style.BRIGHT}{authorization_url}{Style.RESET_ALL}\n\n' f'Sign in and go through the DSA, then copy the url.\n' ) webbrowser.open(authorization_url) authorization_response = _get_authorization_response_url() print() environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1' # allow non-http redirect_uri token = sesh.fetch_token( urljoin(api_url, token_path), authorization_response=authorization_response, client_secret=client_secret, include_client_id=True, ) return str(token['access_token']) def get_oauth2_session( opts: Options, prefs: Preferences, prefs_file_name: str, ) -> OAuth2Session: """Initialize OAuth2 session""" sesh = OAuth2Session( client_id=prefs['oauth2']['client_id'], redirect_uri=prefs['oauth2']['redirect_uri'], scope=' '.join(prefs['oauth2']['scopes']), token={'access_token': prefs['oauth2']['access_token']}, ) sesh.headers.update(HEADERS) return sesh def reauthenticate_oauth2( sesh: OAuth2Session, prefs: Preferences, prefs_file: str, ) -> OAuth2Session: """Get a new oauth token for an existing session Also save it to preferences """ oauth2_prefs = prefs['oauth2'] access_token = _authenticate( sesh, client_secret=prefs['oauth2']['client_secret'], api_url=prefs['api_url'], redirect_uri=oauth2_prefs['redirect_uri'], state=oauth2_prefs['state'], token_path=oauth2_prefs['token_path'], authorize_path=oauth2_prefs['authorize_path'], authorize_params=oauth2_prefs['authorize_params'], ) prefs['oauth2']['access_token'] = access_token save_prefs(prefs, prefs_file) return sesh PK!9\aapi_buddy/network/request.pyimport requests import urllib3 from colorama import Fore, Style from typing import Any, Dict, List, MutableMapping, Optional, Union from yaspin import yaspin from .session import reauthenticate from ..utils.exceptions import ( APIBuddyException, ConnectionException, TimeoutException, ) from ..utils.formatting import api_url_join, format_dict_like_thing from ..utils.http import ( GET, POST, PUT, PATCH, DELETE, ) from ..utils.spin import spin from ..utils.typing import Preferences, Options def _send_request( sesh: requests.Session, method: str, url: str, params: Dict[str, Union[str, List[str]]], data: Any, verify: bool, timeout: int, ) -> requests.Response: with yaspin(spin): if method == GET: return(sesh.get( url, params=params, timeout=timeout, verify=verify, )) elif method == POST: return(sesh.post( url, params=params, data=data, timeout=timeout, verify=verify, )) elif method == PUT: return(sesh.put( url, params=params, data=data, timeout=timeout, verify=verify, )) elif method == PATCH: return(sesh.patch( url, params=params, data=data, timeout=timeout, verify=verify, )) elif method == DELETE: return(sesh.delete( url, params=params, data=data, timeout=timeout, verify=verify, )) else: raise APIBuddyException( title='Something went wrong', message='Try a different http method' ) def print_request( method: str, url: str, headers: MutableMapping[str, str], params: Dict[str, Union[str, List[str]]], data: Dict[str, Any], theme: Optional[str], ) -> None: print( f'\n{Fore.GREEN}{Style.BRIGHT}{method.upper()} ' f'{Fore.BLUE}{url}{Style.RESET_ALL}' ) if headers: print(format_dict_like_thing('Headers', headers, theme)) if params: print(format_dict_like_thing('Query Params', params, theme)) if data is not None: print(format_dict_like_thing('Data', data, theme)) print() def send_request( sesh: requests.Session, prefs: Preferences, opts: Options, prefs_file: str, retry: bool = True, ) -> requests.Response: """Send the http request, reauthenticating if necessary""" timeout = prefs['timeout'] url = api_url_join( prefs['api_url'], prefs['api_version'], opts[''], ) method = opts[''] params = opts[''] data = opts[''] if prefs['verboseness']['request'] is True and retry: print_request(method, url, sesh.headers, params, data, prefs['theme']) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) try: resp = _send_request( sesh, method, url, params, data, prefs['verify_ssl'], timeout, ) except requests.exceptions.ConnectionError: raise ConnectionException() except requests.exceptions.ReadTimeout: raise TimeoutException(timeout) if prefs['auth_type'] is not None: if retry and resp.status_code == prefs['auth_test_status']: sesh = reauthenticate(sesh, prefs, prefs_file) resp = send_request(sesh, prefs, opts, prefs_file, retry=False) return resp PK!a\api_buddy/network/response.pyimport json from bs4 import BeautifulSoup from colorama import Fore, Style from requests import Response from requests.cookies import RequestsCookieJar from typing import MutableMapping, Optional from ..utils.formatting import format_dict_like_thing, highlight_syntax from ..utils.typing import Preferences TAGS_TO_SKIP = [ 'a', 'button' 'footer', 'head', 'header', 'nav', 'script', 'style', ] def _print_response_details( headers: MutableMapping[str, str], cookies: RequestsCookieJar, theme: Optional[str], ) -> None: if headers: print(format_dict_like_thing('Headers', headers, theme)) if cookies: print(format_dict_like_thing('Cookies', cookies, theme)) print() def _strip_html(content: str) -> str: """Parse tags and strip away stuff""" soup = BeautifulSoup(content, features='html.parser') for section in soup(TAGS_TO_SKIP): section.extract() raw_text = soup.get_text() lines = [] for line in raw_text.split('\n'): if line: lines.append(line.strip()) return '\n'.join(lines) def format_response( resp: Response, indent: Optional[int], theme: Optional[str], ) -> str: try: formatted = highlight_syntax( json.dumps(resp.json(), indent=indent), theme, ) except (json.decoder.JSONDecodeError, TypeError): formatted = resp.text if '' in formatted: formatted = _strip_html(formatted) return formatted.rstrip() def print_response(resp: Response, prefs: Preferences) -> None: verbose = prefs['verboseness']['response'] theme = prefs['theme'] arrow = f'{Fore.BLACK}{Style.BRIGHT}=>' if resp.ok: status_color = Fore.GREEN else: status_color = Fore.YELLOW if verbose: status = ( f'{status_color}{resp.status_code} ' f'{Style.NORMAL}{resp.reason}{Style.RESET_ALL}' ) else: status = f'{status_color}{resp.status_code}{Style.RESET_ALL}' print(f'{arrow} {status}') if verbose: _print_response_details(resp.headers, resp.cookies, theme) print(format_response(resp, prefs['indent'], theme)) PK!:Mddapi_buddy/network/session.pyimport requests from typing import Callable, Dict from .auth.oauth2 import get_oauth2_session, reauthenticate_oauth2 from ..utils.typing import Options, Preferences from ..utils.auth import OAUTH2 SessionInitializer = Callable[ [Options, Preferences, str], requests.Session, ] SESSIONS: Dict[str, SessionInitializer] = { OAUTH2: get_oauth2_session, } ReauthenticatStrategy = Callable[ [requests.Session, Preferences, str], requests.Session, ] REAUTHENTICATIONS: Dict[str, ReauthenticatStrategy] = { OAUTH2: reauthenticate_oauth2, } def get_session( opts: Options, prefs: Preferences, prefs_file_name: str, ) -> requests.Session: auth_type = prefs['auth_type'] if auth_type is None: return requests.Session() session_initializer = SESSIONS[auth_type] sesh = session_initializer(opts, prefs, prefs_file_name) sesh.headers.update(prefs['headers']) return sesh def reauthenticate( sesh: requests.Session, prefs: Preferences, prefs_file: str, ) -> requests.Session: auth_type = prefs['auth_type'] if auth_type is None: return sesh reauthenticate_strategy = REAUTHENTICATIONS[auth_type] new_sesh = reauthenticate_strategy(sesh, prefs, prefs_file) new_sesh.headers.update(prefs['headers']) return new_sesh PK!fapi_buddy/utils/__init__.pyfrom os import path VERSION = '0.2.1' PREFS_FILE = '~/.api-buddy.yaml' ROOT_DIR = path.dirname(path.dirname(path.dirname(__file__))) MAX_LINE_LENGTH = 50 PK!MP//api_buddy/utils/auth.pyOAUTH2 = 'oauth2' AUTH_TYPES = ( OAUTH2, ) PK!n^ ^ api_buddy/utils/exceptions.pyimport random from colorama import Fore, Style from typing import NoReturn from api_buddy.utils import PREFS_FILE from api_buddy.config.help import EXAMPLE_PREFS class APIBuddyException(Exception): def __init__(self, title: str, message: str) -> None: self.title = title self.message = message def exit_with_exception(err: APIBuddyException) -> NoReturn: emoji = random.choice(( '⚠️', '😭', '😮', '🙊', '🐛', '🔥', )) pleasantry = random.choice(( 'Oh no', 'Whoops', 'Oops', 'Crikey', 'Dang', 'Bruh', 'Woah', )) print( f'{Fore.YELLOW}{Style.BRIGHT}{pleasantry}! {emoji}\n' f'{Style.NORMAL}{err.title}{Style.RESET_ALL}\n\n' f'{err.message}\n' ) exit(1) class PrefsException(APIBuddyException): TITLE_HEADERS = ( 'There\'s a problem with your preferences', 'Your preferences appear to be borked', 'Your preferences aren\'t quite right', 'Your preferences are a bit off', 'It looks like your preferences are messed up', ) MESSAGE_FOOTERS_1 = ( f'Open up', f'Crack open', f'Check out', ) MESSAGE_FOOTERS_2 = ( 'and have a look', 'and fix it up', 'and make it right', ) def __init__(self, title: str, message: str) -> None: header = random.choice(self.TITLE_HEADERS) footer_1 = random.choice(self.MESSAGE_FOOTERS_1) footer_2 = random.choice(self.MESSAGE_FOOTERS_2) prefs_title = f'{header}\n{title}' prefs_msg = ( f'{message}\n\n{footer_1} ' f'{Fore.MAGENTA}{PREFS_FILE}{Style.RESET_ALL} ' f'{footer_2} -- here\'s an example with all the prefs:\n' f'{EXAMPLE_PREFS}' ) return super().__init__(prefs_title, prefs_msg) class ConnectionException(APIBuddyException): TITLES = ( 'There was a problem connecting that url', 'I can\'t reach that url', 'I can\'t find that url in the interwebs', ) MESSAGES = ( 'Are you on WiFi?', 'Is that even a real API?', 'Maybe try again?', 'Do you have a hotspot or something?', 'I think your WiFi is busted', ) def __init__(self) -> None: msg = random.choice(self.MESSAGES) return super().__init__( title=random.choice(self.TITLES), message=( f'{msg}\n\nCheck your {Fore.MAGENTA}api_url{Style.RESET_ALL} ' 'setting in your preferences and make sure you\'r connected to' ' the internet' ), ) class TimeoutException(APIBuddyException): TITLES = ( 'This is taking forever', 'I can\'t wait for this response anymore', 'Yo what\'s taking so long', ) def __init__(self, timeout: int) -> None: return super().__init__( title=random.choice(self.TITLES), message=( 'Your request timed out. I waited ' f'{Fore.MAGENTA}{Style.BRIGHT}{timeout}{Style.RESET_ALL} ' 'seconds.\n\n' 'If you want to wait longer, you should update the ' f'{Fore.MAGENTA}timeout{Style.RESET_ALL} setting in your ' 'preferences.' ), ) PK!q$$api_buddy/utils/formatting.pyfrom colorama import Fore, Style from pygments import highlight from pygments.formatters import Terminal256Formatter from pygments.lexers.data import JsonLexer, YamlLexer from pygments.styles import get_style_by_name from requests.cookies import RequestsCookieJar from typing import Any, cast, Dict, List, MutableMapping, Optional, Union from urllib.parse import urljoin from .exceptions import APIBuddyException from ..config.themes import SHELLECTRIC, Shellectric VARIABLE_CHARS = '#{}' JSON = 'json' YAML = 'yaml' def format_yaml_line(name: str, val: str) -> str: return ( f' {Fore.YELLOW}{name}{Fore.BLACK}{Style.BRIGHT}:' f' {Fore.RED}{val}{Style.RESET_ALL}' ) def format_yaml_list(things: List[str]) -> str: delim = f'\n {Fore.BLACK}{Style.BRIGHT}- {Fore.RED}' formatted_things = delim.join(sorted(things)) return f'{delim}{formatted_things}{Style.RESET_ALL}' def highlight_syntax( stuff: str, theme: Optional[str], lang: str = JSON ) -> str: """Colorize stuff with syntax highlighting""" if lang == JSON: lexer = JsonLexer() else: lexer = YamlLexer() if theme is None: return stuff elif theme == SHELLECTRIC: pygment_theme = Shellectric else: # theme already validated in preferences loading pygment_theme = get_style_by_name(theme) return cast(str, highlight( stuff, lexer, Terminal256Formatter(style=pygment_theme), )) def api_url_join( api_url: str, api_version: Optional[str], endpoint: str, ) -> str: """Joins base api url with api version and endpoint - a, None, c => a/c - a, b, c => a/b/c """ path = endpoint.lstrip('/') if api_version is not None: path = f'{api_version}/{path}' return urljoin(api_url, path) def format_dict_like_thing( name: str, thing: Union[ Dict[str, Any], MutableMapping[str, str], RequestsCookieJar, ], theme: Optional[str] = None, ) -> str: """Format dictionaries for nice printing""" delim = f'\n - ' formatted = f'{name}:' for key, val in thing.items(): display_val = val if isinstance(val, list): display_val = delim.join(val) display_val = f'{delim}{display_val}' key_val_pair = f'\n {key}: {display_val}' formatted += key_val_pair if theme is not None: formatted = highlight_syntax(formatted, theme, lang=YAML) return formatted.rstrip() def flat_str_dict( thing_name: str, thing: Dict[Any, Any], check_special_chars: bool = False, ) -> Dict[str, str]: """Convert dictionary like thing into strict, flat Dict[str, str]""" processed_vars = {} for name, val in thing.items(): if isinstance(val, (dict, list)): raise APIBuddyException( title=( f'Your {Fore.MAGENTA}{Style.BRIGHT}"{name}" ' f'{Style.NORMAL}{thing_name}{Style.RESET_ALL} ' 'is not gonna fly' ), message='It can\'t be nested, try something simpler', ) # bool capitalization is unpredictable if isinstance(val, bool): display_val = f'\'{str(val).lower()}\'' raise APIBuddyException( title=( f'Your {Fore.MAGENTA}{Style.BRIGHT}"{name}" ' f'{Style.NORMAL}{thing_name}{Style.RESET_ALL} is a boolean' ), message=( 'You\'re going to have to throw some quotes around that ' 'bad boy so I know how to capitalize it. Something like:\n' f' {format_yaml_line(name, display_val)}' ), ) if isinstance(name, bool): display_name = f'\'{str(name).lower()}\'' raise APIBuddyException( title=( f'Yo, you have a boolean for a {thing_name} name "{name}"' ), message=( 'Rename it or throw some quotes around that bad boy so I ' 'know how to capitalize it. Something like:\n' f' {format_yaml_line(display_name, val)}' ), ) if name is None: raise APIBuddyException( title='You must use a string name', message=( f'Using {Fore.MAGENTA}{Style.BRIGHT}null{Style.RESET_ALL} ' f'or {Fore.MAGENTA}{Style.BRIGHT}None{Style.RESET_ALL} as ' 'a name is not a thing.' ), ) if check_special_chars: any_name_has_special_chars = any( special_char in str(name) for special_char in VARIABLE_CHARS ) if any_name_has_special_chars: display_special_chars = ( ' '.join([f'"{c}"' for c in tuple(VARIABLE_CHARS)]) ) raise APIBuddyException( title=f'Your {thing_name} name "{name}" is too funky', message=( f'You can\'t use any of these special characters:\n ' f'{Fore.MAGENTA}{Style.BRIGHT}{display_special_chars}' f'{Style.RESET_ALL}' ), ) processed_vars[str(name)] = str(val) return processed_vars PK!~n'Zapi_buddy/utils/http.pyfrom colorama import Fore, Style from typing import Dict, List, Union from .typing import QueryParams from .exceptions import APIBuddyException GET = 'get' POST = 'post' PATCH = 'patch' PUT = 'put' DELETE = 'delete' HTTP_METHODS = ( GET, POST, PATCH, PUT, DELETE, ) def pack_query_params( params: List[str] ) -> QueryParams: """Convert string query params into key-value pairs ['key1=val1', 'key2=val2', 'key2=val3'] => {'key1': 'val1', 'key2': ['val2', 'val3']} """ keyed_params: Dict[str, Union[str, List[str]]] = {} for param in params: try: key, val = param.split('=') except ValueError: raise APIBuddyException( title='One of your query params is borked', message=( f'{Fore.MAGENTA}{param}{Style.RESET_ALL} should contain ' f'one and only one {Fore.MAGENTA}"="{Style.RESET_ALL}' ), ) prev_val = keyed_params.get(key) if prev_val is None: keyed_params[key] = val elif isinstance(prev_val, str): keyed_params[key] = [prev_val, val] else: prev_val.append(val) return keyed_params def unpack_query_params( params: QueryParams ) -> List[str]: """Convert key-value pairs into string query params {'key1': 'val1', 'key2': ['val2', 'val3']} => ['key1=val1', 'key2=val2', 'key2=val3'] """ unpacked: List[str] = [] for name, value in params.items(): if isinstance(value, list): for item in value: unpacked.append(f'{name}={item}') else: unpacked.append(f'{name}={value}') return unpacked PK! 5ooapi_buddy/utils/spin.pyfrom colorama import Fore, Style from typing import List from yaspin import Spinner INTERVAL = 80 OPEN = f'{Style.BRIGHT}' CLOSE = f'{Style.RESET_ALL}' COLORS = (Fore.GREEN, Fore.BLUE, Fore.MAGENTA, Fore.RED, Fore.YELLOW) frames: List[str] = [] for color in COLORS: frames += ( f'{OPEN}{color}=> {CLOSE}', f'{OPEN}{color}==> {CLOSE}', f'{OPEN}{color}===>{CLOSE}', f'{OPEN}{color} ==={CLOSE}', f'{OPEN}{color} =={CLOSE}', f'{OPEN}{color} ={CLOSE}', f'{OPEN}{color} <{CLOSE}', f'{OPEN}{color} <={CLOSE}', f'{OPEN}{color} <=={CLOSE}', f'{OPEN}{color}<==={CLOSE}', f'{OPEN}{color}=== {CLOSE}', f'{OPEN}{color}== {CLOSE}', f'{OPEN}{color}= {CLOSE}', f'{OPEN}{color} {CLOSE}', f'{OPEN}{color}> {CLOSE}', ) spin = Spinner(frames, INTERVAL) PK! F$$api_buddy/utils/typing.pyfrom mypy_extensions import TypedDict from typing import Any, Dict, Iterable, List, Optional, Union VerbosenessPreferences = TypedDict('VerbosenessPreferences', { 'request': bool, 'response': bool, }) QueryParams = Dict[str, Union[str, List[str]]] OAuth2Preferences = TypedDict('OAuth2Preferences', { 'client_id': str, 'client_secret': str, 'scopes': Iterable[str], 'redirect_uri': str, 'access_token': str, 'state': Optional[str], 'token_path': str, 'authorize_path': str, 'authorize_params': QueryParams, }) RawPreferences = Dict[str, Any] Preferences = TypedDict('Preferences', { 'api_url': str, 'auth_type': Optional[str], 'oauth2': OAuth2Preferences, 'auth_test_status': int, 'api_version': Optional[str], 'verify_ssl': bool, 'timeout': int, 'headers': Dict[str, str], 'verboseness': VerbosenessPreferences, 'indent': Optional[int], 'theme': Optional[str], 'variables': Dict[str, str], }) Options = TypedDict('Options', { '--help': bool, '--version': bool, '': str, '': str, '': QueryParams, '': Any, }) RawOptions = Dict[str, Optional[Union[str, bool]]] # TypedDict currently requires string literal key indexing # RawOptions = TypedDict('RawOptions', { # '--help': bool, # '--version': bool, # 'help': bool, # 'get': bool, # 'post': bool, # 'patch': bool, # 'put': bool, # 'delete': bool, # '': str, # '': List[str], # '': Optional[str], # }) PK! api_buddy/validation/__init__.pyPK!Kapi_buddy/validation/options.pyfrom json import loads, JSONDecodeError from copy import deepcopy from colorama import Fore, Style from typing import Any, cast, List from urllib.parse import urlparse from ..utils.typing import Options, RawOptions from ..utils.exceptions import APIBuddyException from ..utils.http import HTTP_METHODS, GET, pack_query_params def _more_than_one_method_selected(opts: RawOptions) -> bool: return sum( cast(bool, opts[method]) for method in HTTP_METHODS ) > 1 def _validate_method(opts: RawOptions) -> RawOptions: """Converts named bools to str enum Implicitly removed the old docopt booleans from opts Defaults to 'get' """ if _more_than_one_method_selected(opts): raise APIBuddyException( title='These HTTP methods are borked', message=( 'It appears you selected more than one method...\n' 'How did you even do this?' ) ) selected_method = GET for method in HTTP_METHODS: using_this_method = opts[method] del opts[method] if using_this_method is True: selected_method = method opts[''] = selected_method return opts def _validate_endpoint(endpoint: str) -> str: url_parts = urlparse(endpoint) if url_parts.scheme: message = 'You don\'t need to supply the full url, just the path.' path = url_parts.path if path: message += ( f'\nDid you mean {Fore.BLUE}{Style.BRIGHT}{url_parts.path}' f'{Style.RESET_ALL}?' ) raise APIBuddyException( title='Check your endpoint, dude', message=message, ) return endpoint def _validate_data(data: str) -> Any: try: return loads(data) except JSONDecodeError: raise APIBuddyException( title='Your request body data are wack', message=( 'Please use valid json for: ' f'{Fore.MAGENTA}{data}{Style.RESET_ALL}' ), ) def _validate_params_and_data(opts: RawOptions) -> RawOptions: """Validate query params and request body data Because of the cli argument signature, docopt always puts at the end the if it's given, and None in opts[''] no matter what. It can't ever know the difference between the last param and the optional data arg, so we have to check the last param and see if it's data. """ params = cast(List[str], opts['']) data = None if len(params) > 0: maybe_data = params[-1] maybe_params = params[:-1] try: data = _validate_data(maybe_data) except APIBuddyException as data_err: try: # maybe it's a param? pack_query_params([maybe_data]) except APIBuddyException: if opts[''] != GET: # should never be data for GET raise data_err # it's not a param, its bad data else: # maybe_data is data params = maybe_params if data is not None and opts[''] == GET: raise APIBuddyException( title=( 'You can\'t use request body data with ' f'{Fore.MAGENTA}GET{Style.RESET_ALL}' ), message=( 'Did you mean to use ' f'{Fore.MAGENTA}POST{Style.RESET_ALL}?' ), ) opts[''] = pack_query_params(params) # type: ignore opts[''] = data return opts def _validate_help(opts: RawOptions) -> RawOptions: opts['--help'] = opts['help'] del opts['help'] return opts def validate_options(opts: RawOptions) -> Options: """Convert types and validate""" valid_opts = deepcopy(opts) valid_opts[''] = _validate_endpoint( cast(str, valid_opts['']) ) _validate_method(valid_opts) _validate_params_and_data(valid_opts) _validate_help(valid_opts) return cast(Options, valid_opts) PK! E#api_buddy/validation/preferences.pyfrom colorama import Fore, Style from schema import ( Schema, SchemaError, Optional as Maybe, Or ) from pygments.styles import get_all_styles, get_style_by_name, ClassNotFound from typing import Any, cast, List, Optional from urllib.parse import urlparse from ..utils.formatting import flat_str_dict, format_yaml_list from ..utils.auth import AUTH_TYPES, OAUTH2 from ..utils.exceptions import PrefsException from ..utils.http import pack_query_params from ..utils.typing import Preferences, RawPreferences from ..config.themes import SHELLECTRIC DEFAULT_URL_SCHEME = 'https' DEFAULT_OAUTH2_PREFS = { 'redirect_uri': 'http://localhost:8080/', 'state': None, 'access_token': 'can_haz_token', 'token_path': 'token', 'authorize_path': 'authorize', 'authorize_params': {}, } DEFAULT_VERBOSENESS_PREFS = { 'request': False, 'response': False, } DEFAULT_PREFS = { 'auth_type': None, 'oauth2': DEFAULT_OAUTH2_PREFS, 'auth_test_status': 401, 'api_version': None, 'verify_ssl': True, 'timeout': 60, 'headers': {}, 'verboseness': DEFAULT_VERBOSENESS_PREFS, 'indent': 2, 'theme': SHELLECTRIC, 'variables': {}, } DEFAULT_AUTH_PREFS = { OAUTH2: DEFAULT_OAUTH2_PREFS, } NESTED_DEFAULT_PREFS = { **DEFAULT_AUTH_PREFS, # type: ignore 'verboseness': DEFAULT_VERBOSENESS_PREFS, } oauth2_schema = Schema({ 'client_id': str, 'client_secret': str, 'scopes': Schema([str]).validate, Maybe( 'redirect_uri', default=DEFAULT_OAUTH2_PREFS['redirect_uri'], ): str, Maybe( 'state', default=DEFAULT_OAUTH2_PREFS['state'], ): Or(str, None), Maybe( 'access_token', default=DEFAULT_OAUTH2_PREFS['access_token'], ): str, Maybe( 'token_path', default=DEFAULT_OAUTH2_PREFS['token_path'], ): str, Maybe( 'authorize_path', default=DEFAULT_OAUTH2_PREFS['authorize_path'], ): str, Maybe( 'authorize_params', default=DEFAULT_OAUTH2_PREFS['authorize_params'], ): [str], }) verboseness_schema = Schema({ Maybe( 'request', default=DEFAULT_VERBOSENESS_PREFS['request'], ): bool, Maybe( 'response', default=DEFAULT_VERBOSENESS_PREFS['response'], ): bool, }) prefs_schema = Schema({ 'api_url': str, Maybe( 'auth_type', default=DEFAULT_PREFS['auth_type'], ): Or(str, None), Maybe( 'oauth2', default=DEFAULT_PREFS['oauth2'], ): oauth2_schema, Maybe( 'auth_test_status', default=DEFAULT_PREFS['auth_test_status'], ): int, Maybe( 'api_version', default=DEFAULT_PREFS['api_version'], ): Or(str, int, float, None), Maybe( 'verify_ssl', default=DEFAULT_PREFS['verify_ssl'], ): bool, Maybe( 'timeout', default=DEFAULT_PREFS['timeout'], ): int, Maybe( 'headers', default=DEFAULT_PREFS['headers'], ): dict, Maybe( 'verboseness', default=DEFAULT_PREFS['verboseness'], ): verboseness_schema, Maybe( 'indent', default=DEFAULT_PREFS['indent'], ): Or(int, None), Maybe( 'theme', default=DEFAULT_PREFS['theme'], ): Or(str, None), Maybe( 'variables', default=DEFAULT_PREFS['variables'], ): dict, }) def _validate_auth_type(auth_type: Optional[str]) -> Optional[str]: if auth_type is None: return None valid_auth_type = auth_type.lower() if valid_auth_type not in AUTH_TYPES: delim = f'\n {Fore.BLACK}{Style.BRIGHT}- {Fore.MAGENTA}' display_auth_types = ( delim.join(AUTH_TYPES) ) raise PrefsException( title=f'I can\'t recognize your auth_type', message=( f'It should be one of these:' f'{delim}null{delim}' f'{display_auth_types}{Style.RESET_ALL}' ), ) return valid_auth_type def _validate_api_url(api_url: str) -> str: url_parts = urlparse(api_url) valid_url = api_url if not url_parts.scheme: valid_url = f'{DEFAULT_URL_SCHEME}://{api_url}' if '?' in api_url: raise PrefsException( title=f'Your api_url can\'t have query parameters', message=( 'Maybe try ' f'{Fore.BLUE}{Style.BRIGHT}{valid_url.split("?", 1)[0]}' f'{Style.RESET_ALL}' ), ) if '#' in api_url: raise PrefsException( title=f'Your api_url can\'t have hash fragments', message=( 'Maybe try ' f'{Fore.BLUE}{Style.BRIGHT}{valid_url.split("#", 1)[0]}' f'{Style.RESET_ALL}' ), ) return valid_url def _validate_api_version(api_version: Any) -> Optional[str]: if api_version is not None and not isinstance(api_version, str): return str(api_version) return api_version def _validate_theme(theme: Optional[str]) -> Optional[str]: if theme is None: return None elif theme == SHELLECTRIC: return SHELLECTRIC else: try: get_style_by_name(theme) except ClassNotFound: all_styles = [name for name in get_all_styles()] all_styles.append( f'{SHELLECTRIC} {Fore.BLACK}(this one is coolest)' ) raise PrefsException( title=( f'I haven\'t heard of the {Fore.MAGENTA}{theme}' f'{Fore.YELLOW} theme before.' ), message=( 'It sounds cool, but you have to pick one of these ' f'instead:\n{format_yaml_list(all_styles)}' ), ) return theme def validate_preferences(prefs: RawPreferences) -> Preferences: """Wrap errors nicely""" prefs['api_version'] = _validate_api_version(prefs.get('api_version')) try: valid_prefs: Preferences = prefs_schema.validate(prefs) except SchemaError as err: raise PrefsException( title='Something doesn\'t match the schema', message=str(err), ) valid_prefs['api_url'] = _validate_api_url(valid_prefs['api_url']) valid_prefs['variables'] = flat_str_dict( 'variable', valid_prefs['variables'], check_special_chars=True, ) valid_prefs['oauth2']['authorize_params'] = pack_query_params( cast(List[str], valid_prefs['oauth2']['authorize_params']), ) valid_prefs['headers'] = flat_str_dict('header', valid_prefs['headers']) valid_prefs['auth_type'] = _validate_auth_type(valid_prefs['auth_type']) valid_prefs['theme'] = _validate_theme(valid_prefs['theme']) return valid_prefs PK!HX))*api_buddy-0.2.1.dist-info/entry_points.txtN+I/N.,()J,ȴҔJL<..PK!c!!!api_buddy-0.2.1.dist-info/LICENSECopyright (c) 2019 Peter Fonseca Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HڽTUapi_buddy-0.2.1.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HrJ"api_buddy-0.2.1.dist-info/METADATAWoF~_1Ule;ns"ߩMF^PhEčI.+zR`~~%.#|7{*b)tG/UcR4mQ,uQ(یwUn,šS[JLLt}&{e8SRKy_hE=SpDUَ_脱1[nM]& 6M>CQN8Хʣ)ӕĠmnuȅBc)/z?ڲ^5g(*WfWӵWv4c42gu֔F_활˔S֩2a1b +{Wbqy}NӪfq`orC咮h}5J~ʚUE!/TMqi9 TɞǍt"rq[r ףonnIxc^4>?_ &'GFsFE;SWbz?O?0O J^J S1;<=yK2^"۟(W5;p[NW/PK&$T)ά {\;>nWiol3~.pCT<o -AYRs_Fѣn?yުvD.$fV˿ϭ*lR(Թ}pҪs.9^ DUŰE>p8*8j!^^e42eХ IJDv2sӹ?p}*(ᢐ\2Vgi-r]ʦ0@Cmo Jp8:L{P}L7NtNZ10—PyS{y訝hh|ϴ`1S֕sp{@MUPe D n)c"1Jt,y`1rÕ !õV\2,B5KV mwZ#El6W.*]nQ`|9hh=yUq͚YyV9 f3 NjT;d6ߎh!X nPg8* r ߢ)-MdTFc#>oHQSDˍ덩'7@}N7Ћ~n ܤB92C> z,X0/;p 64%mRPy'q3Y\PfP+Py )!֛'q wCSFMn?/_>+?t >Ø{WA^ϡ~sXcFbI%5 hH.}$5tX]|ڿV1=\KhU0F;-i 10]46ӡ CKuMB!}L0[[Ǡ]uyq!\fj!,ö߅I};)LЮ}H[lEDKiGl4\j pjYe#i]4EE fAf{%~Rzw>yM`Pa:Td3CP(Y@Cf/Cv;8:3 bjk 4868o^8}ږǯW}]m&Ku\Jq2$=WAqLwrPa[V]&Q8s.ԝU*iZ9XC%fHl뒭PApxȫd~ RKP?PK!H^5g api_buddy-0.2.1.dist-info/RECORDǒH} ,aB6^xQGGuP^lX>¾)Ih un} :&D0}m>b,:jhc"Br?V\+,T{<63 K|Rحѱi" P\C]wF2@:%N \9m(MLCjW evՙdAkZ+(]T䆑7Bx8RFZ3T@qzXt2Oq۰YÈ0Vt}q΁=,;iikY&qQڊt 3w *`ELG%(݅״B7{l7akfi̼fd`nC}=0Qޛ)r'rcc,e6nfd8v?JoCszav0p܋œ7> d?,khxHoS:xSg(]D/۲4 t:A1`ZSQDQ[t_ӚTQz`o!{޺/KB^z=pLmʃ& ǼoLJp=/agڦKq%.W_.Kw6SVR8_?8ئ'дV,GmEbu鵗(f}.g.+}ׁmqaQ-1}:;[Q;(s_Dޫ4o%H!Cͷ74LKWIj~GRpSYOa hgyOZ| }10R05̈ةd:ҧ%<8:%PTO-d*$M(A>)t>7o'gAq(N)AӤh; 7BX!<]7g#q0J=u:_pUh[6@PK!0e11api_buddy/__init__.pyPK!!22dapi_buddy/cli.pyPK!api_buddy/config/__init__.pyPK!Mfro api_buddy/config/help.pyPK!#zapi_buddy/config/options.pyPK![ssapi_buddy/config/preferences.pyPK!B)!api_buddy/config/themes.pyPK!5SS%api_buddy/config/variables.pyPK!.api_buddy/network/__init__.pyPK!".api_buddy/network/auth/__init__.pyPK!uF .api_buddy/network/auth/oauth2.pyPK!9\a5;api_buddy/network/request.pyPK!a\Kapi_buddy/network/response.pyPK!:MddKTapi_buddy/network/session.pyPK!fYapi_buddy/utils/__init__.pyPK!MP//Zapi_buddy/utils/auth.pyPK!n^ ^ ![api_buddy/utils/exceptions.pyPK!q$$hapi_buddy/utils/formatting.pyPK!~n'Zapi_buddy/utils/http.pyPK! 5oo8api_buddy/utils/spin.pyPK! F$$܉api_buddy/utils/typing.pyPK! 7api_buddy/validation/__init__.pyPK!Kuapi_buddy/validation/options.pyPK! E#api_buddy/validation/preferences.pyPK!HX))*api_buddy-0.2.1.dist-info/entry_points.txtPK!c!!!api_buddy-0.2.1.dist-info/LICENSEPK!HڽTUvapi_buddy-0.2.1.dist-info/WHEELPK!HrJ"api_buddy-0.2.1.dist-info/METADATAPK!H^5g @api_buddy-0.2.1.dist-info/RECORDPKuE