tessl/pypi-twisted
An asynchronous networking framework written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
application.mddocs/
Application Framework
Service-oriented application architecture with hierarchical service management, plugin system, and daemon utilities for building long-running applications. Twisted's application framework provides structure for complex server applications.
Capabilities
Service Interface and Implementations
Core service abstractions and implementations for building modular applications.
class service.IService:
"""
Interface for services that can be started and stopped.
Services form the building blocks of Twisted applications.
"""
def startService():
"""Start the service."""
def stopService():
"""
Stop the service.
Returns:
None or Deferred: Completion indicator
"""
def privilegedStartService():
"""Start service with privileges (before dropping root)."""
class service.Service:
"""
Base implementation of IService.
Attributes:
- name: Service name (str)
- parent: Parent service container
- running: Whether service is running (bool)
"""
name = None
parent = None
running = False
def __init__(self):
"""Initialize service."""
def startService(self):
"""
Start the service.
Sets running=True and calls child services' startService.
"""
def stopService(self):
"""
Stop the service.
Returns:
Deferred: Fires when service and all children are stopped
"""
def privilegedStartService(self):
"""Start service with elevated privileges."""
def setName(self, name):
"""
Set service name.
Args:
name (str): Service name
"""
def setServiceParent(self, parent):
"""
Set parent service container.
Args:
parent: Parent service (IServiceCollection)
"""
def disownServiceParent(self):
"""Remove this service from its parent."""
class service.MultiService:
"""
Service container that manages multiple child services.
Attributes:
- services: List of child services
- namedServices: Dict of named child services
"""
services = None
namedServices = None
def __init__(self):
"""Initialize multi-service container."""
def addService(self, service):
"""
Add a child service.
Args:
service: Service to add (IService)
"""
def removeService(self, service):
"""
Remove a child service.
Args:
service: Service to remove
"""
def getServiceNamed(self, name):
"""
Get child service by name.
Args:
name (str): Service name
Returns:
Service: Named service or None
"""
def privilegeStart(self):
"""Start all child services with privileges."""
def startService(self):
"""Start this service and all children."""
def stopService(self):
"""
Stop this service and all children.
Returns:
Deferred: Fires when all services stopped
"""
class service.Application:
"""
Top-level application container.
Attributes:
- name: Application name
- uid: User ID to run as (int)
- gid: Group ID to run as (int)
"""
name = None
uid = None
gid = None
def __init__(self, name, uid=None, gid=None):
"""
Args:
name (str): Application name
uid (int): User ID to switch to
gid (int): Group ID to switch to
"""
def setComponent(self, interface, component):
"""
Register component for interface.
Args:
interface: Component interface
component: Component implementation
"""
def getComponent(self, interface, default=None):
"""
Get component for interface.
Args:
interface: Component interface
default: Default if not found
Returns:
Component or default
"""Service Usage Example:
from twisted.application import service
from twisted.internet import reactor, endpoints
from twisted.web import server, resource
class WebService(service.Service):
"""Custom web service."""
def __init__(self, port, resource):
service.Service.__init__(self)
self.port = port
self.resource = resource
self.site = None
self.listening_port = None
def startService(self):
service.Service.startService(self)
self.site = server.Site(self.resource)
endpoint = endpoints.TCP4ServerEndpoint(reactor, self.port)
self.listening_port = endpoint.listen(self.site)
print(f"Web service started on port {self.port}")
def stopService(self):
if self.listening_port:
d = self.listening_port.stopListening()
print("Web service stopped")
return d
return service.Service.stopService(self)
# Create application with multiple services
application = service.Application("MyApp")
# Create multi-service container
main_service = service.MultiService()
main_service.setServiceParent(application)
# Add web service
web_resource = resource.Resource()
web_service = WebService(8080, web_resource)
web_service.setName("web")
web_service.setServiceParent(main_service)
# Add another service
other_service = service.Service()
other_service.setName("other")
other_service.setServiceParent(main_service)Internet Services
Pre-built service implementations for common network operations.
class internet.TCPServer:
"""
TCP server service.
"""
def __init__(self, port, factory, backlog=50, interface=''):
"""
Args:
port (int): Port to listen on
factory: Protocol factory
backlog (int): Listen backlog size
interface (str): Interface to bind to
"""
class internet.TCPClient:
"""
TCP client service that maintains persistent connection.
"""
def __init__(self, host, port, factory):
"""
Args:
host (str): Server hostname
port (int): Server port
factory: Client factory
"""
class internet.UDPServer:
"""
UDP server service.
"""
def __init__(self, port, protocol, interface='', maxPacketSize=8192):
"""
Args:
port (int): Port to bind to
protocol: UDP protocol instance
interface (str): Interface to bind to
maxPacketSize (int): Maximum packet size
"""
class internet.SSLServer:
"""
SSL server service.
"""
def __init__(self, port, factory, contextFactory, backlog=50, interface=''):
"""
Args:
port (int): Port to listen on
factory: Protocol factory
contextFactory: SSL context factory
backlog (int): Listen backlog
interface (str): Interface to bind to
"""
class internet.SSLClient:
"""
SSL client service.
"""
def __init__(self, host, port, factory, contextFactory):
"""
Args:
host (str): Server hostname
port (int): Server port
factory: Client factory
contextFactory: SSL context factory
"""
class internet.UNIXServer:
"""
Unix domain socket server service.
"""
def __init__(self, address, factory, backlog=50, mode=0o666, wantPID=0):
"""
Args:
address (str): Socket file path
factory: Protocol factory
backlog (int): Listen backlog
mode (int): Socket file permissions
wantPID (bool): Include PID in address
"""
class internet.TimerService:
"""
Service that calls a function periodically.
"""
def __init__(self, step, callable, *args, **kwargs):
"""
Args:
step (float): Interval between calls in seconds
callable: Function to call
*args, **kwargs: Arguments for function
"""
def startService(self):
"""Start the timer service."""
def stopService(self):
"""
Stop the timer service.
Returns:
Deferred: Fires when stopped
"""Internet Services Usage Example:
from twisted.application import service, internet
from twisted.internet import protocol
from twisted.web import server, resource
# Create application
application = service.Application("ServerApp")
# TCP server service
class EchoProtocol(protocol.Protocol):
def dataReceived(self, data):
self.transport.write(data)
echo_factory = protocol.ServerFactory()
echo_factory.protocol = EchoProtocol
tcp_service = internet.TCPServer(8080, echo_factory)
tcp_service.setServiceParent(application)
# Timer service for periodic tasks
def heartbeat():
print("Heartbeat...")
timer_service = internet.TimerService(30.0, heartbeat)
timer_service.setServiceParent(application)
# Web server service
root = resource.Resource()
site = server.Site(root)
web_service = internet.TCPServer(8000, site)
web_service.setServiceParent(application)String Port Specifications
String-based configuration for network services.
def strports.parse(description, factory, *args, **kw):
"""
Parse port specification string and create listening port.
Args:
description (str): Port specification (e.g., "tcp:8080")
factory: Protocol factory
*args, **kw: Additional arguments
Returns:
Port: Listening port object
"""
def strports.listen(description, factory):
"""
Listen on port specified by string description.
Args:
description (str): Port specification
factory: Protocol factory
Returns:
Port: Listening port object
"""
def strports.service(description, factory, *args, **kw):
"""
Create service from port specification string.
Args:
description (str): Port specification
factory: Protocol factory
*args, **kw: Additional arguments
Returns:
Service: Service object
"""String Port Examples:
tcp:8080 # TCP port 8080, all interfaces
tcp:8080:interface=127.0.0.1 # TCP port 8080, localhost only
ssl:8443 # SSL port 8443
unix:/tmp/socket # Unix domain socket
systemd:domain=INET # systemd socket activationApplication Loading and Running
Functions for loading and running Twisted applications.
def app.run(runApp, ServerOptions):
"""
Run a Twisted application.
Args:
runApp: Application callable
ServerOptions: Server configuration options
"""
def app.startApplication(application, save):
"""
Start a Twisted application.
Args:
application: Application object
save (bool): Whether to save application state
"""
def app.getApplication(config, passphrase):
"""
Load application from configuration.
Args:
config: Configuration object
passphrase: Decryption passphrase
Returns:
Application: Loaded application
"""
class app.ServerOptions:
"""
Command-line options for server applications.
Attributes:
- profile: Enable profiling
- profiler: Profiler module to use
- debug: Enable debug mode
- logfile: Log file path
- pidfile: PID file path
- chroot: Chroot directory
- uid: User ID to switch to
- gid: Group ID to switch to
"""
profile = False
profiler = 'hotshot'
debug = False
logfile = None
pidfile = None
chroot = None
uid = None
gid = NoneReactor Selection
Utilities for selecting and installing reactors.
def reactors.installReactor(shortName):
"""
Install named reactor.
Args:
shortName (str): Reactor name ('select', 'epoll', 'kqueue', etc.)
Returns:
Reactor: Installed reactor instance
"""
def reactors.getReactorTypes():
"""
Get available reactor types.
Returns:
dict: Mapping of names to reactor descriptions
"""
def reactors.default():
"""
Get default reactor for current platform.
Returns:
Reactor: Default reactor
"""Reactor Selection Example:
from twisted.application import reactors
# Get available reactors
reactor_types = reactors.getReactorTypes()
print("Available reactors:", list(reactor_types.keys()))
# Install specific reactor
try:
reactor = reactors.installReactor('epoll')
print("Installed epoll reactor")
except ImportError:
reactor = reactors.installReactor('select')
print("Fallback to select reactor")Plugin System
Plugin discovery and loading infrastructure.
class service.IServiceMaker:
"""
Interface for service maker plugins.
Plugins implementing this interface can create services
from command-line options.
"""
tapname = None # Plugin name
description = None # Plugin description
options = None # Options class
def makeService(options):
"""
Create service from options.
Args:
options: Parsed command-line options
Returns:
Service: Created service
"""
# Plugin registration uses Twisted's plugin system
# Plugins are discovered in twisted/plugins/ directoriesPlugin Example:
# In myapp/plugins/myservice_plugin.py
from twisted.application.service import IServiceMaker
from twisted.plugin import IPlugin
from twisted.python import usage
from zope.interface import implementer
class Options(usage.Options):
optParameters = [
['port', 'p', 8080, 'Port to listen on'],
['interface', 'i', '127.0.0.1', 'Interface to bind to'],
]
@implementer(IServiceMaker, IPlugin)
class MyServiceMaker:
tapname = "myservice"
description = "My custom service"
options = Options
def makeService(self, options):
from myapp.service import MyService
return MyService(options['port'], options['interface'])
# Register the plugin
serviceMaker = MyServiceMaker()TAC Files (Twisted Application Configuration)
Configuration files for Twisted applications.
# Example .tac file: application.tac
from twisted.application import service, internet
from twisted.web import server, resource, static
from twisted.python import log
# Create the application
application = service.Application("WebServer")
# Create a web resource
root = static.File("/var/www")
site = server.Site(root)
# Create and configure services
web_service = internet.TCPServer(8080, site)
web_service.setServiceParent(application)
# Optional: Set up logging
log.msg("Web server application configured")Running TAC files:
# Run with twistd
twistd -y application.tac
# Run in foreground with logging
twistd -n -y application.tac
# Run as daemon with specific reactor
twistd -r epoll -y application.tacInstall with Tessl CLI
npx tessl i tessl/pypi-twisted