Source code for fedex.base_service

"""
The L{base_service} module contains classes that form the low level foundations
of the Web Service API. Things that many different kinds of requests have in
common may be found here.

In particular, the L{FedexBaseService} class handles most of the basic,
repetitive setup work that most requests do.
"""

import os
import logging

import suds
from suds.client import Client
from suds.plugin import MessagePlugin


[docs]class GeneralSudsPlugin(MessagePlugin): """ General Suds Plugin: Adds logging request and response functionality and prunes empty WSDL objects before sending. """ def __init__(self, **kwargs): """Initializes the request and response loggers.""" self.request_logger = logging.getLogger('fedex.request') self.response_logger = logging.getLogger('fedex.response') self.kwargs = kwargs
[docs] def marshalled(self, context): """Removes the WSDL objects that do not have a value before sending.""" context.envelope = context.envelope.prune()
[docs] def sending(self, context): """Logs the sent request.""" self.request_logger.info("FedEx Request {}".format(context.envelope))
[docs] def received(self, context): """Logs the received response.""" self.response_logger.info("FedEx Response {}".format(context.reply))
[docs]class FedexBaseServiceException(Exception): """ Exception: Serves as the base exception that other service-related exception objects are sub-classed from. """ def __init__(self, error_code, value): self.error_code = error_code self.value = value def __unicode__(self): return "%s (Error code: %s)" % (repr(self.value), self.error_code) def __str__(self): return self.__unicode__()
[docs]class FedexFailure(FedexBaseServiceException): """ Exception: The request could not be handled at this time. This is generally a server problem. """ pass
[docs]class FedexError(FedexBaseServiceException): """ Exception: These are generally problems with the client-provided data. """ pass
class SchemaValidationError(FedexBaseServiceException): """ Exception: There is probably a problem in the data you provided. """ def __init__(self, fault): self.error_code = -1 self.value = "suds encountered an error validating your data against this service's WSDL schema. " \ "Please double-check for missing or invalid values, filling all required fields." try: self.value += ' Details: {}'.format(fault) except AttributeError: pass
[docs]class FedexBaseService(object): """ This class is the master class for all Fedex request objects. It gets all of the common SOAP objects created via suds and populates them with values from a L{FedexConfig} object, along with keyword arguments via L{__init__}. @note: This object should never be used directly, use one of the included sub-classes. """ def __init__(self, config_obj, wsdl_name, *args, **kwargs): """ This constructor should only be called by children of the class. As is such, only the optional keyword arguments caught by C{**kwargs} will be documented. @type customer_transaction_id: L{str} @keyword customer_transaction_id: A user-specified identifier to differentiate this transaction from others. This value will be returned with the response from Fedex. """ self.logger = logging.getLogger('fedex') """@ivar: Python logger instance with name 'fedex'.""" self.config_obj = config_obj """@ivar: The FedexConfig object to pull auth info from.""" if not self._version_info: self._version_info = {} """#ivar: Set in each service class. Holds version info for the VersionId SOAP object.""" # If the config object is set to use the test server, point # suds at the test server WSDL directory. if config_obj.use_test_server: self.logger.info("Using test server.") self.wsdl_path = os.path.join(config_obj.wsdl_path, 'test_server_wsdl', wsdl_name) else: self.logger.info("Using production server.") self.wsdl_path = os.path.join(config_obj.wsdl_path, wsdl_name) self.client = Client('file:///%s' % self.wsdl_path.lstrip('/'), plugins=[GeneralSudsPlugin()]) # self.client.options.cache.clear() # Clear the cache, then re-init client when changing wsdl file. self.VersionId = None """@ivar: Holds details on the version numbers of the WSDL.""" self.WebAuthenticationDetail = None """@ivar: WSDL object that holds authentication info.""" self.ClientDetail = None """@ivar: WSDL object that holds client account details.""" self.response = None """@ivar: The response from Fedex. You will want to pick what you want out here here. This object does have a __str__() method, you'll want to print or log it to see what possible values you can pull.""" self.TransactionDetail = None """@ivar: Holds customer-specified transaction IDs.""" self.__set_web_authentication_detail() self.__set_client_detail(*args, **kwargs) self.__set_version_id() self.__set_transaction_detail(*args, **kwargs) self._prepare_wsdl_objects() def __set_web_authentication_detail(self): """ Sets up the WebAuthenticationDetail node. This is required for all requests. """ # Start of the authentication stuff. web_authentication_credential = self.client.factory.create('WebAuthenticationCredential') web_authentication_credential.Key = self.config_obj.key web_authentication_credential.Password = self.config_obj.password # Encapsulates the auth credentials. web_authentication_detail = self.client.factory.create('WebAuthenticationDetail') web_authentication_detail.UserCredential = web_authentication_credential # Set Default ParentCredential if hasattr(web_authentication_detail, 'ParentCredential'): web_authentication_detail.ParentCredential = web_authentication_credential self.WebAuthenticationDetail = web_authentication_detail def __set_client_detail(self, *args, **kwargs): """ Sets up the ClientDetail node, which is required for all shipping related requests. """ client_detail = self.client.factory.create('ClientDetail') client_detail.AccountNumber = self.config_obj.account_number client_detail.MeterNumber = self.config_obj.meter_number client_detail.IntegratorId = self.config_obj.integrator_id if hasattr(client_detail, 'Region'): client_detail.Region = self.config_obj.express_region_code client_language_code = kwargs.get('client_language_code', None) client_locale_code = kwargs.get('client_locale_code', None) if hasattr(client_detail, 'Localization') and (client_language_code or client_locale_code): localization = self.client.factory.create('Localization') if client_language_code: localization.LanguageCode = client_language_code if client_locale_code: localization.LocaleCode = client_locale_code client_detail.Localization = localization self.ClientDetail = client_detail def __set_transaction_detail(self, *args, **kwargs): """ Checks kwargs for 'customer_transaction_id' and sets it if present. """ customer_transaction_id = kwargs.get('customer_transaction_id', None) if customer_transaction_id: transaction_detail = self.client.factory.create('TransactionDetail') transaction_detail.CustomerTransactionId = customer_transaction_id self.logger.debug(transaction_detail) self.TransactionDetail = transaction_detail def __set_version_id(self): """ Pulles the versioning info for the request from the child request. """ version_id = self.client.factory.create('VersionId') version_id.ServiceId = self._version_info['service_id'] version_id.Major = self._version_info['major'] version_id.Intermediate = self._version_info['intermediate'] version_id.Minor = self._version_info['minor'] self.logger.debug(version_id) self.VersionId = version_id def _prepare_wsdl_objects(self): """ This method should be over-ridden on each sub-class. It instantiates any of the required WSDL objects so the user can just print their __str__() methods and see what they need to fill in. """ pass def __check_response_for_fedex_error(self): """ This checks the response for general Fedex errors that aren't related to any one WSDL. """ if self.response.HighestSeverity == "FAILURE": for notification in self.response.Notifications: if notification.Severity == "FAILURE": raise FedexFailure(notification.Code, notification.Message) def _check_response_for_request_errors(self): """ Override this in each service module to check for errors that are specific to that module. For example, invalid tracking numbers in a Tracking request. """ if self.response.HighestSeverity == "ERROR": for notification in self.response.Notifications: if notification.Severity == "ERROR": raise FedexError(notification.Code, notification.Message) def _check_response_for_request_warnings(self): """ Override this in a service module to check for errors that are specific to that module. For example, changing state/province based on postal code in a Rate Service request. """ if self.response.HighestSeverity in ("NOTE", "WARNING"): for notification in self.response.Notifications: if notification.Severity in ("NOTE", "WARNING"): self.logger.warning(FedexFailure(notification.Code, notification.Message))
[docs] def create_wsdl_object_of_type(self, type_name): """ Creates and returns a WSDL object of the specified type. :param type_name: specifies the object's type name from WSDL. """ return self.client.factory.create(type_name)
def _assemble_and_send_request(self): """ This method should be over-ridden on each sub-class. It assembles all required objects into the specific request object and calls send_request. Objects that are not set will be pruned before sending via GeneralSudsPlugin marshalled function. """ pass
[docs] def send_request(self, send_function=None): """ Sends the assembled request on the child object. @type send_function: function reference @keyword send_function: A function reference (passed without the parenthesis) to a function that will send the request. This allows for overriding the default function in cases such as validation requests. """ # Send the request and get the response back. try: # If the user has overridden the send function, use theirs # instead of the default. if send_function: # Follow the overridden function. self.response = send_function() else: # Default scenario, business as usual. self.response = self._assemble_and_send_request() except suds.WebFault as fault: # When this happens, throw an informative message reminding the # user to check all required variables, making sure they are # populated and valid raise SchemaValidationError(fault.fault) # Check the response for general Fedex errors/failures that aren't # specific to any given WSDL/request. self.__check_response_for_fedex_error() # Check the response for errors specific to the particular request. # This method can be overridden by a method on the child class object. self._check_response_for_request_errors() # Check the response for errors specific to the particular request. # This method can be overridden by a method on the child class object. self._check_response_for_request_warnings() # Debug output. (See Request and Response output) self.logger.debug("== FEDEX QUERY RESULT ==") self.logger.debug(self.response)