Модули Python - Продукт Modus BI

null

Блок загрузки данных с помощью Python позволяет использовать возможности этого языка для обработки и загрузки данных из произвольных источников.

Запуск и обработка результатов работы модуля осуществляется с помощью Агента ETL. Окружение Python в текущей версии настраивается самостоятельно.

Модули Python позволяют сделать какой-либо анализ не через SQL, а с помощью сложных анализов, применяя аналитические библиотеки.

Для этого пишется скрипт, через которую делается выгрузка, создаются параметры для пользователя, переменные и внутри кода все это считается. Если SQL просто выгружает, то Python позволяет и получить данные, и работать с ними, например, вы можете запросить анализ методики линейного регрессирования.

Для работы с модулем Python нажмите в верхнем меню «Главное», выберите выпадающий список «Настройки» и перейдите в раздел «Модули Python».

В отобразившемся разделе создайте новый модуль при помощи кнопки «Создать» (1) или «Создать новый элемент копированием текущего» (2).

В отобразившейся форме создания заполнение поле «Наименование» и «Окружение».

В окружении прописываются разные библиотеки, т.е на локальном компьютере может существовать несколько окружений для Python и в них могут быть разные версии Python и разные наборы библиотек.

До выбора окружения необходимо подготовить само окружение.

Порядок установки следующий:

выберите каталог для установки;
запустите консоль, перейдите в каталог;
выполните команды:

python.exe -m venv venv
venv\Scripts\activate
pip install <нужные библиотеки>

Для шаблонов кода в справочнике «Модули Python» потребуются следующие библиотеки:

База данных:
- psycopg2
- sqlalchemy
Веб-сервис:
- requests

Создайте каталог internal/pyread и положите в него файлы api.py и pythonExec.py.

При нажатии на «Показать все» у поля «Окружение» отобразится форма выбора окружения. При необходимости создания нового окружения нажмите на кнопку «Создать».

В отобразившейся форме заполните поля «Код» и «Наименование», а в поле «Исполняемый файл python» загрузите файл с локального компьютера.

Заполните и закройте форму, выберите окружение.

На вкладке «Скрипт» введите код.

При необходимости заполнения кода при помощи шаблона нажмите на кнопку «Заполнить из шаблона» и выберите необходимый вариант, отобразится скрипт:

«База данных»:

Скрипт шаблона базы данных


from typing import Dict, Union

from sqlalchemy import create_engine, text

from api import (
	ParametersClass, ParameterClass, ParameterItemsClass, ColumnsClass, TrivialType,
    ColumnClass, DataSourceClass, TableDataClass, TypeDescription, Source
)


def get_params() -> ParametersClass:
    # Создаем несколько параметров для пользователя

    td_str = TypeDescription(precision=0, length=100, contain_time=False, data_type=TrivialType.ttString )
    td_number = TypeDescription(precision=10, length=100, contain_time=False, data_type=TrivialType.ttNumber )
    td_integer = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttInteger )
    td_date = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttDate )
    td_datetime = TypeDescription(precision=0, length=0, contain_time=True, data_type=TrivialType.ttDatetime )

    variables = [
        ParameterClass(index=0, name="is_active", type_desc=td_integer, value=1),
    ]

    # Оборачиваем в VariableItemsClass
    items = ParameterItemsClass(variables)

    return ParametersClass(items=items)


def get_structure_table(params: ParametersClass) -> DataSourceClass:

    is_active = params[0].Value  # Получаем значение для поля "is_active"

    # Базовые колонки
    td = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttInteger )
    td_str = TypeDescription(precision=0, length=100, contain_time=False, data_type=TrivialType.ttString )
    columns = [
	    ColumnClass(index=0, name="Age", type_desc=td),
	    ColumnClass(index=1, name="Name", type_desc=td_str),
    ]


    # Создаем объект таблицы
    columns_class = ColumnsClass(columns)
    table = DataSourceClass(columns=columns_class)

    # Возвращаем созданную таблицу
    return table


def get_data(
	params: ParametersClass,
	table_structure: DataSourceClass,
	source: Source,
	context: Dict[str, Union[int, float, bool, str]],
) -> TableDataClass:

    query_params = {param.Name: param.Value for param in params.Items}

    engine = create_engine(f"postgresql+psycopg2://{source.Login}:{source.Password}@{source.Address}/{source.Database}")
    table = TableDataClass(table_structure)

    with engine.connect() as engine.connection:
        query = text("SELECT Name, Age FROM table_test")

        result = engine.connection.execute(query, query_params)
        rows = result.mappings().all()

        for row in rows:
            data = []
            for column in table.iter_columns():
                data.append(row.get(column.Name))
            table.AddRow(data)

    return table

«Веб-сервис»:

Скрипт шаблона веб-сервиса

from typing import Dict, Union

import requests

from api import (
ParametersClass, ParameterClass, ParameterItemsClass, ColumnsClass, TrivialType,
ColumnClass, DataSourceClass, TableDataClass, TypeDescription, Source
)


def get_params() -> ParametersClass:
    # Создаем несколько переменных для пользователя

    td_str = TypeDescription(precision=0, length=100, contain_time=False, data_type=TrivialType.ttString )
    td_number = TypeDescription(precision=10, length=100, contain_time=False, data_type=TrivialType.ttNumber )
    td_integer = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttInteger )
    td_date = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttDate )
    td_datetime = TypeDescription(precision=0, length=0, contain_time=True, data_type=TrivialType.ttDatetime )

    variables = [
        ParameterClass(index=0, name="test_param",  type_desc=td_integer, value=1),
    ]

    # Оборачиваем в VariableItemsClass
    items = ParameterItemsClass(variables)

    return ParametersClass(items=items)


def get_structure_table(params: ParametersClass) -> DataSourceClass:

    # Базовые колонки
    td = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttInteger)
    td_str = TypeDescription(precision=0, length=200, contain_time=False, data_type=TrivialType.ttString)
    columns = [
	    ColumnClass(index=0, name="ID", type_desc=td_str),
	    ColumnClass(index=1, name="Name", type_desc=td_str),
	    ColumnClass(index=2, name="Photo", type_desc=td_str),
	    ColumnClass(index=3, name="Status", type_desc=td_str),
	]


    # Создаем объект таблицы
    columns_class = ColumnsClass(columns)
    table = DataSourceClass(columns=columns_class)

    # Возвращаем созданную таблицу
    return table


def get_data(
	params: ParametersClass,
	table_structure: DataSourceClass,
	source: Source,
	context: Dict[str, Union[int, float, bool, str]],
) -> TableDataClass:

    table = TableDataClass(table_structure)
    response = requests.get(source.Address)
    # https://petstore.swagger.io/v2/pet/findByStatus?status=available

    data = response.json()
    for pet in data[:10]:
    	table.AddRow([
    	    pet['id'],
    	    pet['name'],
    	    pet['photoUrls'][0] if pet['photoUrls'] else '',
    	    pet['status'],
    	])

    return table

«Файл»:

Скрипт шаблона файла

from typing import Dict, Union
import re

from api import (
ParametersClass, ParameterClass, ParameterItemsClass, ColumnsClass, TrivialType,
ColumnClass, DataSourceClass, TableDataClass, TypeDescription, Source
)



def get_params() -> ParametersClass:
    # Создаем несколько параметров для пользователя

    td_str = TypeDescription(precision=0, length=100, contain_time=False, data_type=TrivialType.ttString )
    td_number = TypeDescription(precision=10, length=100, contain_time=False, data_type=TrivialType.ttNumber )
    td_integer = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttInteger )
    td_date = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttDate )
    td_datetime = TypeDescription(precision=0, length=0, contain_time=True, data_type=TrivialType.ttDatetime )

    variables = [
        ParameterClass(index=0, name="test_param",  type_desc=td_str, value="some string"),
    ]

    # Оборачиваем в VariableItemsClass
    items = ParameterItemsClass(variables)

    return ParametersClass(items=items)


def get_structure_table(params: ParametersClass) -> DataSourceClass:

    # Базовые колонки
    td = TypeDescription(precision=0, length=0, contain_time=False, data_type=TrivialType.ttInteger)
    td_str = TypeDescription(precision=0, length=100, contain_time=False, data_type=TrivialType.ttString)
    columns = [
    	ColumnClass(index=0, name="Name", type_desc=td_str),
    	ColumnClass(index=1, name="INN", type_desc=td_str),
    	ColumnClass(index=2, name="Phone", type_desc=td_str),
    ]


    # Создаем объект таблицы
    columns_class = ColumnsClass(columns)
    table = DataSourceClass(columns=columns_class)

    # Возвращаем созданную таблицу
    return table


def get_data(params: ParametersClass, table_structure: DataSourceClass, source: Source, context: Dict[str, Union[int, float, bool, str]]) -> TableDataClass:

    table = TableDataClass(table_structure)

    with open(source.Address, 'r', encoding='utf-8') as f:
    	text = f.read()

    table.AddRow([
        re.search('Наименование: (.*)', text).group(0),
        re.search('ИНН: (.*)', text).group(0),
        re.search('Телефон: (.*)', text).group(0),
    ])

    return table

Шаблоны дают представление о том, как должен выглядеть файл python. В нем подготавливаются 3 функции — get_params, get_structure_table и get_data. Остановимся на них подробнее:

Функции get_params, get_structure_table, get_data

1. get_params

Предоставляет 1С список параметров для доступа к данным и их дефолтные значения.

В функции заданы основные типы данных (td_str, td_number, td_integer...) в качестве примера, на их основе можно конструировать свои типы, указывая нужные величины length (длина строки/числа) и precision (точность для дробного числа).

Сами параметры мы помещаем в список variables, каждое значение представляет собой инициализированный класс ParameterClass. Все аргументы инициализации - обязательные.

    variables = [
        ParameterClass(index=0, name="is_active", type_desc=td_integer, value=1),
    ]

Если нам не нужны параметры, то список variables оставляем пустым.

2. get_structure_table

Отдает 1С перечень колонок результирующей таблицы.

Список типов задается так же, как и в функции get_params.

Поля помещаем в список columns.

3. get_data

В функции заполняется и отдается таблица с данными. Здесь должна находится основная логика по работе с данными. Итогом выполнения функции должна стать возвращаемая таблица - экземпляр класса TableDataClass.

Входящие параметры get_data:

params: ParametersClass — параметры сбора данных;
table_structure: DataSourceClass — структура результирующей таблицы;
- Columns: ColumnsClass — список колонок типа ColumnClass;
  - ColumnClass
    - Index: int — индекс колонки (с 0);
    - Name: str — имя колонки;
    - type_desc: TypeDescription — тип колонки:
      - DataType: TrivialType:
        
        ttString = 1;
        
        ttNumber = 2;
        
        ttDate = 3;
        
        ttDatetime = 4;
        
        ttInteger = 5;
        
        ttUndefined = 6.
Length: Optional[int] — длина значения;
Precision: Optional[int] — точность значения;
ContainTime: bool — содержит время (deprecated):
- Data: List[List[Union[bool, int, str, float, datetime.datetime, None]]]
- Get(row: int, col: Union[int, str]) -> Union[bool, int, str, float, datetime.datetime, None] - Возвращает значение ячейки по строке и колонке
- GetColumn(col: Union[int, str]) -> ColumnClass - Возвращает колонку по индексу или имени
- IsNull(row: int, col: Union[int, str]) -> bool - Проверяет, является ли ячейка пустой
source: Source — параметры источника данных:
- Address: str — адрес БД, файла, веб-сервиса, и.т.п.;
- Database: str — база данных;
- Login: str — логин к БД;
- Password: str — пароль к БД;
- Timeout: int — таймаут вызова БД/сервиса;
- SecureMode: str — параметр подключения;
- SkipVerify: bool — параметр подключения;
- MaxConns: int — максимальное число соединений;
- DefSchema: str — схема подключения;
- Protocol: str — протокол подключения;
- AdditionalProperties: Dict[str, str] — дополнительные параметры подключения;
- FileName: str — имя файла;
- Kind: str — тип файла;
- DataId: str — id данных файла во временном хранилище сессии.
context: Dict[str, Union[int, float, bool, str]] — произвольные переменные контекста.

При замене уже имеющегося скрипта отобразиться форма с подтверждением замены скрипта шаблоном. Подтвердите замену нажатием на кнопку «Да» либо нажмите на кнопку «Нет» для отмены.

На вкладке «Описание» возможно написать комментарий с описанием модуля.

Нажмите на кнопку «Записать и закрыть». Модуль создан.

Далее при установке правил выгрузки вы сможете установить вид правила «Модуль python», как описано в разделе «Настройка правила вида «Модуль python».

Категории: Modus ETL: Сбор данных
Дата публикации: 22.04.25
Приоритет: Приоритет: 0.0
Просмотреть счетчик: 79 Просмотров
Дата создания: 22.04.25