Repository: bryzgaloff/airflow-clickhouse-plugin (language: python)

Issue:
Support Airflow 2.1.4
According to this job: https://github.com/bryzgaloff/airflow-clickhouse-plugin/actions/runs/3496964766

Airflow 2.1.4 for some reason does not force installation of `typing-extensions` and an SQL sensor cannot be used.

🟢  **If you are here for a quick solution for Airflow v2.1, simply `pip install typing-extensions`**.

Potential solutions:
* Install `typing-extensions` conditionally. And support in the tests.
* Add a note to README for a user to install manually.


Discussion / hints:

Fix patch:
diff --git a/LICENSE b/LICENSE
index d12c0ab..dcad53f 100644
--- a/LICENSE
+++ b/LICENSE
@@ -1,6 +1,6 @@
 The MIT License
 
-Copyright (c) 2019-2020 Whisk, https://whisk.com
+Copyright (c) 2019-2023 Anton Bryzgalov, tony.bryzgaloff@gmail.com
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal
diff --git a/Makefile b/Makefile
deleted file mode 100644
index 34b3a6a..0000000
--- a/Makefile
+++ /dev/null
@@ -1,34 +0,0 @@
-# define the name of the virtual environment directory
-VENV := venv
-
-# default target, when make executed without arguments
-all: venv
-
-$(VENV)/bin/activate: requirements.txt
-	python3 -m venv $(VENV)
-	./$(VENV)/bin/pip install -r requirements.txt
-	./$(VENV)/bin/pip install pandas
-
-# venv is a shortcut target
-venv: $(VENV)/bin/activate
-
-clean:
-	rm -rf $(VENV)
-	find . | grep -E "(/__pycache__$|\.pyc$|\.pyo$)" | xargs rm -rf
-
-run-clickhouse:
-	docker run -p 9000:9000 --ulimit nofile=262144:262144 -it clickhouse/clickhouse-server
-
-run-clickhouse-dind:
-	docker exec -it $$(docker run --rm -d clickhouse/clickhouse-server) bash
-
-unit: venv
-	./$(VENV)/bin/python3 -m unittest discover -s tests/unit
-
-integration: venv
-	./$(VENV)/bin/python3 -m unittest discover -s tests/integration
-
-tests: venv
-	./$(VENV)/bin/python3 -m unittest discover -s tests
-
-.PHONY: all venv unit clean
diff --git a/README.md b/README.md
index 8ab52c0..f264bab 100644
--- a/README.md
+++ b/README.md
@@ -4,39 +4,72 @@
 ![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/bryzgaloff/airflow-clickhouse-plugin/tests.yml?branch=master)
 ![GitHub contributors](https://img.shields.io/github/contributors/bryzgaloff/airflow-clickhouse-plugin?color=blue)
 
-Provides `ClickHouseOperator`, `ClickHouseHook` and `ClickHouseSqlSensor` for
-    [Apache Airflow][airflow] based on [mymarilyn/clickhouse-driver][ch-driver].
+🔝 The most popular [Apache Airflow][airflow] plugin for ClickHouse, ranked in the top 1% of downloads [on PyPI](https://pypi.org/project/airflow-clickhouse-plugin/).
 
-Top-1% downloads [on PyPI](https://pypi.org/project/airflow-clickhouse-plugin/).
+This plugin provides two families of operators: richer [`clickhouse_driver.Client.execute`-based](#clickhouse-driver-family) and standardized [compatible with Python DB API 2.0](#python-db-api-20-family).
 
-# Features
+Both operators' families are fully supported and covered with tests for different versions of Airflow and Python.
 
-1. SQL queries are templated.
-2. Can run multiple SQL queries per single `ClickHouseOperator`.
-3. Result of the last query of `ClickHouseOperator` instance is pushed to XCom.
-4. Executed queries are logged in a pretty form.
-5. Uses efficient native ClickHouse TCP protocol thanks to 
-    [clickhouse-driver][ch-driver-docs]. **Does not support HTTP protocol.**
-6. Supports extra ClickHouse [connection parameters][ch-driver-connection] such
-    as various timeouts, `compression`, `secure`, etc through Airflow
-    [Connection.extra][airflow-conn-extra] property.
+## `clickhouse-driver` family
 
-# Installation and dependencies
+- `ClickHouseOperator`
+- `ClickHouseHook`
+- `ClickHouseSensor`
+
+These operators are based on [mymarilyn/clickhouse-driver][ch-driver]'s `Client.execute` method and arguments. They offer a full functionality of `clickhouse-driver` and are recommended if you are starting fresh with ClickHouse in Airflow.
+
+### Features
+
+- **SQL Templating**: SQL queries and other parameters are templated.
+- **Multiple SQL Queries**: execute run multiple SQL queries within a single `ClickHouseOperator`. The result of the last query is pushed to XCom (configurable by `do_xcom_push`).
+- **Logging**: executed queries are logged in a visually pleasing format, making it easier to track and debug.
+- **Efficient Native ClickHouse Protocol**: Utilizes efficient _native_ ClickHouse TCP protocol, thanks to [clickhouse-driver][ch-driver-docs]. **Does not support HTTP protocol.**
+- **Custom Connection Parameters**: Supports additional ClickHouse [connection parameters][ch-driver-connection], such as various timeouts, `compression`, `secure`, through the Airflow [Connection.extra][airflow-conn-extra] property.
+
+See reference and examples [below](#usage).
+
+### Installation and dependencies
 
 `pip install -U airflow-clickhouse-plugin`
 
-Dependencies: `apache-airflow` with `apache-airflow-providers-common-sql`
-    (usually pre-packed with Airflow) and `clickhouse-driver`.
+Dependencies: only `apache-airflow` and `clickhouse-driver`.
+
+## Python DB API 2.0 family
+
+- Operators:
+  - `ClickHouseSQLExecuteQueryOperator`
+  - `ClickHouseSQLColumnCheckOperator`
+  - `ClickHouseSQLTableCheckOperator`
+  - `ClickHouseSQLCheckOperator`
+  - `ClickHouseSQLValueCheckOperator`
+  - `ClickHouseSQLIntervalCheckOperator`
+  - `ClickHouseSQLThresholdCheckOperator`
+  - `ClickHouseBranchSQLOperator`
+- `ClickHouseDbApiHook`
+- `ClickHouseSqlSensor`
+
+These operators combine [`clickhouse_driver.dbapi`][ch-driver-db-api] with [apache-airflow-providers-common-sql]. While they have limited functionality compared to `Client.execute` (not all arguments are supported), they provide a standardized interface. This is useful when porting Airflow pipelines to ClickHouse from another SQL provider backed by `common.sql` Airflow package, such as MySQL, Postgres, BigQuery, and others.
+
+The feature set of this version is fully based on `common.sql` Airflow provider: refer to its [reference][common-sql-reference] and [examples][common-sql-examples] for details.
+
+An example is also available [below](#db-api-20-clickhousesqlsensor-and-clickhousesqlexecutequeryoperator-example).
+
+### Installation and dependencies
+
+Add `common.sql` extra when installing the plugin: `pip install -U airflow-clickhouse-plugin[common.sql]` — to enable DB API 2.0 operators.
+
+Dependencies: `apache-airflow-providers-common-sql` (usually pre-packed with Airflow) in addition to `apache-airflow` and `clickhouse-driver`.
 
 ## Python and Airflow versions support
 
 Different versions of the plugin support different combinations of Python and
-    Airflow versions. We _primarily_ support **Airflow 2.0+ and Python 3.7+**.
+    Airflow versions. We _primarily_ support **Airflow 2.0+ and Python 3.8+**.
     If you need to use the plugin with older Python-Airflow combinations, pick a
     suitable plugin version:
 
 | airflow-clickhouse-plugin version | Airflow version         | Python version     |
 |-----------------------------------|-------------------------|--------------------|
+| 1.0.0                             | \>=2.0.0,<2.7.0         | ~=3.8              |
 | 0.11.0                            | ~=2.0.0,\>=2.2.0,<2.7.0 | ~=3.7              |
 | 0.10.0,0.10.1                     | ~=2.0.0,\>=2.2.0,<2.6.0 | ~=3.7              |
 | 0.9.0,0.9.1                       | ~=2.0.0,\>=2.2.0,<2.5.0 | ~=3.7              |
@@ -50,177 +83,130 @@ Different versions of the plugin support different combinations of Python and
 `~=` means compatible release, see [PEP 440][pep-440-compatible-releases] for an
     explanation.
 
-## Note on pandas dependency
-
-Starting from Airflow 2.2 `pandas` is now an [extra requirement][pandas-extra].
-    To install `airflow-clickhouse-plugin` with `pandas` support, use
-    `pip install airflow-clickhouse-plugin[pandas]`.
-
-**Important**: this works only with `pip` 21+. So to handle `pandas` dependency
-    properly  you may need to first upgrade `pip` using `pip install -U pip`.
-
-If you are not able to upgrade `pip` to 21+, install dependency directly using
-    `pip install apache-airflow[pandas]==` (specifying current Airflow version).
-    Simple one-liner:
-    `pip install "apache-airflow[pandas]==$(pip freeze | grep apache-airflow== | cut -d'=' -f3)"`.
+Previous versions of the plugin might require `pandas` extra: `pip install airflow-clickhouse-plugin[pandas]==0.11.0`. Check out earlier versions of `README.md` for details.
 
 # Usage
 
 To see examples [scroll down](#examples). To run them, [create an Airflow connection to ClickHouse](#how-to-create-an-airflow-connection-to-clickhouse).
 
-## ClickHouseOperator Reference
+## ClickHouseOperator reference
+
+To import `ClickHouseOperator` use `from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator`.
 
-To import `ClickHouseOperator` use:
-    `from airflow_clickhouse_plugin.operators.clickhouse_operator import ClickHouseOperator`
+Supported arguments:
+* `sql` (templated, required): query (if argument is a single `str`) or multiple queries (iterable of `str`). Supports files with `.sql` extension.
+* `clickhouse_conn_id`: Airflow connection id. Connection schema is described [below](#clickhouse-connection-schema). Default connection id is `clickhouse_default`.
+* Arguments of [`clickhouse_driver.Client.execute` method][ch-driver-execute-summary]:
+  * `parameters` (templated): passed `params` of the `execute` method. (Renamed to avoid name conflict with Airflow tasks' `params` argument.)
+    * `dict` for `SELECT` queries.
+    * `list`/`tuple`/generator for `INSERT` queries.
+    * If multiple queries are provided via `sql` then the `parameters` are passed to _all_ of them.
+  * `with_column_types` (not templated).
+  * `external_tables` (templated).
+  * `query_id` (templated).
+  * `settings` (templated).
+  * `types_check` (not templated).
+  * `columnar` (not templated).
+  * For the documentation of these arguments, refer to [`clickhouse_driver.Client.execute` API reference][ch-driver-execute-reference].
+* `database` (templated): if present, overrides `schema` of Airflow connection.
+* Other arguments (including a required `task_id`) are inherited from Airflow [BaseOperator][airflow-base-op].
 
-Supported kwargs:
-* `sql`: templated query (if argument is a single `str`) or queries (if iterable
-    of `str`'s).
-* `clickhouse_conn_id`: connection id. Connection schema is described
-    [below](#clickhouse-connection-schema).
-* `parameters`: passed to clickhouse-driver [execute method][ch-driver-execute].
-  * If multiple queries are provided via `sql` then the parameters are passed to
-      _all_ of them.
-  * Parameters are _not_ templated.
-* `database`: if present, overrides database defined by connection.
-* Other kwargs (including the required `task_id`) are inherited from Airflow 
-    [BaseOperator][airflow-base-op].
+Result of the _last_ query is pushed to XCom (disable using `do_xcom_push=False` argument).
 
-The result of the _last_ query is pushed to XCom.
+In other words, the operator simply wraps [`ClickHouseHook.execute` method](#clickhousehook-reference).
 
 See [example](#clickhouseoperator-example) below.
 
-## ClickHouseHook Reference
+## ClickHouseHook reference
 
-To import `ClickHouseHook` use:
-    `from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook`
+To import `ClickHouseHook` use `from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook`.
 
 Supported kwargs of constructor (`__init__` method):
-* `clickhouse_conn_id`: connection id. Connection schema is described
-    [below](#clickhouse-connection-schema).
-* `database`: if present, overrides database defined by connection.
-
-Supports all the methods of the Airflow [BaseHook][airflow-base-hook] including:
-* `get_records(sql: str, parameters: dict=None)`: returns result of the query
-    as a list of tuples. Materializes all the records in memory.
-* `get_first(sql: str, parameters: dict=None)`: returns the first row of the
-    result. Does not load the whole dataset into memory because of using
-    [execute_iter][ch-driver-execute-iter]. If the dataset is empty then returns
-    `None` following [fetchone][python-db-api-2-fetchone] semantics.
-* `run(sql, parameters)`: runs a single query (specified argument of type `str`)
-    or multiple queries (if iterable of `str`). `parameters` can have any form
-    supported by [execute][ch-driver-execute] method of clickhouse-driver.
-  * If single query is run then returns its result. If multiple queries are run
-      then returns the result of the last of them.
-  * If multiple queries are given then `parameters` are passed to _all_ of them.
-  * Materializes all the records in memory (uses simple `execute` but not 
-      `execute_iter`).
-    * To achieve results streaming by `execute_iter` use it directly via
-        `hook.get_conn().execute_iter(…)`
-        (see [execute_iter reference][ch-driver-execute-iter]).
-  * Every `run` call uses a new connection which is closed when finished.
-* `get_conn()`: returns the underlying
-    [clickhouse_driver.Client][ch-driver-client] instance.
+* `clickhouse_conn_id`: Airflow connection id. Connection schema is described [below](#clickhouse-connection-schema). Default connection id is `clickhouse_default`.
+* `database`: if present, overrides `schema` of Airflow connection.
+
+Defines `ClickHouseHook.execute` method which simply wraps [`clickhouse_driver.Client.execute`][ch-driver-execute-reference]. It has all the same arguments, except of:
+* `sql` (instead of `execute`'s `query`): query (if argument is a single `str`) or multiple queries (iterable of `str`).
+
+`ClickHouseHook.execute` returns a result of the _last_ query.
+
+Also, the hook defines `get_conn()` method which returns an underlying [clickhouse_driver.Client][ch-driver-client] instance.
 
 See [example](#clickhousehook-example) below.
 
-## ClickHouseSqlSensor Reference
+## ClickHouseSensor reference
 
-Sensor fully inherits from [Airflow SQLSensor][airflow-sql-sensor] and therefore
-    fully implements its interface using `ClickHouseHook` to fetch the SQL
-    execution result and supports templating of `sql` argument.
+To import `ClickHouseSensor` use `from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor`.
 
-See [example](#clickhousesqlsensor-example) below.
+This class wraps [`ClickHouseHook.execute` method](#clickhousehook-reference) into an [Airflow sensor][airflow-sensor]. Supports all the arguments of [`ClickHouseOperator`](#clickhouseoperator-reference) and additionally:
+* `is_success`: a callable which accepts a single argument — a return value of `ClickHouseHook.execute`. If a return value of `is_success` is truthy, the sensor succeeds. By default, the callable is `bool`: i.e. if the return value of `ClickHouseHook.execute` is truthy, the sensor succeeds. Usually, `execute` is a list of records returned by query: thus, by default it is falsy if no records are returned.
+* `is_failure`: a callable which accepts a single argument — a return value of `ClickHouseHook.execute`. If a return value of `is_failure` is truthy, the sensor raises `AirflowException`. By default, `is_failure` is `None` and no failure check is performed.
 
-## How to create an Airflow connection to ClickHouse
+See [example](#clickhousesensor-example) below.
 
-As a `type` of a new connection, choose **SQLite**. `host` should be set to
-    ClickHouse host's IP or domain name.
+## How to create an Airflow connection to ClickHouse
 
-There is **no** special ClickHouse connection type yet, so we use SQLite as
-    the closest one.
+As a `type` of a new connection, choose **SQLite** or any other SQL database. There is **no** special ClickHouse connection type yet, so we use any SQL as the closest one.
 
-The rest of the connection details may be skipped as they
-    [have defaults](#default-values) defined by `clickhouse-driver`. If
-    you use non-default values, set them according to the
-    [connection schema](#clickhouse-connection-schema).
+All the connection attributes are optional: default host is `localhost` and other credentials [have defaults](#default-values) defined by `clickhouse-driver`. If you use non-default values, set them according to the [connection schema](#clickhouse-connection-schema).
 
-If you use a secure connection to ClickHouse (this requires additional
-    configurations on ClickHouse side), set `extra` to `{"secure":true}`.
+If you use a secure connection to ClickHouse (this requires additional configurations on ClickHouse side), set `extra` to `{"secure":true}`. All `extra` connection parameters are passed to [`clickhouse_driver.Client`][ch-driver-client] as-is.
 
-### ClickHouse Connection schema
+### ClickHouse connection schema
 
-[clickhouse_driver.Client][ch-driver-client] is initialized with attributes stored
-    in Airflow [Connection attributes][airflow-connection-attrs]. The mapping of
-    the attributes is listed below:
+[`clickhouse_driver.Client`][ch-driver-client] is initialized with attributes stored in Airflow [Connection attributes][airflow-connection-howto]:
   
 | Airflow Connection attribute | `Client.__init__` argument |
-| --- | --- |
-| `host` | `host` |
-| `port` | `port` |
-| `schema` | `database` |
-| `login` | `user` |
-| `password` | `password` |
-| `extra` | `**kwargs` |
+|------------------------------|----------------------------|
+| `host`                       | `host`                     |
+| `port` (`int`)               | `port`                     |
+| `schema`                     | `database`                 |
+| `login`                      | `user`                     |
+| `password`                   | `password`                 |
+| `extra`                      | `**kwargs`                 |
 
-`database` argument of `ClickHouseOperator` or `ClickHouseHook` overrides
-    `schema` attribute of the Airflow connection.
+`database` argument of `ClickHouseOperator`, `ClickHouseHook`, `ClickHouseSensor`, and others overrides `schema` attribute of the Airflow connection.
 
 ### Extra arguments
 
-You may also pass [additional arguments][ch-driver-connection], such as
-    timeouts, `compression`, `secure`, etc through
-    [Connection.extra][airflow-conn-extra] attribute. The attribute should
-    contain a JSON object which will be [deserialized][airflow-conn-dejson] and
-    all of its properties will be passed as-is to the `Client`.
+You may set non-standard arguments of [`clickhouse_driver.Client`][ch-driver-client], such as timeouts, `compression`, `secure`, etc. using Airflow's [`Connection.extra`][airflow-conn-extra] attribute. The attribute should contain a JSON object which will be [deserialized][airflow-conn-dejson] and all of its properties will be passed as-is to the `Client`.
 
-For example, if Airflow connection contains `extra={"secure":true}` then
-    the `Client.__init__` will receive `secure=True` keyword argument in
-    addition to other non-empty connection attributes.
+For example, if Airflow connection contains `extra='{"secure": true}'` then the `Client.__init__` will receive `secure=True` keyword argument in addition to other connection attributes.
 
 #### Compression
 
-You should install several packages to support compression. For example, for lz4:
+You should install specific packages to support compression. For example, for lz4:
 
 ```bash
 pip3 install clickhouse-cityhash lz4
 ```
 
-Then you should include `compression` parameter in airflow connection uri: `extra={"compression":"lz4"}`.  You can get 
-additional information about extra options from [official documentation of clickhouse-driver](https://clickhouse-driver.readthedocs.io/en/latest/installation.html#installation-pypi)
+Then you should include `compression` parameter in airflow connection uri: `extra='{"compression":"lz4"}'`. You can get additional information about extra options from [official documentation of clickhouse-driver][ch-driver-pypi-install].
 
-Connection URI should look like in the example below:
+Connection URI with compression will look like `clickhouse://login:password@host:port/?compression=lz4`.
 
-`clickhouse://login:password@host:port/?compression=lz4`
+See [official documentation][airflow-connection-howto] to learn more about connections management in Airflow.
 
-See [official documentation](https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html) to 
-get more info about connections in Airflow.
+### Default Values
 
-### Default values
+If some Airflow connection attribute is not set, it is not passed to `clickhouse_driver.Client`. In such cases, the plugin uses a default value from the corresponding [`clickhouse_driver.Connection`][ch-driver-connection] argument. For instance, `user` defaults to `'default'`.
 
-If the Airflow connection attribute is not set then it is not passed to the
-    `Client` at all. In that case the default value of the corresponding
-    [clickhouse_driver.Connection][ch-driver-connection] argument is used (e.g.
-    `user` defaults to `'default'`).
+This means that the plugin itself does not define any default values for the ClickHouse connection. You may fully rely on default values of the [clickhouse-driver][ch-driver] version you use.
 
-This means that Airflow ClickHouse Plugin does not itself define any default
-    values for the ClickHouse connection. You may fully rely on default values
-    of the [clickhouse-driver][ch-driver] version you use. The only exception is
-    `host`: if the attribute of Airflow connection is not set then `'localhost'`
-    is used.
+The only exception is `host`: if the attribute of Airflow connection is not set then `'localhost'` is used.
 
 ### Default connection
 
-By default, the plugin uses `connection_id='clickhouse_default'`.
+By default, the plugin uses Airflow connection with id `'clickhouse_default'`.
 
 ## Examples
 
-### ClickHouseOperator Example
+### ClickHouseOperator example
 
 ```python
 from airflow import DAG
-from airflow_clickhouse_plugin.operators.clickhouse_operator import ClickHouseOperator
-from airflow.operators.python_operator import PythonOperator
+from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
+from airflow.operators.python import PythonOperator
 from airflow.utils.dates import days_ago
 
 with DAG(
@@ -246,55 +232,52 @@ with DAG(
             ''',
             # result of the last query is pushed to XCom
         ),
+        # query_id is templated and allows to quickly identify query in ClickHouse logs
+        query_id='{{ ti.dag_id }}-{{ ti.task_id }}-{{ ti.run_id }}-{{ ti.try_number }}',
         clickhouse_conn_id='clickhouse_test',
     ) >> PythonOperator(
         task_id='print_month_income',
-        provide_context=True,
-        python_callable=lambda task_instance, **_:
+        python_callable=lambda task_instance:
             # pulling XCom value and printing it
             print(task_instance.xcom_pull(task_ids='update_income_aggregate')),
     )
 ```
 
-### ClickHouseHook Example
+### ClickHouseHook example
 
 ```python
 from airflow import DAG
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-from airflow.hooks.mysql_hook import MySqlHook
-from airflow.operators.python_operator import PythonOperator
+from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
+from airflow.providers.sqlite.hooks.sqlite import SqliteHook
+from airflow.operators.python import PythonOperator
 from airflow.utils.dates import days_ago
 
 
-def mysql_to_clickhouse():
-    mysql_hook = MySqlHook()
+def sqlite_to_clickhouse():
+    sqlite_hook = SqliteHook()
     ch_hook = ClickHouseHook()
-    records = mysql_hook.get_records('SELECT * FROM some_mysql_table')
-    ch_hook.run('INSERT INTO some_ch_table VALUES', records)
+    records = sqlite_hook.get_records('SELECT * FROM some_sqlite_table')
+    ch_hook.execute('INSERT INTO some_ch_table VALUES', records)
 
 
 with DAG(
-        dag_id='mysql_to_clickhouse',
+        dag_id='sqlite_to_clickhouse',
         start_date=days_ago(2),
 ) as dag:
     dag >> PythonOperator(
-        task_id='mysql_to_clickhouse',
-        python_callable=mysql_to_clickhouse,
+        task_id='sqlite_to_clickhouse',
+        python_callable=sqlite_to_clickhouse,
     )
 ```
 
-Important note: don't try to insert values using 
-    `ch_hook.run('INSERT INTO some_ch_table VALUES (1)')` literal form.
-    clickhouse-driver [requires][ch-driver-insert] values for `INSERT` query to
-    be provided via `parameters` due to specifics of the native ClickHouse
-    protocol.
+Important note: don't try to insert values using `ch_hook.execute('INSERT INTO some_ch_table VALUES (1)')` literal form. [`clickhouse-driver` requires][ch-driver-insert] values for `INSERT` query to be provided via `parameters` due to specifics of the native ClickHouse protocol.
 
-### ClickHouseSqlSensor Example
+### ClickHouseSensor example
 
 ```python
 from airflow import DAG
-from airflow_clickhouse_plugin.sensors.clickhouse_sql_sensor import ClickHouseSqlSensor
-from airflow_clickhouse_plugin.operators.clickhouse_operator import ClickHouseOperator
+from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor
+from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
 from airflow.utils.dates import days_ago
 
 
@@ -302,11 +285,11 @@ with DAG(
         dag_id='listen_warnings',
         start_date=days_ago(2),
 ) as dag:
-    dag >> ClickHouseSqlSensor(
+    dag >> ClickHouseSensor(
         task_id='poke_events_count',
         database='monitor',
         sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
-        success=lambda cnt: cnt > 10000,
+        is_success=lambda cnt: cnt > 10000,
     ) >> ClickHouseOperator(
         task_id='create_alert',
         database='alerts',
@@ -317,118 +300,68 @@ with DAG(
     )
 ```
 
-# How to run tests
-
-## Unit tests
-
-From the root project directory
-
-Using `make`:
-
-```bash
-make unit
-```
-
-Using `python`:
-
-```bash
-python -m unittest discover -s tests
-```
-
-## Integration tests
-
-Integration tests require access to ClickHouse server. Tests use connection
-    URI defined [via environment variable][airflow-conn-env]
-    `AIRFLOW_CONN_CLICKHOUSE_DEFAULT` with `clickhouse://localhost` as default.
-
-You can run ClickHouse server in a local Docker container using the following command:
-
-Using `make`:
-
-```bash
-make run-clickhouse
-```
-
-Using `shell`:
-
-```bash
-docker run -p 9000:9000 --ulimit nofile=262144:262144 -it clickhouse/clickhouse-server
-```
-
-And then run from the project root:
+### DB API 2.0: ClickHouseSqlSensor and ClickHouseSQLExecuteQueryOperator example
 
-Using `make`:
-
-```bash
-make integration
-```
-
-Using `python`:
-
-```bash
-python3 -m unittest discover -s tests/integration
-```
-
-## All tests
-
-From the root project directory:
-
-Using `make`:
-
-```bash
-make tests
-```
+```python
+from airflow import DAG
+from airflow_clickhouse_plugin.sensors.clickhouse_dbapi import ClickHouseSqlSensor
+from airflow_clickhouse_plugin.operators.clickhouse_dbapi import ClickHouseSQLExecuteQueryOperator
+from airflow.utils.dates import days_ago
 
-Using `python`:
 
-```bash
-python3 -m unittest discover -s tests
+with DAG(
+        dag_id='listen_warnings',
+        start_date=days_ago(2),
+) as dag:
+    dag >> ClickHouseSqlSensor(
+        task_id='poke_events_count',
+        hook_params=dict(schema='monitor'),
+        sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
+        success=lambda cnt: cnt > 10000,
+        conn_id=None,  # required by common.sql SqlSensor; use None for default
+    ) >> ClickHouseSQLExecuteQueryOperator(
+        task_id='create_alert',
+        database='alerts',
+        sql='''
+            INSERT INTO events SELECT eventDate, count()
+            FROM monitor.warnings WHERE eventDate = '{{ ds }}'
+        ''',
+    )
 ```
 
-### Github Actions
+# How to run tests
 
-[GitHub Action][github-action-src] is set up for this project.
+Unit tests: `python3 -m unittest discover -t tests -s unit`
 
-### Run tests using Docker
+Integration tests require access to a ClickHouse server. Here is how to set up a local test environment using Docker:
+* Run ClickHouse server in a local Docker container: `docker run -p 9000:9000 --ulimit nofile=262144:262144 -it clickhouse/clickhouse-server`
+* Run tests with Airflow connection details set [via environment variable][airflow-conn-env]: `PYTHONPATH=src AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://localhost python3 -m unittest discover -t tests -s integration`
+* Stop the container after running the tests to deallocate its resources.
 
-Run ClickHouse server inside Docker:
+Run all (unit&integration) tests with ClickHouse connection defined: `PYTHONPATH=src AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://localhost python3 -m unittest discover -s tests`
 
-Using `shell`:
+## GitHub Actions
 
-```bash
-docker exec -it $(docker run --rm -d clickhouse/clickhouse-server) bash
-```
+[GitHub Action][github-action-src] is configured for this project.
 
-Using `make`:
+## Run all tests inside Docker
 
-```bash
-make run-clickhouse-dind
-```
+Start a ClickHouse server inside Docker: `docker exec -it $(docker run --rm -d clickhouse/clickhouse-server) bash`
 
 The above command will open `bash` inside the container.
 
 Install dependencies into container and run tests (execute inside container):
 
-Using `python`:
-
 ```bash
 apt-get update
-apt-get install -y python3.10 python3-pip git make
+apt-get install -y python3 python3-pip git make
 git clone https://github.com/whisklabs/airflow-clickhouse-plugin.git
 cd airflow-clickhouse-plugin
-python3.10 -m pip install -r requirements.txt
-python3.10 -m unittest discover -s tests
+python3 -m pip install -r requirements.txt
+PYTHONPATH=src AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://localhost python3 -m unittest discover -s tests
 ```
 
-Using `make`:
-
-```bash
-apt-get update
-apt-get install -y python3.10 python3-pip git make
-git clone https://github.com/whisklabs/airflow-clickhouse-plugin.git
-cd airflow-clickhouse-plugin
-make tests
-```
+Stop the container.
 
 # Contributors
 
@@ -448,25 +381,28 @@ Community contributors:
 * Maxim Tarasov, [@MaximTar](https://github.com/MaximTar)
 * [@dvnrvn](https://github.com/dvnrvn)
 * Giovanni Corsetti, [@CorsettiS](https://github.com/CorsettiS)
+* Dmytro Zhyzniev, [@1ng4lipt](https://github.com/1ng4lipt)
 
 
 [airflow]: https://airflow.apache.org/
 [ch-driver]: https://github.com/mymarilyn/clickhouse-driver
 [ch-driver-docs]: https://clickhouse-driver.readthedocs.io/en/latest/
-[ch-driver-execute]: https://clickhouse-driver.readthedocs.io/en/latest/quickstart.html#selecting-data
-[airflow-base-op]: https://airflow.apache.org/docs/2.1.0/_api/airflow/models/baseoperator/index.html
-[airflow-base-hook]: https://airflow.apache.org/docs/apache-airflow/2.1.0/_api/airflow/hooks/base/index.html#airflow.hooks.base.BaseHook
-[ch-driver-execute-iter]: https://clickhouse-driver.readthedocs.io/en/latest/quickstart.html#streaming-results
+[ch-driver-execute-summary]: https://clickhouse-driver.readthedocs.io/en/latest/quickstart.html#selecting-data
+[ch-driver-execute-reference]: https://clickhouse-driver.readthedocs.io/en/latest/api.html#clickhouse_driver.Client.execute
+[airflow-base-op]: https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/baseoperator/index.html
 [ch-driver-insert]: https://clickhouse-driver.readthedocs.io/en/latest/quickstart.html#inserting-data
 [ch-driver-client]: https://clickhouse-driver.readthedocs.io/en/latest/api.html#client
-[airflow-conn-extra]: https://airflow.apache.org/docs/2.1.0/_api/airflow/models/connection/index.html#airflow.models.connection.Connection.extra
 [ch-driver-connection]: https://clickhouse-driver.readthedocs.io/en/latest/api.html#connection
-[airflow-connection-attrs]: https://airflow.apache.org/docs/apache-airflow/2.1.0/_api/airflow/models/index.html?highlight=connection#airflow.models.Connection
+[airflow-conn-extra]: https://airflow.apache.org/docs/2.1.0/_api/airflow/models/connection/index.html#airflow.models.connection.Connection.extra
+[airflow-connection-howto]: https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html
 [airflow-conn-dejson]: https://airflow.apache.org/docs/apache-airflow/2.1.0/_api/airflow/models/index.html?highlight=connection#airflow.models.Connection.extra_dejson
 [airflow-conn-env]: https://airflow.apache.org/docs/apache-airflow/2.1.0/howto/connection.html#storing-a-connection-in-environment-variables
-[python-db-api-2-fetchone]: https://www.python.org/dev/peps/pep-0249/#fetchone
-[cloud-composer-versions]: https://cloud.google.com/composer/docs/concepts/versioning/composer-versions#supported_versions
-[airflow-sql-sensor]: https://airflow.apache.org/docs/2.1.0/_api/airflow/sensors/sql/index.html
 [github-action-src]: https://github.com/whisklabs/airflow-clickhouse-plugin/tree/master/.github/workflows
-[pandas-extra]: https://github.com/apache/airflow/commit/2c26b15a8087cb8a81eb19fedbc768bd6da92df7#diff-60f61ab7a8d1910d86d9fda2261620314edcae5894d5aaa236b821c7256badd7
 [pep-440-compatible-releases]: https://peps.python.org/pep-0440/#compatible-release
+[apache-airflow-providers-common-sql]: https://airflow.apache.org/docs/apache-airflow-providers-common-sql/stable/index.html
+[db-api-pep]: https://peps.python.org/pep-0249/
+[airflow-sensor]: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html
+[ch-driver-pypi-install]: https://clickhouse-driver.readthedocs.io/en/latest/installation.html#installation-pypi
+[common-sql-reference]: https://airflow.apache.org/docs/apache-airflow-providers-common-sql/stable/_api/airflow/providers/common/sql/index.html
+[common-sql-examples]: https://airflow.apache.org/docs/apache-airflow-providers-common-sql/stable/operators.html
+[ch-driver-db-api]: https://clickhouse-driver.readthedocs.io/en/latest/dbapi.html
diff --git a/airflow_clickhouse_plugin/__version__.py b/airflow_clickhouse_plugin/__version__.py
deleted file mode 100644
index c8efbe9..0000000
--- a/airflow_clickhouse_plugin/__version__.py
+++ /dev/null
@@ -1,16 +0,0 @@
-"""
-__version__.py
-~~~~~~~~~~~~~~
-
-Information about the current version of the airflow_clickhouse_plugin package.
-"""
-
-__title__ = 'airflow-clickhouse-plugin'
-__description__ = \
-    'airflow-clickhouse-plugin - ' \
-    'Airflow plugin to execute ClickHouse commands and queries'
-__version__ = '0.11.0'
-__url__ = 'https://github.com/bryzgaloff/airflow-clickhouse-plugin'
-__license__ = 'MIT License'
-__author__ = 'Anton Bryzgalov'
-__author_email__ = 'tony.bryzgaloff@gmail.com'
diff --git a/airflow_clickhouse_plugin/hooks/clickhouse_hook.py b/airflow_clickhouse_plugin/hooks/clickhouse_hook.py
deleted file mode 100644
index 83cd23b..0000000
--- a/airflow_clickhouse_plugin/hooks/clickhouse_hook.py
+++ /dev/null
@@ -1,130 +0,0 @@
-import contextlib
-from itertools import islice
-from typing import *
-
-from airflow.hooks.base import BaseHook
-from clickhouse_driver import Client
-
-
-class ClickHouseHook(BaseHook):
-    conn_name_attr = 'clickhouse_conn_id'
-    default_conn_name = 'clickhouse_default'
-
-    def __init__(
-            self,
-            clickhouse_conn_id: str = default_conn_name,
-            database: Optional[str] = None,
-    ):
-        self.clickhouse_conn_id = clickhouse_conn_id
-        self.database = database
-
-    def get_conn(self) -> Client:
-        conn = self.get_connection(self.clickhouse_conn_id)
-        connection_kwargs = conn.extra_dejson.copy()
-        if conn.port:
-            connection_kwargs.update(port=int(conn.port))
-        if conn.login:
-            connection_kwargs.update(user=conn.login)
-        if conn.password:
-            connection_kwargs.update(password=conn.password)
-        if self.database:
-            connection_kwargs.update(database=self.database)
-        elif conn.schema:
-            connection_kwargs.update(database=conn.schema)
-        return Client(conn.host or 'localhost', **connection_kwargs)
-
-    def get_records(self, sql: str, parameters: Optional[dict] = None) -> List[Tuple]:
-        self._log_query(sql, parameters)
-        with _disconnecting(self.get_conn()) as client:
-            return client.execute(sql, params=parameters)
-
-    def get_first(self, sql: str, parameters: Optional[dict] = None) -> Optional[Tuple]:
-        self._log_query(sql, parameters)
-        with _disconnecting(self.get_conn()) as client:
-            try:
-                return next(client.execute_iter(sql, params=parameters))
-            except StopIteration:
-                return None
-
-    def get_pandas_df(self, sql: str):
-        import pandas as pd
-        rows, columns_defs = self.run(sql, with_column_types=True)
-        columns = [column_name for column_name, _ in columns_defs]
-        return pd.DataFrame(rows, columns=columns)
-
-    def run(
-            self,
-            sql: Union[str, Iterable[str]],
-            parameters: Union[None, dict, list, tuple, Generator] = None,
-            with_column_types: bool = False,
-            types_check: bool = False,
-    ) -> Any:
-        if isinstance(sql, str):
-            sql = (sql,)
-        with _disconnecting(self.get_conn()) as conn:
-            last_result = None
-            for s in sql:
-                self._log_query(s, parameters)
-                last_result = conn.execute(
-                    s,
-                    params=parameters,
-                    with_column_types=with_column_types,
-                    types_check=types_check,
-                )
-
-        return last_result
-
-    def _log_query(
-            self,
-            sql: str,
-            parameters: Union[None, dict, list, tuple, Generator],
-    ) -> None:
-        self.log.info(
-            '%s%s', sql,
-            f' with {self._log_params(parameters)}' if parameters else '',
-        )
-
-    @staticmethod
-    def _log_params(
-            parameters: Union[dict, list, tuple, Generator],
-            limit: int = 10,
-    ) -> str:
-        if isinstance(parameters, Generator) or len(parameters) <= limit:
-            return str(parameters)
-        if isinstance(parameters, dict):
-            head = dict(islice(parameters.items(), limit))
-        else:
-            head = parameters[:limit]
-        head_str = str(head)
-        closing_paren = head_str[-1]
-        return f'{head_str[:-1]} … and {len(parameters) - limit} ' \
-            f'more parameters{closing_paren}'
-
-
-_InnerT = TypeVar('_InnerT')
-
-
-@contextlib.contextmanager
-def _disconnecting(thing: _InnerT) -> ContextManager[_InnerT]:
-    """
-    Context to automatically disconnect something at the end of a block.
-
-    Similar to ``contextlib.closing`` but calls .disconnect() method on exit.
-
-    Code like this:
-
-    >>> with _disconnecting(<module>.open(<arguments>)) as f:
-    >>>     <block>
-
-    is equivalent to this:
-
-    >>> f = <module>.open(<arguments>)
-    >>> try:
-    >>>     <block>
-    >>> finally:
-    >>>     f.disconnect()
-    """
-    try:
-        yield thing
-    finally:
-        thing.disconnect()
diff --git a/airflow_clickhouse_plugin/operators/clickhouse_operator.py b/airflow_clickhouse_plugin/operators/clickhouse_operator.py
deleted file mode 100644
index a3c66c3..0000000
--- a/airflow_clickhouse_plugin/operators/clickhouse_operator.py
+++ /dev/null
@@ -1,35 +0,0 @@
-from typing import Any, Dict, Iterable, Optional, Sequence, Union
-
-from airflow.models import BaseOperator
-from airflow.utils.decorators import apply_defaults
-
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-
-
-class ClickHouseOperator(BaseOperator):
-    template_fields = ('_sql',)
-    template_ext: Sequence[str] = ('.sql',)
-    template_fields_renderers = {'_sql': 'sql'}
-    default_conn_name = ClickHouseHook.default_conn_name
-
-    def __init__(
-            self,
-            sql: Union[str, Iterable[str]],
-            clickhouse_conn_id: str = default_conn_name,
-            parameters: Optional[Dict[str, Any]] = None,
-            database: Optional[str] = None,
-            *args,
-            **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-        self._sql = sql
-        self._conn_id = clickhouse_conn_id
-        self._parameters = parameters
-        self._database = database
-
-    def execute(self, context: Dict[str, Any]) -> Any:
-        hook = ClickHouseHook(
-            clickhouse_conn_id=self._conn_id,
-            database=self._database,
-        )
-        return hook.run(self._sql, self._parameters)
diff --git a/airflow_clickhouse_plugin/sensors/clickhouse_sql_sensor.py b/airflow_clickhouse_plugin/sensors/clickhouse_sql_sensor.py
deleted file mode 100644
index a530b9e..0000000
--- a/airflow_clickhouse_plugin/sensors/clickhouse_sql_sensor.py
+++ /dev/null
@@ -1,41 +0,0 @@
-from typing import Any, Callable, Dict, Optional, Sequence
-
-from airflow.providers.common.sql.sensors.sql import SqlSensor
-from airflow.utils.decorators import apply_defaults
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-
-
-class ClickHouseSqlSensor(SqlSensor):
-    template_fields: Sequence[str] = ('sql',)
-    template_ext: Sequence[str] = ('.sql',)
-    default_conn_name = ClickHouseHook.default_conn_name
-
-    def __init__(
-        self,
-        sql: str = None,
-        clickhouse_conn_id: str = default_conn_name,
-        parameters: Optional[Dict[str, Any]] = None,
-        database: Optional[str] = None,
-        success: Optional[Callable[[Any], bool]] = None,
-        failure: Optional[Callable[[Any], bool]] = None,
-        fail_on_empty: bool = False,
-        *args,
-        **kwargs,
-    ):
-        super().__init__(
-            conn_id=clickhouse_conn_id,
-            sql=sql,
-            parameters=parameters,
-            success=success,
-            failure=failure,
-            fail_on_empty=fail_on_empty,
-            *args,
-            **kwargs,
-        )
-        self._database = database
-
-    def _get_hook(self) -> ClickHouseHook:
-        return ClickHouseHook(
-            clickhouse_conn_id=self.conn_id,
-            database=self._database,
-        )
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..663bbb2
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,52 @@
+[build-system]
+requires = [
+    "setuptools",
+]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "airflow-clickhouse-plugin"
+version = "1.0.0"
+description = "airflow-clickhouse-plugin — Airflow plugin to execute ClickHouse commands and queries"
+readme = "README.md"
+requires-python = ">=3.8"
+license = { text = "MIT License" }
+authors = [
+    { name = "Anton Bryzgalov", email = "tony.bryzgaloff@gmail.com" },
+    { name = "Viktor Taranenko" },
+]
+keywords = [
+    "clickhouse",
+    "airflow",
+]
+classifiers = [
+    "Development Status :: 5 - Production/Stable",
+    "Environment :: Plugins",
+    "Intended Audience :: Developers",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.8",
+]
+dynamic = [
+    "dependencies",
+]
+
+[project.urls]
+GitHub = "https://github.com/bryzgaloff/airflow-clickhouse-plugin"
+Documentation = "https://github.com/bryzgaloff/airflow-clickhouse-plugin#airflow-clickhouse-plugin"
+Changelog = "https://github.com/bryzgaloff/airflow-clickhouse-plugin/releases"
+Issues = "https://github.com/bryzgaloff/airflow-clickhouse-plugin/issues"
+
+[project.optional-dependencies]
+"common.sql" = [
+    "apache-airflow[common.sql]>=2.2.0,<2.7.0",
+    "clickhouse-driver>=0.2.1",
+]
+
+[tool.setuptools.dynamic]
+dependencies = { file = ["requirements.txt"] }
+
+[tool.setuptools.packages.find]
+where = ["src"]
+namespaces = true  # add directories without __init__.py
diff --git a/requirements.txt b/requirements.txt
index 631d4ec..2f75ec1 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,2 @@
 clickhouse-driver~=0.2.0
 apache-airflow>=2.0.0,<2.7.0
-apache-airflow-providers-common-sql
diff --git a/setup.py b/setup.py
deleted file mode 100755
index 2e53fb4..0000000
--- a/setup.py
+++ /dev/null
@@ -1,46 +0,0 @@
-#!/usr/bin/env python3
-
-import pathlib
-from setuptools import setup, find_packages
-
-import airflow_clickhouse_plugin.__version__ as about
-
-here = pathlib.Path(__file__).parent.absolute()
-with open(here / 'README.md') as readme_file:
-    long_description = readme_file.read()
-with open(here / 'requirements.txt') as requirements_file:
-    install_requires = [
-        requirement_line.rstrip('\n')
-        for requirement_line in requirements_file
-    ]
-
-setup(
-    name=about.__title__,
-    description=about.__description__,
-    long_description=long_description,
-    long_description_content_type='text/markdown',
-    version=about.__version__,
-    author=about.__author__,
-    author_email=about.__author_email__,
-    url=about.__url__,
-    packages=find_packages(exclude=('tests', 'tests.*')),
-    include_package_data=True,
-    python_requires=">=3.7",
-    install_requires=install_requires,
-    extras_require={
-        'pandas': ['apache-airflow[pandas]>=2.0.0,<2.7.0'],
-    },
-    license=about.__license__,
-    zip_safe=False,
-    classifiers=[
-        'Development Status :: 4 - Beta',
-        'Environment :: Plugins',
-        'Intended Audience :: Developers',
-        'Programming Language :: Python :: 3.11',
-        'Programming Language :: Python :: 3.10',
-        'Programming Language :: Python :: 3.9',
-        'Programming Language :: Python :: 3.8',
-        'Programming Language :: Python :: 3.7',
-    ],
-    keywords=['clickhouse', 'airflow'],
-)
diff --git a/src/airflow_clickhouse_plugin/hooks/clickhouse.py b/src/airflow_clickhouse_plugin/hooks/clickhouse.py
new file mode 100644
index 0000000..698352e
--- /dev/null
+++ b/src/airflow_clickhouse_plugin/hooks/clickhouse.py
@@ -0,0 +1,154 @@
+import contextlib
+import typing as t
+from itertools import islice
+
+import clickhouse_driver
+from airflow.hooks.base import BaseHook
+from airflow.models import Connection
+
+# annotated according to clickhouse_driver.Client.execute comments
+_ParamT = t.NewType('_ParamT', t.Union[list, tuple, dict])
+ExecuteParamsT = t.NewType(
+    'ExecuteParamsT',
+    t.Union[
+        # INSERT queries
+        t.List[_ParamT], t.Tuple[_ParamT, ...], t.Generator[_ParamT, None, None],
+        # SELECT queries
+        dict,
+    ],
+)
+ExecuteReturnT = t.NewType(
+    # clickhouse_driver.Client.execute return type
+    'ExecuteReturnT',
+    t.Union[
+        int,  # number of inserted rows
+        t.List[tuple],  # list of tuples with rows/columns
+        t.Tuple[t.List[tuple], t.List[t.Tuple[str, str]]],  # with_column_types
+    ],
+)
+
+
+class ExternalTable(t.TypedDict):
+    name: str
+    structure: t.List[t.Tuple[str, str]]
+    data: t.List[t.Dict[str, t.Any]]
+
+
+default_conn_name = 'clickhouse_default'
+
+
+class ClickHouseHook(BaseHook):
+    def __init__(
+            self,
+            *args,
+            clickhouse_conn_id: str = default_conn_name,
+            database: t.Optional[str] = None,
+            **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self._clickhouse_conn_id = clickhouse_conn_id
+        self._database = database
+
+    def get_conn(self) -> clickhouse_driver.Client:
+        conn = self.get_connection(self._clickhouse_conn_id)
+        return clickhouse_driver.Client(**conn_to_kwargs(conn, self._database))
+
+    def execute(
+            self,
+            sql: t.Union[str, t.Iterable[str]],
+            # arguments of clickhouse_driver.Client.execute
+            params: t.Optional[ExecuteParamsT] = None,
+            with_column_types: bool = False,
+            external_tables: t.Optional[t.List[ExternalTable]] = None,
+            query_id: t.Optional[str] = None,
+            settings: t.Dict[str, t.Any] = None,
+            types_check: bool = False,
+            columnar: bool = False,
+    ) -> ExecuteReturnT:
+        """
+        Passes arguments to ``clickhouse_driver.Client.execute``.
+
+        Allows execution of multiple queries, if ``sql`` argument is an
+        iterable. Returns results of the last query's execution.
+        """
+        if isinstance(sql, str):
+            sql = (sql,)
+        with _disconnecting(self.get_conn()) as conn:
+            last_result = None
+            for query in sql:
+                self.log.info(_format_query_log(query, params))
+                last_result = conn.execute(
+                    query,
+                    params=params,
+                    with_column_types=with_column_types,
+                    external_tables=external_tables,
+                    query_id=query_id,
+                    settings=settings,
+                    types_check=types_check,
+                    columnar=columnar,
+                )
+        return last_result
+
+
+def conn_to_kwargs(conn: Connection, database: t.Optional[str]) -> t.Dict[str, t.Any]:
+    """ Translate Airflow Connection to clickhouse-driver Connection kwargs. """
+    connection_kwargs = conn.extra_dejson.copy()
+    # Connection attributes can be parsed to empty strings by urllib.unparse
+    connection_kwargs['host'] = conn.host or 'localhost'
+    if conn.port:
+        connection_kwargs.update(port=conn.port)
+    if conn.login:
+        connection_kwargs.update(user=conn.login)
+    if conn.password:
+        connection_kwargs.update(password=conn.password)
+    if database is not None:
+        connection_kwargs.update(database=database)
+    elif conn.schema:
+        connection_kwargs.update(database=conn.schema)
+    return connection_kwargs
+
+
+def _format_query_log(query: str, params: ExecuteParamsT) -> str:
+    return ''.join((query, f' with {_format_params(params)}' if params else ''))
+
+
+def _format_params(params: ExecuteParamsT, limit: int = 10) -> str:
+    if isinstance(params, t.Generator) or len(params) <= limit:
+        return str(params)
+    if isinstance(params, dict):
+        head = dict(islice(params.items(), limit))
+    else:
+        head = params[:limit]
+    head_str = str(head)
+    closing_paren = head_str[-1]
+    return f'{head_str[:-1]} … and {len(params) - limit} ' \
+        f'more parameters{closing_paren}'
+
+
+_DisconnectingT = t.TypeVar('_DisconnectingT')
+
+
+@contextlib.contextmanager
+def _disconnecting(thing: _DisconnectingT) -> t.ContextManager[_DisconnectingT]:
+    """
+    Context to automatically disconnect something at the end of a block.
+
+    Similar to ``contextlib.closing`` but calls .disconnect() method on exit.
+
+    Code like this:
+
+    >>> with _disconnecting(<module>.open(<arguments>)) as f:
+    >>>     <block>
+
+    is equivalent to this:
+
+    >>> f = <module>.open(<arguments>)
+    >>> try:
+    >>>     <block>
+    >>> finally:
+    >>>     f.disconnect()
+    """
+    try:
+        yield thing
+    finally:
+        thing.disconnect()
diff --git a/src/airflow_clickhouse_plugin/hooks/clickhouse_dbapi.py b/src/airflow_clickhouse_plugin/hooks/clickhouse_dbapi.py
new file mode 100644
index 0000000..86c26ec
--- /dev/null
+++ b/src/airflow_clickhouse_plugin/hooks/clickhouse_dbapi.py
@@ -0,0 +1,20 @@
+import clickhouse_driver
+from airflow.providers.common.sql.hooks.sql import DbApiHook
+
+from airflow_clickhouse_plugin.hooks.clickhouse import conn_to_kwargs, \
+    default_conn_name
+
+
+class ClickHouseDbApiHook(DbApiHook):
+    conn_name_attr = 'clickhouse_conn_id'
+    clickhouse_conn_id: str  # set by DbApiHook.__init__
+    default_conn_name = default_conn_name
+
+    def __init__(self, *args, schema: str = None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._schema = schema
+
+    def get_conn(self) -> clickhouse_driver.dbapi.Connection:
+        airflow_conn = self.get_connection(self.clickhouse_conn_id)
+        return clickhouse_driver.dbapi \
+            .connect(**conn_to_kwargs(airflow_conn, self._schema))
diff --git a/src/airflow_clickhouse_plugin/operators/clickhouse.py b/src/airflow_clickhouse_plugin/operators/clickhouse.py
new file mode 100644
index 0000000..c780b28
--- /dev/null
+++ b/src/airflow_clickhouse_plugin/operators/clickhouse.py
@@ -0,0 +1,83 @@
+import typing as t
+
+from airflow.models import BaseOperator
+
+from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook, \
+    ExecuteParamsT, ExecuteReturnT, default_conn_name
+
+
+class BaseClickHouseOperator(BaseOperator):
+    """
+    A superclass for operator classes. Defines __init__ with common arguments.
+
+    Includes arguments of clickhouse_driver.Client.execute.
+    """
+
+    template_fields = (  # all str-containing arguments
+        '_sql',
+        '_parameters',
+        '_external_tables',
+        '_query_id',
+        '_settings',
+        '_database',
+    )
+    template_ext: t.Sequence[str] = ('.sql',)
+    template_fields_renderers = {
+        '_sql': 'sql',
+        '_parameters': 'json',
+        '_external_tables': 'json',
+        '_settings': 'json',
+    }
+
+    def __init__(
+            self,
+            *args,
+            sql: t.Union[str, t.Iterable[str]],
+            # arguments of clickhouse_driver.Client.execute
+            parameters: t.Optional[ExecuteParamsT] = None,
+            with_column_types: bool = False,
+            external_tables: t.Optional[t.List[dict]] = None,
+            query_id: t.Optional[str] = None,
+            settings: t.Dict[str, t.Any] = None,
+            types_check: bool = False,
+            columnar: bool = False,
+            # arguments of ClickHouseHook.__init__
+            clickhouse_conn_id: str = default_conn_name,
+            database: t.Optional[str] = None,
+            **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self._sql = sql
+
+        self._parameters = parameters
+        self._with_column_types = with_column_types
+        self._external_tables = external_tables
+        self._query_id = query_id
+        self._settings = settings
+        self._types_check = types_check
+        self._columnar = columnar
+
+        self._clickhouse_conn_id = clickhouse_conn_id
+        self._database = database
+
+    def _hook_execute(self) -> ExecuteReturnT:
+        hook = ClickHouseHook(
+            clickhouse_conn_id=self._clickhouse_conn_id,
+            database=self._database,
+        )
+        return hook.execute(
+            self._sql,
+            self._parameters,
+            self._with_column_types,
+            self._external_tables,
+            self._query_id,
+            self._settings,
+            self._types_check,
+            self._columnar,
+        )
+
+
+class ClickHouseOperator(BaseClickHouseOperator, BaseOperator):
+    def execute(self, context: t.Dict[str, t.Any]) -> ExecuteReturnT:
+        return self._hook_execute()
diff --git a/src/airflow_clickhouse_plugin/operators/clickhouse_dbapi.py b/src/airflow_clickhouse_plugin/operators/clickhouse_dbapi.py
new file mode 100644
index 0000000..de8c611
--- /dev/null
+++ b/src/airflow_clickhouse_plugin/operators/clickhouse_dbapi.py
@@ -0,0 +1,82 @@
+import typing as t
+
+from airflow.providers.common.sql.operators import sql
+
+from airflow_clickhouse_plugin.hooks.clickhouse_dbapi import \
+    ClickHouseDbApiHook
+
+
+class ClickHouseDbApiHookMixin(object):
+    # these attributes are defined in both BaseSQLOperator and SqlSensor
+    conn_id: str
+    hook_params: t.Optional[dict]
+
+    def _get_clickhouse_db_api_hook(self, **extra_hook_params) -> ClickHouseDbApiHook:
+        hook_kwargs = {}
+        if self.conn_id is not None:
+            hook_kwargs['clickhouse_conn_id'] = self.conn_id
+        if self.hook_params is not None:
+            hook_kwargs.update(self.hook_params)
+        hook_kwargs.update(extra_hook_params)
+        return ClickHouseDbApiHook(**hook_kwargs)
+
+
+class ClickHouseBaseDbApiOperator(ClickHouseDbApiHookMixin, sql.BaseSQLOperator):
+    def get_db_hook(self) -> ClickHouseDbApiHook:
+        return self._get_clickhouse_db_api_hook(schema=self.database)
+
+
+class ClickHouseSQLExecuteQueryOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLExecuteQueryOperator,
+):
+    pass
+
+
+class ClickHouseSQLColumnCheckOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLColumnCheckOperator,
+):
+    pass
+
+
+class ClickHouseSQLTableCheckOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLTableCheckOperator,
+):
+    pass
+
+
+class ClickHouseSQLCheckOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLCheckOperator,
+):
+    pass
+
+
+class ClickHouseSQLValueCheckOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLValueCheckOperator,
+):
+    pass
+
+
+class ClickHouseSQLIntervalCheckOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLIntervalCheckOperator,
+):
+    pass
+
+
+class ClickHouseSQLThresholdCheckOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.SQLThresholdCheckOperator,
+):
+    pass
+
+
+class ClickHouseBranchSQLOperator(
+    ClickHouseBaseDbApiOperator,
+    sql.BranchSQLOperator,
+):
+    pass
diff --git a/src/airflow_clickhouse_plugin/sensors/clickhouse.py b/src/airflow_clickhouse_plugin/sensors/clickhouse.py
new file mode 100644
index 0000000..4b57ccc
--- /dev/null
+++ b/src/airflow_clickhouse_plugin/sensors/clickhouse.py
@@ -0,0 +1,31 @@
+import typing as t
+
+from airflow.exceptions import AirflowException
+from airflow.sensors.base import BaseSensorOperator
+
+from airflow_clickhouse_plugin.hooks.clickhouse import ExecuteReturnT
+from airflow_clickhouse_plugin.operators.clickhouse import \
+    BaseClickHouseOperator
+
+
+class ClickHouseSensor(BaseClickHouseOperator, BaseSensorOperator):
+    """ Pokes using clickhouse_driver.Client.execute. """
+
+    def __init__(
+            self,
+            *args,
+            is_failure: t.Callable[[ExecuteReturnT], bool] = None,
+            is_success: t.Callable[[ExecuteReturnT], bool] = None,
+            **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self._is_failure = is_failure
+        self._is_success = bool if is_success is None else is_success
+
+    def poke(self, context: dict) -> bool:
+        result = self._hook_execute()
+        if self._is_failure is not None:
+            is_failure = self._is_failure(result)
+            if is_failure:
+                raise AirflowException(f'is_failure returned {is_failure}')
+        return self._is_success(result)
diff --git a/src/airflow_clickhouse_plugin/sensors/clickhouse_dbapi.py b/src/airflow_clickhouse_plugin/sensors/clickhouse_dbapi.py
new file mode 100644
index 0000000..198ebfb
--- /dev/null
+++ b/src/airflow_clickhouse_plugin/sensors/clickhouse_dbapi.py
@@ -0,0 +1,10 @@
+from airflow.providers.common.sql.sensors.sql import SqlSensor
+
+from airflow_clickhouse_plugin.hooks.clickhouse_dbapi import ClickHouseDbApiHook
+from airflow_clickhouse_plugin.operators.clickhouse_dbapi import \
+    ClickHouseDbApiHookMixin
+
+
+class ClickHouseSqlSensor(ClickHouseDbApiHookMixin, SqlSensor):
+    def _get_hook(self) -> ClickHouseDbApiHook:
+        return self._get_clickhouse_db_api_hook()


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index bebd60e..25930fb 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -1,4 +1,4 @@
-name: Run tests
+name: Unit & Integration Tests
 
 on:
   push:
@@ -10,7 +10,6 @@ on:
       - "tests/**"
       - ".github/workflows/tests.yml"
       - "requirements.txt"
-      - "Makefile"
   pull_request:
     branches: [ "master" ]
     paths:
@@ -20,15 +19,18 @@ on:
       - "tests/**"
       - ".github/workflows/tests.yml"
       - "requirements.txt"
-      - "Makefile"
+
+env:
+  PYTHONPATH: "src"  # for tests to import src/airflow_clickhouse_plugin
 
 jobs:
-  unit-tests:
+
+  unit-tests-no-common-sql:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
-        airflow-version: ["2.0.2", "2.2.5", "2.3.4", "2.4.3", "2.5.0","2.6.0"]
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        airflow-version: ["2.0.2", "2.1.4", "2.2.5", "2.3.4", "2.4.3", "2.5.3", "2.6.3"]
     steps:
     - uses: actions/checkout@v3
     - name: Set up Python ${{ matrix.python-version }}
@@ -41,13 +43,34 @@ jobs:
         python -m pip install -r requirements.txt --ignore-installed apache-airflow==${{ matrix.airflow-version }}
     - name: Run unit tests
       run: |
-        make unit
-  integration-tests:
+        python -m unittest discover -t tests -s unit -p test_clickhouse.py
+
+  unit-tests-common-sql:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        airflow-version: ["2.1.4", "2.2.5", "2.3.4", "2.4.3", "2.5.3", "2.6.3"]
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -r requirements.txt --ignore-installed apache-airflow[common.sql]==${{ matrix.airflow-version }}
+    - name: Run unit tests
+      run: |
+        python -m unittest discover -t tests -s unit
+
+  integration-tests-no-common-sql:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
-        airflow-version: ["2.0.2", "2.2.5", "2.3.4", "2.4.3", "2.5.0","2.6.0"]
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        airflow-version: ["2.0.2", "2.1.4", "2.2.5", "2.3.4", "2.4.3", "2.5.3", "2.6.3"]
     services:
       clickhouse:
         image: yandex/clickhouse-server
@@ -67,4 +90,31 @@ jobs:
       env:
         AIRFLOW_CONN_CLICKHOUSE_DEFAULT: "clickhouse://localhost:${{ job.services.clickhouse.ports['9000'] }}"
       run: |
-        make integration
+        python -m unittest discover -t tests -s integration -p test_clickhouse.py
+
+  integration-tests-common-sql:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        airflow-version: ["2.1.4", "2.2.5", "2.3.4", "2.4.3", "2.5.3", "2.6.3"]
+    services:
+      clickhouse:
+        image: yandex/clickhouse-server
+        ports:
+          - 9000/tcp
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -r requirements.txt --ignore-installed apache-airflow[common.sql]==${{ matrix.airflow-version }}
+    - name: Run tests on ClickHouse server
+      env:
+        AIRFLOW_CONN_CLICKHOUSE_DEFAULT: "clickhouse://localhost:${{ job.services.clickhouse.ports['9000'] }}"
+      run: |
+        python -m unittest discover -t tests -s integration
diff --git a/airflow_clickhouse_plugin/__init__.py b/tests/integration/hooks/__init__.py
similarity index 100%
rename from airflow_clickhouse_plugin/__init__.py
rename to tests/integration/hooks/__init__.py
diff --git a/tests/integration/hooks/test_clickhouse.py b/tests/integration/hooks/test_clickhouse.py
new file mode 100644
index 0000000..81c9f40
--- /dev/null
+++ b/tests/integration/hooks/test_clickhouse.py
@@ -0,0 +1,25 @@
+import unittest
+
+from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
+
+
+class ClickHouseHookTestCase(unittest.TestCase):
+    def test_execute(self):
+        return_value = ClickHouseHook().execute(
+            'SELECT sum(value) * %(multiplier)s AS output FROM ext',
+            params={'multiplier': 2},
+            with_column_types=True,
+            external_tables=[{
+                'name': 'ext',
+                'structure': [('value', 'Int32')],
+                'data': [{'value': 1}, {'value': 2}],
+            }],
+            query_id='airflow-clickhouse-plugin-test',
+            types_check=True,
+            columnar=True,
+        )
+        self.assertTupleEqual(([(6,)], [('output', 'Int64')]), return_value)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/integration/hooks/test_clickhouse_dbapi.py b/tests/integration/hooks/test_clickhouse_dbapi.py
new file mode 100644
index 0000000..2059359
--- /dev/null
+++ b/tests/integration/hooks/test_clickhouse_dbapi.py
@@ -0,0 +1,20 @@
+import unittest
+
+from airflow_clickhouse_plugin.hooks.clickhouse_dbapi import ClickHouseDbApiHook
+
+
+class ClickHouseDbApiHookTestCase(unittest.TestCase):
+    def test_get_records(self):
+        records = ClickHouseDbApiHook().get_records(
+            '''
+                SELECT number * %(multiplier)s AS output
+                FROM system.numbers
+                LIMIT 1 OFFSET 1
+            ''',
+            parameters={'multiplier': 2},
+        )
+        self.assertListEqual([(2,)], records)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/integration/test_hook.py b/tests/integration/test_hook.py
deleted file mode 100644
index 8d2c3dd..0000000
--- a/tests/integration/test_hook.py
+++ /dev/null
@@ -1,125 +0,0 @@
-import unittest
-
-from clickhouse_driver.errors import ServerException, ErrorCodes
-
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-from tests.integration.util import ClickHouseConnectionEnvVarTestCase
-
-
-class BasicTestCase(ClickHouseConnectionEnvVarTestCase):
-    _hook: ClickHouseHook
-
-    def setUp(self):
-        self._hook = ClickHouseHook()
-
-    def test_connection_recreated(self):
-        temp_table_name = 'test_temp_table'
-        result = self._hook.run((
-            f'CREATE TEMPORARY TABLE {temp_table_name} (test_field UInt8)',
-            f'INSERT INTO {temp_table_name} '
-            f'SELECT number FROM system.numbers WHERE number < 5 LIMIT 5',
-            f'SELECT SUM(test_field) FROM {temp_table_name}',
-        ))
-        self.assertListEqual([(10,)], result)
-        try:
-            # a new connection is created and temp table is absent
-            self._hook.run(f'SELECT * FROM {temp_table_name}')
-        except ServerException as err:
-            self.assertEqual(ErrorCodes.UNKNOWN_TABLE, err.code)
-        else:
-            raise AssertionError('server did not raise an error')
-
-    def test_get_first(self):
-        self.assertTupleEqual((1,), self._hook.get_first('SELECT 1'))
-        self.assertIsNone(self._hook.get_first('SELECT 1 WHERE 0'))
-
-
-class BasicInsertTestCase(ClickHouseConnectionEnvVarTestCase):
-    _hook: ClickHouseHook
-
-    def setUp(self):
-        self._hook = ClickHouseHook()
-        self._temp_table_name = f'test_temp_table_for_insert'
-        self._hook.run(f'CREATE TABLE {self._temp_table_name} (test_field UInt8) ENGINE MergeTree() ORDER BY test_field')
-
-    def test_simple_insert(self):
-        result = self._hook.run(f'INSERT INTO {self._temp_table_name} (test_field) VALUES', [(3,)])
-        self.assertEqual(1, result)
-
-        select_result = self._hook.run(f'SELECT test_field FROM {self._temp_table_name} WHERE test_field = 3')
-        self.assertListEqual([(3,)], select_result)
-
-    def test_simple_insert_without_types_check(self):
-        from clickhouse_driver import errors
-        with self.assertRaises(errors.TypeMismatchError) as e:
-            self._hook.run(f'INSERT INTO {self._temp_table_name} (test_field) VALUES', [(-1,)])
-        self.assertNotIn('Expected UInt8', str(e.exception))
-
-    def test_simple_insert_with_types_check(self):
-        from datetime import date
-        from clickhouse_driver import errors
-        with self.assertRaises(errors.TypeMismatchError) as e:
-            self._hook.run(
-                f'INSERT INTO {self._temp_table_name} (test_field) VALUES',
-                [(date(2012, 10, 25), )],
-                types_check=True,
-            )
-        self.assertIn('Expected UInt8', str(e.exception))
-
-    def tearDown(self) -> None:
-        self._hook.run(f'DROP TABLE IF EXISTS {self._temp_table_name}')
-
-
-try:
-    import pandas as pd
-except ImportError:
-    pass
-else:
-    class GetAsPandasDfTestCase(ClickHouseConnectionEnvVarTestCase):
-        def _test(self, sql: str, expected_df):
-            actual_df = ClickHouseHook().get_pandas_df(sql)
-            self.assertListEqual(
-                list(actual_df.columns),
-                list(expected_df.columns),
-            )
-            self.assertListEqual(
-                actual_df.to_dict('records'),
-                expected_df.to_dict('records'),
-            )
-
-        def test(self):
-            self._test(
-                '''
-                    SELECT
-                        number,
-                        concat('result: ', toString(number + number)) AS nSum
-                    FROM system.numbers
-                    WHERE number < 4
-                    LIMIT 3
-                ''',
-                pd.DataFrame.from_dict(dict(
-                    number=(0, 1, 2),
-                    nSum=('result: 0', 'result: 2', 'result: 4'),
-                ))
-            )
-
-        def test_empty_df(self):
-            self._test(
-                '''
-                    SELECT
-                        number,
-                        concat('result: ', toString(number + number)) AS n_sum
-                    FROM (
-                        SELECT number
-                        FROM system.numbers
-                        WHERE number < 4
-                        LIMIT 3
-                    )
-                    WHERE number > 4
-                ''',
-                pd.DataFrame(columns=['number', 'n_sum']),
-            )
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/tests/integration/test_operator.py b/tests/integration/test_operator.py
deleted file mode 100644
index ab2aeba..0000000
--- a/tests/integration/test_operator.py
+++ /dev/null
@@ -1,19 +0,0 @@
-import unittest
-
-from airflow_clickhouse_plugin.operators.clickhouse_operator import ClickHouseOperator
-
-
-class BasicTestCase(unittest.TestCase):
-    def test_operator(self):
-        value = 1
-        operator = ClickHouseOperator(
-            task_id='test',
-            sql=('SELECT %(val0)s', 'SELECT %(val1)s'),
-            parameters={'val0': value + 1, 'val1': value},
-        )
-        last_result = operator.execute(context=dict())
-        self.assertListEqual([(value,)], last_result)
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/tests/integration/test_sql_sensor.py b/tests/integration/test_sql_sensor.py
deleted file mode 100644
index 0c41b19..0000000
--- a/tests/integration/test_sql_sensor.py
+++ /dev/null
@@ -1,23 +0,0 @@
-import unittest
-
-from airflow_clickhouse_plugin.sensors.clickhouse_sql_sensor import ClickHouseSqlSensor
-
-
-class BasicTestCase(unittest.TestCase):
-    def test_sql_sensor(self):
-        self.assertFalse(
-            ClickHouseSqlSensor(task_id='_', sql='SELECT 0').poke(None),
-        )
-        self.assertTrue(
-            ClickHouseSqlSensor(task_id='_', sql='SELECT 1').poke(None),
-        )
-        self.assertFalse(
-            ClickHouseSqlSensor(task_id='_', sql='SELECT NULL').poke(None),
-        )
-        self.assertFalse(
-            ClickHouseSqlSensor(task_id='_', sql='SELECT 1 WHERE 0').poke(None),
-        )
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/tests/integration/util.py b/tests/integration/util.py
deleted file mode 100644
index 923ef46..0000000
--- a/tests/integration/util.py
+++ /dev/null
@@ -1,11 +0,0 @@
-import os
-import unittest
-
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-
-
-class ClickHouseConnectionEnvVarTestCase(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        conn_var = f'AIRFLOW_CONN_{ClickHouseHook.default_conn_name.upper()}'
-        os.environ.setdefault(conn_var, 'clickhouse://localhost')
diff --git a/airflow_clickhouse_plugin/hooks/__init__.py b/tests/unit/hooks/__init__.py
similarity index 100%
rename from airflow_clickhouse_plugin/hooks/__init__.py
rename to tests/unit/hooks/__init__.py
diff --git a/tests/unit/hooks/test_clickhouse.py b/tests/unit/hooks/test_clickhouse.py
new file mode 100644
index 0000000..ee08355
--- /dev/null
+++ b/tests/unit/hooks/test_clickhouse.py
@@ -0,0 +1,191 @@
+import unittest
+from unittest import mock
+
+from airflow.models import Connection
+
+from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook, \
+    _format_query_log
+
+
+class ClickHouseHookTestCase(unittest.TestCase):
+    def test_arguments(self):
+        queries = ['SELECT 1', 'SELECT 2']
+        client_instance_mock = self._client_cls_mock.return_value
+        client_instance_mock.execute.side_effect = [1, 2]
+        self._get_connection_mock.return_value = Connection(
+            conn_id='test-conn-id',
+            host='test-host',
+            port=1234,
+            login='test-login',
+            password='test-pass',
+            schema='test-schema',
+            extra='{"test_extra": "test-extra-value"}',
+        )
+
+        return_value = ClickHouseHook(
+            clickhouse_conn_id='test-conn-id',
+            database='test-database',
+        ).execute(
+            sql=queries,
+            params=[('test-param', 1)],
+            with_column_types=True,
+            external_tables=[{'name': 'ext'}],  # type: ignore
+            query_id='test-query-id',
+            settings={'test-setting': 1},
+            types_check=True,
+            columnar=True,
+        )
+
+        with self.subTest('connection id'):
+            self._get_connection_mock.assert_called_once_with('test-conn-id')
+
+        with self.subTest('Client.__init__'):
+            self._client_cls_mock.assert_called_once_with(
+                host='test-host',
+                port=1234,
+                user='test-login',
+                password='test-pass',
+                database='test-database',
+                test_extra='test-extra-value',
+            )
+
+        for query, mock_call \
+                in zip(queries, client_instance_mock.execute.mock_calls):
+            with self.subTest(f'Client.execute {query}'):
+                self.assertEqual(
+                    mock.call(
+                        query,
+                        params=[('test-param', 1)],
+                        with_column_types=True,
+                        external_tables=[{'name': 'ext'}],
+                        query_id='test-query-id',
+                        settings={'test-setting': 1},
+                        types_check=True,
+                        columnar=True,
+                    ),
+                    mock_call,
+                )
+
+        with self.subTest('Client.disconnect'):
+            client_instance_mock.disconnect.assert_called_once_with()
+
+        with self.subTest('return value'):
+            self.assertEqual(2, return_value)
+
+    def test_defaults(self):
+        client_instance_mock = self._client_cls_mock.return_value
+        client_instance_mock.execute.return_value = 'test-return-value'
+        self._get_connection_mock.return_value = Connection()
+
+        return_value = ClickHouseHook().execute('SELECT 1')
+
+        with self.subTest('connection id'):
+            self._get_connection_mock \
+                .assert_called_once_with('clickhouse_default')
+
+        with self.subTest('Client.__init__'):
+            self._client_cls_mock.assert_called_once_with(host='localhost')
+
+        with self.subTest('Client.execute'):
+            client_instance_mock.execute.assert_called_once_with(
+                'SELECT 1',
+                params=None,
+                with_column_types=False,
+                external_tables=None,
+                query_id=None,
+                settings=None,
+                types_check=False,
+                columnar=False,
+            )
+
+        with self.subTest('Client.disconnect'):
+            client_instance_mock.disconnect.assert_called_once_with()
+
+        with self.subTest('return value'):
+            self.assertEqual('test-return-value', return_value)
+
+    def setUp(self):
+        self._client_cls_patcher = mock.patch('clickhouse_driver.Client')
+        self._client_cls_mock = self._client_cls_patcher.start()
+        self._get_connection_patcher = \
+            mock.patch.object(ClickHouseHook, 'get_connection')
+        self._get_connection_mock = self._get_connection_patcher.start()
+
+    def tearDown(self):
+        self._client_cls_patcher.stop()
+        self._get_connection_patcher.stop()
+
+
+class ClickHouseHookLoggingTestCase(unittest.TestCase):
+    def test(self):
+        test_generator = (_ for _ in range(1))
+        subtests = (
+            # params=None
+            (
+                'SELECT 1', None,
+                'SELECT 1',
+            ),
+            # params: list
+            (
+                'INSERT INTO test2 VALUES', [],
+                'INSERT INTO test2 VALUES',
+            ),
+            (
+                'INSERT INTO test3 VALUES', [3],
+                'INSERT INTO test3 VALUES with [3]',
+            ),
+            (
+                'INSERT INTO test4 VALUES', [val for val in range(11)],
+                ''.join((
+                    'INSERT INTO test4 VALUES with [',
+                    ', '.join(map(str, range(10))),
+                    ' … and 1 more parameters]',
+                )),
+            ),
+            # params: tuple
+            (
+                'INSERT INTO test5 VALUES', (),
+                'INSERT INTO test5 VALUES',
+            ),
+            (
+                'INSERT INTO test6 VALUES', (6,),
+                'INSERT INTO test6 VALUES with (6,)',
+            ),
+            (
+                'INSERT INTO test7 VALUES', tuple(val for val in range(11)),
+                ''.join((
+                    'INSERT INTO test7 VALUES with (',
+                    ', '.join(map(str, range(10))),
+                    ' … and 1 more parameters)',
+                )),
+            ),
+            # params: dict
+            (
+                'SELECT 8', {},
+                'SELECT 8',
+            ),
+            (
+                'SELECT %(param)s', {'param': 9},
+                "SELECT %(param)s with {'param': 9}",
+            ),
+            (
+                'SELECT 10', {k: k for k in range(11)},
+                ''.join((
+                    'SELECT 10 with {',
+                    ', '.join(f'{key}: {key}' for key in range(10)),
+                    ' … and 1 more parameters}',
+                )),
+            ),
+            # params: Generator
+            (
+                'INSERT INTO test11 VALUES', test_generator,
+                f'INSERT INTO test11 VALUES with {test_generator}',
+            ),
+        )
+        for query, params, expected in subtests:
+            with self.subTest((query, params)):
+                self.assertEqual(expected, _format_query_log(query, params))
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit/hooks/test_clickhouse_dbapi.py b/tests/unit/hooks/test_clickhouse_dbapi.py
new file mode 100644
index 0000000..0d04c2a
--- /dev/null
+++ b/tests/unit/hooks/test_clickhouse_dbapi.py
@@ -0,0 +1,59 @@
+import unittest
+from unittest import mock
+
+from airflow.models import Connection
+
+from airflow_clickhouse_plugin.hooks.clickhouse_dbapi import \
+    ClickHouseDbApiHook
+
+
+class ClickHouseDbApiHookTestCase(unittest.TestCase):
+    def test_definition(self):
+        self.assertEqual('clickhouse_conn_id', ClickHouseDbApiHook.conn_name_attr)
+        self.assertEqual('clickhouse_default', ClickHouseDbApiHook.default_conn_name)
+
+    def test_arguments(self):
+        self._get_connection_mock.return_value = Connection(
+            conn_id='test-conn-id',
+            host='test-host',
+            port=1234,
+            login='test-login',
+            password='test-pass',
+            schema='test-schema',
+            extra='{"test_extra": "test-extra"}',
+        )
+        return_value = ClickHouseDbApiHook(
+            clickhouse_conn_id='test-conn-id',
+            schema='test-schema',
+        ).get_conn()
+        self._get_connection_mock.assert_called_once_with('test-conn-id')
+        self._connect_mock.assert_called_once_with(
+            user='test-login',
+            password='test-pass',
+            host='test-host',
+            port=1234,
+            database='test-schema',
+            test_extra='test-extra',
+        )
+        self.assertIs(return_value, self._connect_mock.return_value)
+
+    def test_defaults(self):
+        self._get_connection_mock.return_value = Connection()
+        ClickHouseDbApiHook().get_conn()
+        self._get_connection_mock.assert_called_once_with('clickhouse_default')
+        self._connect_mock.assert_called_once_with(host='localhost')
+
+    def setUp(self) -> None:
+        self._get_connection_patcher = \
+            mock.patch.object(ClickHouseDbApiHook, 'get_connection')
+        self._get_connection_mock = self._get_connection_patcher.start()
+        self._connect_patcher = mock.patch('clickhouse_driver.dbapi.connect')
+        self._connect_mock = self._connect_patcher.start()
+
+    def tearDown(self) -> None:
+        self._get_connection_patcher.stop()
+        self._connect_patcher.stop()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/airflow_clickhouse_plugin/operators/__init__.py b/tests/unit/operators/__init__.py
similarity index 100%
rename from airflow_clickhouse_plugin/operators/__init__.py
rename to tests/unit/operators/__init__.py
diff --git a/tests/unit/operators/test_clickhouse.py b/tests/unit/operators/test_clickhouse.py
new file mode 100644
index 0000000..0d2bc49
--- /dev/null
+++ b/tests/unit/operators/test_clickhouse.py
@@ -0,0 +1,94 @@
+import unittest
+from unittest import mock
+
+from airflow_clickhouse_plugin.operators.clickhouse import \
+    ClickHouseOperator
+
+
+class ClickHouseOperatorTestCase(unittest.TestCase):
+    def test_arguments(self):
+        return_value = ClickHouseOperator(
+            task_id='test1',  # required by Airflow
+            sql='SELECT 1',
+            parameters=[('test-param', 1)],
+            with_column_types=True,
+            external_tables=[{'name': 'ext'}],
+            query_id='test-query-id',
+            settings={'test-setting': 1},
+            types_check=True,
+            columnar=True,
+            clickhouse_conn_id='test-conn-id',
+            database='test-database',
+        ).execute(context={})
+        with self.subTest('ClickHouseHook.__init__'):
+            self._hook_cls_mock.assert_called_once_with(
+                clickhouse_conn_id='test-conn-id',
+                database='test-database',
+            )
+        with self.subTest('ClickHouseHook.execute'):
+            self._hook_cls_mock.return_value.execute.assert_called_once_with(
+                'SELECT 1',
+                [('test-param', 1)],
+                True,
+                [{'name': 'ext'}],
+                'test-query-id',
+                {'test-setting': 1},
+                True,
+                True,
+            )
+        with self.subTest('return value'):
+            self.assertIs(
+                return_value,
+                self._hook_cls_mock.return_value.execute.return_value,
+            )
+
+    def test_defaults(self):
+        ClickHouseOperator(
+            task_id='test2',  # required by Airflow
+            sql='SELECT 2',
+        ).execute(context={})
+        with self.subTest('ClickHouseHook.__init__'):
+            self._hook_cls_mock.assert_called_once_with(
+                clickhouse_conn_id='clickhouse_default',
+                database=None,
+            )
+        with self.subTest('ClickHouseHook.execute'):
+            self._hook_cls_mock.return_value.execute.assert_called_once_with(
+                'SELECT 2',
+                None,
+                False,
+                None,
+                None,
+                None,
+                False,
+                False,
+            )
+
+    def setUp(self):
+        self._hook_cls_patcher = mock.patch('.'.join((
+            'airflow_clickhouse_plugin.operators',
+            'clickhouse.ClickHouseHook',
+        )))
+        self._hook_cls_mock = self._hook_cls_patcher.start()
+
+    def tearDown(self):
+        self._hook_cls_patcher.stop()
+
+
+class ClickHouseOperatorClassTestCase(unittest.TestCase):
+    def test_template_fields(self):
+        self.assertSetEqual(
+            {
+                '_sql',
+                '_parameters',
+                '_external_tables',
+                '_query_id',
+                '_settings',
+                '_database',
+            },
+            frozenset(ClickHouseOperator.template_fields),
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit/operators/test_clickhouse_dbapi.py b/tests/unit/operators/test_clickhouse_dbapi.py
new file mode 100644
index 0000000..7c3df16
--- /dev/null
+++ b/tests/unit/operators/test_clickhouse_dbapi.py
@@ -0,0 +1,41 @@
+import unittest
+from unittest import mock
+
+from airflow_clickhouse_plugin.operators.clickhouse_dbapi import \
+    ClickHouseBaseDbApiOperator
+
+
+class ClickHouseBaseDbApiOperatorTestCase(unittest.TestCase):
+    def test_arguments(self):
+        return_value = ClickHouseBaseDbApiOperator(
+            task_id='test1',  # required by Airflow
+            conn_id='test-conn-id',
+            database='test-database',
+            hook_params={'test_param': 'test-param-value'},
+        ).get_db_hook()
+        self._hook_cls_mock.assert_called_once_with(
+            clickhouse_conn_id='test-conn-id',
+            schema='test-database',
+            test_param='test-param-value',
+        )
+        self.assertIs(return_value, self._hook_cls_mock.return_value)
+
+    def test_defaults(self):
+        ClickHouseBaseDbApiOperator(
+            task_id='test2',  # required by Airflow
+        ).get_db_hook()
+        self._hook_cls_mock.assert_called_once_with(schema=None)
+
+    def setUp(self):
+        self._hook_cls_patcher = mock.patch('.'.join((
+            'airflow_clickhouse_plugin.operators',
+            'clickhouse_dbapi.ClickHouseDbApiHook',
+        )))
+        self._hook_cls_mock = self._hook_cls_patcher.start()
+
+    def tearDown(self):
+        self._hook_cls_patcher.stop()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/airflow_clickhouse_plugin/sensors/__init__.py b/tests/unit/sensors/__init__.py
similarity index 100%
rename from airflow_clickhouse_plugin/sensors/__init__.py
rename to tests/unit/sensors/__init__.py
diff --git a/tests/unit/sensors/test_clickhouse.py b/tests/unit/sensors/test_clickhouse.py
new file mode 100644
index 0000000..e75cf9f
--- /dev/null
+++ b/tests/unit/sensors/test_clickhouse.py
@@ -0,0 +1,116 @@
+import unittest
+from unittest import mock
+
+from airflow import AirflowException
+
+from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor
+
+
+class ClickHouseSensorTestCase(unittest.TestCase):
+    def test_arguments(self):
+        is_success_mock = mock.Mock()
+        is_failure_mock = mock.Mock(return_value=False)
+        execute_mock: mock.Mock = self._hook_cls_mock.return_value.execute
+        return_value = ClickHouseSensor(
+            task_id='test1',  # required by Airflow
+            sql='SELECT 1',
+            parameters=[('test-param', 1)],
+            with_column_types=True,
+            external_tables=[{'name': 'ext'}],
+            query_id='test-query-id',
+            settings={'test-setting': 1},
+            types_check=True,
+            columnar=True,
+            clickhouse_conn_id='test-conn-id',
+            database='test-database',
+            is_success=is_success_mock,
+            is_failure=is_failure_mock,
+        ).poke(context={})
+        with self.subTest('ClickHouseHook.__init__'):
+            self._hook_cls_mock.assert_called_once_with(
+                clickhouse_conn_id='test-conn-id',
+                database='test-database',
+            )
+        with self.subTest('ClickHouseHook.execute'):
+            execute_mock.assert_called_once_with(
+                'SELECT 1',
+                [('test-param', 1)],
+                True,
+                [{'name': 'ext'}],
+                'test-query-id',
+                {'test-setting': 1},
+                True,
+                True,
+            )
+        with self.subTest('is_failure'):
+            is_failure_mock.assert_called_once_with(execute_mock.return_value)
+        with self.subTest('return value'):
+            is_success_mock.assert_called_once_with(execute_mock.return_value)
+            self.assertIs(return_value, is_success_mock.return_value)
+
+    def test_defaults(self):
+        # side_effect is for bool_mock to operate as real bool during __init__
+        bool_mock = mock.Mock(side_effect=bool)
+        with mock.patch('builtins.bool', bool_mock):
+            operator = ClickHouseSensor(
+                task_id='test2',  # required by Airflow
+                sql='SELECT 2',
+            )
+        bool_mock.side_effect = None
+        return_value = operator.poke(context={})
+        with self.subTest('ClickHouseHook.__init__'):
+            self._hook_cls_mock.assert_called_once_with(
+                clickhouse_conn_id='clickhouse_default',
+                database=None,
+            )
+        with self.subTest('ClickHouseHook.execute'):
+            self._hook_cls_mock.return_value.execute.assert_called_once_with(
+                'SELECT 2',
+                None,
+                False,
+                None,
+                None,
+                None,
+                False,
+                False,
+            )
+        with self.subTest('is_success is bool'):
+            self.assertIs(return_value, bool_mock.return_value)
+
+    def test_failure(self):
+        is_failure_mock = mock.Mock(return_value=True)
+        with self.assertRaisesRegexp(AirflowException, 'is_failure returned True'):
+            ClickHouseSensor(
+                task_id='test3',  # required by Airflow
+                sql='SELECT 3',
+                is_failure=is_failure_mock,
+            ).poke(context={})
+
+    def setUp(self):
+        self._hook_cls_patcher = mock.patch('.'.join((
+            'airflow_clickhouse_plugin.operators',
+            'clickhouse.ClickHouseHook',
+        )))
+        self._hook_cls_mock = self._hook_cls_patcher.start()
+
+    def tearDown(self):
+        self._hook_cls_patcher.stop()
+
+
+class ClickHouseSensorClassTestCase(unittest.TestCase):
+    def test_template_fields(self):
+        self.assertSetEqual(
+            {
+                '_sql',
+                '_parameters',
+                '_external_tables',
+                '_query_id',
+                '_settings',
+                '_database',
+            },
+            frozenset(ClickHouseSensor.template_fields),
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit/sensors/test_clickhouse_dbapi.py b/tests/unit/sensors/test_clickhouse_dbapi.py
new file mode 100644
index 0000000..ca142de
--- /dev/null
+++ b/tests/unit/sensors/test_clickhouse_dbapi.py
@@ -0,0 +1,44 @@
+import unittest
+from unittest import mock
+
+from airflow_clickhouse_plugin.sensors.clickhouse_dbapi import \
+    ClickHouseSqlSensor
+
+
+class ClickHouseSqlSensorTestCase(unittest.TestCase):
+    def test_arguments(self):
+        return_value = ClickHouseSqlSensor(
+            task_id='test1',  # required by Airflow
+            sql='SELECT 1',  # required by SqlSensor
+            conn_id='test-conn-id',
+            hook_params={'test_param': 'test-param-value'},
+        )._get_hook()
+        self._hook_cls_mock.assert_called_once_with(
+            clickhouse_conn_id='test-conn-id',
+            test_param='test-param-value',
+        )
+        self.assertIs(return_value, self._hook_cls_mock.return_value)
+
+    def test_defaults(self):
+        ClickHouseSqlSensor(
+            task_id='test2',  # required by Airflow
+            sql='SELECT 2',  # required by SqlSensor
+            conn_id='test-conn-id',  # required by SqlSensor
+        )._get_hook()
+        self._hook_cls_mock.assert_called_once_with(
+            clickhouse_conn_id='test-conn-id',
+        )
+
+    def setUp(self):
+        self._hook_cls_patcher = mock.patch('.'.join((
+            'airflow_clickhouse_plugin.operators',
+            'clickhouse_dbapi.ClickHouseDbApiHook',
+        )))
+        self._hook_cls_mock = self._hook_cls_patcher.start()
+
+    def tearDown(self):
+        self._hook_cls_patcher.stop()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit/test_hook.py b/tests/unit/test_hook.py
deleted file mode 100644
index 4485f95..0000000
--- a/tests/unit/test_hook.py
+++ /dev/null
@@ -1,201 +0,0 @@
-import json
-from typing import Any, Dict
-from unittest import TestCase, mock
-
-import airflow.models
-import clickhouse_driver
-
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-
-
-class ClickHouseConnectionParamsTestCase(TestCase):
-    def test_port(self):
-        port = 8888
-        self._connection_kwargs.update(port=str(port))
-        self._test_client_initiated_with(port=port)
-
-    def test_login(self):
-        login = 'user'
-        self._connection_kwargs.update(login=login)
-        self._test_client_initiated_with(user=login)
-
-    def test_password(self):
-        password = 'password'
-        self._connection_kwargs.update(password=password)
-        self._test_client_initiated_with(password=password)
-
-    def test_database(self):
-        database = 'test-db-from-init-args'
-        ClickHouseHook(database=database).get_conn()
-        self._assert_client_initiated_with(database=database)
-
-    def test_schema(self):
-        schema = 'test-db-from-connection-schema'
-        self._connection_kwargs.update(schema=schema)
-        self._test_client_initiated_with(database=schema)
-
-    def test_host(self):
-        host = 'test-host'
-        self._connection_kwargs.update(host=host)
-        self._test_client_initiated_with(host)
-
-    def test_host_missing(self):
-        self._test_client_initiated_with('localhost')
-
-    def test_extra_params(self):
-        extra_params = dict(
-            int_param=123,
-            float_param=0.456,
-            list_param=['abc', True, 12.34],
-            obj_param={'a': 'b', 'c': [1, 2, 3], 'd': {'e': False}},
-            string_param='value',
-            true_param=True,
-            false_param=False,
-            none_param=None,
-        )
-        for key, value in extra_params.items():
-            with self.subTest(key):
-                self._connection_kwargs.clear()
-                extra_kwarg = {key: value}
-                self._connection_kwargs['extra'] = extra_kwarg
-                self._test_client_initiated_with(**extra_kwarg)
-
-    def _test_client_initiated_with(self, *args, **kwargs) -> None:
-        self._client_mock.reset_mock()
-        ClickHouseHook().get_conn()  # instantiate connection
-        self._assert_client_initiated_with(*args, **kwargs)
-
-    def _assert_client_initiated_with(self, *args, **kwargs) -> None:
-        if not args:
-            args = ('localhost',)  # host argument defaults to localhost
-        args = (clickhouse_driver.Client,) + args
-        self._client_mock.assert_called_once_with(*args, **kwargs)
-
-    _connection_kwargs: Dict[str, Any] = {}
-    _patched_hook_connection: mock._patch
-    _patched_client: mock._patch
-    _client_mock: mock.Mock
-
-    @classmethod
-    def setUpClass(cls):
-        cls._patched_hook_connection = mock.patch(
-            'airflow_clickhouse_plugin.hooks.clickhouse_hook.ClickHouseHook.get_connection',
-            lambda self, conn_id: airflow.models.Connection(**dict(
-                cls._connection_kwargs,
-                extra=json.dumps(cls._connection_kwargs['extra']),
-            ) if 'extra' in cls._connection_kwargs else cls._connection_kwargs),
-        )
-        cls._patched_hook_connection.__enter__()
-
-    @classmethod
-    def tearDownClass(cls) -> None:
-        cls._patched_hook_connection.__exit__(None, None, None)
-
-    def setUp(self):
-        self._connection_kwargs.clear()
-        self._client_mock = mock.Mock()
-        self._patched_client = \
-            mock.patch('clickhouse_driver.Client.__new__', self._client_mock)
-        self._patched_client.__enter__()
-
-    def tearDown(self):
-        self._patched_client.__exit__(None, None, None)
-
-
-class HookLogQueryTestCase(TestCase):
-    def setUp(self) -> None:
-        self.hook = ClickHouseHook()
-
-    def test_log_params_dict(self):
-        self.assertEqual('{}', self.hook._log_params({}))
-        self.assertEqual('{1: 1}', self.hook._log_params({1: 1}))
-        self.assertEqual('{1: 1}', self.hook._log_params({1: 1}, limit=1))
-        self.assertEqual(
-            '{1: 1 … and 1 more parameters}',
-            self.hook._log_params({1: 1, 2: 2}, limit=1),
-        )
-        self.assertEqual(
-            '{1: 1, 2: 2}',
-            self.hook._log_params({1: 1, 2: 2}),
-        )
-        self.assertEqual(
-            '{0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 7: 7, 8: 8, 9: 9 …'
-            ' and 1 more parameters}',
-            self.hook._log_params({k: k for k in range(11)}),
-        )
-        self.assertEqual(
-            '{0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 7: 7, 8: 8, 9: 9 …'
-            ' and 10 more parameters}',
-            self.hook._log_params({k: k for k in range(20)}),
-        )
-        self.assertEqual(
-            '{0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 7: 7, 8: 8, 9: 9 …'
-            ' and 10 more parameters}',
-            self.hook._log_params({k: k for k in range(20)}, limit=10),
-        )
-
-    def test_log_params_generator(self):
-        def gen():
-            yield
-        g = gen()
-        self.assertEqual(str(g), self.hook._log_params(g))
-
-    def test_log_params_tuple(self):
-        self.assertEqual('()', self.hook._log_params(()))
-        self.assertEqual('(1,)', self.hook._log_params((1, )))
-        self.assertEqual('(1,)', self.hook._log_params((1, ), limit=1))
-        self.assertEqual(
-            '(1, … and 1 more parameters)',
-            self.hook._log_params((1, 2), limit=1),
-        )
-        self.assertEqual(
-            '(1, 2)',
-            self.hook._log_params((1, 2)),
-        )
-        self.assertEqual(
-            '(0, 1, 2, 3, 4, 5, 6, 7, 8, 9 … and 1 more parameters)',
-            self.hook._log_params(tuple(range(11))),
-        )
-        self.assertEqual(
-            '(0, 1, 2, 3, 4, 5, 6, 7, 8, 9 … and 10 more parameters)',
-            self.hook._log_params(tuple(range(20))),
-        )
-        self.assertEqual(
-            '(0, 1, 2, 3, 4, 5, 6, 7, 8, 9 … and 10 more parameters)',
-            self.hook._log_params(tuple(range(20)), limit=10),
-        )
-
-    def test_log_params_list(self):
-        self.assertEqual('[]', self.hook._log_params([]))
-        self.assertEqual('[1]', self.hook._log_params([1]))
-        self.assertEqual('[1]', self.hook._log_params([1], limit=1))
-        self.assertEqual(
-            '[1 … and 1 more parameters]',
-            self.hook._log_params([1, 2], limit=1),
-        )
-        self.assertEqual(
-            '[1, 2]',
-            self.hook._log_params([1, 2]),
-        )
-        self.assertEqual(
-            '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9 … and 1 more parameters]',
-            self.hook._log_params(list(range(11))),
-        )
-        self.assertEqual(
-            '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9 … and 10 more parameters]',
-            self.hook._log_params(list(range(20))),
-        )
-        self.assertEqual(
-            '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9 … and 10 more parameters]',
-            self.hook._log_params(list(range(20)), limit=10),
-        )
-
-    def test_log_query(self):
-        _ = self.hook.log  # to initialize .log property
-        with mock.patch.object(self.hook, '_log') as patched:
-            self.hook._log_query('SELECT 1', {})
-            patched.info.assert_called_with('%s%s', 'SELECT 1', '')
-            self.hook._log_query('SELECT 1', {1: 1})
-            patched.info.assert_called_with('%s%s', 'SELECT 1', ' with {1: 1}')
-            self.hook._log_query('SELECT 1', [1])
-            patched.info.assert_called_with('%s%s', 'SELECT 1', ' with [1]')
diff --git a/tests/unit/test_operator.py b/tests/unit/test_operator.py
deleted file mode 100644
index 1d2b0e9..0000000
--- a/tests/unit/test_operator.py
+++ /dev/null
@@ -1,83 +0,0 @@
-import datetime
-import os
-import unittest
-from tempfile import NamedTemporaryFile
-from unittest import mock
-
-from airflow.models import DAG
-
-from airflow_clickhouse_plugin.hooks.clickhouse_hook import ClickHouseHook
-from airflow_clickhouse_plugin.operators.clickhouse_operator import ClickHouseOperator
-
-_TEST_START_DATE = datetime.datetime.now()
-
-
-class ClickHouseOperatorTestCase(unittest.TestCase):
-    @mock.patch(
-        'airflow_clickhouse_plugin'
-        '.operators.clickhouse_operator.ClickHouseHook',
-    )
-    def test(self, clickhouse_hook_mock: mock.MagicMock):
-        sql = object()
-        clickhouse_conn_id = object()
-        parameters = object()
-        database = object()
-        op = ClickHouseOperator(
-            task_id='_', sql=sql, clickhouse_conn_id=clickhouse_conn_id,
-            parameters=parameters, database=database,
-        )
-        op.execute(context=dict())
-        clickhouse_hook_mock.assert_called_once_with(
-            clickhouse_conn_id=clickhouse_conn_id,
-            database=database,
-        )
-        clickhouse_hook_mock().run.assert_called_once_with(sql, parameters)
-
-    @mock.patch(
-        'airflow_clickhouse_plugin'
-        '.operators.clickhouse_operator.ClickHouseHook',
-    )
-    def test_defaults(self, clickhouse_hook_mock: mock.MagicMock):
-        sql = 'SELECT 1'
-        op = ClickHouseOperator(task_id='_', sql=sql)
-        op.execute(context=dict())
-        clickhouse_hook_mock.assert_called_once_with(
-            clickhouse_conn_id=ClickHouseHook.default_conn_name,
-            database=None,
-        )
-        clickhouse_hook_mock().run.assert_called_once_with(sql, None)
-
-    def test_template_fields_overrides(self):
-        assert ClickHouseOperator.template_fields == ('_sql',)
-
-    def test_resolve_template_files_value(self):
-        with NamedTemporaryFile(suffix='.sql') as sql_file:
-            sql_file.write(b'{{ ds }}')
-            sql_file.flush()
-            sql_file_dir = os.path.dirname(sql_file.name)
-            sql_file_name = os.path.basename(sql_file.name)
-
-            with DAG('test-dag', start_date=_TEST_START_DATE, template_searchpath=sql_file_dir):
-                task = ClickHouseOperator(task_id='test_task', sql=sql_file_name)
-
-            task.resolve_template_files()
-
-        assert task._sql == '{{ ds }}'
-
-    def test_resolve_template_files_list(self):
-        with NamedTemporaryFile(suffix='.sql') as sql_file:
-            sql_file.write(b'{{ ds }}')
-            sql_file.flush()
-            sql_file_dir = os.path.dirname(sql_file.name)
-            sql_file_name = os.path.basename(sql_file.name)
-
-            with DAG('test-dag', start_date=_TEST_START_DATE, template_searchpath=sql_file_dir):
-                task = ClickHouseOperator(task_id='test_task', sql=[sql_file_name, 'some_string'])
-
-            task.resolve_template_files()
-
-        assert task._sql == ['{{ ds }}', 'some_string']
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/tests/unit/test_sql_sensor.py b/tests/unit/test_sql_sensor.py
deleted file mode 100644
index c0bb29d..0000000
--- a/tests/unit/test_sql_sensor.py
+++ /dev/null
@@ -1,126 +0,0 @@
-import unittest
-from unittest import mock
-
-from airflow.exceptions import AirflowException
-
-from airflow_clickhouse_plugin.sensors.clickhouse_sql_sensor import ClickHouseSqlSensor
-
-
-class ClickHouseSqlSensorTestCase(unittest.TestCase):
-    def test_clickhouse_sql_sensor_poke(self):
-        op = ClickHouseSqlSensor(task_id='_', sql='',)
-        for return_value, expected_result in (
-                ([], False),
-                ([[None]], False),
-                ([['None']], True),
-                ([[0.0]], False),
-                ([[0]], False),
-                ([['0']], True),
-                ([['1']], True),
-        ):
-            with self.subTest(return_value):
-                self._get_records_mock.return_value = return_value
-                self.assertEqual(expected_result, op.poke(context=None))
-
-    def test_clickhouse_sql_sensor_poke_fail_on_empty(self):
-        op = ClickHouseSqlSensor(task_id='_', sql='', fail_on_empty=True)
-        self._get_records_mock.return_value = []
-        with self.assertRaises(AirflowException):
-            op.poke(context=None)
-
-    def test_clickhouse_sql_sensor_poke_success(self):
-        op = ClickHouseSqlSensor(task_id='_', sql='', success=[1].__contains__)
-        for return_value, expected_result in (
-                ([], False),
-                ([[1]], True),
-                ([['1']], False),
-        ):
-            with self.subTest(return_value):
-                self._get_records_mock.return_value = return_value
-                self.assertEqual(expected_result, op.poke(context=None))
-
-    def test_clickhouse_sql_sensor_poke_failure(self):
-        op = ClickHouseSqlSensor(task_id='_', sql='', failure=[1].__contains__)
-
-        self._get_records_mock.return_value = []
-        with self.subTest(self._get_records_mock.return_value):
-            self.assertFalse(op.poke(None))
-
-        self._get_records_mock.return_value = [[1]]
-        with self.subTest(self._get_records_mock.return_value):
-            with self.assertRaises(AirflowException):
-                op.poke(context=None)
-
-    def test_clickhouse_sql_sensor_poke_failure_success(self):
-        op = ClickHouseSqlSensor(
-            task_id='_', sql='',
-            failure=[1].__contains__, success=[2].__contains__,
-        )
-        for return_value, expected_result in (
-                ([], False),
-                ([[2]], True),
-        ):
-            with self.subTest(return_value):
-                self._get_records_mock.return_value = return_value
-                self.assertEqual(expected_result, op.poke(context=None))
-        self._get_records_mock.return_value = [[1]]
-        with self.subTest(self._get_records_mock.return_value):
-            with self.assertRaises(AirflowException):
-                op.poke(context=None)
-
-    def test_clickhouse_sql_sensor_poke_failure_success_same(self):
-        fn = [2].__contains__
-        op = ClickHouseSqlSensor(task_id='_', sql='', failure=fn, success=fn)
-
-        self._get_records_mock.return_value = []
-        with self.subTest(self._get_records_mock.return_value):
-            self.assertFalse(op.poke(None))
-
-        self._get_records_mock.return_value = [[2]]
-        with self.subTest(self._get_records_mock.return_value):
-            with self.assertRaises(AirflowException):
-                op.poke(context=None)
-
-    def test_clickhouse_sql_sensor_poke_invalid_failure(self):
-        op = ClickHouseSqlSensor(task_id='_', sql='', failure=[1])
-        self._get_records_mock.return_value = [[1]]
-        with self.assertRaises(AirflowException):
-            op.poke(context=None)
-
-    def test_clickhouse_sql_sensor_poke_invalid_success(self):
-        op = ClickHouseSqlSensor(task_id='_', sql='', success=[1])
-        self._get_records_mock.return_value = [[1]]
-        with self.assertRaises(AirflowException):
-            op.poke(context=None)
-
-    _get_records_patch: mock._patch
-    _get_records_mock: mock.MagicMock
-
-    @classmethod
-    def setUpClass(cls):
-        cls._get_records_patch = mock.patch(
-            'airflow_clickhouse_plugin.hooks.clickhouse_hook'
-                '.ClickHouseHook.get_records',
-        )
-        cls._get_records_mock = cls._get_records_patch.__enter__()
-
-    @classmethod
-    def tearDownClass(cls) -> None:
-        cls._get_records_patch.__exit__(None, None, None)
-
-
-class ClickHouseLegacySqlSensorTestCase(unittest.TestCase):
-    @mock.patch('airflow.providers.common.sql.sensors.sql.SqlSensor.poke')
-    def test_get_hook_defined(
-            self,
-            sql_sensor_poke_mock: mock.MagicMock,
-    ):
-        """ Test for case when ``SqlSensor._get_hook`` method is present. """
-        op = ClickHouseSqlSensor(task_id='_', sql='')
-        context = dict(some_test_context=True)
-        op.poke(context)
-        sql_sensor_poke_mock.assert_called_once_with(context)
-
-
-if __name__ == '__main__':
-    unittest.main()