You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
from typing import Union, Awaitable, Dict
|
|
|
|
from .helpers import AsyncHelper, Helper, RpcEndpoint
|
|
from .rpc import RpcException
|
|
|
|
|
|
class _Caller:
|
|
def __init__(self, caller: RpcEndpoint):
|
|
self.caller = caller
|
|
|
|
def get_url(self, js_code):
|
|
try:
|
|
return self.caller.get_url(js_code=js_code)
|
|
except RpcException as e:
|
|
if e.rpc_error_code == -32113:
|
|
endpoints = self.caller.refresh_endpoints()
|
|
self.caller.set_endpoints(endpoints)
|
|
return self.get_url(js_code)
|
|
raise e
|
|
|
|
async def async_get_url(self, js_code):
|
|
try:
|
|
return await self.caller.get_url(js_code=js_code)
|
|
except RpcException as e:
|
|
if e.rpc_error_code == -32113:
|
|
endpoints = await self.caller.refresh_endpoints()
|
|
self.caller.set_endpoints(endpoints)
|
|
return self.get_url(js_code)
|
|
raise e
|
|
|
|
|
|
class _ACaller:
|
|
def __init__(self, caller: RpcEndpoint):
|
|
self.caller = caller
|
|
|
|
async def get_url(self, js_code):
|
|
try:
|
|
return await self.caller.get_url(js_code=js_code)
|
|
except RpcException as e:
|
|
if e.rpc_error_code == -32113:
|
|
endpoints = await self.caller.refresh_endpoints()
|
|
self.caller.set_endpoints(endpoints)
|
|
return self.get_url(js_code)
|
|
raise e
|
|
|
|
|
|
class ZXGKCaller:
|
|
callers: Dict[str, Union[_Caller, _ACaller]] = {}
|
|
|
|
def __init__(self, uri, async_: bool):
|
|
self.async_ = async_
|
|
self.helper = AsyncHelper(uri) if async_ else Helper(uri)
|
|
|
|
def get_endpoint_caller(self, target) -> Union[RpcEndpoint, Awaitable[RpcEndpoint]]:
|
|
return self.helper.enter(target)
|
|
|
|
def set_endpoint_caller(self, name: str, caller: RpcEndpoint):
|
|
self.callers[name] = _ACaller(caller) if self.async_ else _Caller(caller)
|
|
|
|
def __getattr__(self, item):
|
|
return self.callers[item]
|