Запуск скрученного реактора в iPython

Question

Я знаю, что это обычно делается с помощью twistd, но я хочу использовать iPython для проверки кода «live» на скрученном коде.

How to start twisted's reactor from ipython спросил в основном то же самое, но первое решение больше не работает с текущим ipython/twisted, а второе также непригодно (поток вызывает несколько ошибок).

https://gist.github.com/kived/8721434 имеет что-то, называемое TPython, которое предназначено для этого, но работает, похоже, работает, за исключением того, что клиенты никогда не подключаются к серверу (при запуске одних и тех же клиентов работает в оболочке python).

У меня есть, чтобы использовать Conch Manhole, или есть способ заставить iPython играть хорошо (возможно, с _threadedselect).

Для справки, я спрашиваю, используя IPython 5.0.0, 2.7.12 питона, витую 16.4.1

Answer 1

Асинхронный код в целом может быть проблематичным для запуска в реальном интерпретаторе. Лучше всего запустить скрипт async в фоновом режиме и сделать свой материал iPython в отдельном интерпретаторе. Вы можете взаимодействовать с файлами или TCP. Если это пошло по вашей голове, это потому, что это не всегда просто, и лучше всего избегать хлопот.

Тем не менее, вы будете рады узнать, что есть отличный проект под названием crochet для использования Twisted в неасинхронных приложениях. Это действительно один из моих любимых модулей, и я шокирован тем, что он не используется более широко (вы можете это изменить, но это правда). В модуле crochet есть декоратор run_in_reactor, который запускает реактор Twisted в отдельном потоке, управляемом самим crochet. Вот пример быстрого класса, который выполняет запросы к API RESTFul Star Wars, а затем сохраняет ответ JSON в списке.

from __future__ import print_function 
import json 

from twisted.internet import defer, task 
from twisted.web.client import getPage 

from crochet import run_in_reactor, setup as setup_crochet 
setup_crochet() 

class StarWarsPeople(object): 
    people_id = [_id for _id in range(1, 89)] 
    people = [] 

    @run_in_reactor 
    def requestPeople(self): 
     """ 
     Request Star Wars JSON data from the SWAPI site. 
     This occurs in a Twisted reactor in a separate thread. 
     """ 
     for _id in self.people_id: 
      url = 'http://swapi.co/api/people/{0}'.format(_id).encode('utf-8') 
      d = getPage(url) 
      d.addCallback(self.appendJSON) 

    def appendJSON(self, response): 
     """ 
     A callback which will take the response from the getPage() request, 
     convert it to JSON, then append it to self.people, which can be 
     accessed outside of the crochet thread. 
     """ 
     response_json = json.loads(response.decode('utf-8')) 
     #print(response_json) # uncomment if you want to see output 
     self.people.append(response_json) 

Сохранить в файл (например: swapi.py), открытая IPython, импортировать вновь созданный модуль, а затем запустить быстрый тест, как так:

from swapi import StarWarsPeople 
testing = StarWarsPeople() 
testing.requestPeople() 

from time import sleep 
for x in range(5): 
    print(len(testing.people)) 
    sleep(2) 

Как вы можете видеть, что это работает в фон и прочее все еще могут встречаться в основном потоке. Вы можете продолжать использовать интерпретатор iPython, как обычно. У вас даже может быть люк, бегущий в фоновом режиме, для какого-то классного взлома!

Ссылки

• https://crochet.readthedocs.io/en/1.5.0/introduction.html#crochet-use-twisted-anywhere

Answer 2

Хотя это не дает ответа на вопрос, который я думал, что я имел, это не ответ (вроде) вопрос, который я разместил. Встраивание ipython работает в том смысле, что вы получаете доступ к бизнес-объектам с работающим реактором.

from twisted.internet import reactor 
from twisted.internet.endpoints import serverFromString 
from myfactory import MyFactory 

class MyClass(object): 
    def __init__(self, **kwargs): 
     super(MyClass, self).__init__(**kwargs) 
     server = serverFromString(reactor, 'tcp:12345') 
     server.list(MyFactory(self)) 
     def interact(): 
      import IPython 
      IPython.embed() 
     reactor.callInThread(interact) 

if __name__ == "__main__": 
    myclass = MyClass() 
    reactor.run() 

Позвоните выше, используя python myclass.py или аналогичный.