Twisted makes it easy to implement custom network applications.
Here's a TCP server that echoes back everything that's written to it:
from twisted.internet import protocol, reactor, endpoints
class Echo(protocol.Protocol):
def dataReceived(self, data):
self.transport.write(data)
class EchoFactory(protocol.Factory):
def buildProtocol(self, addr):
return Echo()
endpoints.serverFromString(reactor, "tcp:1234").listen(EchoFactory())
reactor.run()
Learn more about
writing servers,
writing clients and the
core networking libraries, including support for
SSL, UDP, scheduled events, unit testing infrastructure, and much more.
Twisted includes an event-driven web server. Here's a sample web application;
notice how the resource object persists in memory, rather than being recreated on each request:
from twisted.web import server, resource
from twisted.internet import reactor, endpoints
class Counter(resource.Resource):
isLeaf = True
numberRequests = 0
def render_GET(self, request):
self.numberRequests += 1
request.setHeader(b"content-type", b"text/plain")
content = u"I am request #{}\n".format(self.numberRequests)
return content.encode("ascii")
endpoints.serverFromString(reactor, "tcp:8080").listen(server.Site(Counter()))
reactor.run()
Learn more about
web application development,
templates and Twisted'
HTTP client.
Here's a simple publish/subscribe server, where clients see all messages posted by other clients:
from twisted.internet import reactor, protocol, endpoints
from twisted.protocols import basic
class PubProtocol(basic.LineReceiver):
def __init__(self, factory):
self.factory = factory
def connectionMade(self):
self.factory.clients.add(self)
def connectionLost(self, reason):
self.factory.clients.remove(self)
def lineReceived(self, line):
for c in self.factory.clients:
source = u"<{}> ".format(self.transport.getHost()).encode("ascii")
c.sendLine(source + line)
class PubFactory(protocol.Factory):
def __init__(self):
self.clients = set()
def buildProtocol(self, addr):
return PubProtocol(self)
endpoints.serverFromString(reactor, "tcp:1025").listen(PubFactory())
reactor.run()
You can test this out by opening two terminals and doing telnet localhost 1025 in each, then typing things.
Twisted includes a sophisticated IMAP4 client library.
import sys
from twisted.internet import protocol, defer, endpoints, task
from twisted.mail import imap4
from twisted.python import failure
async def main(
reactor, username="alice", password="secret", strport="tls:example.com:993"
):
endpoint = endpoints.clientFromString(reactor, strport)
factory = protocol.Factory.forProtocol(imap4.IMAP4Client)
try:
client = await endpoint.connect(factory)
await client.login(username.encode("utf-8"),
password.encode("utf-8"))
await client.select("INBOX")
info = await client.fetchEnvelope(imap4.MessageSet(1))
print("First message subject:", info[1]["ENVELOPE"][1])
except:
print("IMAP4 client interaction failed")
print(failure.Failure().getTraceback())
task.react(lambda *a, **k: defer.ensureDeferred(main(*a, **k)), sys.argv[1:])
Give this a try, supplying your IMAP4 username, app password
(generate one for gmail,
generate one for fastmail), and
client endpoint description
for your IMAP4 server. You'll see the subject of the first message in your mailbox printed.
See the
TwistedMail
documentation for more information.
Twisted includes an SSH client & server, "conch" (i.e.: the Twisted Shell).
import sys, os
from twisted.internet import protocol, defer, endpoints, task
from twisted.conch.endpoints import SSHCommandClientEndpoint
async def main(reactor, username="alice", sshhost="example.com", portno="22"):
envAgent = endpoints.UNIXClientEndpoint(reactor, os.environ["SSH_AUTH_SOCK"])
endpoint = SSHCommandClientEndpoint.newConnection(
reactor, "echo 'hello world'", username, sshhost,
int(portno), agentEndpoint=envAgent,
)
class ShowOutput(protocol.Protocol):
received = b""
def dataReceived(self, data):
self.received += data
def connectionLost(self, reason):
finished.callback(self.received)
finished = defer.Deferred()
factory = protocol.Factory.forProtocol(ShowOutput)
await endpoint.connect(factory)
print("SSH response:", await finished)
task.react(lambda *a, **k: defer.ensureDeferred(main(*a, **k)), sys.argv[1:])
You can use this client to run "hello world" on any SSH server
that your local SSH agent can authenticate to,
if you pass your username, host name, and optionally port number on the command line.