zoukankan      html  css  js  c++  java
  • 用PyRestful快速构建Tornado下REST APIs 的支持

     一、安装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("
    Stop 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("
    Stop 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("
    Stop 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'))
  • 相关阅读:
    hdu2138(求素数)
    hdu2104
    poj1664(放苹果)
    数塔问题给你有哪些启示?
    汉诺塔问题(1)
    算法的力量(转李开复)
    最长子序列问题之系列一
    forward和redirect的区别
    group by 和having
    java中的多态三要素是什么?
  • 原文地址:https://www.cnblogs.com/jsben/p/6027397.html
Copyright © 2011-2022 走看看