Créons un client de socket Web avec Python. (Authentification par jeton d'accès)

J'ai eu la chance de jouer avec l'API de l'application Typetalk de nulab (https://developer.nulab-inc.com/ja/docs/typetalk), j'ai donc implémenté websocket. Notez que je ne l'ai jamais fait avec python

Environnement de développement

MaxOSX 10.10.3 Python 2.7.9

Installation

Installez les packages requis avec python

pip install websocket-client

Pour installer le client websocket de python.

implémentation de websocket

Cette fois, puisque nous avons utilisé Typetalk, nous allons présenter comment obtenir un jeton d'accès à partir de type talk et définir websocketclient. Ce code vous permet de voir tout le comportement des utilisateurs enregistrés.

main.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import urllib
import urllib2
import json
import string

import websocket
import requests

class TypeTalkSample:
    access_token = None

    def __init__(self):
    	#Obtenez un jeton d'accès de typetalk
        client_id = "xxxxxxx"
        client_secret = "xxxxxxx"
        params = {
            'client_id': client_id,
            'client_secret': client_secret,
            'grant_type': 'client_credentials',
            'scope': 'topic.read,topic.post',
        }
        data = urllib.urlencode(params)
        res = urllib2.urlopen(urllib2.Request('https://typetalk.in/oauth2/access_token', data))
        
        #Tenir le jeton d'accès
        self.access_token = json.load(res)['access_token']

		#Définissez un websocket. Vous pouvez définir un websocket à l'aide d'un jeton d'accès en enregistrant une autorisation dans l'en-tête ici.
        websocket.enableTrace(True)
        ws = websocket.WebSocketApp("wss://typetalk.in/api/v1/streaming", header=["Authorization: Bearer %s" % self.access_token], on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close)
        
        #Démarrez websocket. Ctrl+Assurez-vous qu'il se termine par C.
        try:
            ws.run_forever()
        except KeyboardInterrupt:
            ws.close()

	#La méthode définie ici devient la fonction de rappel de websocket.
	#Lors de la réception d'un message
    def on_message(self, ws, message):
        print message

	#Lorsqu'une erreur survient
    def on_error(self, ws, error):
        print error

	#Lorsque vous fermez le socket Web
    def on_close(self, ws):
        print 'disconnected streaming server'

	#Lorsque vous ouvrez le socket Web
    def on_open(self, ws):
        print 'connected streaming server'


if __name__ == "__main__":
    typetalk = TypeTalkSample()

Lien

Typetalk api (https://developer.nulab-inc.com/ja/docs/typetalk)