python rpyc客户端怎么调用服务端

蜗牛 互联网技术资讯 2022-06-18 22 0

这篇文章主要介绍“python rpyc客户端怎么调用服务端”,在日常操作中,相信很多人在python rpyc客户端怎么调用服务端问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”python rpyc客户端怎么调用服务端”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

    python rpyc客户端调用服务端方法

    使用rpyc遇到两个问题:

    问题1

    client调用server的exposed_xxxx()方法时传进去的参数类型,到了server端会改变,

    打印出来是<netref class 'rpyc.core.netref.type'>类型,与原来传进的类型不同

    解决方法:

    def exposed_clentA_will_use(self, hello):
        hello = rpyc.classic.obtain(hello)

    这样就把传进来的hello转变为本地类型,如果本地没有这种类型,需要创建这种类型

    比如传进来的是<class 'numpy.ndarray'>类型,但是server本地没有,则需要import numpy

    问题2

    client调用server的方法时有报错ValueError: pickling is disabled

    解决方法:

    rpyc_config = rpyc.core.protocol.DEFAULT_CONFIG
    rpyc_config['allow_pickle'] = True 
    host_ip = '127.0.0.1' 
    conn = rpyc.connect(host_ip, port = 33333, config = rpyc_config)

    由于是不安全的协议,需要手动允许pickle

    字面意思也明确了:allow_pickle

    python rpyc的使用心得

    6月份,在公司做APP和Web自动化测试开发,碰到一个比较棘手的问题是:APP需要和Web一起操作才能走完整个测试流程。

    但是APP和Web使用的是不同的测试工具,Appium和Selenium,测试框架也是独立的。如何把二者整合起来呢?

    没错,python rpyc模块派上用场了,这里使用python rpyc对Web页面相关的操作做了二次封装。

    from rpyc import Service
    from rpyc.utils.server import ThreadedServer
    class MyService(Service):
        def exposed_bid(xxx):
            back_page.bid(xxx)
    
    s=ThreadedServer(MyService,port=12233,listener_timeout=300,auto_register=False)
    s.start()

    在APP自动化测试脚本里面,这样调用即可:

    c = rpyc.connect('localhost',12233)
    c.root.bid(xxxx)

    但是在APP测试脚本运行过程中,会出现rpyc调用超时的情况,这是因为rpyc默认的调用等待时间只有30s,如果某些rpyc调用的等待时间超过30s就会报错,解决方法很简单,在rpyc包的protocol.py文件中找到
    SYNC_REQUEST_TIMEOUT,改大即可。

    到此,关于“python rpyc客户端怎么调用服务端”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注蜗牛博客网站,小编会继续努力为大家带来更多实用的文章!

    免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:niceseo99@gmail.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

    评论