Calling Twisted recator.run() Inside flask app

Question

I am writing a web service in python flask for Jasmin SMS Gateway to create and delete user from the gateway. In case of POST request I am calling runScenario() and after that I am starting reactor.run() which will create user in the gateway. this code runs perfectly for the first web service call but on second call its giving me this error :

raise error.ReactorNotRestartable()
ReactorNotRestartable

This is my Flask app:

#!/usr/bin/env python
from flask import Flask, jsonify, request, Response, make_response, abort
from JasminIntegration import *

JasminWebservice = Flask(__name__)

@JasminWebservice.errorhandler(404)
def not_found(error):
    return make_response(jsonify({'error': 'Not found'}), 404)

@JasminWebservice.route('/jsms/webservice', methods=['POST'])
def create_user():
    if not request.json or not 'username' in request.json:
        abort(400)

    runScenario(request)
    reactor.run()
    return jsonify({'response':'Success'}), 201

if __name__ == '__main__':    
    JasminWebservice.run(host="0.0.0.0",port=7034,debug=True)

I am calling runScenario() which is defined in JasminIntegration.py

#!/usr/bin/env python
import sys
import pickle
from flask import abort
from twisted.internet import defer, reactor
from jasmin.managers.proxies import SMPPClientManagerPBProxy
from jasmin.routing.proxies import RouterPBProxy
from jasmin.routing.Routes import DefaultRoute
from jasmin.routing.jasminApi import SmppClientConnector, User, Group, MtMessagingCredential, SmppsCredential
from jasmin.protocols.smpp.configs import SMPPClientConfig
from twisted.web.client import getPage


@defer.inlineCallbacks
def runScenario(Request):
    try:
        proxy_router = RouterPBProxy()
        yield proxy_router.connect('127.0.0.1', 8988, 'radmin', 'rpwd')

        if Request.method == 'POST':            
            smppUser = Request.json['username']
            smppPass = Request.json['password']
            smppThroughput = Request.json['tp']
            smppBindSessions = Request.json['sessions']

            if not smppUser:
                abort(400)

            if len(smppPass) == 0 or len(smppPass) > 8:
                abort(400)

            if not smppThroughput.isdigit():
                abort(400)

            if not smppBindSessions.isdigit():
                abort(400)

            # Provisiong router with users
            smpp_cred = SmppsCredential()
            yield smpp_cred.setQuota('max_bindings',int(smppBindSessions))

            mt_cred = MtMessagingCredential()
            yield mt_cred.setQuota('smpps_throughput' , smppThroughput)
            #yield mt_cred.setQuota('submit_sm_count' , 500)

            g1 = Group('clients')
            u1 = User(uid = smppUser, group = g1, username = smppUser, password = smppPass, mt_credential = mt_cred, smpps_credential = smpp_cred)
            yield proxy_router.group_add(g1)
            yield proxy_router.user_add(u1)

        if Request.method == 'DELETE':

            smppUser = Request.json['username']

            if not smppUser:
                abort(404)

            yield proxy_router.user_remove(smppUser) 
    except Exception, e:
        yield "%s" %str(e)

    finally:
        print "Stopping Reactor"
        reactor.stop()

Please help me out to solve this issue:

Answer 1

Reactor is not restartable in Twisted by design, it performs initialization and finalization that isn't easily restartable.

In the example provided you're using a development WSGI server (flask's default one, http://werkzeug.pocoo.org/docs/0.11/serving/) which appears to be single-threaded by default.

Your problem would go away if you avoid using threads and switch to a multi-process server instead. i.e. it would work if you just run it like this (see processes=2 => each request will be handled in a new process, but no more than 2 concurrent ones):

if __name__ == '__main__':                                                      
    JasminWebservice.run(host="0.0.0.0", port=7034, debug=True, processes=2)

But I wouldn't rely on that – you'll run into similar troubles when writing unit tests and restricting your app to run in multi-process environment only is not a good approach.

But looks like the problem stems from your app design – why would you need Flask and an additional WSGI server? You can build REST API fully in twisted ending up running only a single reactor which would handle both queries to your API and incoming requests.

Answer 2

You can't stop the reactor and run it again. The error you're getting is by design. Consider using klein it uses werkzeug like flask and leverages twisted for networking as well. The syntax is even similar. Translating your code into klein would look a bit like this:

import json
from klein import Klein
from exception werkzeug.exceptions import NotFound
from JasminIntegration import *

JasminWebservice = Klein()

@JasminWebservice.handle_errors(NotFound)
def not_found(request, error):
    request.setResponseCode(404)
    return json.dumps({'error': 'Not found'})

@JasminWebservice.route('/jsms/webservice', methods=['POST'])
def create_user(request):
    try:
        data = json.loads(request.content.read())
        if not data or if 'username' not in data:
            raise NotFound()
    except:     # yeah I know this isn't best practice
        raise NotFound()

    runScenario(request)
    request.setResponseCode(201)
    return json.dumps({'response':'Success'})

if __name__ == '__main__':    
    JasminWebservice.run("0.0.0.0", port=7034)

As a side note, you shouldn't stop the reactor unless you want to exit your app entirely.

Answer 3

@ffeast I also tried to do it in twisted but encountered the same problem, as reactor is not restartable.In twisted I have done something like this:

#!/usr/bin/env python
from pprint import pprint
import json
from twisted.web import server, resource
from twisted.internet import reactor
from JasminIntegration import *
import ast

class Simple(resource.Resource):
    isLeaf = True
    def render_GET(self, request):
        return "<html>Hello, world!</html>"

    def render_POST(self, request):
        pprint(request.__dict__)
        newdata = request.content.getvalue()
        newdata = ast.literal_eval(newdata)
        ret = runScenario(newdata)
        #print  request.content
        #print newdata
        return ''
site = server.Site(Simple())
reactor.listenTCP(7034, site)
reactor.run()