Перейти к содержанию

Пример: модель «Магазин»

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

Модель покажет пользователям AW строки заказов с наименованиями товаров, покупателями и суммами, а для анализа периодов — иерархию «год → месяц → день».

Исходные данные

Модель потребляет три источника данных:

Источник Идентификатор Тип Содержание
Заказы orders sql Заказы: id, product_id, customer_id, summa, created_at
Товары products table Товары: id, name
Покупатели customers table Покупатели: id, name

Метаданные

Начнём с файла model.yml — опишем модель, источники, поля и иерархию дат:

model.yml
model:
  name: Модель товары-покупатели-Заказы (shop)

  data_sources:
    - name: customers
      alias: Покупатели
      type: table

    - name: products
      alias: Товары
      type: table

    - name: orders
      alias: Заказы
      type: sql

  fields:
    - name: id
      alias: Идентификатор заказа
      type: number
    - name: summa
      alias: Сумма заказа
      type: float
    - name: summa_with_discount
      alias: Сумма со скидкой
      type: float

  hiers:
    - name: order_date
      alias: Дата заказа
      fields: ["order_year", "order_month", "order_day"]

Источники описаны с идентификаторами customers, products, orders — под этими именами они будут доступны в коде модели через data_sources.<name>. Соответствие идентификаторов реальным источникам данных AW технический администратор задаст при подключении репозитория.

Иерархия order_date ссылается на три вычисляемых поля — мы добавим их в коде модели. Сами поля перечислены в fields.

Код модели

Теперь код модели:

model.py
from aw_etl.models.code import use_data_source, apply_join, add_calc_field


def main(data_sources):
    """Строит граф вычислений кодовой модели"""
    products = use_data_source(data_sources.products)
    customers = use_data_source(data_sources.customers)
    orders = use_data_source(data_sources.orders, sql="select * from orders")

    # соединяем заказы с товарами и покупателями
    n = apply_join(orders, products, "product_id", "id", how="left")
    n = apply_join(n, customers, "customer_id", "id", how="left")

    # добавляем вычисляемые поля
    n = add_calc_field(n, "order_day", "extract(day from created_at)")
    n = add_calc_field(n, "order_month", "extract(month from created_at)")
    n = add_calc_field(n, "order_year", "extract(year from created_at)")
    n = add_calc_field(n, "summa_with_discount", "summa - 0.3 * summa")

    return n

Разберём код по шагам:

  1. Из источника products получаем весь справочник товаров.
  2. Из источника customers получаем справочник покупателей.
  3. Из источника orders получаем заказы SQL-запросом — параметр sql операции use_data_source работает для источников типа sql.
  4. Операцией apply_join соединяем заказы с товарами по ключу product_id → id (тип left — заказы без товара не теряем), затем с покупателями по customer_id → id.
  5. Операциями add_calc_field добавляем поля дат (order_day, order_month, order_year) и сумму со скидкой.
  6. Возвращаем итоговый узел — он и станет датафреймом модели.

Одноимённые колонки name (у товаров и покупателей) при соединении не конфликтуют: колонка правого узла переименовывается с суффиксом _1 (rename='right' по умолчанию), поэтому в результате будут name и name_1.

Тесты

Проверим модель локальными тестами:

test_model_shop.py
import datetime
from pathlib import Path

import pytest

from aw_client.data_source import DataSourceTable
from aw_client.models.code import DataSourceBindings, DataSourceBindingToTestTable
from aw_client.models.code.runtime import run_code_model, read_code_model_meta


@pytest.fixture
def data_sources():
    """Привязка источников к тестовым таблицам"""
    return DataSourceBindings(
        bindings={
            "products": DataSourceBindingToTestTable(
                table=DataSourceTable(
                    rows=[
                        {"id": 1, "name": "Товар 1"},
                    ]
                )
            ),
            "customers": DataSourceBindingToTestTable(
                table=DataSourceTable(
                    rows=[
                        {"id": 1, "name": "Покупатель 1"},
                    ]
                )
            ),
            "orders": DataSourceBindingToTestTable(
                table=DataSourceTable(
                    rows=[
                        {
                            "id": 1,
                            "product_id": 1,
                            "customer_id": 1,
                            "summa": 1000,
                            "created_at": datetime.datetime(2025, 1, 1, 0, 0, 0),
                        }
                    ]
                )
            ),
        }
    )


def test_meta():
    """Проверка метаданных модели"""
    read_code_model_meta(Path(__file__).parent)


def test_model(data_sources):
    """Проверка модели на тестовых данных"""
    model_root = run_code_model(Path(__file__).parent, data_sources=data_sources)

    assert set(model_root.df.columns).issuperset(
        {
            "id",
            "product_id",
            "customer_id",
            "summa",
            "name",
            "name_1",
            "order_day",
            "order_month",
            "order_year",
            "summa_with_discount",
        }
    )
    assert model_root.df.count() == 1
  1. Фикстура data_sources описывает тестовые таблицы для всех трёх источников: один товар, один покупатель и один заказ.
  2. Тест test_meta проверяет корректность model.yml.
  3. Тест test_model запускает модель функцией run_code_model() и проверяет результат: состав колонок (включая вычисляемые) и число строк после соединения.

Запускаем тесты:

(.venv) $ pytest model_shop/test_model_shop.py -v

Структура репозитория

Готовый пример в репозитории:

code-models/
├── model_shop/
│   ├── model.py
│   ├── model.yml
│   └── test_model_shop.py
└── README.md

Осталось опубликовать модель: коммит в feature-ветку → merge в production-ветку → репозиторий подключает технический администратор. После синхронизации модель появится в аккаунте AW и её смогут использовать пользователи.

Что развить дальше