added clickhouse integration

.github/workflows/ci.yml

@@ -152,6 +152,44 @@ jobs:
 
       - name: Run CLI tests
         run: uv run pytest tests/cli/ -v
+
+  clickhouse-cli-tests:
+    runs-on: ubuntu-latest
+
+    env:
+      JETBASE_SQLALCHEMY_URL: clickhouse://default:@localhost:8123/default
+
+    services:
+      clickhouse:
+        image: clickhouse/clickhouse-server:latest
+        env:
+          CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
+          CLICKHOUSE_PASSWORD: ""
+        ports:
+          - 8123:8123
+          - 9000:9000
+        options: >-
+          --health-cmd "wget --spider -q http://localhost:8123/ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v6
+
+      - name: Install dependencies
+        run: uv sync --dev --extra clickhouse
+
+      - name: Run CLI tests
+        run: uv run pytest tests/cli/ -v
 
   # TEMP REMOVE: due to snowflake account being down
   # snowflake-integration-tests:

jetbase/commands/lock_status.py

@@ -1,3 +1,6 @@
+from jetbase.config import get_config
+from jetbase.database.queries.base import detect_db
+from jetbase.enums import DatabaseType
 from jetbase.logging import logger
 from jetbase.models import LockStatus
 from jetbase.repositories.lock_repo import fetch_lock_status, lock_table_exists
@@ -11,9 +14,16 @@ def lock_status_cmd() -> None:
     Queries the jetbase_lock table to check if migrations are currently
     locked. If locked, displays the timestamp when the lock was acquired.
 
+    For ClickHouse, displays a message that locking is not supported.
+
     Returns:
         None: Prints "LOCKED" with timestamp or "UNLOCKED" to stdout.
     """
+    # Check for ClickHouse first
+    sqlalchemy_url: str = get_config(required={"sqlalchemy_url"}).sqlalchemy_url
+    if detect_db(sqlalchemy_url) == DatabaseType.CLICKHOUSE:
+        logger.info("ClickHouse does not support database locking.")
+        return
 
     if not lock_table_exists() or not migrations_table_exists():
         logger.info("Status: UNLOCKED")

jetbase/commands/unlock.py

@@ -1,3 +1,6 @@
+from jetbase.config import get_config
+from jetbase.database.queries.base import detect_db
+from jetbase.enums import DatabaseType
 from jetbase.logging import logger
 from jetbase.repositories.lock_repo import (
     lock_table_exists,
@@ -14,9 +17,16 @@ def unlock_cmd() -> None:
     Use this only if you are certain that no migration is currently running,
     as unlocking during an active migration can cause database corruption.
 
+    For ClickHouse, displays a message that locking is not supported.
+
     Returns:
         None: Prints "Unlock successful." to stdout.
     """
+    # Check for ClickHouse first
+    sqlalchemy_url: str = get_config(required={"sqlalchemy_url"}).sqlalchemy_url
+    if detect_db(sqlalchemy_url) == DatabaseType.CLICKHOUSE:
+        logger.info("ClickHouse does not support database locking. No lock to release.")
+        return
 
     if not lock_table_exists() or not migrations_table_exists():
         logger.info("Unlock successful.")

jetbase/database/queries/clickhouse.py

@@ -0,0 +1,185 @@
+from sqlalchemy import TextClause, text
+
+from jetbase.database.queries.base import BaseQueries
+from jetbase.enums import MigrationType
+
+
+class ClickHouseQueries(BaseQueries):
+    @staticmethod
+    def create_migrations_table_stmt() -> TextClause:
+        return text(
+            """
+            CREATE TABLE IF NOT EXISTS jetbase_migrations (
+                order_executed UInt64,
+                version Nullable(String),
+                description String,
+                filename String,
+                migration_type String,
+                applied_at DateTime64(6) DEFAULT now64(6),
+                checksum String
+            ) ENGINE = MergeTree()
+            ORDER BY order_executed
+            """
+        )
+
+    @staticmethod
+    def insert_version_stmt() -> TextClause:
+        return text(
+            """
+            INSERT INTO jetbase_migrations (order_executed, version, description, filename, migration_type, checksum, applied_at) 
+            SELECT 
+                (SELECT COALESCE(MAX(order_executed), 0) + 1 FROM jetbase_migrations),
+                :version, 
+                :description, 
+                :filename, 
+                :migration_type, 
+                :checksum,
+                now64(6)
+            """
+        )
+
+    @staticmethod
+    def check_if_migrations_table_exists_query() -> TextClause:
+        return text(
+            """
+            SELECT count() > 0 AS table_exists
+            FROM system.tables
+            WHERE database = currentDatabase()
+              AND name = 'jetbase_migrations'
+            """
+        )
+
+    @staticmethod
+    def check_if_lock_table_exists_query() -> TextClause:
+        return text("SELECT 0 AS table_exists")
+
+    @staticmethod
+    def latest_version_query() -> TextClause:
+        return text(
+            f"""
+            SELECT 
+                version 
+            FROM 
+                jetbase_migrations
+            WHERE
+                migration_type = '{MigrationType.VERSIONED.value}'
+            ORDER BY 
+                order_executed DESC
+            LIMIT 1
+            """
+        )
+
+    @staticmethod
+    def latest_versions_query() -> TextClause:
+        return text(
+            f"""
+            SELECT 
+                version 
+            FROM 
+                jetbase_migrations
+            WHERE
+                migration_type = '{MigrationType.VERSIONED.value}'
+            ORDER BY 
+                order_executed DESC
+            LIMIT :limit
+            """
+        )
+
+    @staticmethod
+    def latest_versions_by_starting_version_query() -> TextClause:
+        return text(
+            f"""
+            SELECT
+                version
+            FROM
+                jetbase_migrations
+            WHERE order_executed > 
+                (SELECT order_executed FROM jetbase_migrations 
+                    WHERE version = :starting_version AND migration_type = '{MigrationType.VERSIONED.value}')
+            AND migration_type = '{MigrationType.VERSIONED.value}'
+            ORDER BY 
+                order_executed DESC
+            """
+        )
+
+    @staticmethod
+    def delete_version_stmt() -> TextClause:
+        return text(
+            f"""
+            ALTER TABLE jetbase_migrations DELETE 
+            WHERE version = :version
+            AND migration_type = '{MigrationType.VERSIONED.value}'
+            SETTINGS mutations_sync = 1
+            """
+        )
+
+    @staticmethod
+    def update_repeatable_migration_stmt() -> TextClause:
+        return text(
+            """
+            ALTER TABLE jetbase_migrations
+            UPDATE checksum = :checksum, applied_at = now64(6)
+            WHERE filename = :filename
+            AND migration_type = :migration_type
+            SETTINGS mutations_sync = 1
+            """
+        )
+
+    @staticmethod
+    def repair_migration_checksum_stmt() -> TextClause:
+        return text(
+            f"""
+            ALTER TABLE jetbase_migrations
+            UPDATE checksum = :checksum
+            WHERE version = :version
+            AND migration_type = '{MigrationType.VERSIONED.value}'
+            SETTINGS mutations_sync = 1
+            """
+        )
+
+    @staticmethod
+    def delete_missing_version_stmt() -> TextClause:
+        return text(
+            f"""
+            ALTER TABLE jetbase_migrations DELETE
+            WHERE version = :version
+            AND migration_type = '{MigrationType.VERSIONED.value}'
+            SETTINGS mutations_sync = 1
+            """
+        )
+
+    @staticmethod
+    def delete_missing_repeatable_stmt() -> TextClause:
+        return text(
+            f"""
+            ALTER TABLE jetbase_migrations DELETE
+            WHERE filename = :filename
+            AND migration_type IN ('{MigrationType.RUNS_ALWAYS.value}', '{MigrationType.RUNS_ON_CHANGE.value}')
+            SETTINGS mutations_sync = 1
+            """
+        )
+
+    @staticmethod
+    def migration_records_query(
+        ascending: bool = True,
+        all_repeatables: bool = False,
+        migration_type: MigrationType | None = None,
+    ) -> TextClause:
+        query: str = f"""
+            SELECT
+                order_executed, 
+                version, 
+                description,
+                filename,
+                migration_type,
+                applied_at,
+                checksum  
+            FROM
+                jetbase_migrations
+            {"WHERE migration_type = " + f"'{migration_type.value}'" if migration_type else ""}
+            {"WHERE migration_type IN ('RUNS_ON_CHANGE', 'RUNS_ALWAYS')" if all_repeatables else ""}
+            ORDER BY
+                order_executed {"ASC" if ascending else "DESC"}
+        """
+
+        return text(query)

jetbase/database/queries/query_loader.py

@@ -2,6 +2,7 @@
 
 from jetbase.config import get_config
 from jetbase.database.queries.base import BaseQueries, QueryMethod
+from jetbase.database.queries.clickhouse import ClickHouseQueries
 from jetbase.database.queries.databricks import DatabricksQueries
 from jetbase.database.queries.mysql import MySQLQueries
 from jetbase.database.queries.postgres import PostgresQueries
@@ -37,6 +38,8 @@ def get_database_type() -> DatabaseType:
         return DatabaseType.MYSQL
     elif dialect_name == "databricks":
         return DatabaseType.DATABRICKS
+    elif dialect_name == "clickhouse":
+        return DatabaseType.CLICKHOUSE
     else:
         raise ValueError(f"Unsupported database type: {dialect_name}")
 
@@ -66,6 +69,8 @@ def get_queries() -> type[BaseQueries]:
         return MySQLQueries
     elif db_type == DatabaseType.DATABRICKS:
         return DatabricksQueries
+    elif db_type == DatabaseType.CLICKHOUSE:
+        return ClickHouseQueries
     else:
         raise ValueError(f"Unsupported database type: {db_type}")
 

jetbase/engine/lock.py

@@ -4,9 +4,18 @@
 
 from sqlalchemy.engine import CursorResult
 
+from jetbase.config import get_config
+from jetbase.database.queries.base import detect_db
+from jetbase.enums import DatabaseType
 from jetbase.repositories.lock_repo import lock_database, release_lock
 
 
+def _is_clickhouse() -> bool:
+    """Check if the current database is ClickHouse."""
+    sqlalchemy_url: str = get_config(required={"sqlalchemy_url"}).sqlalchemy_url
+    return detect_db(sqlalchemy_url) == DatabaseType.CLICKHOUSE
+
+
 def acquire_lock() -> str:
     """
     Acquire the migration lock immediately.
@@ -46,6 +55,8 @@ def migration_lock() -> Generator[None, None, None]:
     even if an exception occurs. Fails immediately if the lock is
     already held by another process.
 
+    For ClickHouse, locking is not supported.
+
     Yields:
         None: Yields control to the context block.
 
@@ -56,6 +67,11 @@ def migration_lock() -> Generator[None, None, None]:
         >>> with migration_lock():
         ...     run_migration()
     """
+    # ClickHouse doesn't support reliable locking - skip entirely
+    if _is_clickhouse():
+        yield
+        return
+
     process_id: str | None = None
     try:
         process_id = acquire_lock()

jetbase/enums.py

@@ -24,3 +24,4 @@ class DatabaseType(Enum):
     SNOWFLAKE = "snowflake"
     MYSQL = "mysql"
     DATABRICKS = "databricks"
+    CLICKHOUSE = "clickhouse"

jetbase/repositories/lock_repo.py

@@ -3,12 +3,20 @@
 from sqlalchemy import Result, Row
 from sqlalchemy.engine import CursorResult
 
+from jetbase.config import get_config
 from jetbase.database.connection import get_db_connection
-from jetbase.database.queries.base import QueryMethod
+from jetbase.database.queries.base import QueryMethod, detect_db
 from jetbase.database.queries.query_loader import get_query
+from jetbase.enums import DatabaseType
 from jetbase.models import LockStatus
 
 
+def _is_clickhouse() -> bool:
+    """Check if the current database is ClickHouse."""
+    sqlalchemy_url: str = get_config(required={"sqlalchemy_url"}).sqlalchemy_url
+    return detect_db(sqlalchemy_url) == DatabaseType.CLICKHOUSE
+
+
 def lock_table_exists() -> bool:
     """
     Check if the jetbase_lock table exists in the database.
@@ -34,9 +42,16 @@ def create_lock_table_if_not_exists() -> None:
     Creates the lock table and initializes it with a single unlocked
     record. This table is used to prevent concurrent migrations.
 
+    Skipped for ClickHouse as its does not support
+    reliable database locking.
+
     Returns:
         None: Table is created as a side effect.
     """
+    # ClickHouse doesn't support reliable locking - skip lock table creation
+    if _is_clickhouse():
+        return
+
     with get_db_connection() as connection:
         connection.execute(get_query(query_name=QueryMethod.CREATE_LOCK_TABLE_STMT))