python服务化接口的新尝试

九 9th, 2013

       是不是有这么一个场景,对外提供一堆数据或者是要返回给用户一个结果。但是不想把内部的一些数据和逻辑暴露给对方。。。简单点来说,就是想以服务的方式对外提供一个接口。对于这种有很多处理方式,RPC,搭建一个web服务啥的。。。。但是这些毕竟都太重量级了,操作起来很麻烦。我这里给出的一种非常easy的方式来处理。使用bottle解决问题。

需求: 检查一个zookeeper服务中的某些节点是否存在,如果存在返回OK,不存在则给出不存的节点信息。要求返回的信息是和pyunit的结果信息一致。

实现环境:

  • 1.  python 2.7  以及自带的pyunit
  • 2.  bottle          作为一个python的简易服务器     pip    install       bottle
  • 3.  kazoo         一个python的zookeeper客户端   pip install kazoo

1. 创建一个python的测试类 zk_check.py

 -*- coding: utf-8 -*-                                                                                                                                                          
                                                                              
from kazoo.client import KazooClient                                         
import unittest                                                              
class zktest(unittest.TestCase):                                              
    def runTest(self):                                                       
        zknamespace = “/app/zktest_performance_1”                             
        zkhosts = “127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183”    
        ZKTEST_DRIVERS = [“ip1”, “ip2”]           
        ZKTEST_NODES = [“ip3”, “ip4”, “ip5”, “ip6”]
                                                                             
        driverChildren = []                                                  
        nodeChildren = []                                                    
        badDrivers = []                                                      
        badNodes = []                                                        
        # checking                                                           
        zk = KazooClient(hosts=zkhosts, read_only=True)                      
        zk.start()                                                           
        driverFatherPath = zknamespace + “/status/drivers”                   
        nodeFatherPath = zknamespace + “/status/nodes”                                                                            
                                               
        if zk.exists(driverFatherPath):                                      
            driverChildren = zk.get_children(driverFatherPath)               
            if len(driverChildren) >                                                                                                  
 
                for driver in zktest_DRIVERS:                                 
                    if driver not in driverChildren:                         
                        badDrivers.append(driver)                                                                      
                                                                              
        if zk.exists(nodeFatherPath):                                        
            nodeChildren = zk.get_children(nodeFatherPath)                   
            if len(nodeChildren) >                                                                                                  
 
                for node in zktest_NODES:                                     
                    if node not in nodeChildren:                             
                        badNodes.append(node)                                                                                                                                               
        zk.stop()
       if (len(badNodes)==0) and (len(badDrivers)==0):                      
            self.assertEquals(1,1,”pass”)                                    
        else:                                                                
            if len(badDrivers) > 0:                                          
                self.assertEquals(1,2,’len : %d , error : %s’ % (len(badDrivers),badDrivers))
            if len(badNodes) > 0:                                            
                self.assertEquals(1,2,’len : %d , error : %s’ % (len(badNodes),badNodes))                                                                             
if __name__ == ‘__main__’:                                                   
    unittest.main()

2. 写一个bottle服务,将结果输出

import commands                                                                                                                                                                   
from bottle import route, run, template                                       
@route(‘/alisa’)                                                              
 def index():                                                                  
      command = “python /Users/metaboy/script/zk_check.py”                  
 
      #output = os.popen(command)                                               
      return template(‘<b>{{text}}</b>’, text=commands.getoutput(command))                                                  
                                                                                
run(host=’localhost’, port=8888) 

3. 后台启动bottle服务,提供外部访问ip

现在可以直接通过  http://localhost:8888/alisa   进行访问。





除非注明,本站文章均为原创。本文基于 BY-NC-SA 协议进行授权,欢迎转载,演绎或用于商业目的,但是必须保留本文的署名 metaboy(包含链接).

本文链接地址: http://blog.wangyuxiong.com/archives/52058

订阅本站:http://www.wangyuxiong.com/feed

分类: 技术尝新 语言编程         标签: , ,
目前还没有任何评论.

无觅相关文章插件,快速提升流量