您的位置:首页 > 其它

用PyRestful快速构建Tornado下REST APIs 的支持

2016-11-03 17:16 931 查看
一、安装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'))
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: