一、安装PyRestful库
$ pip install pyrestful
二、使用案例
（一）books_service.py
# -*- coding: utf-8 -*- import tornado.ioloop
import pyrestful.rest from pyrestful import mediatypes
from pyrestful.rest import get, post, put, delete class Book(object):
isbn = int
title = str class BookResource(pyrestful.rest.RestHandler):
@get(_path="/books/json/{isbn}", _types=[int], _produces=mediatypes.APPLICATION_JSON)
def getBookJSON(self, isbn):
book = Book()
book.isbn = isbn
book.title = "My book for isbn "+str(isbn) return book @get(_path="/books/xml/{isbn}", _types=[int], _produces=mediatypes.APPLICATION_XML)
def getBookXML(self, isbn):
book = Book()
book.isbn = isbn
book.title = "My book for isbn "+str(isbn) return book @post(_path="/books/xml",_types=[Book],_consumes=mediatypes.APPLICATION_XML, _produces=mediatypes.APPLICATION_XML)
def postBookXML(self, book):
""" this is an echo...returns the same xml document """
return book @post(_path="/books/json",_types=[Book],_consumes=mediatypes.APPLICATION_JSON, _produces=mediatypes.APPLICATION_JSON)
def postBookJSON(self, book):
""" this is an echo...returns the same json document """
return book @post(_path="/books",_types=[Book])
def postBook(self, book):
""" this is an echo, returns json or xml depending of request content-type """
return book if __name__ == '__main__':
try:
print("Start the service")
app = pyrestful.rest.RestService([BookResource])
app.listen(8080)
tornado.ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
print("\nStop the service")
（二）create_customer.py
# -*- coding: utf-8 -*- import json
import sys if sys.version_info > (3,):
raw_input = input
import http.client as httplib
import urllib.parse as urllib
else:
import httplib
import urllib print('Create customer')
print('===============')
name_customer = raw_input('Customer Name : ')
address_customer = raw_input('Customer Address : ') if len(name_customer) == 0 and len(address_customer) == 0:
print('You must indicates name and address of customer')
else:
params = urllib.urlencode({'name_customer':name_customer,'address_customer':address_customer})
headers = {"Content-Type": "application/x-www-form-urlencoded"}
conn = httplib.HTTPConnection("localhost:8080") conn.request('POST','/customer',params,headers) resp = conn.getresponse()
data = resp.read()
if resp.status == 200:
json_data = json.loads(data.decode('utf-8'))
print(json_data)
else:
print(data)
（三）customer_service.py
# -*- coding: utf-8 -*- import tornado.ioloop
import pyrestful.rest from pyrestful import mediatypes
from pyrestful.rest import get, post, put, delete class Customer(object):
id_customer = int
name_customer = str
address_customer = str
def __init__(self,id_customer=0, name_customer=None, address_customer=None):
self.id_customer = id_customer
self.name_customer = name_customer
self.address_customer = address_customer
# Setters
def setId_Customer(self,id_customer):
self.id_customer = id_customer
def setName_Customer(self,name_customer):
self.name_customer = name_customer
def setAddress_Customer(self,address_customer):
self.address_customer = address_customer
# Getters
def getId_Customer(self):
return self.id_customer
def getName_Customer(self):
return self.name_customer
def getAddress_Customer(self):
return self.address_customer class CustomerDataBase(object):
customerDB = dict()
id_seq = 1 def insert(self, name_customer, address_customer):
sequence = self.id_seq
customer = Customer(sequence, str(name_customer), str(address_customer))
self.customerDB[sequence] = customer
self.id_seq += 1 return sequence def update(self,id_customer, name_customer, address_customer):
if self.exists(id_customer):
customer = self.customerDB[id_customer]
customer.setName_Customer(str(name_customer))
customer.setAddress_Customer(str(address_customer))
self.customerDB[id_customer] = customer
return True
else:
return False def delete(self,id_customer):
if self.exists(id_customer):
del self.customerDB[id_customer]
return True
else:
return False def find(self,id_customer):
if self.exists(id_customer):
return self.customerDB[id_customer]
else:
return None def exists(self,id_customer):
if id_customer in self.customerDB:
return True
else:
return False def all(self):
return self.customerDB class CustomerResource(pyrestful.rest.RestHandler):
def initialize(self, database):
self.database = database @get(_path="/customer", _produces=mediatypes.APPLICATION_JSON)
def getListCustomer(self):
customers = self.database.all() response = dict()
for k in customers.keys():
cust = dict()
cust['id_customer'] = customers[k].getId_Customer()
cust['name_customer'] = customers[k].getName_Customer()
cust['address_customer'] = customers[k].getAddress_Customer()
response[k] = { k : cust } return response @get(_path="/customer/{id_customer}", _types=[int], _produces=mediatypes.APPLICATION_JSON)
def getCustomer(self, id_customer):
if not self.database.exists(id_customer):
self.gen_http_error(404,"Error 404 : do not exists the customer : %d"%id_customer)
return customer = self.database.find(id_customer) response = dict()
response['id_customer'] = customer.getId_Customer()
response['name_customer'] = customer.getName_Customer()
response['address_customer'] = customer.getAddress_Customer()
print(response)
return response @post(_path="/customer", _types=[str,str], _produces=mediatypes.APPLICATION_JSON)
def createCustomer(self, name_customer, address_customer):
id_customer = self.database.insert(name_customer, address_customer) return {"created_customer_id": id_customer} @put(_path="/customer/{id_customer}", _types=[int,str,str], _produces=mediatypes.APPLICATION_JSON)
def updateCustomer(self, id_customer, name_customer, address_customer):
if not self.database.exists(id_customer):
self.gen_http_error(404,"Error 404 : do not exists the customer : %d"%id_customer)
return updated = self.database.update(id_customer,name_customer,address_customer) return {"updated_customer_id": id_customer, "success":updated} @delete(_path="/customer/{id_customer}", _types=[int], _produces=mediatypes.APPLICATION_JSON)
def deleteCustomer(self,id_customer):
if not self.database.exists(id_customer):
self.gen_http_error(404,"Error 404 : do not exists the customer : %d"%id_customer)
return deleted = self.database.delete(id_customer) return {"delete_customer_id": id_customer, "success":deleted} if __name__ == '__main__':
try:
print("Start the service")
database = CustomerDataBase()
app = pyrestful.rest.RestService([CustomerResource], dict(database=database))
app.listen(8080)
tornado.ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
print("\nStop the service")
（四）delete_customer.py
# -*- coding: utf-8 -*- import json
import sys if sys.version_info > (3,):
raw_input = input
import http.client as httplib
import urllib.parse as urllib
else:
import httplib
import urllib print('Delete customer')
print('===============')
id_customer = raw_input('Id Customer : ') if len(id_customer) == 0:
print('You must indicates id of customer')
else:
conn = httplib.HTTPConnection("localhost:8080") conn.request('DELETE','/customer/%s'%id_customer) resp = conn.getresponse()
data = resp.read()
if resp.status == 200:
json_data = json.loads(data.decode('utf-8'))
print(json_data)
else:
print(data.decode('utf-8'))
（五）echo_service.py
import tornado.ioloop
import pyrestful.rest from pyrestful import mediatypes
from pyrestful.rest import get class EchoService(pyrestful.rest.RestHandler):
@get(_path="/echo/{name}", _produces=mediatypes.APPLICATION_JSON)
def sayHello(self, name):
return {"Hello":name} if __name__ == '__main__':
try:
print("Start the echo service")
app = pyrestful.rest.RestService([EchoService])
app.listen(8080)
tornado.ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
print("\nStop the echo service")
（六）update_customer.py
# -*- coding: utf-8 -*- import sys
import json if sys.version_info > (3,):
raw_input = input
import http.client as httplib
import urllib.parse as urllib
else:
import httplib
import urllib print('Update customer')
print('===============')
id_customer = raw_input('Id Customer : ')
name_customer = raw_input('Customer Name : ')
address_customer = raw_input('Customer Address : ') if len(id_customer) == 0 and len(name_customer) == 0 and len(address_customer) == 0:
print('You must indicates id, name and address of customer')
else:
params = urllib.urlencode({'name_customer':str(name_customer),'address_customer':str(address_customer)})
headers = {"Content-type": "application/x-www-form-urlencoded"}
conn = httplib.HTTPConnection("localhost:8080") conn.request('PUT','/customer/%s'%id_customer,params,headers) resp = conn.getresponse()
data = resp.read()
if resp.status == 200:
json_data = json.loads(data.decode('utf-8'))
print(json_data)
else:
print(data.decode('utf-8'))