test_subscription_transport.py 文件源码

def create_app(sub_mgr, schema, options):
    app = Flask(__name__)
    sockets = Sockets(app)

    app.app_protocol = lambda environ_path_info: 'graphql-subscriptions'

    app.add_url_rule(
        '/graphql',
        view_func=GraphQLView.as_view('graphql', schema=schema, graphiql=True))

    @app.route('/publish', methods=['POST'])
    def sub_mgr_publish():
        sub_mgr.publish(*request.get_json())
        return jsonify(request.get_json())

    @sockets.route('/socket')
    def socket_channel(websocket):
        subscription_server = SubscriptionServer(sub_mgr, websocket, **options)
        subscription_server.handle()
        return []

    return app