Transmissão de voz super rápida com Python assíncrono e Sanic

O SMS se tornou o método padrão para enviar notificações quando o push não está disponível. Tanto é assim que eu quase não recebo mais SMS de uma “pessoa” de verdade. Meus colegas usam o Slack, meus amigos usam o Facebook Messenger, os amigos mais preocupados com segurança usam o Telegram e os amigos mais paranóicos usam o Signal. Até minha mãe, que só comprou seu primeiro smartphone no ano passado, agora me manda fotos fofas da minha sobrinha pelo WhatsApp, em vez de e-mail.

À medida que fui deixando de ver as mensagens SMS como uma forma de me comunicar com amigos e familiares e passando a vê-las mais como um canal de notificações de serviços, percebi que não reagia mais da mesma maneira ao novo som das mensagens SMS. Ele não transmite a mesma urgência de antes; agora sei que é mais provável que seja um cupom da pizzaria da minha região do que qualquer coisa que exija minha atenção imediata.

E, muitas vezes, isso é suficiente. Nem toda notificação é urgente ou exige uma ação imediata. Mas e aquelas mensagens que exigem isso? Os alertas críticos? Notificações que exigem uma ação imediata, como interrupções no serviço ou alertas de condições climáticas extremas. Para esses casos, precisamos de algo que seja mais difícil de ignorar: um telefone tocando.

Não é engraçado? Você ouve um telefone tocando, e pode ser qualquer pessoa. Mas quando o telefone toca, a gente tem que atender, não é mesmo? — O Interlocutor, “Phone Booth” (2002)

Antes de começarmos

Há algumas coisas de que você vai precisar antes de começarmos.

• Python 3.5 ou versão posterior; vamos utilizar alguns dos recursos assíncronos mais recentes recursos assíncronos do Python, por isso precisaremos de uma versão bastante atualizada. Também recomendo o virtualenv, já que precisaremos instalar algumas dependências.

• MongoDB instalado localmente

• ngrok ou uma forma semelhante de expor seu aplicativo à internet

Configurando nosso banco de dados

Vamos usar o MongoDB para armazenar os números das pessoas para quem precisamos ligar. Nossos documentos serão bem simples:

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

Então, antes de começarmos, vamos abrir nosso shell e adicionar alguns documentos à nossa coleção. Inicie o shell do MongoDB executando mongodb.

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

O .insert() método aceita uma lista de documentos; portanto, adicione alguns números diferentes que você gostaria de ligar como parte deste exemplo. Eles também podem ser Numbers virtuais da Nexmo, caso você precise de alguns extras para testes.

Configurando nosso projeto de transmissão de voz em Python

Todo o código deste exemplo está na GitHub da Comunidade Nexmo. Devemos cloná-lo agora e instalar os pré-requisitos. A partir de agora, certifique-se de executar todos esses comandos dentro do seu ambiente virtual. À medida que avançarmos, você precisará de várias janelas de terminal; portanto, lembre-se de ativar o ambiente em cada uma delas.

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

Agora que temos todo o nosso código localmente e instalamos todas as nossas dependências com pip, temos mais uma etapa final de configuração a concluir. Para usar a Voice API da Nexmo, precisaremos do ID do nosso aplicativo, da chave privada do aplicativo e do número virtual da Nexmo para usar como número do chamador.

Copie a chave privada que a Nexmo gerou quando você criou seu aplicativo de voz e coloque-a no seu python-sanic-voice-broadcast/ diretório. Você precisará renomeá-la para broadcast.key também.

A seguir, vamos salvar o ID do nosso aplicativo e o número virtual como variáveis de ambiente para que possamos acessá-los em nosso código. Você precisará export definir essas variáveis toda vez que abrir um novo Terminal; ou, se estiver usando o virtualenvwrapper, pode usar postactivate para defini-las automaticamente ao ativar seu ambiente virtual.

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

Chamadas de voz de saída e objetos de controle de chamadas do Nexmo

Precisamos instruir a API da Nexmo sobre quais ações ela deve realizar sempre que o usuário atender nossa chamada. Assim como na minha postagem anterior sobre conversão de texto em fala , vamos usar a talk ação e uma voz sintetizada para ler nossa notificação para o usuário.

A API da Nexmo fará uma solicitação GET para a URL de resposta que você forneceu ao criar seu aplicativo. Essa URL precisará estar acessível pela Nexmo; portanto, se você for executar seu servidor localmente, precisará usar uma ferramenta como o ngrok para expor seu servidor local à internet pública.

Tem que ser rápido! Criando um servidor Sanic assíncrono em Python

Sanic running in a Terminal

Você pode encontrar o código do servidor no server.py arquivo, mas vamos nos concentrar na answer rota por enquanto.

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

Nosso NCCO é incrivelmente simples. Temos uma única ação que exibirá o texto “Esta é uma mensagem do sistema de transmissão da Nexmo” sempre que um usuário atender nossa chamada de saída. Estamos usando o método Sanic json para garantir que enviemos os cabeçalhos HTTP corretos com nossa resposta JSON.

Vamos tentar agora. No seu terminal, execute python server.py e acesse http://127.0.0.1:8000 no seu navegador.

Este servidor só pode ser acessado localmente, mas precisamos que ele esteja disponível para a API da Nexmo. Se você estiver usando o ngrok para criar um túnel para o seu localhost, este seria um bom momento para abrir outro Terminal e iniciar o ngrok:

ngrok http 8000

Lembre-se de atualizar as URLs de resposta e de eventos do seu aplicativo de voz para que correspondam ao seu endereço do ngrok. Você pode encontrar a rota do evento no server.py arquivo.

Fazendo uma chamada de voz síncrona

Neste exemplo, vamos usar o cliente Nexmo para Python. Mas o código é praticamente o mesmo para JavaScript, Java, PHP, Ruby ou 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)

O processo é simples: conectamos-nos ao nosso banco de dados, selecionamos um número de telefone para ligar e criamos uma nova chamada de saída usando a API da Nexmo e a biblioteca cliente do Python.

Usar o cliente Python é a maneira mais simples de fazer uma chamada de saída. Mas também é síncrona. Nos bastidores, nosso cliente Python usa a biblioteca requests, que é realmente incrível. Mas, infelizmente, a requests não é uma biblioteca assíncrona, embora isso esteja a caminho!

![Captura de tela do cliente Nexmo em Python sendo executado no Terminal" class](https://s3.amazonaws.com/a.storyblok.com/f/270183/385282/7d148c684e/making-outbound-voice-call-terminal.png "Captura de tela do cliente Nexmo em Python em execução no Terminal" class")

Se você tiver um número pequeno de notificações para enviar, isso provavelmente já é suficiente. A API da Nexmo é rápida, mas você ainda precisa se conectar a ela pela internet. Haverá alguma latência. Do meu escritório em Glasgow, na Escócia, leva aproximadamente 1 segundo para a solicitação e a resposta da API da Nexmo. Portanto, enviar algumas notificações dessa forma provavelmente seria adequado, mas se eu quisesse enviar milhares de notificações, ou mesmo centenas de milhares de notificações de forma síncrona, essa provavelmente não seria a melhor maneira.

Fazendo uma chamada de voz assíncrona

Todo o código desta próxima seção está no broadcast.py arquivo. É um pouco mais complexo do que o anterior, pois vamos recriar uma pequena parte do cliente Nexmo em Python, de forma a suportar chamadas assíncronas.

Vamos dar uma olhada primeiro no nosso ciclo de eventos.

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()

Aqui temos um `future` e uma `coroutine`; isso Future vai ser executado até que nossa broadcast método sinalizar que está concluído chamando future.set_result. É essa broadcast corrotina que reunirá todas as chamadas que precisamos fazer. Vamos dar uma olhada nisso a seguir.

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

A primeira coisa a se notar é que se trata de uma corrotina, e estamos usando a async def sintaxe, o que significa que precisaremos de uma versão do Python >= 3.5

Python Motor Mascot

Nosso código do cliente MongoDB também sofreu algumas alterações. Agora estamos usando o motor em vez da biblioteca pymongo. Fizemos essa mudança porque o pymongo não é assíncrono. Para evitar que nossas chamadas ao MongoDB fiquem bloqueadas, precisamos usar o motor.

A parte mais importante dessa corrotina é a introdução do aiohttp. Esse módulo nos permite fazer solicitações HTTP assíncronas. Vamos criar uma sessão de cliente assíncrona e passá-la para nossa create_call corrotina.

Escrevendo nosso método `create_call`

Você deve ter percebido, no primeiro exemplo de bloqueio, que o cliente Python da Nexmo possui um método create_call; vamos usar o mesmo nome para nossa corrotina que chamamos acima para cada um dos números em nosso MongoDB.

No início da nossa corrotina, estamos criando uma nova BroadcastClient.

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

A Voice API da Nexmo utiliza JSON Web Tokens (JWT) para autenticação. Isso BroadcastClient irá gerar nosso token e nos fornecer métodos para criar os cabeçalhos e a carga úteis corretos que devemos enviar com nossa solicitação de API. Vamos dar uma olhada nisso primeiro, antes de voltarmos ao nosso create_call método.

Tokens JSON da Web e nossa carga útil de alertas de Voice

Se você observar o BroadcastClient, você vai perceber que ela não possui um __init__ método, e isso é proposital. Precisamos que a criação do nosso cliente seja não-bloqueante, mas também precisamos ler o conteúdo da nossa chave privada do disco. Normalmente, faríamos essa configuração no __init__ , mas os métodos mágicos do Python não foram projetados para funcionar com async/await. Em vez disso, vamos usar o padrão de fábrica.

O BroadcastClient possui um create método, que é assíncrono e retorna um BroadcastClient objeto, que tem os atributos de classe corretos definidos, incluindo o conteúdo do nosso arquivo de chave privada. Você pode ver como o usamos para instanciar nosso cliente no exemplo de código anterior.

client = await BroadcastClient.create(number_to=number)

Depois de instanciar nosso cliente, podemos gerar os cabeçalhos para nossa solicitação, que incluirão nosso token.

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

Se você já trabalhou com JWT, esse código lhe parecerá familiar. Caso contrário, recomendo ler a especificação completa para entender como ele funciona.

A chave privada usada para codificar o token deve corresponder à chave pública configurada para seu aplicativo de voz. Outro ponto a ser observado é o User-Agent ; ele é usado para identificar seu aplicativo e deve ser único.

A carga útil do alerta de voz

Você vai perceber como isso é semelhante ao dicionário passado para o cliente Nexmo em nosso primeiro exemplo síncrono:

# 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]
    }

As bibliotecas de cliente da Nexmo são todas muito envelopamentos bem simples sobre uma API REST. Como você pode ver, mesmo ao escrever nosso código sem usar o cliente Python, ele fica muito parecido.

Tudo bem, vamos voltar ao nosso create_call método e ver como usamos nosso BroadcastClient para enviar nossas notificações urgentes.

Chamando a Voice API da Nexmo

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

Depois de instanciar nosso cliente, usamos a sessão do aiohttp para enviar uma solicitação POST ao calls ponto de extremidade da API do Nexmo. Nossos cabeçalhos agora incluirão nosso token JWT, e a carga útil é uma representação em JSON do dicionário retornado por get_payload.

Depois de enviarmos a solicitação POST para a API da Nexmo, devemos verificar se há um 429 status HTTP. Se tivermos excedido nosso limite de taxa, esse é o código que a Nexmo retornará. Portanto, se recebermos um 429, devemos backoff. Atualmente, o limite de taxa da API para solicitações POST à Voice API é de duas solicitações por segundo. Veremos os backoff decoradores daqui a pouco.

Por fim, nossa create_call corrotina retornará True ou False, dependendo se a string JSON contém ou não o status “started”. Embora essa verificação pareça um pouco rudimentar, veremos como ela é importante na próxima seção.

Manter a distância e ser um usuário educado da API

Temos os seguintes decoradores em nosso create_call método.

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

Aqui estamos usando a biblioteca backoff para reexecutar nossa chamada à API caso ela falhe, mas ela também aguardará um tempo antes de tentar novamente, para que não sobrecarreguemos o endpoint da API.

Temos dois decoradores, cada um aguardando um tipo diferente de erro da API. O on_exception decorador é acionado quando a corrotina lança uma NexmoRateError; essa exceção ocorre sempre que temos um status HTTP de 429. Não definimos um número máximo de tentativas para este decorador, mas o configuramos para usar uma duração exponencial para o nosso backoff, com jitter.

Chamadas com backoff exponencial e sem jitter

Graph showing clustering with no jitter

Chamadas com backoff exponencial e jitter total

Graph with no clustering as jitter is applied

Gráficos extraídos de “Retrocesso exponencial e jitter”, do blog de arquitetura da AWS

Como podemos ver no segundo gráfico, há muito menos aglomerações de chamadas quando adicionamos jitter ao nosso algoritmo. O blog de arquitetura da AWS explica isso muito bem em sua postagem Backoff exponencial e jitter.

Nosso segundo decorador on_predicate é acionado sempre que a corrotina retorna um Falsey valor. Nosso gerador para o tempo de espera neste exemplo é fibo, que irá gerar os números da série de Fibonacci, novamente com alguma variação para evitar aglomeração.

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

Se os geradores exponenciais ou de Fibonacci não forem adequados ao seu caso de uso, é muito simples criar o seu próprio. Conforme xkcd 221 , aqui está um gerador que sempre produz um tempo de espera aleatório.

XKCD 221 - Random number generator

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

Nossa create_call corrotina retornará um Falsey valor sempre que a resposta JSON da API da Nexmo não contiver started. Existem diversos motivos possíveis para isso: nossa chave privada pode estar incorreta, podemos ter valores inválidos em nossa carga útil, não temos crédito suficiente em nossa Account da Nexmo e assim por diante. Esses não são problemas que serão resolvidos apenas chamando a API novamente. Portanto, neste exemplo, fazemos algumas tentativas para compensar uma pequena falha de conectividade ou algo semelhante e, depois, simplesmente desistimos.

Experimentando tudo

Screencast showing multiple async tasks making outbound voice calls

Antes de tentar executar qualquer um dos scripts, lembre-se de que você precisará ter Sanic e ngrok em execução para que o Nexmo possa buscar seu arquivo NCCO!

Você também precisará concluir a seção de configuração no início deste artigo. Certifique-se de ter o MongoDB em execução com vários documentos em seu contactsCollection, que definiu as variáveis de ambiente necessárias, que salvou sua chave privada na raiz do projeto como broadcast.key, e que tenha instalado todos os pré-requisitos em seu ambiente virtual usando o pip.

Depois de concluir todas as etapas acima, você pode executar a tarefa síncrona com:

python blocking_broadcast.py

E para testar a versão assíncrona, execute:

python broadcast.py

Observe a saída do script assíncrono; a ordem dos calling number e call requested to number provavelmente é diferente, já que se trata de um processo assíncrono.