This is a pre-production deployment of Warehouse. Changes made here affect the production instance of PyPI (
Help us improve Python packaging - Donate today!

LaunchKey Asynchronous SDK for Twisted

Project Description


The LaunchKey Twisted library allows for asynchronous communication with the LaunchKey Engine API via a Python library and Twisted. The public interface is the same as the synchronous SDK with the two exceptions:

  1. The first argument of the init script for launchkey_twisted.TwistedAPI is a Twisted Web Client Agent.
  2. All public methods return Twisted Deferred objects.


$ easy_install launchkey-twisted


$ pip install launchkey-twisted

The following example is a functional representation of a simple session based launch request and subsequent de-orbit if applicable. The use of polling is not suggested for production server based implementations.

import sys
from twisted.python import log
from twisted.internet import task, reactor, defer
from twisted.web.client import Agent
from launchkey_twisted import TwistedAPI, PendingResponse

# agent is needed to link the SDK to the reactor as an HTTP client
agent = Agent(reactor)

# app_key will be provided in the dashboard
app_key = 1234567890

# app_secret will be provided in the dashboard once, or a new one may be generated
app_secret = "abcdefghijklmnopqrstuvwxyz123456"

# private key location will will be the path to the private key file you received from the dashboard or the
# private key file you used to generate the public key you uploaded to the dashboard
private_key_location = "/path/to/private.key"

# Your LaunchKey username
username = "myusername"

# Poll interval determines the delay in seconds between poll requests
poll_interval = 1.0

# Log out delay determines the delay in seconds between an accepted authentication and logout
log_out_delay = 3.0

# If you are having issues, set this to true for verbose logging onthe reactor
debug = False

def handle_auth_response(auth_request):
    Receives the auth_request identifier and begin the polling process
    :param auth_request: Identifier for this authentication request returned by the LaunchKey Engine API
    :return: Deferred
    print("Authentication request successfully initiated with identifier: {0}".format(auth_request))
    return poll_for_user_repsonse(auth_request)

def poll_for_user_repsonse(auth_request):
    Schedule the next poll operation.   Adds callback for success and error situations
    :param auth_request: Identifier for this authentication request returned by the LaunchKey Engine API
    :return: Deferred
    print("Poll for user response in {0} seconds".format(poll_interval))
    d = task.deferLater(reactor, poll_interval, api.poll_request, auth_request)
    d.addCallback(handle_poll_response, auth_request)
    return d

def handle_poll_error(failure):
    Trap PendingResponse errors and schedule another poll
    :param failure:
    :return: Deferred
    return poll_for_user_repsonse(failure.value.auth_request)

def handle_poll_response(response, auth_request):
    Receive the poll response and the original auth_request to validate the poll response package and detyermine the
    :param response: Poll response object
    :param auth_request: Identifier for this authentication request returned by the LaunchKey Engine API
    :return: Deferred
    print("User response received, checking the response package")
    deferred = api.is_authorized(auth_request, response['auth'])
    deferred.addCallback(handle_is_authorized_response, auth_request)
    return deferred

def handle_is_authorized_response(authorized, auth_request):
    Receive the is_authorized response and schedule the logout if authorized
    :param authorized: Boolean value depicting the users response to the auth_request.
    (True: accepted | False: declined)
    :param auth_request: Identifier for this authentication request returned by the LaunchKey Engine API
    :return: Deferred
    if authorized:
        print("User accepted, scheduling logout for {0} seconds from now".format(log_out_delay))
        deferred = task.deferLater(reactor, log_out_delay, log_user_out, auth_request)
        print("User declined")
        deferred = defer.succeed(True)

    return deferred

def log_user_out(auth_request):
    End the user session for the provided authentication request
    :param auth_request: Identifier for this authentication request returned by the LaunchKey Engine API
    :return: Deferred
    print("Logging user out")
    deferred = api.logout(auth_request)
    return deferred

def handle_error(failure):
    Print an error
    :param failure:
    print("Completed with error: ", failure)
    return failure

def done(*args):
    Stop the reactor
    :param args:
    print("Reactor stopped")

agent = Agent(reactor)

api = TwistedAPI(
    open(private_key_location, "r").read()

print("Starting authentication request for user: {0}".format(username))
auth = api.authorize(username, True)

if debug:

print("Starting reactor")


$ python nosetests


  1. Fork it
  2. Create your feature branch (git checkout -b my-new-feature)
  3. Commit your changes (git commit -am 'Add some feature')
  4. Push to the branch (git push origin my-new-feature)
  5. Create new Pull Request
Release History

Release History

This version
History Node


Download Files

Download Files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

File Name & Checksum SHA256 Checksum Help Version File Type Upload Date
launchkey-twisted-1.0.0.tar.gz (5.9 kB) Copy SHA256 Checksum SHA256 Source Aug 25, 2015

Supported By

WebFaction WebFaction Technical Writing Elastic Elastic Search Pingdom Pingdom Monitoring Dyn Dyn DNS Sentry Sentry Error Logging CloudAMQP CloudAMQP RabbitMQ Heroku Heroku PaaS Kabu Creative Kabu Creative UX & Design Fastly Fastly CDN DigiCert DigiCert EV Certificate Rackspace Rackspace Cloud Servers DreamHost DreamHost Log Hosting