Source code for mws.mws

# -*- coding: utf-8 -*-
"""Main module for python-amazon-mws package."""

import base64
import hashlib
import hmac
import warnings
from enum import Enum
from urllib.parse import quote

from requests import request
from requests.exceptions import HTTPError

from mws.errors import MWSError, MWSRequestError
from mws.response import MWSResponse
from mws.utils.crypto import response_md5_is_valid
from mws.utils.params import (
from mws.utils.timezone import mws_utc_now

__version__ = "1.0dev16"
PAM_USER_AGENT = f"python-amazon-mws/{__version__} (Language=Python)"
"""See recommended user agent string format:


__all__ = [

class Marketplaces(Enum):
    """Enumeration for MWS marketplaces, containing endpoints and marketplace IDs.

    Example, endpoint and ID for UK marketplace:
        endpoint = Marketplaces.UK.endpoint
        marketplace_id = Marketplaces.UK.marketplace_id

    AE = ("", "A2VIGQ35RCS4UG")
    AU = ("", "A39IBJ37TRP1C6")
    BR = ("", "A2Q3Y263D00KWC")
    CA = ("", "A2EUQ1WTGCTBG2")
    DE = ("", "A1PA6795UKMFR9")
    EG = ("", "ARBP9OOSHTCHU")
    ES = ("", "A1RKKUPIHCS9HS")
    FR = ("", "A13V1IB3VIYZZH")
    GB = ("", "A1F83G8C2ARO7P")
    IN = ("", "A21TJRUUN4KGV")
    IT = ("", "APJ6JRA9NG5V4")
    JP = ("", "A1VC38T7YXB528")
    MX = ("", "A1AM78C64UM0Y8")
    NL = ("", "A1805IZSGTT6HS")
    SA = ("", "A17E79C6D8DWNP")
    SE = ("", "A2NODRKZP88ZB9")
    SG = ("", "A19VAU5U5O7RUS")
    TR = ("", "A33AVAJ2PDY3EV")
    UK = ("", "A1F83G8C2ARO7P")  # alias for GB
    US = ("", "ATVPDKIKX0DER")
    PL = ("", "A1C3SOZRARQ6R3")
    BE = ("", "AMEN7PMS3EDWL")

    def __init__(self, endpoint, marketplace_id):
        """Easy dot access like: Marketplaces.endpoint ."""
        self.endpoint = endpoint
        self.marketplace_id = marketplace_id

    def value(self):
        return self.marketplace_id

def canonicalized_query_string(params):
    """Builds the canonicalized query string from the set of params,
    according to `Creating a Canonicalized Query String

    Each key-value pair takes the form "key=value"
    Sets of "key=value" pairs are joined by "&".
    Keys should appear in alphabetical order in the result string.

      params = {'foo': 1, 'bar': 4, 'baz': 'potato'}
    description_items = []
    encoded_params = clean_params_dict(params, urlencode=True)
    for item in sorted(encoded_params.keys()):
        encoded_val = encoded_params[item]
    return "&".join(description_items)

class MWS(object):
    """Base Amazon API class.


    A new parser, ``mws.utils.parsers.MWSResponse``, can be used to process requests
    made to MWS. To test the new parser:

    .. code-block:: python

        # instantiate your API class as usual
        api = Products(access_key, secret_key, account_id, ...)
        # Then set the `_use_feature_mwsresponse` flag on the instance.
        api._use_feature_mwsresponse = True

    Now all requests you make with this instance of the API class
    will run through ``MWSResponse``.

    # This is used to post/get to the different uris used by amazon per api
    # ie. /Orders/2011-01-01
    # All subclasses must define their own URI only if needed
    URI = "/"

    # The API version varies in most amazon APIs
    VERSION = "2009-01-01"

    # There seem to be some xml namespace issues. therefore every api subclass
    # is recommended to define its namespace, so that it can be referenced
    # like so AmazonAPISubclass.NAMESPACE.
    # For more information see
    NAMESPACE = ""

    # In here we name each of the operations available to the subclass
    # that have 'ByNextToken' operations associated with them.
    # If the Operation is not listed here, self.action_by_next_token
    # will raise an error.

    # Some APIs are available only to either a "Merchant" or "Seller"
    # the type of account needs to be sent in every call to the amazon MWS.
    # This constant defines the exact name of the parameter Amazon expects
    # for the specific API being used.
    # All subclasses need to define this if they require another account type
    # like "Merchant" in which case you define it like so.
    # ACCOUNT_TYPE = "Merchant"
    # Which is the name of the parameter for that specific account type.

    # For using proxy you need to init this class with one more parameter proxies. It must look like 'ip_address:port'
    # if proxy without auth and 'login:password@ip_address:port' if proxy with auth

    ACCOUNT_TYPE = "SellerId"

    def __init__(  # nosec No password default is provided, only auth_token empty value (where it may not be needed)
        self.access_key = access_key
        self.secret_key = secret_key
        self.account_id = account_id
        self.auth_token = auth_token
        self.version = version or self.VERSION
        self.uri = uri or self.URI
        self.proxy = proxy
        self.user_agent_str = user_agent_str or PAM_USER_AGENT
        self.extra_headers = headers or {}
        self.force_response_encoding = force_response_encoding

        # * TESTING FLAGS * #
        self._test_request_params = False
        self._use_feature_mwsresponse = False

        if region in Marketplaces.__members__:
            self.domain = Marketplaces[region].endpoint
            regions = ", ".join(Marketplaces.__members__.keys())
            error_msg = (
                f"Incorrect region supplied: {region}. "
                f"Must be one of the following: {regions}"
            raise ValueError(error_msg)

    def get_default_params(self, action, timestamp):
        """Get the params required in all MWS requests."""
        params = {
            "Action": action,
            "AWSAccessKeyId": self.access_key,
            self.ACCOUNT_TYPE: self.account_id,
            "SignatureVersion": "2",
            "Timestamp": timestamp,
            "Version": self.version,
            "SignatureMethod": "HmacSHA256",
        if self.auth_token:
            params["MWSAuthToken"] = self.auth_token
        # TODO current tests only check for auth_token being set.
        # need a branch test to check for auth_token being skipped (no key present)
        return params

    def make_request(
        """Make request to Amazon MWS API with these params.

        ``action`` is a string matching the name of the request action
        (i.e. "ListOrders").

        ``params`` is a flat dict containing params to pass to the operation.

        ``method`` is a string, matching an HTTP verb ("GET", "POST", etc.),
        which sets the method for a `requests.request` call.

        ``timeout`` passes to `requests.request`, setting the timeout for this request.

        ``kwargs`` may include:

        - `result_key`, providing a custom key to use as the root for results
          returned by `response.parsed`.
        - `body`, primarily used in Feeds requests to send a data file in the request.
        params = params or {}

        request_timestamp = mws_utc_now()
        request_params = self.get_default_params(action, request_timestamp)
        proxies = self.get_proxies()

        # Remove empty keys and clean values before transmitting
        request_params = remove_empty_param_keys(request_params)
        request_params = clean_params_dict(request_params)
        # Create a canonical query string
        canonical_query = canonicalized_query_string(request_params)
        if self._test_request_params:
            # For tests: return the params from this request before the request is made.
            return request_params
        # TODO: All current testing stops here. More branches needed.

        # Sign the request using the canonical string.
        signature = self.calc_signature(method, canonical_query)
        request_params["Signature"] = signature
        headers = {"User-Agent": self.user_agent_str}
        headers.update(kwargs.get("extra_headers", {}))

        result_key = kwargs.get("result_key", f"{action}Result")

        request_args = {
            "method": method,
            "url": self.endpoint,
            "headers": headers,
            "proxies": proxies,
            "timeout": timeout,
        body = kwargs.get("body")
        if body:
            # Force POST method: no other choice in this case.
            method = "POST"
            # Typically for a SubmitFeed operation, our data is in the body,
            # and other params need to be set in query parameters.
            request_args["data"] = body
            request_args["params"] = request_params
        elif method == "POST":
            request_args["data"] = request_params
            request_args["params"] = request_params

            response = request(**request_args)
            # When retrieving data from the response object,
            # be aware that response.content returns the content in bytes while response.text calls
            # response.content and converts it to unicode.

            if self._use_feature_mwsresponse:
                # Turn on the new response parser and DotDict parsed output
                # (will be made standard in v1.0)
                if not response_md5_is_valid(response):
                    raise ValueError(
                        "MD5 hash validation failed: wrong content length for response"

                parsed_response = MWSResponse(
                parsed_response.timestamp = request_timestamp
                ### DEPRECATED ###
                # Remove in v1.0
                from defusedxml.ElementTree import ParseError as XMLError

                from mws.utils.parsers import DataWrapper, DictWrapper

                data = response.content
                        parsed_response = DictWrapper(data, result_key)
                    except TypeError:
                        # When we got CSV as result, we will got error on this
                        parsed_response = DictWrapper(response.text, result_key)

                except XMLError:
                    parsed_response = DataWrapper(data, response.headers)
                parsed_response.response = response

        except HTTPError as exc:
            raise MWSRequestError(exc)

        # Store the response object in the parsed_response for quick access
        return parsed_response

    def endpoint(self):
        return f"{self.domain}{self.uri}"

    def get_proxies(self):
        """Return a dict of http and https proxies, as defined by `self.proxy`."""
        proxies = {"http": None, "https": None}
        if self.proxy:
            # TODO need test to enter here
            proxies = {
                "http": f"http://{self.proxy}",
                "https": f"https://{self.proxy}",
        return proxies

    def get_service_status(self):
        """Returns MWS service status.

        Typical return values (embedded within `response.parsed`) are:

        - GREEN
        - GREEN_I
        - YELLOW
        - RED

        The same request can be used for any MWS API subclass, and MWS may respond
        differently for each endpoint. Best to use this method from the same API
        subclass you intend to use for other requests!

        Docs (from Orders API example):
        return self.make_request("GetServiceStatus")

    def action_by_next_token(self, action, next_token):
        """Run a '...ByNextToken' action for the given action.

        If the action is not listed in self.NEXT_TOKEN_OPERATIONS, MWSError is raised.
        Action is expected NOT to include 'ByNextToken'
        at the end of its name for this call: function will add that by itself.
        if action not in self.NEXT_TOKEN_OPERATIONS:
            # TODO Would like a test entering here.
            # Requires a dummy API class to be written that will trigger it.
            raise MWSError(
                f"{action} action not listed in this API's NEXT_TOKEN_OPERATIONS. "
                "Please refer to documentation."

        action = f"{action}ByNextToken"

        return self.make_request(action, {"NextToken": next_token})

    def calc_signature(self, method, canonical_query):
        """Calculate MWS signature to interface with Amazon

            method (str)
            canonical_query (str)
        sig_data = "\n".join(
                self.domain.replace("https://", "").lower(),
        return base64.b64encode(
                self.secret_key.encode(), sig_data.encode(), hashlib.sha256

    def enumerate_param(self, param, values):
        """DEPRECATED, alias for `utils.params.enumerate_param`."""
        from mws.utils.deprecation import RemovedInPAM10Warning

                "MWS.enumerate_param is deprecated. "
                "Please use methods in 'mws.utils.params', instead."
        return enumerate_param(param, values)

[docs] def generic_request( self, action, params=None, method="POST", timeout=PAM_DEFAULT_TIMEOUT, **kwargs, ): """Builds a generic request with arbitrary parameter arguments. This method should be called from an API subclass (``Orders``, ``Feeds``, etc.), else the ``uri`` attribute of the class instance must be set manually. This method's signature matches that of ``.make_request``, as the two methods are similar. However, ``params`` is expected to be either the default ``None`` or a nested dictionary, that is then passed to :py:func:`flat_param_dict() <mws.utils.params.flat_param_dict>`. """ # NOTE you may be asking why this method exists. Why not simply put the logic # of `flat_param_dict` into `make_request`, and let every request method # pass nested objects freely? # Well, I tried that. Turns out giving up that kind of control has some # unintended consequences. # For instance, say you know that a given parameter for your request only takes # one value, such as `ReportType=_SOME_REPORT_TYPE_`. # If the user passes a list wrapping that string, `flat_param_dict` will # happily enumerate that value: `ReportType.1=_SOME_REPORT_TYPE_`. # For that particular request, we would know this to be an error: MWS will # not accept that entry. Surfacing that error to the end user would be quite # difficult, and trying to prevent it would introduce the hassle of verifying # argument types in every request method. # It is better to allow that list object to pass through the params to # `make_request`, where it can raise an error in our code, rather than have # the request get sent to MWS. if not self.uri or self.uri == "/": raise ValueError( ( "Cannot send generic request to URI '%s'. " "Please use one of the API classes " "(`mws.apis.Reports`, `mws.apis.Feeds`, etc.) " "to initiate this request." ) % self.uri ) if not isinstance(params, dict): raise ValueError("`params` must be a dict.") data = flat_param_dict(params) return self.make_request( action=action, params=data, method=method, timeout=timeout, **kwargs )