Python SDKBETANEW

В этой статье

Python SDK

Структура проекта:

src
- main
  - sdk
    - __init__.py
    - abstract_group.py — абстрактная группа
    - abstract_block.py — абстрактный блок
    - abstract_connection.py — абстрактное подключение
    - argument_processor.py — класс для обработки запроса в блок
    - block_params.py — класс для хранения параметров блоков
    - default_info.py — шаблоны ответа по контракту
    - input_processor.py — класс для обработки старта блоков
    - message_wrapper.py — класс формирования сообщений
    - utils.py — базовые функции
  - packages
    - <user_package_name>
      - blocks
        
        __init__.py
      - connections
        
        __init__.py
      - group.py
      - python_group_<group_hash>.png — иконка блока

Заметка

Пользовательские блоки должны храниться в папке blocks, подключения — в папке connections.

Точка входа (entrypoint.py)

import sys
import logging

from sdk.argument_processor import ArgumentProcessor
from sdk.input_processor import InputProcessor

def main():
    logging.basicConfig(
        level=logging.DEBUG,  # Уровень логирования (DEBUG, INFO, WARNING, ERROR, CRITICAL)
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',  # Формат сообщений
        datefmt='%Y-%m-%d %H:%M:%S',  # Формат времени
        filename=r'debug_app.log',  # Файл для записи логов
        filemode='w',  # Режим записи в файл (w - перезапись, a - добавление)
        # encoding='utf-8' #TODO Python3.8 Error
    )
    logger = logging.getLogger(__name__)
    logger.info('Started!')

    sys.stdout.reconfigure(encoding='utf-8')
    sys.stdin.reconfigure(encoding='utf-8')

    argument_processor = ArgumentProcessor()
    argument_processor.add_arguments()
    exit_process = argument_processor.parse_args()

    logger.info('Input completed!')

    if not exit_process:
        input_processor = InputProcessor(
                                            argument_processor.block_params,
                                            logger
                                            )
        input_processor.start_thread()

if __name__ == "__main__":
    main()

Абстрактная группа (abstract_group.py)

Код абстрактной группы

from abc import abstractmethod

class AbstractGroup():
    def __init__(self):
        pass

    @abstractmethod
    def get_uuid(self)->str:
        return ""

    @abstractmethod
    def get_name(self)->dict:
        return {}

    @abstractmethod
    def get_icon(self)->str:
        return ""

    @abstractmethod
    def get_category(self):
        return ""

    def get_info(self):
        return {
            "uuid":self.get_uuid(),
            "name":self.get_name(),
            "category":self.get_category(),
            "icon":self.get_icon()
        }

Пример реализации абстрактной группы

from sdk.abstract_group import AbstractGroup
import os

class ExampleGroup(AbstractGroup):
    def __init__(self):
        super().__init__()

    def get_uuid(self):
        return "python_group_ed4bf34d-b75c-4ebb-968d-344e0f9c5ec5"

    def get_name(self):
        return {
        "en" : "Python group",
        "ru" : "Группа Python",
    }

    def get_category(self):
        return "tools"

    def get_icon(self):
        return os.path.join(os.path.dirname(os.path.realpath(__file__)), "python_group_ed4bf34d-b75c-4ebb-968d-344e0f9c5ec5.png")

Важно

Особенности при реализации группы:
- get_uuid() — идентификатор подключения в поле info_output содержит пользовательское имя блока и уникальный UUID4 хэш
- get_icon() — имя файла иконки должно быть таким же, как идентификатор блока

Абстрактная блок (abstract_block.py)

Код абстрактного блока

from abc import abstractmethod

class AbstractBlock():
    def __init__(self, block_input=None, block_output=None, info_output=None):
        self.block_input = block_input # Формат входных полей блока
        self.block_output_dict = block_output # Список конфигураций выходных полей блока
        self.block_output = None # Формат выходных полей блока. Выбирается из block_output_dict
        self.info_output = info_output # Информация о блоке при вызове --get-info

        self.aggr_mode = False # Режим работы блока (итеративный/агрегационный)

    def get_info(self):
        return self.info_output

    @abstractmethod
    def process_data(self, **data):
        """
        Абстрактный метод, вызываемый для обработки входных данных. Выходные данные должны соответствовать формату атрибута block_output
        """
        return NotImplementedError

    def _set_block_output(self, mode:str):
        """
        Приватный метод изменения выходной структуры
        """
        assert isinstance(mode, str), f'Invalid block "mode" param type: {type(mode)}'
        self.block_output = self.block_output_dict[mode]

    @abstractmethod
    def set_block_output(self, data_element = None):
        """
        Изменяет структуру выходных полей (атрибут block_output) в соответствии с атрибутом-словарем block_output_dict 
        В качестве условия может использоваться data_element (случайная строка входных данных)
        """
        self._set_block_output('default')

    def _set_block_aggr_mode(self, aggr_mode:bool):
        """
        Приватный метод смены поведения блока
        """
        assert isinstance(aggr_mode,bool ), f'Invalid "aggr_mode" param type: {type(aggr_mode)}'
        self.aggr_mode = aggr_mode

    @abstractmethod
    def set_block_aggr_mode(self, data_element = None):
        """
        Меняет поведение блока. Может быть итеративным (False) и агрегационным (True)
        В качестве условия может использоваться data_element (случайная строка входных данных)
        """
        self._set_block_aggr_mode(False)

Пример реализации абстрактного блока

from sdk.abstract_block import AbstractBlock
from packages.example.connections.default_connection import DefaultConnection

class DefaultBlock(AbstractBlock):
    def __init__(self):
        info_output = {
            "uuid" : "default_block_3159a8e8-decf-4c8d-b664-2edf5a011cdd",
            "type" : "action",
            "name" : {
                "en" : "Testing block info",
                "ru" : "Стандартный блок",
            },
            "description" : {
                "en" : "Testing block description",
                "ru" : "Тестовой блок",
            },
            "compatible_connections" : [DefaultConnection().uuid],
            "optionals" : {
                "is_save_data_on_fail_block_type" : False,
                "is_system_block_type" : False
            },
            "fields" : """[{
                    key: 'var_1',
                    type: 'text',
                    label: 'Введи 1 число',
                    description: "Поле для ввода первого числа",
                    required: true
                }, {
                    key: 'var_2',
                    type: 'text',
                    label: 'Введи 2 число',
                    description: "Поле для ввода второго числа",
                    required: true
                }, {
                    key: 'sw_1',
                    type: 'switcher',
                    label: 'Просто для показа',
                    required: true
                }, (z, bundle) => {
                    if (bundle.inputData.sw_1) {
                        return [{
                            key: 'f_1',
                            type: 'text',
                            label: 'Поле 1 из набора 1',
                            required: false
                        }, {
                            key: 'f_2',
                            type: 'text',
                            label: 'Поле 2 из набора 1',
                            required: false
                        }];
                    } else {
                        return [];
                    }
                }]"""
        }

        block_input = {
            "var_1":'String',
            "var_2":'String',
            "sw_1":'Boolean',
            "f_1":'String',
            "f_2":'String'
        }

        block_output = {
        "default": [
            {
                "name": "amount_str",
                "type": "String"
            },
            {
                "name": "Целое2",
                "type": "Long"
            },
            {
                "name": "массив3",
                "type": "ObjectArray",
                "struct": [{
                    "name": "sub",
                    "type": "Long",
                },{
                    "name": "dup",
                    "type": "Long",
                }]
            }
            ],
        }
        super().__init__(block_input, block_output, info_output)

    def set_block_output(self, data_element:list):
        self._set_block_output('default')

    def set_block_aggr_mode(self, data_element:list):
        self._set_block_aggr_mode(False)

    def process_data(self, **data):
        connection_data = data['connection_data']
        block_data = data['block_data']

        result = []
        for row in block_data:
            result.append([[
                connection_data['var_1'] + connection_data['var_2'],
                int(connection_data['var_1']) + int(connection_data['var_2']),
                [
                    {
                        "sub": int(connection_data['var_1']) - int(connection_data['var_2']),
                        "dup": 4 * int(connection_data['var_1'])
                    }
                ]
                ],[
                row['var_2'] + row['var_1'],
                int(row['var_2']) + int(row['var_1']),
                [
                    {
                        "sub": int(row['var_2']) - int(row['var_1']),
                        "dup": 4 * int(row['var_2'])
                    }
                ]
            ]])
        return result

Важно

Особенности при разработке блока:
- info_output
  - Идентификатор блока в поле info_output содержит пользовательское имя блока и уникальный UUID4 хэш
  - В поле compatible_connections может содержаться несколько подключений. Эти подключения должны быть реализациями абстрактного подключения и импортированы в секции импорта
  - Поле fields содержит строковое представление JavaScript-описания полей пользовательских интеграций
- block_input
  - Перечисляет поля, которые указаны в значении ключа fields в info_output
  - Названия типов данных соответствуют типам данных Java
- block_output
  - Содержит словарь со структурами выходных переменных автоматизации
  - Названия типов данных соответствуют типам данных Java
- process_data
  - Входная переменная data метода process_data может содержать ключи:
    - connection_data — содержит словарь формата "имя_поля_подключения":"значение_поля" из переданного подключения. Доступен только при наличии подключений
    - block_data — содержит список, элементом которого является строка данных в формате "имя_поля_блока":"значение_поля"
  - Результат вызова метода функции — список, глубина вложенности — 3. Уровни вложенности по порядку:
    - 1-й уровень — каждый элемент соответствует одной входящей записи
    - 2-ой уровень — каждый элемент соответствует одной исходящей записи. Набор элементов представляет собой ответ на входящую запись
    - 3-й уровень — каждый элемент соответствует значению поля исходящей записи. Порядок и типы полей соответствуют описанию в output_variables

Абстрактное подключение (abstract_connection.py)

Код абстрактного подключения

from abc import abstractmethod

class AbstractConnection():
    def __init__(self,
                 uuid:str = None,
                 name:dict = None,
                 description:str = None,
                 optionals:dict = None,
                 fields:str = None,
                 connection_input:dict = None
                 ):
        self.uuid = uuid
        self.name = name
        self.description = description
        self.optionals = optionals
        self.fields = fields
        self.connection_input = connection_input

    def get_info(self) -> dict:
        """
        Возвращает информацию о подключении при вызове --get-info
        """
        info_output =  {
            "uuid":self.uuid,
            "name":self.name,
            "description":self.description,
            "optionals":self.optionals,
            "fields":self.fields
        }
        return info_output

Пример реализации абстрактного подключения

from sdk.abstract_connection import AbstractConnection

class DefaultConnection(AbstractConnection):
    def __init__(self):
        uuid = "default_connection_7d9cbbcc-07b1-4207-9df6-4d7c3420dd13"
        name = {
                    "en" : "Testing connection info",
                    "ru" : "Тестовое подключение",
                }
        description = {
                    "en" : "Testing connection description",
                    "ru" : "Описание тестового подключения",
                }
        optionals = {
                    "is_category_1" : False,
                    "option_2" : "category_2"
                }
        fields = """[{
                    "key" : "var_1",
                    "type" : "textPlain",
                    "label" : "Введи число",
                    "description" : "Поле для ввода 1",
                    "is_required" : true,
                    "optionals" : {
                        "visible" : true
                    }
                },{
                    "key" : "var_2",
                    "type" : "textPlain",
                    "label" : "Введите что-нибудь",
                    "description" : "Поле для ввода 2",
                    "is_required" : true,
                }]"""

        connection_input = {
            "var_1":'String',
            "var_2":'String',
        }

        super().__init__(uuid, name, description, optionals, fields, connection_input)

Важно

Особенности при разработке подключения:
- uuid — идентификатор подключения в поле info_output содержит пользовательское имя блока и уникальный UUID4 хэш
- fields — содержит строковое представление JavaScript описания полей пользовательских интеграций
- connection_input
  - Перечисляет поля, которые указаны в значении ключа fields
  - Названия типов данных соответствуют типам данных Java

Была ли статья полезна?

Да

Нет

Модуль реализации блоков на Python

Примеры реализации Python-блоков