document.erp5.StripeConnector.py 4.18 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59
##############################################################################
#
# Copyright (c) 2022 Nexedi SARL and Contributors. All Rights Reserved.
#                    Gabriel Monnerat <gabriel@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
#
##############################################################################

from six.moves import urllib

import requests
import six

from AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject

TIMEOUT = 10 # 10 seconds

class StripeConnector(XMLObject):
  """
    Holds an Stripe connection to a remote Stripe API.
  """
  # CMF Type Definition
  meta_type = 'Stripe Connector'
  portal_type = 'Stripe Connector'

  # Declarative security
  security = ClassSecurityInfo()
  security.declareObjectProtected(Permissions.AccessContentsInformation)

  # Default Properties
  property_sheets = (
    PropertySheet.Base,
    PropertySheet.XMLObject,
    PropertySheet.CategoryCore,
    PropertySheet.DublinCore
  )

60 61 62 63 64 65 66 67
  def serializeSessionParameter(self, request_data, key, value):
    if isinstance(value, list) or isinstance(value, dict):
      iterator = six.iteritems(value) if isinstance(value, dict) else enumerate(value)
      for subkey, subvalue in iterator:
        self.serializeSessionParameter(
          request_data,
          "{}[{}]".format(key, subkey),
          subvalue)
68
    else:
69
      request_data[key] = value
70 71 72 73 74 75 76 77 78 79 80 81 82 83 84

  def createSession(self, data, **kw):
    """
      Create Session in Stripe using Stripe API and return a checkout.session

      data is a dict, see https://stripe.com/docs/api/checkout/sessions/create
    """
    end_point = "checkout/sessions"
    header_dict = {
      'Content-Type': 'application/x-www-form-urlencoded',
    }
    url_string = self.getUrlString()
    if not url_string.endswith("/"):
      url_string += "/"
    url_string += end_point
85 86 87 88 89 90 91
    request_data = {}
    for key, value in six.iteritems(data):
      self.serializeSessionParameter(request_data, key, value)

    if "mode" not in request_data:
      request_data["mode"] = "payment"

92 93 94 95 96 97 98
    response = requests.post(
      url_string,
      headers=header_dict,
      data=urllib.parse.urlencode(request_data),
      auth=((self.getPassword() or "").strip(), ''),
      timeout=self.getTimeout() or TIMEOUT)
    if not response.ok:
99
      __traceback_info__ = (
100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119
        response.request.url,
        response.request.body,
        response.text
      )
      response.raise_for_status()
    return response.json()

  def retrieveSession(self, session_id, **kw):
    """
      Retrieve Session in Stripe using Stripe API and return a checkout.session
    """
    url_string = self.getUrlString()
    if not url_string.endswith("/"):
      url_string += "/"
    url_string += "checkout/sessions/%(session_id)s" % {"session_id": session_id}
    response = requests.get(
      url_string,
      auth=((self.getPassword() or "").strip(), ''),
      timeout=self.getTimeout() or TIMEOUT)
    if not response.ok:
120
      __traceback_info__ = (
121 122 123 124 125 126
        response.request.url,
        response.request.body,
        response.text
      )
      response.raise_for_status()
    return response.json()