Twisted Examples
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from twisted.internet import reactor | |
from twisted.web import client | |
def loadData(url): | |
response = client.getPage(url) | |
return response | |
if __name__ == "__main__": | |
print 'I think we have some data!? It is:', loadData('http://google.com') | |
reactor.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from twisted.internet import reactor | |
from twisted.web import client | |
def loadData(url): | |
response = client.getPage(url) | |
return response | |
if __name__ == "__main__": | |
def gotData(data): | |
print 'We do have some data! Length is:', len(data) | |
reactor.stop() | |
loadData('http://www.google.com').addCallback(gotData) | |
reactor.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from termcolor import colorstr | |
from twisted.internet.protocol import ServerFactory | |
from twisted.protocols.basic import LineReceiver | |
from twisted.internet import reactor | |
class Chattr(LineReceiver): | |
connections = {} | |
username = None | |
def connectionMade(self): | |
self.sendLine(colorstr("Please type your nickname: ",'RED')) | |
def lineReceived(self, line): | |
if self.username is None: | |
un = line.strip() | |
if not len(un): | |
self.sendLine(colorstr('I said, please enter your nickname! ','RED')) | |
else: | |
self.username = un | |
Chattr.connections[un] = self | |
self.sendLine("Welcome, %s" % un) | |
self.sendLine("Type quit to leave") | |
return | |
if line in ('quit', 'die', 'exit'): | |
self.quit() | |
return | |
self.broadcast(colorstr(self.username + ': ', 'GREEN') + line.strip()) | |
def quit(self): | |
self.broadcast(colorstr(self.username, 'GREEN') + ' ' + colorstr('has left the building', 'PURPLE')) | |
self.sendLine(colorstr("I'm going to miss you, %s" % self.username, 'BLUE')) | |
self.transport.loseConnection() | |
if self.username in Chattr.connections: | |
del Chattr.connections[self.username] | |
self.username = None | |
def broadcast(self, message, toself=False): | |
for k,v in Chattr.connections.iteritems(): | |
if k == self.username and not toself: | |
continue | |
v.sendLine(message) | |
class ChatFactor(ServerFactory): | |
protocol = Chattr | |
if __name__ == '__main__': | |
factory = ChatFactor() | |
reactor.listenTCP(5555, factory) | |
reactor.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from twisted.internet import defer, reactor | |
from twisted.web import client | |
@defer.inlineCallbacks | |
def getStuff(): | |
# we want to get two pages and concatenate them together | |
data = yield client.getPage('http://google.com') | |
moredatea = yield client.getPage('http://yahoo.com') | |
defer.returnValue(data + moredatea) | |
def handleStuff(data): | |
# Now we are going to do stuff with the data | |
print "We got the data, and the len is: %s" % len(data) | |
# This example is done, so stop the reactor | |
reactor.stop() | |
if __name__ == '__main__': | |
# getStuff is decorated by inlineCallbacks, so it returns a deferred | |
d = getStuff() | |
d.addCallback(handleStuff) | |
reactor.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from termcolor import colorstr | |
from twisted.internet.protocol import ServerFactory | |
from twisted.protocols.basic import LineReceiver | |
from twisted.internet import reactor | |
import twisted.manhole.telnet | |
class Chattr(LineReceiver): | |
connections = {} | |
username = None | |
def connectionMade(self): | |
self.sendLine(colorstr("Please type your nickname: ",'RED')) | |
def lineReceived(self, line): | |
if self.username is None: | |
un = line.strip() | |
if not len(un): | |
self.sendLine(colorstr('I said, please enter your nickname! ','RED')) | |
else: | |
self.username = un | |
Chattr.connections[un] = self | |
self.sendLine("Welcome, %s" % un) | |
self.sendLine("Type quit to leave") | |
return | |
if line in ('quit', 'die', 'exit'): | |
self.quit() | |
return | |
self.broadcast(colorstr(self.username + ': ', 'GREEN') + line.strip()) | |
def broadcast(self, message, toself=False): | |
for k,v in Chattr.connections.iteritems(): | |
if k == self.username and not toself: | |
continue | |
v.sendLine(message) | |
def quit(self): | |
self.broadcast(colorstr(self.username, 'GREEN') + ' ' + colorstr('has left the building', 'PURPLE')) | |
self.sendLine(colorstr("I'm going to miss you, %s" % self.username, 'BLUE')) | |
self.transport.loseConnection() | |
if self.username in Chattr.connections: | |
del Chattr.connections[self.username] | |
self.username = None | |
class ChatFactor(ServerFactory): | |
protocol = Chattr | |
if __name__ == '__main__': | |
mf = twisted.manhole.telnet.ShellFactory() | |
mf.username = 'admin' | |
mf.password = 'pancakes are gross' | |
mf.namespace['chattr'] = Chattr | |
reactor.listenTCP(8007, mf) | |
factory = ChatFactor() | |
reactor.listenTCP(5555, factory) | |
reactor.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python | |
import sys | |
import time | |
for line in sys.stdin: | |
time.sleep(1) | |
print 'len(%s) = %s' % (line.strip(), len(line)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from twisted.internet import protocol | |
from twisted.internet import reactor | |
class countProto(protocol.ProcessProtocol): | |
def connectionMade(self): | |
print "connection made!" | |
self.transport.write("this is some stuff...\nThis is some more stuff!") | |
self.transport.closeStdin() | |
def outReceived(self, data): | |
print "this is what we got back: %s" % data | |
def processEnded(self, reason): | |
reactor.stop() | |
cp = countProto() | |
reactor.spawnProcess(cp, "process.py", ["process.py"], {}) | |
reactor.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import urllib2 | |
def loadData(url): | |
request = urllib2.urlopen(url) | |
data = request.read() | |
request.close() | |
return data | |
if __name__ == "__main__": | |
print 'we got some data! Length is:', len(loadData('http://google.com')) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# Borrowed from: http://stackoverflow.com/questions/3696430/print-colorful-string-out-to-console-with-python | |
CODE={ | |
'ENDC':0, # RESET COLOR | |
'BOLD':1, | |
'UNDERLINE':4, | |
'BLINK':5, | |
'INVERT':7, | |
'CONCEALD':8, | |
'STRIKE':9, | |
'GREY30':90, | |
'GREY40':2, | |
'GREY65':37, | |
'GREY70':97, | |
'GREY20_BG':40, | |
'GREY33_BG':100, | |
'GREY80_BG':47, | |
'GREY93_BG':107, | |
'DARK_RED':31, | |
'RED':91, | |
'RED_BG':41, | |
'LIGHT_RED_BG':101, | |
'DARK_YELLOW':33, | |
'YELLOW':93, | |
'YELLOW_BG':43, | |
'LIGHT_YELLOW_BG':103, | |
'DARK_BLUE':34, | |
'BLUE':94, | |
'BLUE_BG':44, | |
'LIGHT_BLUE_BG':104, | |
'DARK_MAGENTA':35, | |
'PURPLE':95, | |
'MAGENTA_BG':45, | |
'LIGHT_PURPLE_BG':105, | |
'DARK_CYAN':36, | |
'AUQA':96, | |
'CYAN_BG':46, | |
'LIGHT_AUQA_BG':106, | |
'DARK_GREEN':32, | |
'GREEN':92, | |
'GREEN_BG':42, | |
'LIGHT_GREEN_BG':102, | |
'BLACK':30, | |
} | |
def termcode(num): | |
return '\033[%sm'%num | |
def colorstr(astr,color): | |
return termcode(CODE[color])+astr+termcode(CODE['ENDC']) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment