Custom Exceptions

Custom exceptions for Django models

Define Custom Exceptions

# errors.py

class Error(Exception):
    pass
    

class ExceedsLimit(Error):
    pass
    
class InvalidAmount(Error):
    def __init__(self, amount):
        self.amount = amount
    
    def __str__(self):
        return f'Invalid Amount: {self.amount}'


class InsufficientFunds(Error):
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        
    def __str__(self):
        return f'amount={self.amount}, current balance: {self.balance}'

We define a base Error class the inherits from Exception. This is something we found very useful and we use it a lot. A base error class allows us to catch all errors coming from a certain module:

from account.errors import Error as AccountError

try:
    # action on account
except AccountError:
    # Handle all errors from account

Handling Errors

Define a custom errors to catch both transport and remote application errors. There are two types of errors we need to handle"

• HTTP errors such as connection errors, timeout or connection refused.

• Remote payment errors such as refusal or stolen card.

# errors.py

class Error(Exception):
    pass
    

class Unavailable(Error):
    pass 
    
    
class PaymentGatewayError(Error):
    def __init__(self, code, message):
        self.code = code
        self.message = message

class Refused(PaymentGatewayError):
    pass
    
class Stolen(PaymentGatewayError):
    pass

import uuid
import logging
import requests
import http.cookiejar

from collections import namedtuple
from datetime import datetime


logger = logging.getLogger('payments')


ChargeResponse = namedtuple('ChargeResponse', [
    'uid',
    'amount',
    'token',
    'expiration',
    'transaction_id',
])


RefundResponse = namedtuple('RefundResponse', [
    'transaction_id',
    'refunded_transaction_id',
])


class BlockAll(http.cookiejar.CookiePolicy):
    def set_ok(self, cookie, request):
        return False

payment_session = requests.Session()
payment_session.cookies.policy = BlockAll()


def make_payment_request(path, payload, timeout=5):
    """Make a request to the payment gateway.

    path (str):
        Path to post to.
    payload (object):
        JSON-serializable request payload.
    timeout (int):
        Timeout in seconds.

    Raises
        Unavailable
        requests.exceptions.HTTPError

    Returns (response)
    """
    headers = {
        "Authorization": "Bearer " + PAYMENT_GATEWAY_TOKEN,
    }

    try:
        response = payment_session.post(
            PAYMENT_GATEWAY_BASE_URL + path,
            json=payload,
            headers=headers,
            timeout=timeout,
        )
    except (requests.ConnectionError, requests.Timeout) as e:
        raise errors.Unavailable() from e

    response.raise_for_status()
    return response.json()


def charge(amount, token):
    """Charge credit card.

    amount (int):
        Amount to charge in cents.
    token (str):
        Credit card token.

    Raises
        Unavailable
        Refused
        Stolen
        PaymentGatewayError

    Returns (ChargeResponse)
    """
    try:
        data = make_payment_request('/charge', {
            'uid': str(uuid.uuid4()),
            'amount': amount,
            'token': token,
        })

    except requests.HTTPError as e:
        if e.response.status_code == 400:
            error = e.response.json()
            code = error['code']
            message = error['message']

            if code == 1:
                raise Refused(code, message) from e

            elif code == 2:
                raise Stolen(code, message) from e

            else:
                raise PaymentGatewayError(code, message) from e

        logger.exception("Payment service had internal error")
        raise errors.Unavailable() from e

    return ChargeResponse(
        uid=uuid.UID(data['uid']),
        amount=data['amount'],
        token=data['token'],
        expiration=datetime.strptime(data['expiration'], "%Y-%m-%dT%H:%M:%S.%f"),
        transaction_id=data['transaction_id'],
    )
    
    
def refund(transaction_id):
    """Refund charged transaction.

    transaction_id (str):
        Transaction id to refund.

    Raises:

    Return (RefundResponse)
    """
    try:
        data = make_payment_request('/refund', {
            'uid': str(uuid.uuid4()),
            'transaction_id': transaction_id,
        })

    except requests.HTTPError as e:
        # TODO: Handle refund remote errors

    return RefundResponse(
        'transaction_id': data['transaction_id'],
        'refunded_transaction_id': data['refunded_transaction_id'],
    )