Source code for robin_stocks.gemini.helper

from base64 import urlsafe_b64decode as b64d
from base64 import urlsafe_b64encode as b64e
from functools import wraps
from inspect import signature
from zlib import compress, decompress

from robin_stocks.gemini.globals import (LOGGED_IN, NONCE,
                                         RETURN_PARSED_JSON_RESPONSE,
                                         SECRET_API_KEY, SESSION,
                                         USE_SANDBOX_URLS)


def increment_nonce():
    """ Increase nonce by one.
    """
    global NONCE
    NONCE += 1


def get_nonce():
    """ return the nonce.
    """
    return NONCE


def set_secret_key(data):
    """ Encodes the secret api key before storing it as a global variable.
    """
    global SECRET_API_KEY
    SECRET_API_KEY = b64e(compress(data, 9))


def get_secret_key():
    """ Decodes the secret api key from the global variable.
    """
    return decompress(b64d(SECRET_API_KEY))


def format_inputs(func):
    """ A decorator for formatting inputs. For any function decorated by this,
        the value of jsonify=None will be replaced with the global value stored at 
        RETURN_PARSED_JSON_RESPONSE.
    """
    @wraps(func)
    def format_wrapper(*args, **kwargs):
        bound_args = signature(func).bind(*args, **kwargs)
        bound_args.apply_defaults()
        target_args = dict(bound_args.arguments)
        if target_args['jsonify'] is None:
            kwargs["jsonify"] = get_default_json_flag()
        return(func(*args, **kwargs))
    return(format_wrapper)


def set_default_json_flag(parse_json):
    """ Sets whether you want all functions to return the json parsed response or not.

    :param parse_json: Set to change value of global variable.
    :type parse_json: bool
    """
    global RETURN_PARSED_JSON_RESPONSE
    RETURN_PARSED_JSON_RESPONSE = parse_json


def get_default_json_flag():
    """ Gets the boolean flag on the default JSON setting.
    """
    return RETURN_PARSED_JSON_RESPONSE


def get_sandbox_flag():
    """ Gets whether all functions are to use the sandbox base url.
    """
    return USE_SANDBOX_URLS


def use_sand_box_urls(use_sandbox):
    """ Sets whether you want all functions to use the sandbox base url.

    :param use_sandbox: Set to change value of global variable.
    :type use_sandbox: bool
    """
    global USE_SANDBOX_URLS
    USE_SANDBOX_URLS = use_sandbox


def update_session(key, value):
    """Updates the session header used by the requests library.

    :param key: The key value to update or add to session header.
    :type key: str
    :param value: The value that corresponds to the key.
    :type value: str

    """
    SESSION.headers[key] = value


def set_login_state(logged_in):
    """ Sets the login state

    :param logged_in: Set to change value of global variable.
    :type logged_in: bool
    """
    global LOGGED_IN
    LOGGED_IN = logged_in


def get_login_state():
    """ Gets the login state
    """
    return LOGGED_IN


def login_required(func):
    """ A decorator for indicating which methods require the user to be logged in.
    """
    @wraps(func)
    def login_wrapper(*args, **kwargs):
        global LOGGED_IN
        if not LOGGED_IN:
            raise Exception('{} can only be called when logged in'.format(
                func.__name__))
        return(func(*args, **kwargs))
    return(login_wrapper)


[docs]def request_get(url, payload, parse_json): """ Generic function for sending a get request. :param url: The url to send a get request to. :type url: str :param payload: Dictionary of parameters to pass to the url. Will append the requests url as url/?key1=value1&key2=value2. :type payload: dict :param parse_json: Requests serializes data in the JSON format. Set this parameter true to parse the data to a dictionary \ using the JSON format. :type parse_json: bool :returns: Returns a tuple where the first entry is the response and the second entry will be an error message from the \ get request. If there was no error then the second entry in the tuple will be None. The first entry will either be \ the raw request response or the parsed JSON response based on whether parse_json is True or not. """ response_error = None try: response = SESSION.get(url, params=payload) response.raise_for_status() except Exception as e: response_error = e # Return either the raw request object so you can call response.text, response.status_code, response.headers, or response.json() # or return the JSON parsed information if you don't care to check the status codes. if parse_json: return response.json(), response_error else: return response, response_error
[docs]def request_post(url, payload, parse_json): """ Generic function for sending a post request. :param url: The url to send a post request to. :type url: str :param payload: Dictionary of parameters to pass to the url. Will append the requests url as url/?key1=value1&key2=value2. :type payload: dict :param parse_json: Requests serializes data in the JSON format. Set this parameter true to parse the data to a dictionary \ using the JSON format. :type parse_json: bool :returns: Returns a tuple where the first entry is the response and the second entry will be an error message from the \ get request. If there was no error then the second entry in the tuple will be None. The first entry will either be \ the raw request response or the parsed JSON response based on whether parse_json is True or not. """ response_error = None try: response = SESSION.post(url, params=payload) response.raise_for_status() except Exception as e: response_error = e # Return either the raw request object so you can call response.text, response.status_code, response.headers, or response.json() # or return the JSON parsed information if you don't care to check the status codes. if parse_json: return response.json(), response_error else: return response, response_error