Подключение к ExtSdk2 через OLE - Сценарии ExtSdk2 — различия между версиями
Строка 27: | Строка 27: | ||
== Пример реализации получения ответов от COM-объекта == | == Пример реализации получения ответов от COM-объекта == | ||
<source lang="json"> | <source lang="json"> | ||
+ | |||
def read_all_xml(self, requests): | def read_all_xml(self, requests): | ||
''' | ''' |
Версия 09:53, 14 апреля 2021
Для работы со СБИС3 Плагин из внешних систем, таких как 1С или SAP, требуется получить COM-объект, зарегистрированный Плагином в Вашей системе. Его название "Tensor.SbisPluginClientCOM".
Получение COM-объекта
class OleSbis3Plugin:
def __init__(self, **kwargs):
self.awaited_answers = {}
self.ole = win32com.client.Dispatch("Tensor.SbisPluginClientCOM")
self.version = None
self._wait_connected()
self.host = None
self.account_id = None
self.guid_module = None
def _wait_connected(self):
event = self.read_all_xml(['Event_connected'])
self.version = event[0]['data']['Version']
return
Из данного примера кода видно, что получение COM-объекта занимает одну строку "self.ole = win32com.client.Dispatch("Tensor.SbisPluginClientCOM")" с использованием библиотеки pywin32. Далее Плагин подготавливает необходимые внутренние инструменты для подключения и в случае успеха пробрасывает событие с именем "Event_connected". Успешное подключение следует обработать в отдельной функции "_wait_connected()" как указано в примере. Она ожидает результат работы функции "read_all_xml()"
Пример реализации получения ответов от COM-объекта
def read_all_xml(self, requests):
'''
Чтение и обработка событий сконвертированных в json_xml
:param requests: массив идентификаторов запросов, на которые мы ждем ответ.
:return:
'''
supported_events = {
'Message': self.parse_Message,
'Error': self.parse_Error,
'Event': self.parse_Event
}
result = []
while len(result) == 0:
xml_string = self.ole.ReadAllXml2()
events = XmlJsonSerializer.decode(xml_string)
for event in events:
event_type = event.get('type')
if event_type not in supported_events:
raise Exception(f'Warning not supported event {event_type}')
supported_events[event_type](event)
if requests is not None:
try:
query_id = requests.index(event['queryID'])
if query_id >= 0:
requests.pop(query_id)
result.append(self.pars_event(event))
except ValueError:
pass
pass
return result