Twisted 包含一个事件驱动的 Web 服务器。这是一个示例 Web 应用程序；请注意资源对象如何保存在内存中，而不是在每次请求时重新创建

                  
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()
                
                

了解更多关于 Web 应用程序开发、 模板 和 Twisted 的 HTTP 客户端.

这是一个简单的发布/订阅服务器，客户端可以查看其他客户端发布的所有消息

                  
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()
                  
                

您可以通过打开两个终端并在每个终端中执行 telnet localhost 1025 来测试它，然后输入内容。

Twisted 包含一个复杂的 IMAP4 客户端库。

                  
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:])
                  
                

尝试一下，提供您的 IMAP4 用户名、应用程序密码 (为 gmail 生成一个、为 fastmail 生成一个) 和 您的 IMAP4 服务器的客户端端点描述 。您将看到您邮箱中第一封邮件的主题被打印出来。

查看 TwistedMail 文档以获取更多信息。

Twisted 包含一个 SSH 客户端和服务器，“conch”（即：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:])
                  
                

您可以使用此客户端在您的本地 SSH 代理可以验证的任何 SSH 服务器上运行“hello world”，如果您在命令行上传递您的用户名、主机名和可选的端口号。