Metadata-Version: 2.4
Name: al-data-core
Version: 0.6.4
Summary: Generic data pipeline engine for Airflow (config, processor, DAG factory).
Author-email: AL Data Labs <pypi@aldatalabs.com>
License-Expression: MIT
Project-URL: Homepage, https://aldatalabs.com
Keywords: airflow,data-engineering,etl,dag
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: al-data-utils>=0.3.2
Requires-Dist: pandas
Dynamic: license-file

# al-data-core

Generic Airflow data pipeline engine. Install once, configure per project.

## Install

```bash
pip install al-data-core
```

Python 3.10+. `al-data-utils` is installed automatically. `airflow`, `pandas`, `sqlalchemy`, `pyodbc` are expected from the host environment.

## Modules

- `config_processor` — `DataCoreConfig`
- `tasks_processor` — `TaskProcessor`
- `azure_as_processor` — `run_azure_as_refresh`
- `external_handler_base` — `ExternalHandlerBase`, `discover_handlers` (used for XLS and API handlers)
- `dags_installer` — CLI for copying DAG templates

## DAG Templates

After installing, copy the ready-to-use DAG templates to your Airflow dags folder:

```bash
al-data-core copy-dags                          # latest → /opt/airflow/dags
al-data-core copy-dags /my/dags                 # custom path
al-data-core copy-dags --version v2.0.0         # specific Airflow version
al-data-core list-versions                      # show available versions
```

Available versions:

- `v2.0.0` — Airflow 2.x
- `v3.0.0` — Airflow 3.x

Two files are copied:

- `data_core_dag.py` — main DAG file, reads `data_core_tasks_config.yaml` and registers all project DAGs
- `data_core_config_engine_dag.py` — runs every 15 min, queries config DB and YAML, writes `data_core_tasks_config.yaml`

## Project Layout (per-project repo)

```text
dags/
├── data_core_dag.py                    <- copied from template
├── data_core_config_engine_dag.py      <- copied from template
└── data_core/
    ├── data_core_config.yaml           <- project config
    ├── .configs/data_core/data_core_tasks_config.yaml  <- generated by config engine dag
    ├── data_core_handler_sql/
    │   └── sql_queries/
    │       └── <scope>/
    │           └── <param_code>.sql
    ├── data_core_handler_xls/          <- optional, project-specific
    └── data_core_handler_api/          <- optional, project-specific
```

## data_core_config.yaml

```yaml
CONFIG:
  config_connection_variable: my_mssql_secret   # Airflow variable name
  mappings:
    config_get_tables_list:          integrations_db.config.get_tables_list
    config_parameters:               integrations_db.config.parameters
    config_fn_get_param_values:      integrations_db.config.fn_get_param_values
    config_calc_date_ranges:         integrations_db.config.calc_date_ranges
    config_update_param_values:      integrations_db.config.update_param_values

DAGS:
  my_project_ods_dag:
    schedule_interval: '5 */1 * * *'
    scopes: ['ODS_SOURCE']
    tags: ['my_project']
  my_project_dw_dag:
    schedule_interval: '30 */1 * * *'
    scopes: ['DW_TARGET']
    tags: ['my_project']
    azure_as:                         # optional: refresh semantic model after ETL
      connection_variable: my_azure_as_secret
      models:
        - name: MyModel
          refresh_request: {Type: Full, CommitMode: transactional, MaxParallelism: 2, RetryCount: 2, Objects: []}

ODS_SOURCE:
  enabled: true
  source_type: mssql
  source_connection_variable: source_mssql_secret
  destination_connection_variable: dest_mssql_secret

DW_TARGET:
  enabled: true
  destination_connection_variable: dest_mssql_secret

XLS:
  enabled: true
  handler: data_core_handler_xls   # opts this scope into handler-based dispatch instead of seq_code-based copy/merge

ods_britix_api:
  enabled: true
  handler: data_core_handler_api/britix24
```

## Handlers

A scope opts into handler-based dispatch by setting `handler` to its handlers folder (relative to `data_core_config.yaml`, or an absolute path) — the value *is* the path, so `TaskProcessor` doesn't hardcode scope_code names and adding a new handler scope needs only config, never a template change. Two layouts are supported:

* **One file per param_code** — file name matches the task's `param_code` exactly.
* **One generic handler per scope** — when the folder holds a single handler file, it serves every param_code in the scope and distinguishes tasks via `self.context.param_code`.

```python
# dags/data_core/data_core_handler_xls/fact_sales_plan.py
from al_data_core.external_handler_base import ExternalHandlerBase

class MySalesHandler(ExternalHandlerBase):
    def proceed(self) -> None:
        ...
```

```python
# dags/data_core/data_core_handler_api/britix24/bitrix_ingest.py
from al_data_core.external_handler_base import ExternalHandlerBase

class BitrixIngestHandler(ExternalHandlerBase):
    def proceed(self) -> None:
        scope = self.context.scope_config
        entity = scope['entities'][self.context.param_code]
        ...
```

`TaskProcessor` constructs the handler with no arguments, sets `handler.context` to a `HandlerContext` (scope_code, param_code, scope_config, config_file_path, params_details, debug), then calls `proceed()` — handlers read run-time settings from `self.context` instead of hardcoding them. Discovery happens at task-run time, not DAG-parse time, so a broken handler file can't fail the whole DAG file's parse and handler modules aren't re-imported every scheduler cycle. Each file must define exactly one `ExternalHandlerBase` subclass.

## Changelog

See `CHANGELOG.md` in the repository for release notes.
