0

特定のモジュールをインポートする RpyC サーバーがあります。このモジュール内のクラスを公開して、このクラスをクライアント側から継承できるようにする必要があります。

テスト目的で、モジュールのインポート/公開を削除し、RPyC サービス内に exposed_TestClass という単純なクラスを作成しました。

サーバー側: rpyc_server.py

import rpyc
from rpyc.utils.server import ThreadedServer

class MyService(rpyc.Service):
    class exposed_TestClass:
        def exposed_Exec(self):
            print("original print of class")

t = ThreadedServer(MyService, port=12345)
t.start()

クライアント側: python3 シェル

>>> import rpyc
>>> conn = rpyc.connect("localhost", 12345)
>>> conn.root.TestClass
<class 'exposed_TestClass'>
>>> conn.root.TestClass()
<exposed_TestClass object at 0x7f2dda642588>
>>> #calling the Exec function also works, prints at server side
>>> conn.root.TestClass().Exec()
>>>
>>>
>>> # test inheriting class
>>> class MyClass(conn.root.TestClass):
...     def NewMethod(self):
...         print("printing from new method")
... 
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/python3/dist-packages/rpyc/core/netref.py", line 220, in method
    return syncreq(_self, consts.HANDLE_CALLATTR, name, args, kwargs)
  File "/usr/lib/python3/dist-packages/rpyc/core/netref.py", line 74, in syncreq
    conn = object.__getattribute__(proxy, "____conn__")
AttributeError: 'str' object has no attribute '____conn__'
4

1 に答える 1

0

次のドキュメントに従って: https://rpyc.readthedocs.io/en/latest/tutorial/tut2.html

出来上がり、netrefs (ネットワーク参照、透過オブジェクト プロキシとも呼ばれます) は、ローカルで実行されるすべてを対応するリモート オブジェクトに委譲する特別なオブジェクトです。Netref は関数やモジュールの実際のリストではないかもしれませんが、それらが指し示すオブジェクトのように見えるように「最善を尽くします」…実際、それらは Python の内省メカニズムをだますことさえあります!

クライアント側で次のことを行う場合:

import rpyc
conn = rpyc.connect("localhost", 12345)

o=conn.root.TestClass()
type_o = type(o)
print(type_o)

印刷されます：

<netref class 'rpyc.core.netref.__main__.exposed_TestClass'>

だからやろうとするとき：

class MyClass(type_o):
    def NewMethod(self):
        print("printing from new method")

プロキシ クラス定義から継承しますが、リモート クラス定義からは継承しません。したがって、ドキュメントの状態

彼らは「最善を尽くします」</p>

、リモートクラスからの継承を行う可能性を排除すると思います。誰かがあなたの要求を満たす方法を見つけることができれば、私はとてもうれしいです.

次の行を使用してサーバーをセットアップしてテストを行ったことに注意してください。

t = ThreadedServer(MyService, port=12345, protocol_config={ 'allow_all_attrs': True })
于 2020-11-20T04:54:42.180 に答える