Python >> Tutoriel Python >  >> Python

Travailler avec les API à la manière Pythonique

La communication avec des services externes fait partie intégrante de tout système moderne. Qu'il s'agisse d'un service de paiement, d'authentification, d'analyse ou d'un service interne - les systèmes doivent communiquer entre eux .

Dans ce court article, nous allons implémenter un module pour communiquer avec une passerelle de paiement inventée, étape par étape.

Le service externe

Commençons par définir un service de paiement imaginaire.

Pour débiter une carte de crédit, nous avons besoin d'un jeton de carte de crédit, d'un montant à débiter (en centimes) et d'un identifiant unique fourni par le client (nous) :

POST
{
 token: <string>,
 amount: <number>,
 uid: <string>,
}

Si la charge a réussi, nous obtenons un statut 200 OK avec les données de notre demande, un délai d'expiration pour la charge et un ID de transaction :

200 OK
{
 uid: <string>,
 amount: <number>,
 token: <string>,
 expiration: <string, isoformat>,
 transaction_id: <number>
}

Si la charge a échoué, nous obtenons un statut 400 avec un code d'erreur et un message informatif :

400 Bad Request
{
 uid: <string>,
 error: <number>,
 message: <string>
}

Il y a deux codes d'erreur que nous voulons gérer - 1 =refusé et 2 =volé.

Implémentation naïve

Pour démarrer, nous commençons par une implémentation naïve et construisons à partir de là :

# payments.py

import uuid
import requests

PAYMENT_GATEWAY_BASE_URL = 'https://gw.com/api'
PAYMENT_GATEWAY_TOKEN = 'topsecret'

def charge(
 amount,
 token,
 timeout=5,
):
 """Charge.

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

 Returns (dict):
 New payment information.
 """
 headers = {
 "Authorization": "Bearer " + PAYMENT_GATEWAY_TOKEN,
 }

 payload = {
 "token": token,
 "amount": amount,
 "uid": str(uuid.uuid4()),
 }

 response = requests.post(
 PAYMENT_GATEWAY_BASE_URL + '/charge',
 json=payload,
 headers=headers,
 timeout=timeout,
 )
 response.raise_for_status()

return response.json()

90 % des développeurs s'arrêteront là, alors quel est le problème ?

Gestion des erreurs

Nous devons gérer deux types d'erreurs :

  • Erreurs HTTP telles que les erreurs de connexion, le délai d'attente ou la connexion refusée.
  • Erreurs de paiement à distance telles que refus ou carte volée.

Notre décision d'utiliser requests est un détail d'implémentation interne. Le consommateur de notre module ne devrait pas avoir à en être conscient.

Pour fournir une API complète, notre module doit communiquer les erreurs.

Commençons par définir des classes d'erreurs personnalisées :

# 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

J'ai déjà écrit sur les avantages de l'utilisation d'une classe d'erreur de base.

Ajoutons la gestion des exceptions et la journalisation à notre fonction :

import logging

from . import errors

logger = logging.getLogger('payments')

def charge(
 amount,
 token,
 timeout=5,
):

 # ...

 try:
 response = requests.post(
 PAYMENT_GATEWAY_BASE_URL + '/charge',
 json=payload,
 headers=headers,
 timeout=timeout,
 )
 response.raise_for_status()

 except (requests.ConnectionError, requests.Timeout) as e:
 raise errors.Unavailable() from e

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

 if code == 1:
 raise errors.Refused(code, message) from e
 elif code == 2:
 raise errors.Stolen(code, message) from e
 else:
 raise errors.PaymentGatewayError(code, message) from e

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

Super! Notre fonction ne lève plus requests exceptions. Des erreurs importantes telles qu'une carte volée ou un refus sont signalées comme des exceptions personnalisées.

Définir la réponse

Notre fonction renvoie un dict. Un dict est une structure de données excellente et flexible, mais lorsque vous avez un ensemble défini de champs, il est préférable d'utiliser un type de données plus ciblé.

Dans chaque cours de POO, vous apprenez que tout est un objet. Bien que cela soit vrai dans Java land, Python a une solution légère qui fonctionne mieux dans notre cas - namedtuple .

Un tuple nommé est juste comme il sonne, un tuple où les champs ont des noms. Vous l'utilisez comme une classe et elle consomme moins d'espace (même par rapport à une classe avec des slots).

Définissons un tuple nommé pour la réponse de charge :

from collections import namedtuple

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

Si la charge a réussi, nous créons un ChargeResponse objet :

from datetime import datetime

# ...

def charge(
 amount,
 token,
 timeout=5,
):

 # ...

 data = response.json()

 charge_response = 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'],
 )

 return charge_response

Notre fonction retourne maintenant un ChargeResponse objet. Des traitements supplémentaires tels que le casting et les validations peuvent être ajoutés facilement.

Dans le cas de notre passerelle de paiement imaginaire, nous convertissons la date d'expiration en un objet datetime. Le consommateur n'a pas à deviner le format de date utilisé par le service distant (en ce qui concerne les formats de date, je suis sûr que nous avons tous rencontré une bonne part d'horreurs).

En utilisant une "classe" personnalisée comme valeur de retour, nous réduisons la dépendance au format de sérialisation du fournisseur de paiement. Si la réponse était un XML, renverrions-nous toujours un dict ? C'est juste gênant.

Utiliser une session

Pour extraire quelques millisecondes supplémentaires des appels d'API, nous pouvons utiliser une session. La session Requests utilise un pool de connexions en interne. Les demandes adressées au même hôte peuvent en bénéficier. Nous en profitons également pour ajouter des paramètres utiles comme le blocage des cookies :

import http.cookiejar

# A shared requests session for payment requests.
class BlockAll(http.cookiejar.CookiePolicy):
 def set_ok(self, cookie, request):
 return False

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

# ...

def charge(
 amount,
 token,
 timeout=5,
):
 # ...
 response = payment_session.post( ... )
 # ...

Plus d'actions

Tout service externe, et un service de paiement en particulier, a plus d'une action.

La première section de notre fonction s'occupe de l'autorisation, de la requête et des erreurs HTTP. La deuxième partie gère les erreurs de protocole et la sérialisation propres à l'action de charge.

La première partie est pertinente pour toutes les actions tandis que la seconde partie est spécifique uniquement à la charge.

Séparons la fonction pour pouvoir réutiliser la première partie :

import uuid
import logging
import requests
import http.cookiejar
from datetime import datetime


logger = logging.getLogger('payments')


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'],
 )

Ceci est le code entier.

Il existe une séparation claire entre le "transport", la sérialisation, l'authentification et le traitement des requêtes. Nous avons également une interface bien définie avec notre fonction de niveau supérieur charge .

Pour ajouter une nouvelle action, nous définissons un nouveau type de retour, appelez make_payment_request et traitez la réponse de la même manière :

RefundResponse = namedtuple('RefundResponse', [
 'transaction_id',
 'refunded_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'],
 )

Profitez !

Tests

Le défi avec les API externes est que vous ne pouvez pas (ou du moins ne devriez pas) les appeler dans des tests automatisés. Je souhaite me concentrer sur le code de test qui utilise notre module de paiement plutôt que de tester le module réel.

Notre module a une interface simple, il est donc facile de se moquer. Testons une fonction composée appelée charge_user_for_product :

# test.py

from unittest import TestCase
from unittest.mock import patch

from payment.payment import ChargeResponse
from payment import errors

def TestApp(TestCase):

 @mock.patch('payment.charge')
 def test_should_charge_user_for_product(self, mock_charge):
 mock_charge.return_value = ChargeResponse(
 uid='test-uid',
 amount=1000,
 token='test-token',
 expiration=datetime.datetime(2017, 1, 1, 15, 30, 7),
 transaction_id=12345,
 )
 charge_user_for_product(user, product)
 self.assertEqual(user.approved_transactions, 1)

 @mock.patch('payment.charge')
 def test_should_suspend_user_if_stolen(self, mock_charge):
 mock_charge.side_effect = errors.Stolen
 charge_user_for_product(user, product)
 self.assertEqual(user.is_active, False)

Assez simple - pas besoin de se moquer de la réponse de l'API. Les tests sont contenus dans des structures de données que nous avons définies nous-mêmes et dont nous avons le contrôle total.

Remarque sur l'injection de dépendance

Une autre approche pour tester un service consiste à fournir deux implémentations :la vraie et une fausse. Ensuite pour les tests, injectez le faux.

C'est bien sûr le fonctionnement de l'injection de dépendance. Django ne fait pas de DI mais il utilise le même concept avec des "backends" (email, cache, template, etc). Par exemple, vous pouvez tester les e-mails dans Django en utilisant un backend de test, tester la mise en cache en utilisant un backend en mémoire, etc.

Cela présente également d'autres avantages dans la mesure où vous pouvez avoir plusieurs "vrais" backends.

Que vous choisissiez de vous moquer des appels de service comme illustré ci-dessus ou d'injecter un "faux" service, vous devez avoir une interface appropriée.

Résumé

Nous avons un service externe que nous voulons utiliser dans notre application. Nous voulons implémenter un module pour communiquer avec ce service externe et le rendre robuste, résilient et réutilisable.

Nous avons travaillé les étapes suivantes :

  1. Mise en œuvre naïve - Récupérer à l'aide de requêtes et renvoyer une réponse JSON.
  2. Erreurs gérées - Définition des erreurs personnalisées pour détecter les erreurs de transport et d'application distante. Le consommateur est indifférent au transport (HTTP, RPC, Web Socket) et aux détails d'implémentation (requêtes).
  3. Formaliser la valeur de retour - Utilisation d'un tuple nommé pour renvoyer un type de type classe qui représente une réponse du service distant. Le consommateur est désormais également indifférent au format de sérialisation.
  4. Ajout d'une session - Écrémé quelques millisecondes de la demande et ajouté un emplacement pour la configuration globale de la connexion.
  5. Séparer la demande de l'action - La partie requête est réutilisable et de nouvelles actions peuvent être ajoutées plus facilement.
  6. Tester - Appels simulés à notre module et les ont remplacés par nos propres exceptions personnalisées.