Dummy Flights web services tester

console.py

#!/usr/bin/python

import httplib, time
from flights_pb2 import SearchRequest

class FlightsService(object):
	"""Basic Flights service implementation with common methods"""
	
	def __init__(self,host="10.0.0.210",port=80):
		self.__ws = httplib.HTTPConnection(host,port)
		self.__serviceurl = "/servicestack/search"
		self.__serviceurlget = "/servicestack/search/{}"
		self._headers = {}
		self._message = None
		self.dump = False
		
	def enablecompression(self):
		self._headers["Accept-Encoding"] = "gzip"
		
	def search(self):
		self.__ws.request("POST",self.__serviceurl,self._message,self._headers)
		self.__showresponse()

	def list(self):
		self.__wsget(self.__serviceurl)
    
	def get(self,id):
		self.__wsget(self.__serviceurlget.format(id))
    
	def __wsget(self,url):
		start = time.time()
		self.__ws.request("GET",url,headers=self._headers)
		self.__showresponse()
		end = time.time()
		print("Elapsed {}".format(end-start))
	
	def __showresponse(self):
		response = self.__ws.getresponse()
		data = response.read()

		print(response.status, response.reason, response.getheader("Content-Type"),response.getheader("Content-Encoding"),len(data))
		if self.dump:
			print(data)

class FlightsServiceJson(FlightsService):
	"""Flights Service Implementation using JSON"""
	
	def __init__(self):
		super(FlightsServiceJson, self).__init__()
		self._message = """{
	"Origin":"ATH",
	"Destination":"ATL",
	"DepartureDate":"2011-12-01",
	"OneWay":false,
	"ReturnDate":"2011-12-12"
	"NumOfAdults":1}"""
		self._headers = {
			"Content-Type":"application/json; charset=utf8",
			"Accept":"application/json"
		}
class FlightsServicePb(FlightsService):
	"""Flights Service Implementation using Protocol Buffers
	
	This class needs the .proto compiled to python code to run.
	"""

	def __init__(self):
		super(FlightsServicePb, self).__init__()
		sr = SearchRequest()
		sr.Origin="LON"
		sr.Destination="ATH"
		sr.DepartureDate="2011-12-01"
		sr.OneWay=False
		sr.ReturnDate="2011-12-12"
		sr.NumOfAdults=1
	
		self._message = sr.SerializeToString()
		self._headers = {
			"Content-Type":"application/x-protobuf",
			"Accept":"application/x-protobuf",
		}
		
def FlightServiceFactory(type):
	if type == 'json':
		return FlightsServiceJson()
	elif type == 'pb':
		return FlightsServicePb()

def parseargs():
	import argparse

	parser = argparse.ArgumentParser(description='Flights Shell')
	parser.add_argument('-l','--list',action='store_true',help='List available searches')
	parser.add_argument('-c','--compress',action='store_true',help='Use gzip')
	parser.add_argument('-g','--get', dest='id',help='Gets an specific search id')
	parser.add_argument('-d','--dump', action='store_true',help='Print contents of the response')
	parser.add_argument('-f','--format', choices=('json','pb'),help='Serialization format',default='json')
	return parser.parse_args()
	
if __name__ == "__main__":
	args = parseargs()
	
	# Content Type
	service = FlightServiceFactory(args.format)

	# Compression
	if args.compress:
		service.enablecompression()
		
	# Dump to stdout
	service.dump = args.dump

	# Action
	if args.list:
		service.list()
	elif args.id != None:
		service.get(args.id)
	else:
		service.search()