mnemosynecore 1.1.5

Internal analytics toolkit for data pipelines

mnemosynecore

mnemosynecore — Python-библиотека для аналитиков и data-инженеров.

Главная идея: писать меньше служебного кода в DAG/ноутбуках и быстрее решать рабочие задачи:

• достать секреты и подключения,
• выполнить SQL в Vertica,
• отправить отчёт в Mattermost,
• сделать скриншот Superset,
• скачать файл из SharePoint,
• добавить надёжные retry.

Зачем это аналитикам

Эта библиотека убирает рутину:

• не нужно каждый раз вручную писать код авторизации,
• одинаковый подход в проде и локально,
• проще проверять логику до выкладки в Airflow,
• меньше копипаста между задачами команды.

Установка

pip install mnemosynecore

Опциональные зависимости:

# Airflow helpers
pip install "mnemosynecore[airflow]"

# Superset API
pip install "mnemosynecore[superset]"

# SharePoint NTLM
pip install "mnemosynecore[sharepoint]"

# ClickHouse через un_conn
pip install "mnemosynecore[clickhouse]"

# Всё вместе
pip install "mnemosynecore[airflow,superset,sharepoint,clickhouse]"

Быстрый старт (2 минуты)

import mnemosynecore as mn

# 1) взять секрет
cfg = mn.get_secret("VERTICA_PROD")
print(cfg["host"])

# 2) выполнить SQL
df = mn.vertica_select(
    conn_id="VERTICA_PROD",
    sql="SELECT CURRENT_DATE AS dt",
)
print(df)

Готовые скрипты с рабочими примерами лежат в папке examples/:

• examples/01_secrets_and_test_mode.py
• examples/02_vertica_examples.py
• examples/03_mattermost_examples.py
• examples/04_superset_examples.py
• examples/05_sharepoint_examples.py
• examples/06_superset_links_and_threads.py

Обычные функции и _test-функции

Почти для всех интеграционных функций есть пара:

• обычная: работает с ENV/Vault/Airflow,
• _test: для локального запуска с JSON-кредами.

Пример пары:

• vertica_select(...)
• vertica_select_test(..., dir_path=... | creds_path=...)

Как передавать креды локально

Есть 2 способа:

1. dir_path — путь к папке, где лежит <CONN_ID>.json
2. creds_path — прямой путь к файлу JSON

Пример:

# Файл: ./local_secrets/VERTICA_DEV.json
df = mn.vertica_select_test(
    conn_id="VERTICA_DEV",
    sql="SELECT 1",
    dir_path="./local_secrets",
)

# Файл: /Users/me/secrets/vertica_dev.json
df = mn.vertica_select_test(
    conn_id="VERTICA_DEV",
    sql="SELECT 1",
    creds_path="/Users/me/secrets/vertica_dev.json",
)

Формат JSON-кредов

Универсальный формат:

{
  "host": "example.host",
  "login": "user",
  "password": "secret",
  "port": 443,
  "schema": "https",
  "extra": "{\"basepath\":\"/api/v4\"}"
}

Mattermost

{
  "host": "mattermost.company.ru",
  "password": "BOT_TOKEN",
  "schema": "https",
  "port": 443,
  "extra": "{\"basepath\":\"/api/v4\"}"
}

Superset

{
  "host": "https://ss.company.ru",
  "login": "svc_user",
  "password": "svc_password",
  "extra": "{\"auth_provider\":\"ldap\"}"
}

Для Vertica/коннектов можно указывать и login, и user:

• если есть user, библиотека использует его,
• если user нет, берется login.

SharePoint

{
  "host": "https://sharepoint.company.ru",
  "login": "svc_sharepoint",
  "password": "svc_password",
  "schema": "O3"
}

Примеры: секреты и подключения

import mnemosynecore as mn

# Прод: ENV/Vault/Airflow
secret = mn.get_secret("MM_BOT")

# Локально: из JSON
secret_test = mn.get_secret_test("MM_BOT", dir_path="./secrets")
secret_test2 = mn.get_secret_test("MM_BOT", creds_path="./secrets/mm_bot.json")

# Универсально: сначала локально, если не нашли — прод
cfg = mn.resolve_secret("MM_BOT", dir_path="./secrets")

# Проверка обязательных полей
mn.require_secret_fields("MM_BOT", ["host", "password"])

Примеры: Vertica

import pandas as pd
import mnemosynecore as mn

# 1) SELECT
sales_df = mn.vertica_select(
    conn_id="VERTICA_PROD",
    sql="""
    SELECT dt, revenue
    FROM mart.daily_sales
    WHERE dt >= CURRENT_DATE - INTERVAL '7 day'
    """,
)

# 2) Scalar
row_count = mn.vertica_select_scalar(
    conn_id="VERTICA_PROD",
    sql="SELECT COUNT(*) FROM mart.daily_sales",
)

# 3) Выполнить SQL из файла
mn.vertica_sql_file(
    file_path="./sql/rebuild_mart.sql",
    conn_id="VERTICA_PROD",
)

# 4) Upsert из DataFrame
new_rows = pd.DataFrame([
    {"id": 1, "dt": "2026-03-01", "revenue": 10.0},
    {"id": 2, "dt": "2026-03-01", "revenue": 20.0},
])
mn.vertica_upsert(
    df=new_rows,
    table_name="mart.daily_sales",
    unique_keys=["id", "dt"],
    conn_id="VERTICA_PROD",
)

# 5) Локальный тест через JSON
test_df = mn.vertica_select_test(
    conn_id="VERTICA_DEV",
    sql="SELECT 1 AS x",
    dir_path="./secrets",
)

Примеры: Mattermost

import pandas as pd
import mnemosynecore as mn

# Текстовое сообщение
mn.send_message(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    text="Ежедневный отчёт готов",
)

# Файл
mn.send_file(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    file_path="./reports/daily_sales.xlsx",
    text="Отчёт во вложении",
)

# DataFrame как CSV
df = pd.DataFrame([{"country": "RU", "value": 100}])
mn.send_dataframe_as_csv(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    df=df,
    file_name="daily_metrics.csv",
)

# Локально (_test) с creds_path
mn.send_message_test(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    text="Проверка локальной отправки",
    creds_path="./secrets/mm_bot.json",
)

# Ответ в тред обычным текстом
mn.send_thread_message(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    root_id="post_id_основного_сообщения",
    text="Дополнительный комментарий в тред",
)

# Можно и файл/таблицу отправить прямо в тред через root_id
mn.send_dataframe_as_csv(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    root_id="post_id_основного_сообщения",
    df=df,
    file_name="details.csv",
)

Примеры: Superset

import mnemosynecore as mn

# Скриншот дашборда (bytes)
png = mn.superset_dashboard_thumbnail(
    superset_conn="SUPERSET_CONN",
    dashboard_name="Финансы - ежедневный обзор",
)

# Отправить скриншот дашборда в Mattermost
mn.send_superset_dashboard_screenshot(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    superset_conn="SUPERSET_CONN",
    dashboard_name="Финансы - ежедневный обзор",
    text="Скриншот утреннего дашборда",
)

# Локально (_test) с JSON кредами
mn.superset_chart_thumbnail_test(
    superset_conn="SUPERSET_CONN",
    chart_id=31729,
    creds_path="./secrets/superset_conn.json",
)

# Ссылка на дашборд (markdown)
dash_link = mn.superset_dashboard_link(
    superset_conn="SUPERSET_CONN",
    dashboard_id=145,
    dashboard_name="Daily KPI",
)

# Отправка только ссылки в канал
mn.send_superset_dashboard_link(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    superset_conn="SUPERSET_CONN",
    dashboard_id=145,
    dashboard_name="Daily KPI",
    text="Отчёт на сегодня",
)

# Ссылки на дашборды в тред + обычные текстовые сообщения в этом же треде
mn.send_superset_dashboard_links_to_thread(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    superset_conn="SUPERSET_CONN",
    dashboards=[
        {"dashboard_name": "Main", "id": 145},
        {"dashboard_name": "Finance", "id": 146},
    ],
    header_text="Ежедневная подборка дашбордов",
    thread_messages=[
        "Если график выглядит странно, пишите в этот тред.",
        "Детали по фильтрам приложу чуть позже.",
    ],
)

# Сохранить скриншот в конкретный файл
mn.superset_dashboard_screenshot_to_file(
    superset_conn="SUPERSET_CONN",
    dashboard_id=145,
    output_path="./tmp/daily_kpi_dashboard.png",
)

# Новый быстрый сценарий:
# взяли permalink из UI Superset (с фильтрами) и отправили скрин + ссылку
mn.send_superset_dashboard_permalink_screenshot(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    superset_conn="SUPERSET_CONN",
    dashboard_name="Daily KPI (filtered)",
    dashboard_url="https://ss.company.ru/superset/dashboard/p/AbCdEf123/?native_filters_key=...",
    text="Борд с применёнными фильтрами из UI",
)

Примеры: SharePoint

import mnemosynecore as mn

# Скачать файл в память
content = mn.sharepoint_download_file(
    sharepoint_conn="SP_CONN",
    file_url="/sites/analytics/reports/plan.csv",
)

# Прочитать CSV сразу в DataFrame
df = mn.sharepoint_read_csv(
    sharepoint_conn="SP_CONN",
    file_url="/sites/analytics/reports/plan.csv",
)

# Сохранить файл локально
mn.sharepoint_download_to_file(
    sharepoint_conn="SP_CONN",
    file_url="/sites/analytics/reports/plan.csv",
    output_path="./tmp/plan.csv",
)

# Локально (_test)
df_local = mn.sharepoint_read_csv_test(
    sharepoint_conn="SP_CONN",
    file_url="/sites/analytics/reports/plan.csv",
    dir_path="./secrets",
)

Примеры готовых сценариев для аналитика

Сценарий 1: загрузить данные из SharePoint и отправить summary в Mattermost

import mnemosynecore as mn

df = mn.sharepoint_read_csv(
    sharepoint_conn="SP_CONN",
    file_url="/sites/analytics/input/sales.csv",
)

summary = df.groupby("country", as_index=False)["amount"].sum()

mn.send_dataframe_preview(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    df=summary,
    title="Сумма продаж по странам",
)

Сценарий 2: выполнить SQL и отправить CSV в Mattermost

import mnemosynecore as mn

df = mn.vertica_select(
    conn_id="VERTICA_PROD",
    sql="SELECT * FROM mart.kpi_daily WHERE dt = CURRENT_DATE - 1",
)

mn.send_dataframe_as_csv(
    channel_id="channel_id_here",
    bot_id="MM_BOT",
    df=df,
    file_name="kpi_daily.csv",
    text="KPI за вчера",
)

Сценарий 3: nightly-скриншоты Superset в разные каналы

import mnemosynecore as mn

mn.send_superset_dashboards_to_channels(
    bot_id="MM_BOT",
    superset_conn="SUPERSET_CONN",
    dashboards=[
        {
            "dashboard_name": "Продажи - общий",
            "channel_id": "sales_channel",
            "text": "Утренний срез",
        },
        {
            "dashboard_name": "Логистика - SLA",
            "channel_id": "ops_channel",
            "text": "SLA на утро",
        },
    ],
)

Основные группы API

Секреты

• get_connection_as_json, get_connection_as_json_test
• get_secret, get_secret_test
• resolve_secret, resolve_secret_test
• has_connection, get_secret_field, require_secret_fields, get_secret_with_defaults

Vertica

• vertica_conn, vertica_conn_test
• vertica_sql, vertica_sql_test
• vertica_select, vertica_select_test
• vertica_select_scalar, vertica_select_scalar_test
• vertica_insert_dataframe, vertica_insert_dataframe_test
• vertica_sql_file, vertica_sql_file_test
• vertica_sql_dir, vertica_sql_dir_test
• vertica_dedupe, vertica_dedupe_test
• vertica_upsert, vertica_upsert_test

Mattermost

• post_message, post_message_test
• send_message, send_message_test
• send_thread_message, send_thread_message_test
• send_thread_messages, send_thread_messages_test
• send_file, send_file_test
• send_file_bytes, send_file_bytes_test
• send_files, send_files_test
• send_dataframe_as_csv, send_dataframe_as_csv_test
• send_dataframe_preview, send_dataframe_preview_test

Superset

• get_superset_client, get_superset_client_test
• superset_request, superset_request_test
• superset_dashboard_url, superset_dashboard_url_test
• superset_chart_url, superset_chart_url_test
• superset_dashboard_link, superset_dashboard_link_test
• superset_chart_link, superset_chart_link_test
• superset_dashboard_thumbnail, superset_dashboard_thumbnail_test
• superset_chart_thumbnail, superset_chart_thumbnail_test
• superset_dashboard_screenshot_to_file, superset_dashboard_screenshot_to_file_test
• superset_chart_screenshot_to_file, superset_chart_screenshot_to_file_test
• superset_screenshot_dashboard, superset_screenshot_dashboard_test
• superset_screenshot_charts, superset_screenshot_charts_test
• send_superset_dashboard_screenshot, send_superset_dashboard_screenshot_test
• send_superset_dashboard_permalink_screenshot, send_superset_dashboard_permalink_screenshot_test
• send_superset_chart_screenshot, send_superset_chart_screenshot_test
• send_superset_dashboard_link, send_superset_dashboard_link_test
• send_superset_chart_link, send_superset_chart_link_test
• send_superset_dashboard_links_to_thread, send_superset_dashboard_links_to_thread_test
• send_superset_chart_links_to_thread, send_superset_chart_links_to_thread_test
• send_superset_dashboards_to_channels, send_superset_dashboards_to_channels_test
• send_superset_charts_to_channels, send_superset_charts_to_channels_test

SharePoint

• sharepoint_download_file, sharepoint_download_file_test
• sharepoint_download_to_file, sharepoint_download_to_file_test
• sharepoint_read_text, sharepoint_read_text_test
• sharepoint_read_dataframe, sharepoint_read_dataframe_test
• sharepoint_read_csv, sharepoint_read_csv_test
• sharepoint_read_excel, sharepoint_read_excel_test
• sharepoint_read_json, sharepoint_read_json_test
• sharepoint_read_sql, sharepoint_read_sql_test
• sharepoint_download_many, sharepoint_download_many_test

Retry

• retry_call, retry

Локальная проверка перед релизом

pytest -q

Релиз и публикация

В проекте есть скрипт release.sh:

• повышает версию,
• запускает тесты,
• собирает пакет,
• публикует в TestPyPI/PyPI,
• делает commit/push.

Примеры:

# Только TestPyPI
./release.sh --target testpypi

# TestPyPI + PyPI
./release.sh --target both

# Явно задать версию
./release.sh --version 1.2.0 --target both

Для публикации нужен токен Twine (~/.pypirc или переменные TWINE_*).