Superschnelle Voice-Übertragung mit asynchronem Python und Sanic

Die SMS hat sich als De-facto-Methode für den Versand von Benachrichtigungen etabliert, wenn Push nicht verfügbar ist. So sehr, dass ich nur noch selten eine SMS von einer echten "Person" erhalte. Meine Kollegen nutzen Slack, Freunde den Facebook Messenger, sicherheitsbewusste Freunde Telegram und paranoide Freunde Signal. Sogar meine Mutter, die erst im letzten Jahr ihr erstes Smartphone bekommen hat, schickt mir jetzt süße Bilder von meiner Nichte per WhatsApp statt per E-Mail.

Da ich die SMS nicht mehr als Mittel zur Kommunikation mit Freunden und Familie betrachte, sondern eher als Kanal für die Benachrichtigung über Dienste, reagiere ich nicht mehr in gleicher Weise auf den neuen SMS-Ton. Er vermittelt nicht mehr die gleiche Dringlichkeit wie früher; ich weiß jetzt, dass es sich eher um einen Gutschein für meine örtliche Pizzeria handelt als um etwas, das meine sofortige Aufmerksamkeit erfordert.

Und oft ist das auch gut so. Nicht jede Meldung ist zeitkritisch oder erfordert dringende Maßnahmen. Aber was ist mit den Meldungen, die dies tun? Die kritischen Warnungen? Benachrichtigungen, auf die sofort reagiert werden muss, wie z. B. Serviceausfälle oder extreme Wetterwarnungen. Für diese Meldungen brauchen wir etwas, das nicht so leicht zu ignorieren ist: ein klingelndes Telefon.

Ist das nicht komisch? Sie hören ein Telefon klingeln, und es könnte jeder sein. Aber ein klingelndes Telefon muss doch abgenommen werden, oder nicht? - Der Anrufer, Phone Booth (2002)

Bevor wir loslegen

Bevor wir loslegen, benötigen Sie einige Dinge.

• Python 3.5 oder besser, wir werden einige der neueren async-Funktionen von Pythonverwenden, also brauchen wir eine ziemlich moderne Version. Ich würde auch empfehlen, virtualenv, wie wir gehen zu müssen, um ein paar Abhängigkeiten zu installieren.

• MongoDB lokal installiert

• ngrok oder eine ähnliche Art der Ihre Anwendung dem Internet zur Verfügung zu stellen

Vonage API-Konto

Um dieses Tutorial durchzuführen, benötigen Sie ein Vonage API-Konto. Wenn Sie noch keines haben, können Sie sich noch heute anmelden und mit einem kostenlosen Guthaben beginnen. Sobald Sie ein Konto haben, finden Sie Ihren API-Schlüssel und Ihr API-Geheimnis oben auf dem Vonage-API-Dashboard.

In diesem Lernprogramm wird auch eine virtuelle Telefonnummer verwendet. Um eine zu erwerben, gehen Sie zu Rufnummern > Rufnummern kaufen und suchen Sie nach einer Nummer, die Ihren Anforderungen entspricht.

Sind Sie bereit, mit dem Bau zu beginnen?

Erleben Sie nahtlose Konnektivität, Echtzeit-Messaging und kristallklare Sprach- und Videoanrufe - alles auf Knopfdruck.

Einrichten unserer Datenbank

Wir werden MongoDB verwenden, um die Nummern der Personen zu speichern, die wir anrufen müssen. Unsere Dokumente werden besonders einfach sein:

{
    "_id" : ObjectId("599d3f2c736544a32f48d3c4"),
    "number" : "<NUMBER TO CALL>"
}

Bevor wir also beginnen, öffnen wir unsere Shell und fügen ein paar Dokumente zu unserer Sammlung hinzu. Starten Sie die MongoDB-Shell, indem Sie mongodb.

use contactsDatabase
db.contactsCollection.insert([{"number": "<NUMBER TO CALL>"}, {"number": "<NUMBER TO CALL>"}])

Die Methode .insert() Methode akzeptiert eine Liste von Dokumenten, also fügen Sie ein paar verschiedene Numbers hinzu, die Sie in diesem Beispiel anrufen möchten. Dies können auch virtuelle Nexmo Numbers sein, wenn Sie einige zusätzliche Nummern für Tests benötigen.

Einrichten unseres Python Voice Broadcast Projekts

Der gesamte Code für dieses Beispiel ist in der Nexmo Gemeinschaft Github. Wir sollten diesen nun klonen und unsere Anforderungen installieren. Stellen Sie sicher, dass Sie von nun an alle diese Befehle in Ihrer virtuellen Umgebung ausführen. Im weiteren Verlauf werden Sie mehrere Terminalfenster benötigen, also denken Sie daran, die Umgebung in jedem einzelnen zu aktivieren.

git clone
cd python-sanic-voice-broadcast/
pip install -r requirements.txt

Da wir nun unseren gesamten Code lokal haben und alle Abhängigkeiten mit pipinstalliert haben, müssen wir noch einen letzten Teil der Einrichtung abschließen. Zur Verwendung der Nexmo Voice APIzu verwenden, benötigen wir unsere Anwendungs-ID, den privaten Schlüssel für die Anwendung und die virtuelle Nexmo-Nummer, die als Nummer des Anrufers verwendet werden soll.

Kopieren Sie den privaten Schlüssel, den Nexmo generiert hat, als Sie Ihre Voice-Anwendung erstellt haben, und legen Sie ihn in Ihrem python-sanic-voice-broadcast/ Verzeichnis. Sie müssen ihn umbenennen in broadcast.key umbenennen.

Als Nächstes speichern wir unsere Anwendungs-ID und virtuelle Nummer als Umgebungsvariablen, damit wir in unserem Code darauf zugreifen können. Sie müssen export Sie müssen diese Variablen jedes Mal neu setzen, wenn Sie ein neues Terminal öffnen, oder wenn Sie virtualenvwrapper verwenden, können Sie postactivate verwenden, um sie automatisch zu setzen, wenn Sie Ihre virtuelle Umgebung aktivieren.

export BROADCAST_APPLICATION_ID="<YOUR APPLICATION ID>"
export BROADCAST_NUMBER_FROM="<YOUR NEXMO VIRTUAL NUMBER>"

Ausgehende Voice-Anrufe und Nexmo Call Control Objects

Wir müssen die Nexmo-API anweisen, welche Aktionen sie durchführen soll, wenn der Benutzer unseren Anruf beantwortet. Genau wie in meinem vorherigen Text-to-Speech-Blogpost werden wir die talk Aktion und eine synthetische Stimme um dem Nutzer unsere Benachrichtigung vorzulesen.

Die Nexmo-API wird eine GET-Anfrage an die Antwort-URL stellen, die Sie bei der Erstellung Ihrer Anwendung angegeben haben. Diese URL muss für Nexmo erreichbar sein. Wenn Sie also Ihren Server lokal betreiben wollen, müssen Sie ein Tool wie ngrok verwenden, um Ihren lokalen Server für das öffentliche Internet zugänglich zu machen.

Es muss schnell gehen! Erstellen eines asynchronen Python-Sanic-Servers

Sanic running in a Terminal

Den Code für den Server finden Sie in der Datei server.py Datei, aber konzentrieren wir uns zunächst auf die answer Route konzentrieren.

@app.route("/")
async def answer(request):
    return json([{
        'action': 'talk',
        'text': 'This is a message from the Nexmo broadcast system'
    }])

Unser NCCO ist denkbar einfach. Wir haben eine einzige Aktion, die den Text "Dies ist eine Nachricht vom Nexmo-Rundfunksystem" vorliest, sobald ein Benutzer unseren ausgehenden Anruf beantwortet. Wir verwenden die Sanic json Methode, um sicherzustellen, dass wir die richtigen HTTP-Header mit unserer JSON-Antwort senden.

Probieren wir es jetzt aus. Führen Sie in Ihrem Terminal python server.py und besuchen Sie http://127.0.0.1:8000 in Ihrem Browser.

Dieser Server ist nur lokal erreichbar, aber wir brauchen ihn, um für die Nexmo API verfügbar zu sein. Wenn Sie ngrok verwenden, um einen Tunnel zu Ihrem Localhost zu erstellen, wäre dies ein guter Zeitpunkt, um ein anderes Terminal zu öffnen und ngrok zu starten:

ngrok http 8000

Denken Sie daran, Ihre Voice-Anwendung Antwort und Ereignis-URLs zu aktualisieren, damit sie mit Ihrer ngrok-Adresse übereinstimmen. Die Route des Ereignisses finden Sie in der Datei server.py Datei.

Synchrone Sprachanrufe tätigen

In diesem Beispiel werden wir den Python Nexmo-Client. Aber der Code ist so ziemlich derselbe für JavaScript, Java, PHP, Ruby oder ASP.NET.

import os
import nexmo
from pymongo import MongoClient


if __name__ == '__main__':
    # Connect to our mongo database
    db_client = MongoClient('mongodb://localhost:27017/')
    collection = db_client.contactsDatabase.contactsCollection

    # Create our Nexmo client
    nexmo_client = nexmo.Client(
        application_id=os.environ['BROADCAST_APPLICATION_ID'],
        private_key='broadcast.key'
    )

    # Grab a single contact from our database
    contact = collection.find_one()

    # Create an outbound call to the selected user
    response = nexmo_client.create_call({
        'to': [{'type': 'phone', 'number': contact['number']}],
        'from': {'type': 'phone', 'number': os.environ['BROADCAST_NUMBER_FROM']},
        'answer_url': ['https://nexmo-broadcast.ngrok.io']
    })

    print(response)

Der Prozess ist einfach: Wir stellen eine Verbindung zu unserem Datenspeicher her, wählen eine Telefonnummer aus, die angerufen werden soll, erstellen einen neuen ausgehenden Anruf mit Hilfe der Nexmo-API und der Python-Client-Bibliothek.

Die Verwendung des Python-Clients ist die einfachste Art, einen ausgehenden Anruf zu tätigen. Aber es ist auch synchron. Unter der Haube verwendet unser Python-Client die ziemlich unglaubliche requests-Bibliothek. Aber leider ist requests keine asynchrone Bibliothek, obwohl das bald der Fall sein wird!

![Screenshot des Nexmo-Python-Clients, der in der Klasse Terminal" läuft](https://d226lax1qjow5r.cloudfront.net/blog/blogposts/super-fast-voice-broadcast-with-asynchronous-python-and-sanic/making-outbound-voice-call-terminal.png "Screenshot des Nexmo-Python-Clients, der in der Klasse Terminal" läuft")

Wenn Sie nur eine kleine Anzahl von Benachrichtigungen versenden müssen, ist dies wahrscheinlich gut genug. Die Nexmo-API ist schnell, aber Sie müssen sich trotzdem online mit ihr verbinden. Es wird eine gewisse Latenzzeit geben. Von meinem Büro in Glasgow, Schottland, dauert es ungefähr 1 Sekunde für die Anfrage und die Antwort der Nexmo API. Wenn ich jedoch Tausende oder gar Hunderttausende von Benachrichtigungen synchron senden möchte, ist dies wahrscheinlich nicht der beste Weg.

Einen asynchronen Voice-Anruf tätigen

Der gesamte Code für diesen nächsten Abschnitt befindet sich in der Datei broadcast.py Datei. Er ist etwas komplexer als der obige, da wir einen winzigen Teil des Nexmo-Python-Clients nachbilden werden, und zwar so, dass asynchrone Aufrufe unterstützt werden.

Werfen wir zunächst einen Blick auf unsere Ereignisschleife.

def run_event_loop():
    loop = asyncio.get_event_loop()
    future = asyncio.Future()

    asyncio.ensure_future(broadcast(future, loop))
    loop.run_until_complete(future)

    logger.debug(future.result())
    loop.close()

Hier haben wir eine Zukunft und eine Coroutine, die Future wird laufen, bis unsere broadcast Methode signalisiert, dass sie fertig ist, indem sie future.set_result. Es ist diese broadcast Coroutine wird alle Aufrufe sammeln, die wir machen müssen. Schauen wir sie uns als nächstes an.

async def broadcast(future, loop):
    # Connect to MongoDB
    client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')
    contacts_collection = client.contactsDatabase.contactsCollection
    cursor = contacts_collection.find()

    # Use the aiohttp client which is async
    async with aiohttp.ClientSession(loop=loop) as session:

        # Use a list comprehension to call create_call with each number
        tasks = [
            create_call(session=session, number=document['number'])
            for document in await cursor.to_list(length=100)
        ]
        await asyncio.gather(*tasks)

    # Signal that our future is now complete
    future.set_result(f'attempted to ring {len(tasks)} people')

Das erste, was zu beachten ist, ist, dass dies eine Coroutine ist und wir die async def Syntax, was bedeutet, dass wir eine Python-Version >= 3.5 benötigen.

Python Motor Mascot

Unser MongoDB-Client-Code hat sich ebenfalls etwas verändert. Wir verwenden jetzt motor anstelle der pymongo-Bibliothek verwendet. Wir haben diese Umstellung vorgenommen, da pymongo nicht asynchron ist. Um zu verhindern, dass unsere Mongo-Aufrufe blockiert werden, müssen wir stattdessen motor verwenden.

Der wichtigste Teil dieser Coroutine ist die Einführung von aiohttp. Dieses Modul ermöglicht es uns, asynchrone HTTP-Anfragen zu stellen. Wir werden eine asynchrone Client-Sitzung erstellen und diese an unsere create_call Coroutine übergeben.

Schreiben unserer create_call-Methode

Im ersten Blocking-Beispiel haben Sie vielleicht bemerkt, dass der Nexmo-Python-Client eine Methode create_callverwendet, werden wir denselben Namen für unsere Coroutine verwenden, die wir oben für jede der Numbers in unserer MongoDB aufgerufen haben.

Zu Beginn unserer Coroutine erstellen wir eine neue BroadcastClient.

# Wrap our JWT generation in a new class
client = await BroadcastClient.create(number_to=number)
headers = client.get_headers()
payload = client.get_payload()

Die Nexmo Voice API verwendet JSON Web Tokens (JWT) für die Authentifizierung. Diese BroadcastClient generiert unser Token und stellt uns Methoden zur Verfügung, mit denen wir die richtigen Header und Payloads erstellen können, die wir mit unserer API-Anfrage senden müssen. Schauen wir uns das zuerst an, bevor wir zu unserer create_call Methode zurückkehren.

JSON-Web-Tokens und unser Voice-Alarm-Payload

Wenn Sie sich den BroadcastClientbetrachten, werden Sie feststellen, dass es keine __init__ Methode hat, das ist Absicht. Wir müssen die Erstellung unseres Clients nicht blockieren, aber wir müssen auch den Inhalt unseres privaten Schlüssels von der Festplatte lesen. Normalerweise würden wir diese Einstellungen in der __init__ Methode durchführen, aber Pythons magische Methoden sind nicht für die Arbeit mit async/await ausgelegt. Stattdessen werden wir das das Fabrikmuster.

Die BroadcastClient hat eine create Methode, die asynchron ist und ein BroadcastClient Objekt zurück, das die richtigen Klassenattribute hat, einschließlich des Inhalts unserer privaten Schlüsseldatei. Sie können sehen, wie wir sie verwenden, um unseren Client im vorherigen Codebeispiel zu instanziieren.

client = await BroadcastClient.create(number_to=number)

Sobald wir unseren Client instanziiert haben, können wir die Header für unsere Anfrage generieren, die unser Token enthalten.

def get_headers(self):
    iat = int(time.time())
    payload = {
        'iat': iat,
        'application_id': self.APPLICATION_ID,
        'exp': iat + 60,
        'jti': str(uuid.uuid4())
    }

    token = jwt.encode(payload, self.PRIVATE_KEY, algorithm='RS256')

    headers = {
        'User-Agent': self.USER_AGENT,
        'Authorization': 'Bearer ' + token.decode('utf-8')
    }

    return headers

Wenn Sie bereits mit JWT gearbeitet haben, wird Ihnen dieser Code bekannt vorkommen. Falls nicht, empfehle ich die Lektüre der vollständige Spezifikation zu lesen, um zu verstehen, wie es funktioniert.

Der zur Verschlüsselung des Tokens verwendete private Schlüssel muss mit dem öffentlichen Schlüssel übereinstimmen, der für Ihre Sprachanwendung. Ein weiterer wichtiger Punkt ist das User-Agent Attribut; es wird zur Identifizierung Ihrer Anwendung verwendet und sollte eindeutig sein.

Die Nutzlast des Voice-Alarms

Sie werden feststellen, wie ähnlich dies dem Wörterbuch ist, das dem Nexmo-Client in unserem allerersten synchronen Beispiel übergeben wird:

# Nexmo Python client (synchronous)
response = nexmo_client.create_call({
    'to': [{'type': 'phone', 'number': contact['number']}],
    'from': {'type': 'phone', 'number': os.environ['BROADCAST_NUMBER_FROM']},
    'answer_url': ['https://nexmo-broadcast.ngrok.io']
})

# Get payload method for our asynchronous example
def get_payload(self):
    return {
        'to': [{'type': 'phone', 'number': self.NUMBER_TO}],
        'from': {'type': 'phone', 'number': self.NUMBER_FROM},
        'answer_url': [self.ANSWER_URL]
    }

Die Nexmo-Client-Bibliotheken sind alle sehr dünne Wrapper über eine REST-API. Wie Sie sehen können, sieht unser Code, auch wenn wir ihn ohne den Python-Client schreiben, sehr ähnlich aus.

Okay, kehren wir zurück zu unserer create_call Methode zurück und sehen, wie wir unsere BroadcastClient um unsere dringenden Benachrichtigungen zu senden.

Aufrufen der Nexmo Voice API

async def create_call(session, number):
    logger.info(f'calling {number}')

    # Wrap our JWT generation in a new class
    client = await BroadcastClient.create(number_to=number)
    headers = client.get_headers()
    payload = client.get_payload()

    # POST to the Nexmo API
    async with session.post('https://api.nexmo.com/v1/calls', headers=headers, json=payload) as response:
        status = response.status
        nexmo_response = await response.text()

        # 429 == rate limited, need to back off
        if status == 429:
            raise NexmoRateError

        logger.info(f'call requested to {number} ({status})')

    # The Nexmo JSON response will contain 'started' as the status
    # if everything has gone to plan
    return 'started' in nexmo_response

Sobald wir unseren Client instanziiert haben, verwenden wir die aiohttp-Sitzung, um an den calls Endpunkt der Nexmo API. Unsere Header enthalten nun unser JWT-Token, und die Nutzlast ist eine JSON-Darstellung des Wörterbuchs, das von get_payload.

Nachdem wir die POST-Anfrage an die Nexmo-API gestellt haben, müssen wir prüfen, ob ein 429 HTTP-Status. Wenn wir unsere Ratenbegrenzung überschritten haben, ist dies der Code, den Nexmo zurückgibt. Wenn wir also einen 429erhalten, sollten wir backoff. Zurzeit liegt die API-Ratenbegrenzung für POST-Anfragen an die Voice API bei zwei Anfragen pro Sekunde. Wir werden uns die backoff Dekoratoren an.

Schließlich wird unsere create_call Coroutine entweder True oder False zurück, je nachdem, ob die JSON-Zeichenfolge den Status "started" enthält oder nicht. Auch wenn diese Prüfung etwas rudimentär erscheint, werden wir im nächsten Abschnitt sehen, wie wichtig sie ist.

Sich zurückhalten und ein höflicher API-Benutzer sein

Wir haben die folgenden Dekoratoren in unserer create_call Methode.

@backoff.on_exception(backoff.expo, NexmoRateError, on_backoff=backoff_exception_handler)
@backoff.on_predicate(backoff.fibo, on_backoff=backoff_predicate_handler, max_tries=5)

Hier verwenden wir die Backoff-Bibliothek um unseren API-Aufruf erneut auszuführen, wenn er fehlschlägt, aber es wird auch gewartet, bevor es erneut versucht wird, damit wir den API-Endpunkt nicht beschädigen.

Wir haben zwei Dekoratoren, die jeweils auf eine andere Art von Fehler von der API warten. Der on_exception Dekorator wird ausgelöst, wenn die Coroutine eine NexmoRateErrorauslöst; diese Ausnahme tritt immer dann auf, wenn wir einen HTTP-Status von 429. Wir haben keine maximale Anzahl von Wiederholungen für diesen Dekorator festgelegt, aber wir haben ihn angewiesen, eine exponentielle Dauer für unseren Backoff zu verwenden, mit Jitter.

Anrufe mit exponentiellem Backoff und ohne Jitter

Graph showing clustering with no jitter

Anrufe mit exponentiellem Backoff und vollem Jitter

Graph with no clustering as jitter is applied

Diagramme aus "Exponential Backoff und Jitter" von AWS Architektur-Blog

Wie im zweiten Diagramm zu sehen ist, gibt es viel weniger Anrufcluster, wenn wir unserem Algorithmus Jitter hinzufügen. Der AWS-Architektur-Blog erklärt dies besonders gut in seinem Beitrag Exponential Backoff und Jitter.

Unser zweiter Dekorator on_predicate wird immer dann ausgelöst, wenn die Coroutine einen Falsey Wert zurückgibt. Unser Generator für die Wartezeit in diesem Beispiel ist fiboder die Numbers der Fibonacci-Folgeergibt, wiederum mit einem gewissen Jitter, um Clustering zu verhindern.

def fibo(max_value=None):
    a = 1
    b = 1
    while True:
        if max_value is None or a < max_value:
            yield a
            a, b = b, a + b
        else:
            yield max_value

Wenn die Exponential- oder Fibonacci-Generatoren für Ihren Anwendungsfall nicht geeignet sind, können Sie ganz einfach Ihre eigenen Generatoren schreiben. Wie in xkcd 221 ist hier ein Generator, der immer eine zufällige Wartedauer liefert.

XKCD 221 - Random number generator

def xkcd():
    while True:
        yield 4  # chosen by fair dice roll, guaranteed to be random

Unser create_call Coroutine gibt einen Falsey Wert zurück, wenn die JSON-Antwort von der Nexmo-API keine started. Es gibt viele verschiedene Gründe, warum dies der Fall sein kann: unser privater Schlüssel könnte falsch sein, wir haben ungültige Werte in unserem Payload, wir haben nicht genug Guthaben auf unserem Nexmo Account, und so weiter. Dies sind keine Probleme, die durch einen erneuten Aufruf der API gelöst werden können. In diesem Beispiel versuchen wir es also ein paar Mal, um eine kleine Störung in der Konnektivität oder ähnliches auszugleichen, und geben dann einfach auf.

Alles ausprobieren

Screencast showing multiple async tasks making outbound voice calls

Bevor Sie versuchen, eines der beiden Skripte auszuführen, sollten Sie daran denken, dass Sie über Sanic und ngrok laufen, damit Nexmo Ihre NCCO-Datei abrufen kann!

Sie müssen auch den Abschnitt zur Einrichtung am Anfang dieses Artikels ausfüllen. Stellen Sie sicher, dass Sie MongoDB mit mehreren Dokumenten in Ihrem contactsCollectionSie haben die erforderlichen Umgebungsvariablen gesetzt, Sie haben Ihren privaten Schlüssel in Ihrem Projektstamm als broadcast.keygespeichert haben, und Sie haben alle Anforderungen in Ihrer virtuellen Umgebung mit pip installiert.

Wenn Sie alles oben genannte erledigt haben, können Sie die synchrone Aufgabe mit ausführen:

python blocking_broadcast.py

Und um die asynchrone Version zu testen, führen Sie aus:

python broadcast.py

Beachten Sie die Ausgabe des asynchronen Skripts; die Reihenfolge der calling number und call requested to number sind wahrscheinlich unterschiedlich, da es sich um ein asynchrones Skript handelt.