SSIS вызывает Python: как правильно передать параметры

Антипаттерн: креды в коде

Вот как обычно выглядит первая версия Python-скрипта, который кто-то написал «на коленке» и потом уже не переписывал:

# load_excel_plan.py
import pyodbc, pandas as pd

SERVER = 'sql-prod-01.corp.local'
DATABASE = 'DWH'
USERNAME = 'etl_user'
PASSWORD = 'Pa$$w0rd2023!'
FILE = r'\\fs\share\plans\WHS_2026_04.xlsx'

conn = pyodbc.connect(
    f'Driver={{ODBC Driver 17 for SQL Server}};SERVER={SERVER};'
    f'DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD}'
)
df = pd.read_excel(FILE)
df.to_sql('stg_plan', conn, if_exists='replace')

Что с этим не так:

• Файл попадёт в git. Кто-то из команды сделает git log -p — увидит пароль прод-сервера.
• Пароль ротировали — надо лезть в *.py, править вручную, разворачивать. На 5 разных скриптов это 5 правок.
• Нельзя переиспользовать скрипт для тестового сервера или другой базы — надо копировать файл.
• Имя файла тоже в коде. Каждый новый месяц — правка кода.
• Тип плана (WHS / AUTO / BUDGET) в лучшем случае if-else по имени файла. В худшем — три копии скрипта.

Фикс простой: всё, что меняется между запусками — вынести в параметры, а передавать их из SSIS.

Правильный паттерн: sys.argv

Python-скрипт ничего не знает про сервер до запуска. SSIS передаёт всё через командную строку:

python.exe load_excel_plan.py  SERVER  DATABASE  USER  PASSWORD  FILEPATH  PLANTYPE

В Python sys.argv — список аргументов, где [0] это путь к самому скрипту, а дальше — всё что пришло после него. Скелет:

import sys
import datetime
import urllib.parse
import sqlalchemy as sa
import pandas as pd

# ─── 1. Параметры из SSIS ─────────────────────────
ARGS = sys.argv[1:]
if len(ARGS) == 6:
    SERVER, DATABASE, USERNAME, PASSWORD, FILEPATH, PLANTYPE = ARGS
else:
    # Режим отладки — когда запускаем руками из IDE.
    # На прод эта ветка не попадает, SSIS всегда даёт 6 аргументов.
    SERVER   = "localhost"
    DATABASE = "DWH_dev"
    USERNAME = ""
    PASSWORD = ""
    FILEPATH = r"C:\tmp\plan_test.xlsx"
    PLANTYPE = "WHS"

# ─── 2. Подключение ───────────────────────────────
def connect(server, database, username, password):
    """
    Возвращает sqlalchemy engine с fast_executemany.
    Если нужен Windows-auth — закомментировать UID/PWD
    и поставить Trusted_Connection=yes.
    """
    conn_str = (
        'Driver={ODBC Driver 17 for SQL Server};'
        f'SERVER=tcp:{server};'
        f'Database={database};'
        f'UID={username};PWD={password};'
        'Trusted_Connection=no;'
        'Encrypt=no;'
    )
    uri = 'mssql+pyodbc:///?odbc_connect=' + urllib.parse.quote_plus(conn_str)
    return sa.create_engine(uri, fast_executemany=True)

engine = connect(SERVER, DATABASE, USERNAME, PASSWORD)

# ─── 3. Работа ────────────────────────────────────
def log2db(engine, message):
    with engine.begin() as conn:
        conn.execute(
            sa.text(
                "INSERT INTO [dbo].[te Журнал загрузки из Excel] "
                "([Дата], [Сообщение]) VALUES (CAST(:d AS DATETIME), :m)"
            ),
            {"d": datetime.datetime.now(), "m": message}
        )

try:
    log2db(engine, f"Старт загрузки: {PLANTYPE}, файл {FILEPATH}")

    xl = pd.ExcelFile(FILEPATH)
    for sheet in xl.sheet_names:
        df = pd.read_excel(xl, sheet_name=sheet)
        df["PlanType"] = PLANTYPE
        df.to_sql(
            f"stg_plan_{PLANTYPE.lower()}",
            engine, if_exists="append", index=False
        )
        log2db(engine, f"Лист {sheet} загружен, строк: {len(df)}")

    log2db(engine, f"Готово: {PLANTYPE}")
    sys.exit(0)

except Exception as e:
    log2db(engine, f"ОШИБКА {PLANTYPE}: {e}")
    sys.exit(1)

Что важно:

• Нет ни одного литерала подключения в коде. Всё из argv. Можно смело класть в git.
• Отладочная ветка с пустыми значениями. Когда запускаешь из VS Code без аргументов — падает рано и понятно, а не лезет в прод случайно.
• Named parameters в sa.text. Позиционные ? при переходе sqlalchemy → pyodbc иногда ломаются на русских именах таблиц — через :name надёжнее.
• Try/except на всё тело — чтобы исключение писалось в журнал, а не терялось в stderr.
• sys.exit(1) на ошибке. Ниже расскажу, почему это критично.

Как это вызывать из SSIS

В SSIS-пакете есть стандартный Execute Process Task — таск, который умеет запускать любой exe и ждать результат. Его настройки:

Executable        C:\Python311\python.exe
WorkingDirectory  D:\ssis\scripts
Arguments         (через Expression, см. ниже)
WindowStyle       Hidden
FailTaskIfReturnCodeIsNotSuccessValue  True
SuccessValue      0
TimeOut           0  (или 1800 если надо страховку)

В SSIS-пакете создаём переменные (Variables):

• User::Server — строка
• User::Database — строка
• User::Username — строка
• User::Password — строка, Sensitive = True
• User::FilePath — строка
• User::PlanType — строка

Значения подтягиваем через SSIS Configurations, Project Parameters или из таблицы настроек в SQL — как принято в вашем DWH. Главное — НЕ хардкодом в пакете (иначе мы просто переехали из одного тупика в другой).

В свойстве Arguments таска ставим Expression (крестик справа от поля в Properties):

"load_excel_plan.py " +
@[User::Server] + " " +
@[User::Database] + " " +
@[User::Username] + " " +
@[User::Password] + " " +
"\"" + @[User::FilePath] + "\" " +
@[User::PlanType]

Дальше — как обычно:

• Перед Execute Process Task — Script Task или Execute SQL Task, который готовит путь к файлу (подставляет текущую дату, достаёт из File System Task найденный файл и т.п.).
• Пакет в SSIS Catalog, запуск через SQL Agent.
• Success/Failure стрелки ветвятся как надо — Python вернул 0 → зелёная, вернул 1 → красная.

Журнал загрузки: чтобы SQL Agent видел статус

SSIS сам пишет в SSISDB, но это неудобно читать — ошибка Python-скрипта закопана в OnError-логе глубоко. Делаем параллельно свою таблицу, куда пишет и Python, и сам SSIS:

CREATE TABLE [dbo].[te Журнал загрузки из Excel] (
    [ID]         INT IDENTITY(1,1) PRIMARY KEY,
    [Дата]       DATETIME    NOT NULL DEFAULT GETDATE(),
    [Источник]   NVARCHAR(64),
    [Сообщение]  NVARCHAR(1000)
);
CREATE INDEX IX_journal_date ON [dbo].[te Журнал загрузки из Excel] ([Дата] DESC);

Python пишет туда каждую итерацию — старт, каждый успешный лист, финальное «готово» или «ошибка». После Python в SSIS делаем Execute SQL Task, который достаёт последнюю запись:

SELECT TOP 1 [Сообщение]
FROM [dbo].[te Журнал загрузки из Excel]
WHERE [Дата] >= ?   -- стартовое время пакета
ORDER BY [ID] DESC;

Результат присваиваем в SSIS-переменную — и если там «ОШИБКА …», пакет уходит по красной стрелке на Send Mail Task с человеческим текстом. Без необходимости лезть в SSISDB.

Почему отдельная таблица, а не просто print() в stdout и чтение stderr из Execute Process Task: потому что Execute Process Task при больших объёмах stdout может молча обрезать или блокироваться. Таблица в SQL надёжнее и переживает пакет.

Обработка ошибок и exit-коды

Эта часть не очевидна для тех, кто из Python-мира пришёл в SSIS впервые.

Execute Process Task по умолчанию смотрит на return code процесса. Если он равен SuccessValue (по умолчанию 0) — таск зелёный. Иначе (при FailTaskIfReturnCodeIsNotSuccessValue = True) — красный.

Что в Python возвращает return code:

Последний случай — типичный косяк. Разработчик пишет try/except: log(e), забывает про sys.exit(1) — Python отловил исключение, записал в журнал и вышел с кодом 0. SSIS думает: всё хорошо. SQL Agent рассылает «пакет выполнен», никто не лезет в журнал, пока через неделю не спросят «а где план на апрель».

Как этот паттерн масштабируется

Шесть параметров — это только для примера с Excel. Универсально набор зависит от задачи. Где мы применяли этот же паттерн:

• Excel-планы с несколькими форматами. Один скрипт, шесть аргументов, три SSIS-таска — по одному на формат (WHS, AUTO, BUDGET). Разница только в значении PLANTYPE и пути к файлу. Без дублирования кода.
• Яндекс.Метрика API. Параметры: токен, counter ID, дата от, дата до, тип отчёта. Python дёргает https://api-metrika.yandex.net, возвращает CSV, кладёт в stg_yam_*. SSIS запускает раз в сутки с date_from = GETDATE()-7 (скользящее окно для перезагрузки задним числом).
• amoCRM → факт продаж. Параметры: subdomain, long-lived token, pipeline ID, date_from. Python делает пагинацию по API, разворачивает кастомные поля сделок в колонки, пишет в stg_amo_leads.
• ЦРПТ / Честный знак. Похоже, но там поверх этого добавляется shift-cipher для UIT-токена и CAdES-подпись запросов. Отдельная тема — разберём в следующей статье.
• Загрузка курсов ЦБ РФ. Вообще без кредов (публичное API), но SSIS передаёт server, database, date_from, date_to — чтобы можно было перелить куда угодно.

Везде один и тот же каркас: sys.argv[1:], connect(), log2db(), try/except c sys.exit. Меняется только середина — что именно делаем с данными.

"""
template_etl.py — каркас параметризованного Python-ETL для SSIS.
Заменить тело под задачу: REST-API, Excel, CSV, XML, что угодно.
"""
import sys
import datetime
import urllib.parse
import sqlalchemy as sa

ARGS = sys.argv[1:]
if len(ARGS) >= 4:
    SERVER, DATABASE, USERNAME, PASSWORD, *EXTRA = ARGS
else:
    SERVER = DATABASE = USERNAME = PASSWORD = ""
    EXTRA = []

def connect(server, database, username, password):
    cs = (
        'Driver={ODBC Driver 17 for SQL Server};'
        f'SERVER=tcp:{server};Database={database};'
        f'UID={username};PWD={password};'
        'Trusted_Connection=no;Encrypt=no;'
    )
    uri = 'mssql+pyodbc:///?odbc_connect=' + urllib.parse.quote_plus(cs)
    return sa.create_engine(uri, fast_executemany=True)

def log2db(engine, message, source='python'):
    with engine.begin() as c:
        c.execute(
            sa.text(
                "INSERT INTO [dbo].[te Журнал загрузки] "
                "([Дата],[Источник],[Сообщение]) "
                "VALUES (CAST(:d AS DATETIME),:s,:m)"
            ),
            {"d": datetime.datetime.now(), "s": source, "m": message[:900]}
        )

def main():
    engine = connect(SERVER, DATABASE, USERNAME, PASSWORD)
    try:
        log2db(engine, f"start, args={EXTRA}")
        # ────────────────────────────────
        #   здесь основная работа скрипта
        #   — разбор Excel / REST / XML
        #   — трансформации
        #   — вставка в stg_* таблицы
        # ────────────────────────────────
        log2db(engine, "done")
        return 0
    except Exception as e:
        log2db(engine, f"ERROR: {type(e).__name__}: {e}")
        return 1

if __name__ == "__main__":
    sys.exit(main())

Нюансы, которые ловятся не сразу

1. Пароль в Arguments виден в логах. SSIS по умолчанию логирует параметры Execute Process Task в информационный лог. Если в SSISDB включена повышенная детализация или пакет упал с ошибкой — пароль окажется и в SSISDB, и в Event Log, и в email-рассылке SQL Agent. Базовое решение: отдельная SQL-учётная запись для ETL с минимальными правами. Если этого мало — см. отдельный раздел про обфускацию аргументов ниже.
2. Альтернатива — Integrated Security. Если SSIS и SQL на одном домене, лучше вообще не передавать пароль. Запускаем SQL Agent Job под service-account, в Python убираем UID/PWD, добавляем Trusted_Connection=yes. Минус один параметр из шести.
3. Python на сервере с SSIS. Должен быть установлен на той же машине, где крутится SQL Agent, и быть доступен пользователю, под которым агент работает. Частый косяк: поставили в %USERPROFILE%\AppData под своей учётной записью — SSIS падает с «python.exe not found». Ставьте в C:\Python311 или аналог, доступный всем.
4. ODBC Driver 17 vs 18. Семнадцатый — проверенный, с ним меньше сюрпризов. Восемнадцатый по умолчанию требует Encrypt=yes и валидный сертификат — на внутреннем SQL без TLS упадёт. Либо ставьте 17, либо добавляйте Encrypt=no;TrustServerCertificate=yes.
5. fast_executemany не работает с NVARCHAR(MAX). Если грузите большие текстовые поля и ODBC ругается — явно ограничьте тип в dtype={} у to_sql, либо переключайтесь на BULK INSERT через промежуточный CSV.
6. Кодировки Excel. Windows-пакеты 1С часто выгружают Excel в CP1251 в некоторых ячейках. Pandas читает нормально, но в SQL через to_sql без явного NVARCHAR летят знаки вопроса. Ставьте dtype={'col': sa.types.NVARCHAR(255)}.

Обфускация аргументов: когда пароль вообще нельзя светить в журнале

Базовой меры с отдельной учётной записью и Basic-логированием хватает не везде. Типичные случаи, когда пароль всё равно окажется в логах:

• Пакет упал с ошибкой на запуске Python — SSIS фиксирует в отчёте полный текст Arguments.
• У SQL Agent настроена рассылка «по падению» — email уходит вместе с текстом команды, включая пароль.
• Инцидентный расчёт: кто-то из админов смотрит SSISDB execution report, а пароль от рабочей учётной записи ETL не предназначен даже для внутреннего ИБ.
• Передаётся не пароль SQL, а токен внешнего API (Яндекс.Метрика, amoCRM, Wildberries, ЦРПТ) — отдельной учётной записи «минимальных прав» там нет, токен даёт полный доступ.

Правильное решение — класть секреты в Integrated Security (если SSIS и SQL в одном домене) или в отдельный хранитель секретов: Azure Key Vault, Windows Credential Manager, HashiCorp Vault. Там пароль вообще не появляется в Arguments — Python тянет его напрямую по API хранителя.

На практике у клиентов с локальным SQL, без Active Directory и без бюджета на Key Vault мы используем промежуточное решение — shift-cipher. В SSIS хранится обфусцированная строка, Python раскодирует её на лету. В журнале светится абракадабра, а не настоящий пароль.

Как это выглядит в коде

В SSIS-переменной вместо настоящего пароля MyP@ss2024 лежит обфусцированная версия, например N|Cx!tz2402 (реальное содержимое — alphanumerics со сдвигом и парным swap'ом). Python получает её через sys.argv, прогоняет декодер и подставляет в connection string. Функция симметричная — encoder и decoder отличаются знаком сдвига.

# decoder — используется в проде
def decode_shift_cipher(encoded_text, shift):
    # 1. Разворачиваем парный swap, который сделал encoder
    shuffled = list(encoded_text)
    unshuffled = shuffled[:]
    for i in range(0, len(shuffled) - 1, 2):
        unshuffled[i], unshuffled[i + 1] = shuffled[i + 1], shuffled[i]

    # 2. Убираем добивочный пробел, если encoder его добавил для чётной длины
    if unshuffled and unshuffled[-1] == " ":
        unshuffled.pop()

    # 3. Обратный сдвиг Цезаря по буквам (цифры/символы — как есть)
    decoded = []
    for ch in unshuffled:
        if ch.isalpha():
            base = 65 if ch.isupper() else 97
            decoded.append(chr(((ord(ch) - base - shift) % 26) + base))
        else:
            decoded.append(ch)
    return ''.join(decoded)

# encoder — запускаем один раз на машине разработчика при смене пароля
def encode_shift_cipher(plain_text, shift):
    # 1. Добиваем пробелом до чётной длины (чтобы swap работал ровно)
    text = plain_text if len(plain_text) % 2 == 0 else plain_text + " "

    # 2. Сдвиг Цезаря по буквам
    shifted = []
    for ch in text:
        if ch.isalpha():
            base = 65 if ch.isupper() else 97
            shifted.append(chr(((ord(ch) - base + shift) % 26) + base))
        else:
            shifted.append(ch)

    # 3. Попарный swap соседних символов
    lst = list(''.join(shifted))
    for i in range(0, len(lst) - 1, 2):
        lst[i], lst[i + 1] = lst[i + 1], lst[i]
    return ''.join(lst)

# Использование в рабочем скрипте
import sys
ARGS = sys.argv[1:]
SERVER, DATABASE, USERNAME = ARGS[0], ARGS[1], ARGS[2]
PASSWORD = decode_shift_cipher(ARGS[3], shift=3)   # shift=3 — договор между encoder и скриптом
TOKEN    = decode_shift_cipher(ARGS[4], shift=3)

Процесс работы с секретами

1. Разработчик локально запускает encode_shift_cipher("MyP@ss2024", shift=3) → получает обфусцированную строку.
2. Эту строку кладёт в SSIS-переменную PasswordObf (можно даже не как Sensitive, смысл обфускации в том, что строка в журналах ничего не выдаёт глазом).
3. В Execute Process Task Arguments собираем строку с этой переменной:
   @[User::Server] + " " + @[User::Database] + " " + @[User::Username] + " " + @[User::PasswordObf]
4. В Python читаем sys.argv[4], прогоняем через decode_shift_cipher, подставляем в connection string.
5. В журнале SSIS Arguments выглядят как MY-SERVER DWH etl_user N|Cx!tz2402 — на глаз пароль невосстановим.

Где особенно удобно применять

• Токены внешних API — API Яндекс.Метрики, ключи amoCRM, токены Wildberries. У них нет отдельной учётной записи «минимальных прав»: украденный токен = полный доступ к аккаунту. Обфускация в логах — минимум, который надо сделать.
• Старые on-premise среды — клиент на локальном SQL Server без AD, без бюджета на отдельный хранитель секретов. Shift-cipher — дёшево и поднимает планку инцидентного риска с «пароль читает каждый, кто видел лог» до «нужно знать, что это обфускация, и иметь код декодера».
• Временная мера на время миграции — пока подключается Key Vault или домен.

Где не подойдёт

• ИБ-регламент требует полноценного управления секретами с ротацией и аудитом доступа.
• Атакующая модель включает инсайдера с доступом к репозиторию (он увидит код декодера и значение shift).
• Compliance: PCI DSS, банковский ИБ, ИСПДн уровня К1/К2.

В этих сценариях — только Key Vault / Credential Manager / домен.

Итог — чек-лист

Если переносите один из своих скриптов на этот паттерн:

• Вытащили все креды и пути из *.py в sys.argv.
• Добавили Execute Process Task, указали python.exe и expression для Arguments.
• Создали SSIS-переменные, в т.ч. Password как Sensitive.
• Создали таблицу te Журнал загрузки, функцию log2db в Python.
• Логируете старт, каждый значимый шаг и финал.
• В блоке except есть sys.exit(1).
• FailTaskIfReturnCodeIsNotSuccessValue = True, SuccessValue = 0.
• После Python-таска — Execute SQL Task, читающий последний журнал для ветвления/уведомлений.
• Python установлен системно (не в профиль пользователя).
• Пробовали падение: убили файл → Python вернул 1 → SSIS красный → SQL Agent прислал письмо.
• Если пароль/токен ценный — применена обфускация shift-cipher (или подключён Key Vault / Integrated Security), чтобы в журнале SSIS и email-уведомлениях не светилось открытое значение.

Плюсы, которые получаете сразу:

• Один скрипт покрывает все свои варианты запуска (прод/тест, разные форматы файлов, разные базы).
• Смена пароля — в одном месте в конфиге SSIS.
• Логи в одной таблице, сортировка по дате, поиск по источнику.
• SQL Agent корректно падает и зовёт человека.

Связанные материалы

• Интеграция 1С и Power BI — архитектура ETL-слоя поверх 1С, где SSIS оркестрирует Python-задачи выгрузки
• Как построить DWH поэтапно — где в общей архитектуре живут SSIS-пакеты и Python-скрипты
• DWH на SQL Server — стек Microsoft, куда этот паттерн ложится нативно
• Wildberries API: грабли при выгрузке — ещё один Python-скрипт по тому же паттерну