null

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

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

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

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

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

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

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

Примечание — до выбора окружения необходимо подготовить само окружение. Это необходимо делать на сервере с Агентом ETL.  

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

  • выберите каталог для установки;
  • запустите консоль, перейдите в каталог;
  • выполните команды:
python.exe -m venv venv
venv\Scripts\activate
pip install <нужные библиотеки>

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

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

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

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

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

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

В качестве примеров скриптов можно использовать шаблоны. Для этого нажмите на кнопку «Заполнить из шаблона» и выберите необходимый вариант, отобразится скрипт:

  • «База данных»: 
Скрипт шаблона базы данных 

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

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

В функции заданы основные типы данных (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

Предоставляет пользователю Modus ETL перечень колонок результирующей таблицы.

Список типов задается так же, как и в функции 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».

Связи контента