Deep support ddl

Author: long2ice

.github/workflows/pypi.yml

@@ -1,17 +1,22 @@
 name: pypi
 on:
-  push:
-    paths:
-      - 'mysql2ch/version.py'
-      - '.github/workflows/pypi.yml'
+  release:
+    types:
+      - created
 jobs:
   build:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - uses: actions/setup-python@v1
+      - uses: actions/setup-python@v2
         with:
           python-version: '3.x'
+      - uses: actions/cache@v1
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
       - name: Build dists
         run: |
           python3 setup.py sdist

.github/workflows/pytest.yml

@@ -0,0 +1,23 @@
+name: pytest
+on: [push, pull_request]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+        with:
+          python-version: '3.x'
+      - uses: actions/cache@v1
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements-dev.txt') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements-dev.txt
+      - name: Test with pytest
+        run: |
+          pytest

CHANGELOG.md

@@ -1,7 +1,9 @@
 # ChangeLog
 
-## 0.4.0
+## 0.4.3
+- Deep support ddl.
 
+## 0.4.0
 - Most of the rewrite.
 - Remove read config from env,instead of config.json.
 - Remove ui module.

Makefile

@@ -0,0 +1,55 @@
+checkfiles = mysql2ch/ tests/
+black_opts = -l 100 -t py38
+py_warn = PYTHONDEVMODE=1
+
+help:
+	@echo "mysl2ch development makefile"
+	@echo
+	@echo  "usage: make <target>"
+	@echo  "Targets:"
+	@echo  "    up			Updates dev/test dependencies"
+	@echo  "    deps        Ensure dev/test dependencies are installed"
+	@echo  "    check		Checks that build is sane"
+	@echo  "    lint		Reports all linter violations"
+	@echo  "    test		Runs all tests"
+	@echo  "    style		Auto-formats the code"
+
+deps:
+	@which pip-sync > /dev/null || pip install -q pip-tools
+	@pip install -r requirements-dev.txt
+
+up:
+	CUSTOM_COMPILE_COMMAND="make up" pip-compile -o requirements-dev.txt -U
+	sed -i "s/^-e .*/-e ./" requirements.txt
+
+style: deps
+	isort -rc $(checkfiles)
+	black $(black_opts) $(checkfiles)
+
+check: deps
+ifneq ($(shell which black),)
+	black --check $(black_opts) $(checkfiles) || (echo "Please run 'make style' to auto-fix style issues" && false)
+endif
+	flake8 $(checkfiles)
+	mypy $(checkfiles)
+	pylint -d C,W,R $(checkfiles)
+	bandit -r $(checkfiles)
+	python setup.py check -mrs
+
+lint: deps
+ifneq ($(shell which black),)
+	black --check $(black_opts) $(checkfiles) || (echo "Please run 'make style' to auto-fix style issues" && false)
+endif
+	flake8 $(checkfiles)
+	mypy $(checkfiles)
+	pylint $(checkfiles)
+	bandit -r $(checkfiles)
+	python setup.py check -mrs
+
+test: deps
+	$(py_warn) py.test
+
+publish: deps
+	rm -fR dist/
+	python setup.py sdist
+	twine upload dist/*

mysql2ch/__init__.py

@@ -1,75 +1 @@
-import logging
-import sys
-from typing import List, Dict, Union, Optional
-from pydantic import BaseSettings, HttpUrl, validator
-
-
-def partitioner(key_bytes, all_partitions, available_partitions):
-    """
-    custom partitioner depend on settings
-    :param key_bytes:
-    :param all_partitions:
-    :param available_partitions:
-    :return:
-    """
-    key = key_bytes.decode()
-    partition = Global.settings.schema_table.get(key).get('kafka_partition')
-    return all_partitions[partition]
-
-
-def init_logging(debug):
-    logger = logging.getLogger('mysql2ch')
-    if debug:
-        logger.setLevel(logging.DEBUG)
-    else:
-        logger.setLevel(logging.INFO)
-    sh = logging.StreamHandler(sys.stdout)
-    sh.setLevel(logging.DEBUG)
-    sh.setFormatter(logging.Formatter(
-        fmt='%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s',
-        datefmt='%Y-%m-%d %H:%M:%S'
-    ))
-    logger.addHandler(sh)
-
-
-class Settings(BaseSettings):
-    debug: bool = True
-    environment: str = 'development'
-    mysql_host: str = '127.0.0.1'
-    mysql_port: int = 3306
-    mysql_user: str = 'root'
-    mysql_password: str = '123456'
-    mysql_server_id: int = 1
-    redis_host: str = '120.0.0.1'
-    redis_port: int = 6379
-    redis_password: str = None
-    redis_db: int = 0
-    clickhouse_host: str = '127.0.0.1'
-    clickhouse_port: int = 9000
-    clickhouse_user: str = 'default'
-    clickhouse_password: str = None
-    kafka_server: str = '127.0.0.1:9092'
-    kafka_topic: str = 'test'
-    sentry_dsn: Optional[HttpUrl]
-    schema_table: Dict[str, Dict[str, Union[List[str], int]]]
-    init_binlog_file: str
-    init_binlog_pos: int
-    log_pos_prefix: str = 'mysql2ch'
-    skip_delete_tables: List[str]
-    skip_update_tables: List[str]
-    skip_dmls: List[str]
-    insert_num: int = 20000
-    insert_interval: int = 60
-
-    @validator('schema_table')
-    def check_kafka_partition(cls, v):
-        partitions = list(map(lambda x: v.get(x).get('kafka_partition'), v))
-        if len(partitions) != len(set(partitions)):
-            raise ValueError('kafka_partition must be unique for schema!')
-        return v
-
-
-class Global:
-    settings: Optional['Settings'] = None
-    reader = None
-    writer = None
+__version__ = "0.4.3"

mysql2ch/cli.py

@@ -4,71 +4,79 @@
 import sentry_sdk
 from sentry_sdk.integrations.redis import RedisIntegration
 
-import mysql2ch
-from mysql2ch import Settings
+from mysql2ch.common import Global, Settings, init_logging
 from mysql2ch.consumer import consume
 from mysql2ch.producer import produce
 from mysql2ch.reader import MysqlReader
 from mysql2ch.replication import make_etl
 from mysql2ch.writer import ClickHouseWriter
 
-logger = logging.getLogger('mysql2ch.manage')
+logger = logging.getLogger("mysql2ch.manage")
 
 
 def run(args):
     config = args.config
-    settings = mysql2ch.Global.settings = Settings.parse_file(config)
+    settings = Global.settings = Settings.parse_file(config)
 
     sentry_sdk.init(
-        settings.sentry_dsn,
-        environment=settings.environment,
-        integrations=[RedisIntegration()]
+        settings.sentry_dsn, environment=settings.environment, integrations=[RedisIntegration()]
     )
 
-    mysql2ch.init_logging(settings.debug)
-    mysql2ch.Global.writer = ClickHouseWriter(
+    init_logging(settings.debug)
+    Global.writer = ClickHouseWriter(
         host=settings.clickhouse_host,
         port=settings.clickhouse_port,
         password=settings.clickhouse_password,
-        user=settings.clickhouse_user
+        user=settings.clickhouse_user,
     )
-    mysql2ch.Global.reader = MysqlReader(
+    Global.reader = MysqlReader(
         host=settings.mysql_host,
         port=settings.mysql_port,
         password=settings.mysql_password,
-        user=settings.mysql_user
+        user=settings.mysql_user,
     )
 
     args.func(args)
 
 
 def cli():
-    parser = argparse.ArgumentParser(
-        description='Sync data from MySQL to ClickHouse.',
+    parser = argparse.ArgumentParser(description="Sync data from MySQL to ClickHouse.",)
+    parser.add_argument("-c", "--config", required=True, help="Config json file.")
+    subparsers = parser.add_subparsers(title="subcommands")
+    parser_etl = subparsers.add_parser("etl")
+    parser_etl.add_argument("--schema", required=True, help="Schema to full etl.")
+    parser_etl.add_argument(
+        "--tables",
+        required=False,
+        help="Tables to full etl,multiple tables split with comma,default read from environment.",
+    )
+    parser_etl.add_argument(
+        "--renew",
+        default=False,
+        action="store_true",
+        help="Etl after try to drop the target tables.",
     )
-    parser.add_argument('-c', '--config', required=True, help='Config json file.')
-    subparsers = parser.add_subparsers(title='subcommands')
-    parser_etl = subparsers.add_parser('etl')
-    parser_etl.add_argument('--schema', required=True, help='Schema to full etl.')
-    parser_etl.add_argument('--tables', required=False,
-                            help='Tables to full etl,multiple tables split with comma,default read from environment.')
-    parser_etl.add_argument('--renew', default=False, action='store_true',
-                            help='Etl after try to drop the target tables.')
     parser_etl.set_defaults(run=run, func=make_etl)
 
-    parser_producer = subparsers.add_parser('produce')
+    parser_producer = subparsers.add_parser("produce")
     parser_producer.set_defaults(run=run, func=produce)
 
-    parser_consumer = subparsers.add_parser('consume')
-    parser_consumer.add_argument('--schema', required=True, help='Schema to consume.')
-    parser_consumer.add_argument('--skip-error', action='store_true', default=False, help='Skip error rows.')
-    parser_consumer.add_argument('--auto-offset-reset', required=False, default='earliest',
-                                 help='Kafka auto offset reset,default earliest.')
+    parser_consumer = subparsers.add_parser("consume")
+    parser_consumer.add_argument("--schema", required=True, help="Schema to consume.")
+    parser_consumer.add_argument(
+        "--skip-error", action="store_true", default=False, help="Skip error rows."
+    )
+    parser_consumer.add_argument(
+        "--auto-offset-reset",
+        required=False,
+        default="earliest",
+        help="Kafka auto offset reset,default earliest.",
+    )
     parser_consumer.set_defaults(run=run, func=consume)
 
     parse_args = parser.parse_args()
     parse_args.run(parse_args)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     cli()