Cookiecutter scaffold: стартер для нового adapter

Писать adapter с нуля — boilerplate. 50+ файлов, точная структура каталогов, правильные setup.py ключи, корректный __init__.py для регистрации в dbt. dbt Labs maintains official cookiecutter template, который генерирует всё это за одну команду.

В этом уроке разбираем, что генерирует scaffold, как его использовать, и что нужно изменить под ваш warehouse.

Что такое cookiecutter

Cookiecutter — Python tool для project templating. Принимает template directory с placeholder’ами ({{ adapter_name }}) и генерирует конкретный проект на их основе.

pip install cookiecutter
cookiecutter https://github.com/dbt-labs/dbt-database-adapter-scaffold

Cookiecutter спросит:

adapter_name [my_adapter]: oceanbase
project_name [dbt-oceanbase]: dbt-oceanbase
author_name [your_name]: Lev Neganov
author_email [[email protected]]: [email protected]
copyright_year [2026]: 2026
adapter_description [A dbt adapter for ...]: A dbt adapter for OceanBase
license [Apache-2.0]: Apache-2.0
python_requires [>=3.9]: >=3.9

И сгенерирует ~50 файлов в ./dbt-oceanbase/.

Что генерируется: общая структура

dbt-oceanbase/
├── README.md
├── LICENSE
├── setup.py
├── pyproject.toml
├── MANIFEST.in
├── .gitignore
├── dev-requirements.txt
├── .github/
│   └── workflows/
│       └── ci.yml
├── dbt/
│   ├── adapters/
│   │   ├── __init__.py
│   │   └── oceanbase/
│   │       ├── __init__.py
│   │       ├── __version__.py
│   │       ├── connections.py     # ← ConnectionManager
│   │       ├── credentials.py     # ← Credentials dataclass
│   │       ├── relation.py        # ← Relation subclass
│   │       ├── column.py          # ← Column subclass
│   │       ├── impl.py            # ← Adapter implementation
│   │       └── exceptions.py
│   └── include/
│       └── oceanbase/
│           ├── __init__.py
│           ├── dbt_project.yml    # ← Default project config для adapter
│           ├── profile_template.yml  # ← Шаблон profiles.yml
│           ├── sample_profiles.yml
│           └── macros/
│               ├── adapters.sql   # ← Adapter-specific macros
│               ├── catalog.sql
│               ├── materializations/
│               └── relations/
└── tests/
    ├── __init__.py
    ├── conftest.py
    └── functional/
        └── adapter/
            ├── test_basic.py     # ← Имплементирует dbt-tests-adapter suite
            ├── test_simple_copy.py
            └── ...

Это production-grade scaffold: setup.py с правильными dependencies, GitHub Actions CI, тестовая инфраструктура подключённая к dbt-tests-adapter, документация.

Главные файлы и что в них

setup.py / pyproject.toml

# setup.py
from setuptools import setup, find_namespace_packages

setup(
    name='dbt-oceanbase',
    version='1.0.0',
    description='A dbt adapter for OceanBase',
    long_description=open('README.md').read(),
    long_description_content_type='text/markdown',
    author='Lev Neganov',
    author_email='[email protected]',
    license='Apache-2.0',
    url='https://github.com/levn/dbt-oceanbase',

    packages=find_namespace_packages(include=['dbt.*']),
    package_data={
        'dbt': [
            'include/oceanbase/dbt_project.yml',
            'include/oceanbase/sample_profiles.yml',
            'include/oceanbase/profile_template.yml',
            'include/oceanbase/macros/*.sql',
            'include/oceanbase/macros/**/*.sql',
        ],
    },

    install_requires=[
        'dbt-adapters>=1.7,<2.0',
        'dbt-common>=1.10,<2.0',
        'dbt-core>=1.8,<2.0',
        # Warehouse client library:
        'pymysql>=1.0',   # ← OceanBase compatible с MySQL protocol
    ],

    python_requires='>=3.9',
)

Ключевые моменты:

• find_namespace_packages(include=['dbt.*']) — это namespace package pattern. dbt-core видит все adapter’ы под dbt.adapters.*.
• package_data — включить SQL macros в pip-package. Без этого dbt не найдёт их в runtime.
• install_requires: dbt-adapters, dbt-common, dbt-core — обязательно. Plus warehouse client library (pymysql для OceanBase, snowflake-connector для Snowflake, etc).
• python_requires — supported Python versions.

dbt/adapters/oceanbase/__init__.py

from dbt.adapters.oceanbase.connections import OceanBaseConnectionManager
from dbt.adapters.oceanbase.connections import OceanBaseCredentials
from dbt.adapters.oceanbase.impl import OceanBaseAdapter

from dbt.adapters.base import AdapterPlugin
from dbt.include import oceanbase

Plugin = AdapterPlugin(
    adapter=OceanBaseAdapter,
    credentials=OceanBaseCredentials,
    include_path=oceanbase.PACKAGE_PATH,
)

Это регистрационная точка для dbt. Когда dbt вызывает pip show dbt-oceanbase, он находит Plugin = AdapterPlugin(...) и регистрирует:

• adapter: главный класс
• credentials: dataclass для profiles.yml
• include_path: путь к SQL macros

Без этого dbt не знает, что adapter существует.

dbt/adapters/oceanbase/credentials.py

from dataclasses import dataclass
from typing import Optional

from dbt.adapters.contracts.connection import Credentials


@dataclass
class OceanBaseCredentials(Credentials):
    host: str
    port: int = 2881
    user: str = 'root'
    password: str = ''
    database: str = 'test'
    schema: str = ''

    @property
    def type(self) -> str:
        return 'oceanbase'

    @property
    def unique_field(self) -> str:
        return self.host

    def _connection_keys(self):
        return ('host', 'port', 'user', 'database', 'schema')

Это dataclass, который описывает структуру profiles.yml для adapter’а. Detailed разбор в следующем уроке.

dbt/adapters/oceanbase/connections.py

from contextlib import contextmanager
from typing import Optional

import pymysql

from dbt.adapters.base import BaseConnectionManager
from dbt.adapters.contracts.connection import (
    AdapterResponse,
    Connection,
    ConnectionState,
)
from dbt.adapters.events.logging import AdapterLogger
from dbt.adapters.exceptions import FailedToConnectError


logger = AdapterLogger('OceanBase')


class OceanBaseConnectionManager(BaseConnectionManager):
    TYPE = 'oceanbase'

    @classmethod
    def open(cls, connection: Connection) -> Connection:
        if connection.state == ConnectionState.OPEN:
            return connection

        credentials = connection.credentials
        try:
            handle = pymysql.connect(
                host=credentials.host,
                port=credentials.port,
                user=credentials.user,
                password=credentials.password,
                database=credentials.database,
            )
        except pymysql.Error as e:
            raise FailedToConnectError(str(e))

        connection.handle = handle
        connection.state = ConnectionState.OPEN
        return connection

    @classmethod
    def get_response(cls, cursor) -> AdapterResponse:
        rows_affected = cursor.rowcount
        return AdapterResponse(
            _message=f'OK',
            rows_affected=rows_affected,
        )

    @contextmanager
    def exception_handler(self, sql: str):
        try:
            yield
        except pymysql.Error as e:
            self.release()
            raise RuntimeError(str(e))

    def cancel(self, connection: Connection):
        connection.handle.close()

Это минимальный ConnectionManager. Детальный разбор в уроке 4.

dbt/adapters/oceanbase/impl.py

from dbt.adapters.sql import SQLAdapter

from dbt.adapters.oceanbase.connections import OceanBaseConnectionManager
from dbt.adapters.oceanbase.relation import OceanBaseRelation
from dbt.adapters.oceanbase.column import OceanBaseColumn


class OceanBaseAdapter(SQLAdapter):
    ConnectionManager = OceanBaseConnectionManager
    Relation = OceanBaseRelation
    Column = OceanBaseColumn

    @classmethod
    def date_function(cls) -> str:
        return 'CURDATE()'

    @classmethod
    def convert_text_type(cls, agate_table, col_idx):
        return 'VARCHAR(255)'

    @classmethod
    def convert_number_type(cls, agate_table, col_idx):
        decimals = agate_table.aggregate(agate.MaxPrecision(col_idx))
        return 'DECIMAL(15, 2)' if decimals else 'BIGINT'

    @classmethod
    def convert_datetime_type(cls, agate_table, col_idx):
        return 'DATETIME'

    @classmethod
    def convert_boolean_type(cls, agate_table, col_idx):
        return 'BOOLEAN'

    # Inherit from SQLAdapter — другие methods наследуются

dbt/include/oceanbase/macros/adapters.sql

-- macros/adapters.sql

{% macro oceanbase__list_schemas(database) %}
  {% call statement('list_schemas', fetch_result=True, auto_begin=False) %}
    SHOW DATABASES
  {% endcall %}
  {{ return(load_result('list_schemas').table) }}
{% endmacro %}


{% macro oceanbase__check_schema_exists(information_schema, schema) %}
  {% call statement('check_schema_exists', fetch_result=True, auto_begin=False) %}
    SELECT COUNT(*)
    FROM information_schema.schemata
    WHERE schema_name = '{{ schema }}'
  {% endcall %}
  {{ return(load_result('check_schema_exists').table) }}
{% endmacro %}


{% macro oceanbase__create_schema(relation) %}
  {%- call statement('create_schema') -%}
    CREATE DATABASE IF NOT EXISTS {{ relation.without_identifier() }}
  {%- endcall -%}
{% endmacro %}


{% macro oceanbase__drop_schema(relation) %}
  {%- call statement('drop_schema') -%}
    DROP DATABASE IF EXISTS {{ relation.without_identifier() }}
  {%- endcall -%}
{% endmacro %}


{% macro oceanbase__get_columns_in_relation(relation) %}
  {% call statement('get_columns_in_relation', fetch_result=True) %}
    SHOW COLUMNS FROM {{ relation }}
  {% endcall %}
  {% set table = load_result('get_columns_in_relation').table %}
  {{ return(...) }}
{% endmacro %}

OceanBase использует MySQL-compatible syntax: SHOW DATABASES, SHOW COLUMNS. Это override default’ов (которые используют ANSI information_schema).

После генерации: что менять

Cookiecutter генерирует scaffold, не working adapter. Дальше вам надо:

Шаг 1: Customize credentials

В credentials.py добавьте поля специфичные для вашего warehouse:

@dataclass
class OceanBaseCredentials(Credentials):
    host: str
    port: int = 2881
    user: str = 'root'
    password: str = ''
    database: str = 'test'
    schema: str = ''
    
    # OceanBase-specific:
    tenant: Optional[str] = None
    cluster: Optional[str] = None
    
    # Standard:
    threads: int = 4

Шаг 2: Implement connection

В connections.py:

@classmethod
def open(cls, connection: Connection) -> Connection:
    credentials = connection.credentials
    
    # OceanBase connection строка: user@tenant#cluster
    if credentials.tenant:
        user_str = f'{credentials.user}@{credentials.tenant}'
        if credentials.cluster:
            user_str += f'#{credentials.cluster}'
    else:
        user_str = credentials.user
    
    handle = pymysql.connect(
        host=credentials.host,
        port=credentials.port,
        user=user_str,
        password=credentials.password,
        database=credentials.database,
    )
    
    connection.handle = handle
    connection.state = ConnectionState.OPEN
    return connection

Шаг 3: Configure types

В impl.py:

@classmethod
def convert_text_type(cls, agate_table, col_idx):
    # OceanBase: max VARCHAR length 65535
    return 'VARCHAR(65535)'

@classmethod
def convert_number_type(cls, agate_table, col_idx):
    return 'DOUBLE'

Шаг 4: Run dbt-tests-adapter suite

cd dbt-oceanbase/
pip install -e .
pip install -r dev-requirements.txt
pytest tests/functional/adapter/

Most tests падут изначально. Read failures, fix one by one. Это test-driven adapter development.

Шаг 5: Setup local development

# Запустить OceanBase локально (Docker)
docker run -d --name oceanbase \
    -p 2881:2881 \
    oceanbase/oceanbase-ce:latest

# Создать test profiles.yml
mkdir -p ~/.dbt
cat > ~/.dbt/profiles.yml << EOF
dbt-oceanbase-test:
  target: dev
  outputs:
    dev:
      type: oceanbase
      host: localhost
      port: 2881
      user: root
      password: ''
      database: test
EOF

# Test через example project
git clone https://github.com/dbt-labs/dbt-jaffle-shop
cd dbt-jaffle-shop
dbt debug --profiles-dir ~/.dbt --profile dbt-oceanbase-test
dbt run

Какие тесты в scaffold

Scaffold включает tests/functional/adapter/ — это suite от dbt-tests-adapter package, который dbt Labs maintains для testing любого adapter’а.

Главные test classes:

• BaseSimpleMaterializations — view, table, ephemeral базовые тесты
• BaseIncremental — incremental тесты
• BaseSeeds — seed CSV -> table
• BaseSnapshots — snapshot strategies
• BaseDocsGenerate — docs generation
• BaseSingularTestTimezones — singular tests
• BaseDocsGenerateBase — catalog generation

Каждый класс — это base class с тестами. Ваш scaffold имеет local subclasses, например:

# tests/functional/adapter/test_basic.py
from dbt.tests.adapter.basic.test_base import BaseSimpleMaterializations
from dbt.tests.adapter.basic.test_singular_tests import BaseSingularTests

class TestSimpleMaterializationsOceanBase(BaseSimpleMaterializations):
    pass  # Inherit all tests, nothing custom

class TestSingularTestsOceanBase(BaseSingularTests):
    pass

Running pytest tests/functional/adapter/test_basic.py запускает full suite. Каждый test:

1. Создаёт temp dbt project
2. Запускает dbt commands (run, test, seed)
3. Проверяет результат (table created, rows correct, etc.)

Если ваш adapter работает correctly — tests pass. Иначе failures указывают на bugs.

Полная suite в dbt-tests-adapter package — ~50 test classes. Это specification, что должен делать adapter для dbt-compatibility.

CI Setup (GitHub Actions)

Scaffold генерирует .github/workflows/ci.yml:

name: CI

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      oceanbase:
        image: oceanbase/oceanbase-ce:latest
        ports:
          - 2881:2881
    
    strategy:
      matrix:
        python-version: ['3.9', '3.10', '3.11']
    
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}
      - run: pip install -e .
      - run: pip install -r dev-requirements.txt
      - run: pytest tests/functional/adapter/ -v

Run dbt-tests-adapter suite на каждый push. Production-grade.

Когда НЕ использовать scaffold

Scaffold — отправная точка. В нескольких сценариях лучше начать с zero:

1. Existing similar adapter — если делаете fork существующего (например, dbt-postgres -> dbt-greenplum) — fork существующий repo, не scaffold.

2. Не SQL warehouse — scaffold optimized для SQLAdapter. Для BaseAdapter (REST API, Spark-like) — лучше скопировать структуру с dbt-spark или dbt-bigquery.

3. Learning purposes — если изучаете adapter API, scaffold скрывает много deta’el. Лучше написать с нуля для понимания.

Но для production adapter scaffold — best starting point.

Попробуй сам

1. Установите cookiecutter:

   pip install cookiecutter

2. Сгенерируйте scaffold для гипотетического warehouse:

   cookiecutter https://github.com/dbt-labs/dbt-database-adapter-scaffold
   # adapter_name: testdb
   # project_name: dbt-testdb
   # ... остальные defaults

3. Изучите generated файлы:

   cd dbt-testdb/
   find . -type f -name "*.py" -o -name "*.sql" -o -name "*.yml" | head -30

4. Откройте dbt/adapters/testdb/impl.py. Это MyAdapter template. Найдите все TODO комментарии — это места, где нужно ваш код.

5. Прочитайте tests/functional/adapter/test_basic.py. Это инкорпорированные base tests от dbt-tests-adapter.

6. Запустите pip install -e .. Adapter зарегистрирован в pip namespace. Проверьте:

   python -c "from dbt.adapters.testdb import Plugin; print(Plugin)"

   Должно вывести AdapterPlugin object.

7. Bonus: подключите к реальному warehouse (например, SQLite через sqlite3 library) и сделайте hello-world. Это часы работы.

Ключевые выводы

1. dbt-database-adapter-scaffold — official cookiecutter template для adapter’ов. Генерирует ~50 файлов: src, tests, CI, packaging.

2. AdapterPlugin в __init__.py — регистрационная точка. dbt находит adapter через namespace package pattern.

3. package_data в setup.py — включить SQL macros в pip-package. Без этого dbt не находит них в runtime.

4. dbt-tests-adapter — official test suite. Scaffold подключает её. Прохождение всех тестов — критерий Trusted Adapter.

5. 5 steps после scaffold: customize credentials, implement connection, configure types, run tests, setup local dev.

6. Scaffold — отправная точка, не финал. Production adapter — месяцы работы после scaffold.