Crear un webhook

Conoce cómo escribir el código para manejar las notificaciones al webhook.

Para agregar webhooks a tu integración debes crear un endpoint para exponerlo en internet a través de una URL (por ejemplo, https://myproduct.com/webhooks). Crear un endpoint de webhook en tu servidor no es tan diferente a crear cualquier página en su sitio web o aplicación. Puede estar escrito en Ruby, PHP, NodeJS, Python usando o no un framework de Python como Flask.

Consideraciones

• En cada activación de un evento, Piriod envía los datos del webhook a su endpoint en formato JSON.

• Se incluyen dos datos:

  • event corresponde al nombre del evento.

  • object_id corresponde al id del objeto que activó el evento.

Por lo tanto, como mínimo, el endpoint del webhook debe esperar datos a través de una solicitud POST y confirmar la recepción exitosa de esos datos.

Respuesta HTTP y desactivación automática

Tu endpoint debe devolver un código de estado HTTP 2xx a Piriod. Todos los códigos de respuesta fuera de este rango, incluidos los códigos 3xx, le indican a Piriod que la recepción del evento fue fallida. Si tu endpoint retorna el mismo código de estado fallido repetido en las 3 últimas llamadas el webhook se deshabilitará automáticamente, lo que implica que debas volver a habilitar el webhook manualmente desde Piriod. Debido a que es muy importante que la notificación de webhook se reciba exitosamente, tu endpoint debe devolver un código de estado HTTP 2xx antes de que cualquier lógica de negocio compleja que pueda causar un error o la finalización del tiempo de espera que es de 6 segundos.

Código de ejemplo

Python

import requests
from flask import Flask

app = Flask(__name__)

PIRIOD_API = 'https://api.piriod.com'

credentials = {
    'Authorization': f'Token {your_piriod_token}',
    'x-simple-workspace': 'your_piriod_organization_id'
}

# using Flask
@app.route('/webhooks', methods=['POST'])
def webhooks():
    payload = request.json
    if not payload:
        abort(400)
    if payload.get('event') == 'payment.created':
        payment = _get_payment(payload.get('id'))
        ##
        # do anything with your new payment
        ##
    return True

def _get_payment(id):
    r = requests.get(f'{PIRIOD_API}/payments/{id}/', headers=credentials)
    if not r.status_code == requests.codes.ok:
        abort(r.status_code)
    return r.json()

if __name__ == '__main__':
    app.run(debug=True)