Metadata-Version: 2.5
Name: cdc-1c
Version: 0.1.24
Summary: Change data capture (CDC) from 1C:Enterprise to your data warehouse
Project-URL: Homepage, https://github.com/pavel-v-sobolev/cdc_1C
Project-URL: Repository, https://github.com/pavel-v-sobolev/cdc_1C
Project-URL: Issues, https://github.com/pavel-v-sobolev/cdc_1C/issues
Author-email: Pavel Sobolev <pavel-v-sobolev@yandex.ru>
License-Expression: MIT
License-File: LICENSE
Keywords: 1c,1c-enterprise,cdc,data-engineering,dwh,etl,postgres,python
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Intended Audience :: Science/Research
Classifier: Intended Audience :: System Administrators
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: croniter>=6.0.0
Requires-Dist: dbmerge>=1.0.22
Requires-Dist: requests>=2.33.0
Requires-Dist: sqlalchemy>=2.0.49
Requires-Dist: xmltodict>=1.0.4
Provides-Extra: dev
Requires-Dist: psycopg2-binary>=2.9.12; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9.12; extra == 'postgres'
Description-Content-Type: text/markdown

[![PyPI version](https://img.shields.io/pypi/v/cdc-1c.svg)](https://pypi.org/project/cdc-1c/)
[![Python versions](https://img.shields.io/pypi/pyversions/cdc-1c.svg)](https://pypi.org/project/cdc-1c/)


**cdc-1c** is a docker container and a Python library, that provides 1C system data loading to data warehouse using Change Data Capture apporach. \
It engages standard ODATA mechanism and standard 1C exchange plan mechanism to extract data from 1C system and upsert changes to the target DB.

**cdc-1c** - это докер контейнер и python-библиотека, предназначенные для получения данных из 1С, использующий подход CDC (загрузка изменений данных). \
Продукт использует стандартный интерфейс ODATA и механизм планов обмена для выгрузки изменений данных из системы 1С и обновления данных в целевой БД.

# Общий принцип действия
1) Основной объект библиотеки это оркестратор `Replicator1C`, который циклично читает изменения из 1С (через OData + план обмена) и
пишет их в целевую БД Postgres, подтверждая приём пакета только после успешного сохранения.
2) В Postgres объекты сохраняются в виде отдельных таблиц на каждый регистр, справочник, документ, табличные части документа или справочника. Структура соответствует структуре хранения в 1С, но имена таблиц и полей автоматически переводятся в транслит.
3) Полученные таблицы вы можете сами использовать для построения запросов, но библиотека предлагает также механизм обработчиков (handlers). Когда из 1С приходят новые данные, обработчик обновляет соответствующую часть в витрине данных. Витрину вы описываете сами — представлением (view) в базе данных. В примере показано как сделать обновление витрины быстрым, только по изменениям. Также механизм обработчиков это по сути ваш код на python, в который вы можете вставить что нужно, например, отправка в RabbitMQ или еще что-то.


# Что нужно для работы
1) опубликовать базу 1с на web
2) настроить план обмена в конфигураторе и включить в его состав нужные объекты 1с
3) создать пользователя для доступа к odata и дать ему необходимые права (`чтение` и `изменение` к плану обмена, `чтение` к загружаемым объектам).
4) дать роль `чтение` всем пользователям к плану обмена (иначе будут ошибки при сохранении объектов)
5) создать узел обмена с использованием внешней обработки `cdc-1c.odt`
6) запустить загрузку: докер-образом `sobolevp/cdc-1c` (см. «Запуск в docker») либо python-библиотекой `cdc-1c` (см. ниже)

# Использование библиотеки python

## Установка
```bash
pip install cdc-1c
```

Для записи изменений необходим **PostgreSQL** (Другие СУБД не тестировались, хотя в теории возможны).
Для записи используется библиотека dbmerge. Все необходимые схемы, таблицы и поля модуль создает сам.


```python
from sqlalchemy import create_engine
from cdc_1c import Replicator1C

# pool_size >= full_load_workers + 2 (+1 на каждого обработчика и каждое расписание) —
# почему столько, см. README_DB.md, «Сколько нужно соединений к БД»
engine = create_engine("postgresql+psycopg2://user:pass@localhost:5432/cdc_1c", pool_size=5)

rep = Replicator1C(
    odata_url="http://host/base/odata/standard.odata",
    odata_auth=("odata", "secret"),        # (user, password) либо None без авторизации
    exchange_name="ВашПланОбмена",              # имя плана обмена в 1С
    queue_guid="aaaaaaaa-aaaa-aaaa-aaa-aaaaaaaaaaaa",  # Ref_Key узла обмена
    engine=engine,
    db_schema="cdc_1c",                    # None → схема БД по умолчанию (public у Postgres)
    db_temp_schema="cdc_1c_tmp",           # схема промежуточных таблиц merge; None → схема данных
    request_timeout=60,                    # таймаут HTTP-запросов к 1С, сек (по умолчанию 60 на коннект, 900 на ответ)
    full_load_workers=2,                   # число фоновых потоков полной выгрузки
)

rep.run_forever(interval=60)               # цикл опроса раз в 60 секунд
```

Вариант многопоточного запуска, дающий возможность добавления нескольких обработчиков (Handler) и 
дополнительных репликаторов (Replicator) для других планов обмена, можно посмотреть в этом файле: [runner.py](config/runner.py)



## Запуск в docker

```bash
docker run --rm --network host \
  -e CDC1C_ODATA_URL="http://server/base/odata/standard.odata" \
  -e CDC1C_ODATA_USER=odata -e CDC1C_ODATA_PASSWORD=secret \
  -e CDC1C_EXCHANGE_NAME="ВашПланОбмена" \
  -e CDC1C_QUEUE_GUID="aaaaaaaa-aaaa-aaaa-aaa-aaaaaaaaaaaa" \
  -e CDC1C_DB_URL="postgresql+psycopg2://user:pass@database_host:5432/cdc" \
  -e CDC1C_DB_SCHEMA=cdc_1c -e CDC1C_DB_TEMP_SCHEMA=cdc_1c_tmp \
  -v "$PWD/config:/config:ro" \
  sobolevp/cdc-1c:latest
```

`-v "$PWD/config:/config:ro"` — монтирует папку `config` из текущего каталога. В ней лежит файл [runner.py](config/runner.py), а также подпапка с примерами обработчиков (config/handlers).

**docker compose.** Пример — [docker-compose.yml](docker-compose.yml) в репозитории.


## Запуск из окружения

Если хочется не писать код вовсе, есть готовый entrypoint — `python -m cdc_1c` (он же команда
`cdc-1c`). Он читает те же параметры из переменных окружения, см описание: [README_ENV.md](README_ENV.md)



### Как узнать guid узла обмена

`queue_guid` — это `Ref_Key` узла плана обмена, того самого, на который 1С регистрирует изменения
(`ЭтотУзел` не подходит: он описывает саму базу-источник). Если guid неизвестен, оставьте параметр
пустым (`queue_guid=""`, или просто не задавайте `CDC1C_QUEUE_GUID`) и запустите: чтение изменений
выведет в лог список узлов плана обмена и остановится.

```
ERROR cdc_1c.change_reader: queue_guid is not set. Available nodes of exchange plan ДляODATA:
    a9bc23c5-3689-11f1-926c-0800270bc6cb  CDC  Витрина
```

Guid из первой колонки и есть искомый `queue_guid`.

## Режимы: `run_once` и `run_forever`

```python
rep.run_once()                 # один цикл: read → save → notify (подтверждение только после save)
rep.run_forever(interval=60)   # основной режим работы. бесконечный цикл run_once с паузой; фоном — полные выгрузки
```

- `run_once(notify_changes=False)` — не подтверждать приём (пакет останется в очереди 1С; сделано для отладки).
- `run_forever(interval, max_iterations=0)` — `max_iterations>0` ограничивает число итераций.


## Полная (первоначальная) выгрузка

При работе `run_forever` объекты, впервые встреченные в пакете изменений, автоматически ставятся в
очередь на полную выгрузку и грузятся фоновыми потоками. 
- Можно запустить выгрузку и вручную, поставив флаг в таблице metadata_objects_1c - full_load_is_required=`True`
- Полная выгрузка спроектирована так, чтобы работать параллельно с получением изменений объекта и не затирать свежие изменения объекта. Также она автоматически разбивает объект на отрезки по времени и выгружает его порциями с автоматическим подбором размера (batch_size).


## Что появляется в целевой БД

На каждый объект 1С заводится своя таблица: имя транслитерируется, к полям добавляются служебные
(`merged_on`, `inserted_on`, `is_deleted_or_empty`, `exchange_message_no`). Схемы, таблицы и новые
колонки библиотека создаёт сама. Рядом появляются служебные таблицы, по которым видно состояние
загрузки: журнал `replicator_1c_log`, реестр объектов `metadata_objects_1c`, состояние обработчиков
`handlers_1c` и реестр незавершённых записей `writes_in_process_1c`.

Подробно — [README_DB.md](README_DB.md): состав полей, что означает `is_deleted_or_empty`, зачем
отдельная схема промежуточных таблиц и что лежит в каждой служебной таблице.

## Классы библиотеки

`Replicator1C` читает изменения и пишет их в БД, `Handler1C` + `HandlerLoop` дают возможность запуска своего кода
по событию изменения, `FullLoadCron` — полную выгрузку по расписанию. Что каждый из них принимает и
что у него можно вызвать — [README_API.md](README_API.md).

## Логирование

Из коробки библиотека вешает вывод на логгер `cdc_1c` (INFO), если приложение не настроило логирование
само. Настроили своё — библиотека молчит и пишет через стандартный `logging`.

## Шум со стороны 1С

1С регистрирует изменение объекта на любую перезапись, поэтому в пакет приезжает масса записей, у которых поменялись только `DataVersion` и номер пакета. Такие записи библиотека изменением не считает: строка не обновляется, `merged_on` остаётся прежним — и обработчик, который выбирает данные по `merged_on`, впустую не запускается.


## Дальнейшая обработка и материализация

1С хранит данные нормализованно: чтобы дотянуться из регистра до кода товара, нужен `JOIN` со
справочником по guid. Для КХД обычно удобнее менее строгая нормализация — витрины, собранные
заранее.

Считать их вхолостую по расписанию не нужно: репликатор знает, когда данные изменились, и сам
вызывает ваш код. Такой код называется **обработчиком** — это класс с двумя строчками объявления:

```python
from cdc_1c import Handler1C, HandlerContext

class ZakazyKlientov(Handler1C):
    ON = ["AccumulationRegister_ZakazyKlientov"]   # имена ТАБЛИЦ в БД, не объектов 1С

    def handle(self, context: HandlerContext) -> None:
        self.execute(context, "insert into ... where merged_on > :last_run_at",
                     last_run_at=context.last_run_at)
```

Тем же способом изменения отправляют во внешнюю систему — очередь, вебхук, другую БД.

Начинают обычно с **представления (view)**: в нём собран `SELECT` со всеми `JOIN` по guid-ключам.
DDL можно прописать прямо в обработчике (`setup`) или завести отдельно. Дальше обработчик работает так:

- библиотека вызывает его, передавая дату и время `last_run_at` — с них и нужно обновлять;
- обработчик пишет `SELECT`, который находит по представлению изменившиеся ключи условием
  `merged_on > last_run_at` (`merged_on` есть и у таблиц, подключённых через `JOIN`, — см. пример);
- этот `SELECT` уходит в условия обновления
  [dbmerge](https://github.com/pavel-v-sobolev/dbmerge): `source_condition` говорит, какие ключи
  взять из источника, `delete_condition` — в каком множестве ключей чистить строки, если что-то
  удалилось. `delete_condition` нужен не всегда: строки, выпавшие из набора движений или
  табличной части, репликатор не удаляет, а помечает (см. «Шум со стороны 1С» и `is_deleted_or_empty`),
  поэтому витрине с тем же ключом, что у источника, достаточно обычного обновления. Удаление
  нужно там, где ключ витрины свой, — например, у агрегата по `GROUP BY`.

В примерах разобраны два варианта — с кодом и подробными пояснениями:

1) ключ витрины совпадает с первичным ключом объекта 1С (суррогатный guid). Годится для таблицы
   фактов: [config/handlers/zakazy_klientov.py](config/handlers/zakazy_klientov.py);
2) ключ витрины заменён на бизнес-ключ — например, номер документа вместо guid. Связь с «сырыми»
   данными хранится в колонках-массивах (`ARRAY`) с индексами *GIN*: по ним обработчик быстро
   находит, какие агрегированные ключи задело изменение в источнике.
   [config/handlers/zakazy_klientov_grouped.py](config/handlers/zakazy_klientov_grouped.py).


---

## Ссылки

| файл | описание |
|---|---|
| [README.md](README.md) | этот файл: обзор, установка, запуск |
| [README_ENV.md](README_ENV.md) | запуск из окружения: все переменные `CDC1C_*`, значения по умолчанию, ошибки конфигурации |
| [README_API.md](README_API.md) | классы, которые собирает пользователь: `Replicator1C`, `Handler1C`, `HandlerLoop`, `FullLoadCron`; обработчики — окно изменений, когда их вызывают, пересборка витрины |
| [README_DB.md](README_DB.md) | что появляется в целевой БД: таблицы, служебные поля, служебные таблицы |
| [DESIGN.md](DESIGN.md) | внутреннее устройство для тех, кто правит код: интерфейс OData, цикл изменений, пагинация полной выгрузки, гонки со снимком, механика обработчиков |
| [CHANGELOG.md](CHANGELOG.md) | что менялось от версии к версии |
| [config/runner.py](config/runner.py) | шаблон точки входа: репликатор, обработчики и расписания в одном файле |
| [config/handlers](config/handlers/) | примеры обработчиков |
