Changeset View
Changeset View
Standalone View
Standalone View
looper/gateways.py
| import abc | import abc | ||||
| import enum | import enum | ||||
| import logging | import logging | ||||
| from typing import ( | from typing import ( | ||||
| Any, | Any, | ||||
| Dict, | Dict, | ||||
| Iterable, | Iterable, | ||||
| Mapping, | |||||
| Optional, | Optional, | ||||
| Set, | Set, | ||||
| Union, | Union, | ||||
| cast, | cast, | ||||
| ) | ) | ||||
| import attr | import attr | ||||
| import braintree | import braintree | ||||
| ▲ Show 20 Lines • Show All 192 Lines • ▼ Show 20 Lines | class AbstractPaymentGateway(metaclass=abc.ABCMeta): | ||||
| gateway_name: str = '' # set in each subclass. | gateway_name: str = '' # set in each subclass. | ||||
| supported_collection_methods: Set[str] = set() | supported_collection_methods: Set[str] = set() | ||||
| supports_refunds = True | supports_refunds = True | ||||
| supports_transactions = True | supports_transactions = True | ||||
| _log = log.getChild('AbstractPaymentGateway') | _log = log.getChild('AbstractPaymentGateway') | ||||
| def __init__(self) -> None: | def __init__(self) -> None: | ||||
| gateway_settings = settings.GATEWAYS.get(self.gateway_name) | gateway_settings: Optional[object] = settings.GATEWAYS.get(self.gateway_name) | ||||
| if gateway_settings is None: | if gateway_settings is None: | ||||
| raise looper.exceptions.GatewayConfigurationMissing( | raise looper.exceptions.GatewayConfigurationMissing( | ||||
| f'Missing settings for Gateway {self.gateway_name!r}' | f'Missing settings for Gateway {self.gateway_name!r}' | ||||
| ) | ) | ||||
| if not isinstance(gateway_settings, dict): | if not isinstance(gateway_settings, dict): | ||||
| raise looper.exceptions.GatewayConfigurationError( | raise looper.exceptions.GatewayConfigurationError( | ||||
| f'Expected a dict for Gateway {self.gateway_name}, found: {gateway_settings}' | f'Expected a dict for Gateway {self.gateway_name}, found: {gateway_settings}' | ||||
| ▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Lines | def transact_sale(self, payment_method_token: str, amount: Money) -> str: | ||||
| :raise looper.exceptions.CurrencyNotSupported: when the amount has a | :raise looper.exceptions.CurrencyNotSupported: when the amount has a | ||||
| currency for which there is no configured Merchant Account ID. | currency for which there is no configured Merchant Account ID. | ||||
| :raise looper.exceptions.GatewayError: when there was an error | :raise looper.exceptions.GatewayError: when there was an error | ||||
| performing the transaction at the payment gateway. | performing the transaction at the payment gateway. | ||||
| """ | """ | ||||
| pass | pass | ||||
| @abc.abstractmethod | @abc.abstractmethod | ||||
| def find_transaction(self, transaction_id) -> dict: | def find_transaction(self, transaction_id: str) -> Dict[str, object]: | ||||
| """Lookup a transaction and return it as a dictionary.""" | """Lookup a transaction and return it as a dictionary.""" | ||||
| pass | pass | ||||
| @abc.abstractmethod | @abc.abstractmethod | ||||
| def refund(self, gateway_transaction_id: str, amount: Money): | def refund(self, gateway_transaction_id: str, amount: Money) -> None: | ||||
| """Refund a transaction. | """Refund a transaction. | ||||
| :param: gateway_transaction_id: the transaction ID on the gateway. | :param: gateway_transaction_id: the transaction ID on the gateway. | ||||
| :param amount: The amount to refund, must not be more than the original | :param amount: The amount to refund, must not be more than the original | ||||
| transaction amount. | transaction amount. | ||||
| Note that the amount to refund is non-optional, contrary to, for | Note that the amount to refund is non-optional, contrary to, for | ||||
| example, the Braintree API. This is to make the amount explicit. | example, the Braintree API. This is to make the amount explicit. | ||||
| Show All 12 Lines | class MockableGateway(AbstractPaymentGateway): | ||||
| and unexpected calls to non-mocked functions to be detected. | and unexpected calls to non-mocked functions to be detected. | ||||
| """ | """ | ||||
| gateway_name: str = 'mock' | gateway_name: str = 'mock' | ||||
| supported_collection_methods = {'automatic'} | supported_collection_methods = {'automatic'} | ||||
| _log = log.getChild('MockableGateway') | _log = log.getChild('MockableGateway') | ||||
| # noinspection PyMissingConstructor | # noinspection PyMissingConstructor | ||||
| def __init__(self): | def __init__(self) -> None: | ||||
| # Dont' call the superclass, it'll expect configuration which we don't need nor want. | # Dont' call the superclass, it'll expect configuration which we don't need nor want. | ||||
| self.settings = {} | self.settings: Dict[str, object] = {} | ||||
| def generate_client_token( | def generate_client_token( | ||||
| self, for_currency: str, gateway_customer_id: Optional[str] = None | self, for_currency: str, gateway_customer_id: Optional[str] = None | ||||
| ) -> str: | ) -> str: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to MockableGateway.generate_client_token({gateway_customer_id!r})' | f'Unexpected call to MockableGateway.generate_client_token({gateway_customer_id!r})' | ||||
| ) | ) | ||||
| Show All 23 Lines | def find_customer_payment_method(self, payment_method_token: str) -> PaymentMethodInfo: | ||||
| ) | ) | ||||
| def transact_sale(self, payment_method_token: str, amount: Money) -> str: | def transact_sale(self, payment_method_token: str, amount: Money) -> str: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to MockableGateway.transact_sale' | f'Unexpected call to MockableGateway.transact_sale' | ||||
| f'({payment_method_token!r}, {amount!r})' | f'({payment_method_token!r}, {amount!r})' | ||||
| ) | ) | ||||
| def find_transaction(self, transaction_id) -> dict: | def find_transaction(self, transaction_id: str) -> Dict[str, object]: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to MockableGateway.find_transaction({transaction_id!r})' | f'Unexpected call to MockableGateway.find_transaction({transaction_id!r})' | ||||
| ) | ) | ||||
| def refund(self, gateway_transaction_id: str, amount: Money): | def refund(self, gateway_transaction_id: str, amount: Money) -> None: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to MockableGateway.refund({gateway_transaction_id!r}, {amount})' | f'Unexpected call to MockableGateway.refund({gateway_transaction_id!r}, {amount})' | ||||
| ) | ) | ||||
| class BraintreeGateway(AbstractPaymentGateway): | class BraintreeGateway(AbstractPaymentGateway): | ||||
| gateway_name = 'braintree' | gateway_name = 'braintree' | ||||
| supported_collection_methods = {'automatic'} | supported_collection_methods = {'automatic'} | ||||
| TRANSACTION_SUCCESS_STATUSES = [ | TRANSACTION_SUCCESS_STATUSES = [ | ||||
| braintree.Transaction.Status.Authorized, | braintree.Transaction.Status.Authorized, | ||||
| braintree.Transaction.Status.Authorizing, | braintree.Transaction.Status.Authorizing, | ||||
| braintree.Transaction.Status.Settled, | braintree.Transaction.Status.Settled, | ||||
| braintree.Transaction.Status.SettlementConfirmed, | braintree.Transaction.Status.SettlementConfirmed, | ||||
| braintree.Transaction.Status.SettlementPending, | braintree.Transaction.Status.SettlementPending, | ||||
| braintree.Transaction.Status.Settling, | braintree.Transaction.Status.Settling, | ||||
| braintree.Transaction.Status.SubmittedForSettlement, | braintree.Transaction.Status.SubmittedForSettlement, | ||||
| ] | ] | ||||
| def __init__(self): | def __init__(self) -> None: | ||||
| super().__init__() | super().__init__() | ||||
| self.braintree = braintree.BraintreeGateway( | self.braintree = braintree.BraintreeGateway( | ||||
| braintree.Configuration( | braintree.Configuration( | ||||
| environment=self.settings['environment'], | environment=self.settings['environment'], | ||||
| merchant_id=self.settings['merchant_id'], | merchant_id=self.settings['merchant_id'], | ||||
| public_key=self.settings['public_key'], | public_key=self.settings['public_key'], | ||||
| private_key=self.settings['private_key'], | private_key=self.settings['private_key'], | ||||
| ) | ) | ||||
| Show All 34 Lines | ) -> str: | ||||
| """ | """ | ||||
| params = {} | params = {} | ||||
| if gateway_customer_id: | if gateway_customer_id: | ||||
| params['customer_id'] = gateway_customer_id | params['customer_id'] = gateway_customer_id | ||||
| if for_currency: | if for_currency: | ||||
| merchant_account_id = self._merchant_account_id(for_currency) | merchant_account_id = self._merchant_account_id(for_currency) | ||||
| params['merchant_account_id'] = merchant_account_id | params['merchant_account_id'] = merchant_account_id | ||||
| return self.braintree.client_token.generate(params) | |||||
| client_token = self.braintree.client_token.generate(params) | |||||
| assert isinstance(client_token, str) | |||||
| return client_token | |||||
| def customer_create(self) -> str: | def customer_create(self) -> str: | ||||
| """Create a customer at BrainTree. | """Create a customer at BrainTree. | ||||
| See https://developers.braintreepayments.com/reference/request/customer/create/python | See https://developers.braintreepayments.com/reference/request/customer/create/python | ||||
| """ | """ | ||||
| self._log.debug('Creating new customer at Braintree') | self._log.debug('Creating new customer at Braintree') | ||||
| result = self.braintree.customer.create({}) | result = self.braintree.customer.create({}) | ||||
| if not result.is_success: | if not result.is_success: | ||||
| self._log.debug( | self._log.debug( | ||||
| 'Error creating new customer: message=%s errors=%s', result.message, result.errors | 'Error creating new customer: message=%s errors=%s', result.message, result.errors | ||||
| ) | ) | ||||
| raise BraintreeError(result) | raise BraintreeError(result) | ||||
| return result.customer.id | customer_id = result.customer.id | ||||
| assert isinstance(customer_id, str) | |||||
| return customer_id | |||||
| def payment_method_create( | def payment_method_create( | ||||
| self, | self, | ||||
| payment_method_nonce: str, | payment_method_nonce: str, | ||||
| gateway_customer_id: str, | gateway_customer_id: str, | ||||
| verification_data: Optional[Any] = None, | verification_data: Optional[Any] = None, | ||||
| ) -> PaymentMethodInfo: | ) -> PaymentMethodInfo: | ||||
| """Create a payment method at BrainTree. | """Create a payment method at BrainTree. | ||||
| ▲ Show 20 Lines • Show All 90 Lines • ▼ Show 20 Lines | def transact_sale(self, payment_method_token: str, amount: Money) -> str: | ||||
| raise BraintreeError(result) | raise BraintreeError(result) | ||||
| self._log.info( | self._log.info( | ||||
| 'Transaction %r for amount %s was successful with status=%r', | 'Transaction %r for amount %s was successful with status=%r', | ||||
| result.transaction.id, | result.transaction.id, | ||||
| amount, | amount, | ||||
| result.transaction.status, | result.transaction.status, | ||||
| ) | ) | ||||
| return result.transaction.id | transaction_id = result.transaction.id | ||||
| assert isinstance(transaction_id, str) | |||||
| return transaction_id | |||||
| def find_transaction(self, transaction_id): | def find_transaction(self, transaction_id: str) -> Dict[str, object]: | ||||
| transaction = self.braintree.transaction.find(transaction_id) | transaction = self.braintree.transaction.find(transaction_id) | ||||
| return {'transaction_id': transaction.id} | return {'transaction_id': transaction.id} | ||||
| def refund(self, gateway_transaction_id: str, amount: Money): | def refund(self, gateway_transaction_id: str, amount: Money) -> None: | ||||
| """Refund a transaction. | """Refund a transaction. | ||||
| :param: gateway_transaction_id: the transaction ID on the gateway. | :param: gateway_transaction_id: the transaction ID on the gateway. | ||||
| :param amount: The amount to refund, may not be more than the original | :param amount: The amount to refund, may not be more than the original | ||||
| transaction amount. | transaction amount. | ||||
| See https://developers.braintreepayments.com/reference/request/transaction/refund/python | See https://developers.braintreepayments.com/reference/request/transaction/refund/python | ||||
| """ | """ | ||||
| ▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines | def find_customer_payment_method(self, payment_method_token: str) -> PaymentMethodInfo: | ||||
| return PaymentMethodInfo('bank', PaymentMethodInfo.Type.BANK_ACCOUNT) | return PaymentMethodInfo('bank', PaymentMethodInfo.Type.BANK_ACCOUNT) | ||||
| def transact_sale(self, payment_method_token: str, amount: Money) -> str: | def transact_sale(self, payment_method_token: str, amount: Money) -> str: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to BankGateway.transact_sale' | f'Unexpected call to BankGateway.transact_sale' | ||||
| f'({payment_method_token!r}, {amount!r})' | f'({payment_method_token!r}, {amount!r})' | ||||
| ) | ) | ||||
| def find_transaction(self, transaction_id) -> dict: | def find_transaction(self, transaction_id: str) -> Dict[str, object]: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to BankGateway.find_transaction({transaction_id!r})' | f'Unexpected call to BankGateway.find_transaction({transaction_id!r})' | ||||
| ) | ) | ||||
| def refund(self, gateway_transaction_id: str, amount: Money): | def refund(self, gateway_transaction_id: str, amount: Money) -> None: | ||||
| raise NotImplementedError( | raise NotImplementedError( | ||||
| f'Unexpected call to BankGateway.refund({gateway_transaction_id!r}, {amount})' | f'Unexpected call to BankGateway.refund({gateway_transaction_id!r}, {amount})' | ||||
| ) | ) | ||||