Разработка логики выполнения блока (component.py)

В python-файле находится код выполнения блока.

Код должен содержать класс CUserCustomComponent с методами:

Инициализация - с обязательными параметрами:
- settings - передаются настройки блока
- inputs - передаются входные данные
- log - лог
- context - дополнительные параметры:
  - block_id - id блока
  - libId - id библиотеки блока
  - user_id - id пользователя
execute - выполнение блока. Данный метод выполняется при расчете - во время расчета графа или блока
visualizer - формирование данных для dataset визуализатора, если блок визуализатор
event_name_event - обработка события блока, если есть события, где name_event это наименования события (Например: event_on_click)

Пример

# -*- coding: utf-8 -*-

import json
import requests

from abc import ABC
from datetime import datetime
from typing import Any, Dict, List


class CUserCustomComponent(ABC):

    def __init__(self, settings: Dict, inputs: Dict, log: Any, context: Dict):
        self.settings = settings
        self.inputs = inputs
        self.log = log
        self.context = context

        self.result_requests: List = []

    def execute(self) -> Dict:
        """Расчет пользовательского блока"""
        settings_geocoder = self.settings.get('settings_geocoder', {})
        api_geocoder = settings_geocoder.get('api_geocoder', '')

        self.log.info(f'read settings api_geocoder: {api_geocoder}')

        coord = self.inputs.get('coord', '')

        if not coord:
            self.log.raise_error('No coord')

        self.log.info(f'input coord: {coord}')

        try:
            r = requests.get(
                url='{0}/search'.format(api_geocoder),
                params={"address": coord},
                timeout=10,
            )
        except requests.exceptions.Timeout:
            self.log.raise_error('Timed out')

        if r.status_code != 200:
            self.log.raise_error(r.text)

        if isinstance(r.text, str):
            self.result_requests = json.loads(r.text)
        else:
            self.result_requests = r.text

        if self.result_requests:
            for row in self.result_requests:
                for k, v in row.items():
                    if v is None:
                        row[k] = ''

        return {'address': self.result_requests, "OnClick": {"select_address": "Default"}}

    def visualizer(self) -> Dict:
        """Формирование данных для dataset"""
        self.log.info('Create visualizer')

        # data - обязательный ключ, в который вкладывается весь payload
        return {
            "data": self.result_requests
        }

    def event_on_click(self) -> Dict:
        """Обработка события OnClick"""
        address = self.inputs.get('address', '')
        self.log.info(f'Exec event OnClick')
        return {"OnClick": {"select_address": address}}

    def event_on_check_date(self) -> Dict:
        """Обработка пользовательского события OnCheckDate"""
        dt = self.inputs.get('data', '')
        self.log.info(f'Exec event OnCheckDate')

        return {"OnCheckDate": {"now": str(datetime.utcnow())}}

Обработка событий визуализатора

Формат данных для payload события выглядит следующим образом:

def event_on_selection_change(self) -> Dict[str, Any]:
    value = self.inputs.get("checked")

    return {
        "OnSelectionChange": {
            "checked": value
        }
    }

Важная информация

Результатом метода execute должен быть справочник с ключами id-выхода блока и их значениями. Для событийных выходов, задаются дефолтные значения, с которыми будут считаться последующие блоки.

Важная информация

Наименования класса для обработки логики блока должно совпадать с наименованием из manifest.json

Важная информация

Обработчики событий визуализатор должны быть в виде event_<ключ_события>, где ключ события - id события в manifest.json

Управление зависимостями

В качестве менеджера зависимостями python используется poetry.

pyproject.toml

Если проект инициализировался с помощью шаблонного .env или копированием, то важно заполнить поля "name", "version", "description", "authors" в pyproject.toml

Полезные команды для poetry:

Добавить зависимость

poetry add <package_name> # последняя версия
poetry add <package_name>@<version> # конкретная версия
poetry add <package_name> --platform <platform_name> # зависимость для конкретной платформы

2. Удалить зависимость

poetry remove <package_name> # удалить пакет из списка зависимостей

Запустить скрипт
```
poetry run script.py
```

Разработка логики выполнения блока (component.py)

Управление зависимостями

Полезные ссылки