Skip to content

Instantly share code, notes, and snippets.

@trenton42
Created October 10, 2012 15:35
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save trenton42/1be05d53900598fdcc44 to your computer and use it in GitHub Desktop.
Twisted Examples
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()
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()
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()
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()
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()
#!/usr/bin/python
import sys
import time
for line in sys.stdin:
time.sleep(1)
print 'len(%s) = %s' % (line.strip(), len(line))
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()
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'))
# 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