[DOP-19896] Add Clickhouse API schema

Author: Ilyas Gasanov

README.rst

@@ -33,6 +33,7 @@ Data.SyncMaster is as low-code ETL tool for transfering data between databases a
 List of currently supported connections:
 
 * Apache Hive
+* Clickhouse
 * Postgres
 * Oracle
 * HDFS

syncmaster/schemas/v1/connection_types.py

@@ -8,6 +8,7 @@
 POSTGRES_TYPE = Literal["postgres"]
 S3_TYPE = Literal["s3"]
 HDFS_TYPE = Literal["hdfs"]
+CLICKHOUSE_TYPE = Literal["clickhouse"]
 
 
 class ConnectionType(str, Enum):
@@ -16,3 +17,4 @@ class ConnectionType(str, Enum):
     ORACLE = "oracle"
     S3 = "s3"
     HDFS = "hdfs"
+    CLICKHOUSE = "clickhouse"

syncmaster/schemas/v1/connections/clickhouse.py

@@ -0,0 +1,47 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from pydantic import BaseModel, Field, SecretStr
+
+from syncmaster.schemas.v1.connection_types import CLICKHOUSE_TYPE
+
+
+class ClickhouseBaseSchema(BaseModel):
+    type: CLICKHOUSE_TYPE
+
+    class Config:
+        from_attributes = True
+
+
+class ReadClickhouseConnectionSchema(ClickhouseBaseSchema):
+    host: str
+    port: int
+    database: str | None = None
+    additional_params: dict = Field(default_factory=dict)
+
+
+class ReadClickhouseAuthSchema(ClickhouseBaseSchema):
+    user: str
+
+
+class UpdateClickhouseConnectionSchema(ClickhouseBaseSchema):
+    host: str | None = None
+    port: int | None = None
+    database: str | None = None
+    additional_params: dict | None = Field(default_factory=dict)
+
+
+class UpdateClickhouseAuthSchema(ClickhouseBaseSchema):
+    user: str | None = None  # noqa: F722
+    password: SecretStr | None = None
+
+
+class CreateClickhouseConnectionSchema(ClickhouseBaseSchema):
+    host: str
+    port: int
+    database: str | None = None
+    additional_params: dict = Field(default_factory=dict)
+
+
+class CreateClickhouseAuthSchema(ClickhouseBaseSchema):
+    user: str
+    password: SecretStr

syncmaster/schemas/v1/connections/connection.py

@@ -3,6 +3,14 @@
 
 from pydantic import BaseModel, Field, model_validator
 
+from syncmaster.schemas.v1.connections.clickhouse import (
+    CreateClickhouseAuthSchema,
+    CreateClickhouseConnectionSchema,
+    ReadClickhouseAuthSchema,
+    ReadClickhouseConnectionSchema,
+    UpdateClickhouseAuthSchema,
+    UpdateClickhouseConnectionSchema,
+)
 from syncmaster.schemas.v1.connections.hdfs import (
     HDFSCreateAuthSchema,
     HDFSCreateConnectionSchema,
@@ -51,12 +59,14 @@
     | HDFSReadConnectionSchema
     | ReadOracleConnectionSchema
     | ReadPostgresConnectionSchema
+    | ReadClickhouseConnectionSchema
     | S3ReadConnectionSchema
 )
 CreateConnectionDataSchema = (
     CreateHiveConnectionSchema
     | CreateOracleConnectionSchema
     | CreatePostgresConnectionSchema
+    | CreateClickhouseConnectionSchema
     | HDFSCreateConnectionSchema
     | S3CreateConnectionSchema
 )
@@ -66,15 +76,31 @@
     | S3UpdateConnectionSchema
     | UpdateOracleConnectionSchema
     | UpdatePostgresConnectionSchema
+    | UpdateClickhouseConnectionSchema
 )
 ReadConnectionAuthDataSchema = (
-    ReadHiveAuthSchema | ReadOracleAuthSchema | ReadPostgresAuthSchema | S3ReadAuthSchema | HDFSReadAuthSchema
+    ReadHiveAuthSchema
+    | ReadOracleAuthSchema
+    | ReadPostgresAuthSchema
+    | ReadClickhouseAuthSchema
+    | S3ReadAuthSchema
+    | HDFSReadAuthSchema
 )
 CreateConnectionAuthDataSchema = (
-    CreateHiveAuthSchema | CreateOracleAuthSchema | CreatePostgresAuthSchema | S3CreateAuthSchema | HDFSCreateAuthSchema
+    CreateHiveAuthSchema
+    | CreateOracleAuthSchema
+    | CreatePostgresAuthSchema
+    | CreateClickhouseAuthSchema
+    | S3CreateAuthSchema
+    | HDFSCreateAuthSchema
 )
 UpdateConnectionAuthDataSchema = (
-    UpdateHiveAuthSchema | UpdateOracleAuthSchema | UpdatePostgresAuthSchema | S3UpdateAuthSchema | HDFSUpdateAuthSchema
+    UpdateHiveAuthSchema
+    | UpdateOracleAuthSchema
+    | UpdatePostgresAuthSchema
+    | UpdateClickhouseAuthSchema
+    | S3UpdateAuthSchema
+    | HDFSUpdateAuthSchema
 )
 
 

syncmaster/schemas/v1/transfers/__init__.py

@@ -7,6 +7,7 @@
 from syncmaster.schemas.v1.connections.connection import ReadConnectionSchema
 from syncmaster.schemas.v1.page import PageSchema
 from syncmaster.schemas.v1.transfers.db import (
+    ClickhouseReadTransferSourceAndTarget,
     HiveReadTransferSourceAndTarget,
     OracleReadTransferSourceAndTarget,
     PostgresReadTransferSourceAndTarget,
@@ -31,6 +32,7 @@
     | HDFSReadTransferSource
     | HiveReadTransferSourceAndTarget
     | OracleReadTransferSourceAndTarget
+    | ClickhouseReadTransferSourceAndTarget
     | S3ReadTransferSource
 )
 
@@ -39,6 +41,7 @@
     | HDFSReadTransferTarget
     | HiveReadTransferSourceAndTarget
     | OracleReadTransferSourceAndTarget
+    | ClickhouseReadTransferSourceAndTarget
     | S3ReadTransferTarget
 )
 
@@ -47,6 +50,7 @@
     | HDFSCreateTransferSource
     | HiveReadTransferSourceAndTarget
     | OracleReadTransferSourceAndTarget
+    | ClickhouseReadTransferSourceAndTarget
     | S3CreateTransferSource
 )
 
@@ -55,6 +59,7 @@
     | HDFSCreateTransferTarget
     | HiveReadTransferSourceAndTarget
     | OracleReadTransferSourceAndTarget
+    | ClickhouseReadTransferSourceAndTarget
     | S3CreateTransferTarget
 )
 
@@ -63,6 +68,7 @@
     | HDFSReadTransferSource
     | HiveReadTransferSourceAndTarget
     | OracleReadTransferSourceAndTarget
+    | ClickhouseReadTransferSourceAndTarget
     | S3CreateTransferSource
     | None
 )
@@ -72,6 +78,7 @@
     | HDFSReadTransferSource
     | HiveReadTransferSourceAndTarget
     | OracleReadTransferSourceAndTarget
+    | ClickhouseReadTransferSourceAndTarget
     | S3CreateTransferTarget
     | None
 )

syncmaster/schemas/v1/transfers/db.py

@@ -4,7 +4,12 @@
 
 from pydantic import BaseModel
 
-from syncmaster.schemas.v1.connection_types import HIVE_TYPE, ORACLE_TYPE, POSTGRES_TYPE
+from syncmaster.schemas.v1.connection_types import (
+    CLICKHOUSE_TYPE,
+    HIVE_TYPE,
+    ORACLE_TYPE,
+    POSTGRES_TYPE,
+)
 
 
 class ReadDBTransfer(BaseModel):
@@ -21,3 +26,7 @@ class OracleReadTransferSourceAndTarget(ReadDBTransfer):
 
 class PostgresReadTransferSourceAndTarget(ReadDBTransfer):
     type: POSTGRES_TYPE
+
+
+class ClickhouseReadTransferSourceAndTarget(ReadDBTransfer):
+    type: CLICKHOUSE_TYPE

tests/test_unit/test_connections/test_create_all_connection/test_create_clickhouse_connection.py

@@ -0,0 +1,143 @@
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from syncmaster.db.models import AuthData, Connection
+from syncmaster.db.repositories.utils import decrypt_auth_data
+from syncmaster.settings import Settings
+from tests.mocks import MockGroup, UserTestRoles
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.backend, pytest.mark.clickhouse]
+
+
+async def test_developer_plus_can_create_clickhouse_connection(
+    client: AsyncClient,
+    group: MockGroup,
+    session: AsyncSession,
+    settings: Settings,
+    role_developer_plus: UserTestRoles,
+):
+    # Arrange
+    user = group.get_member_of_role(role_developer_plus)
+
+    # Act
+    result = await client.post(
+        "v1/connections",
+        headers={"Authorization": f"Bearer {user.token}"},
+        json={
+            "group_id": group.id,
+            "name": "New connection",
+            "description": "",
+            "connection_data": {
+                "type": "clickhouse",
+                "host": "127.0.0.1",
+                "port": 1521,
+                "database": "database_name",
+            },
+            "auth_data": {
+                "type": "clickhouse",
+                "user": "user",
+                "password": "secret",
+            },
+        },
+    )
+    connection = (
+        await session.scalars(
+            select(Connection).filter_by(
+                name="New connection",
+            ),
+        )
+    ).first()
+
+    creds = (
+        await session.scalars(
+            select(AuthData).filter_by(
+                connection_id=connection.id,
+            ),
+        )
+    ).one()
+
+    # Assert
+    decrypted = decrypt_auth_data(creds.value, settings=settings)
+    assert result.status_code == 200
+    assert result.json() == {
+        "id": connection.id,
+        "name": connection.name,
+        "description": connection.description,
+        "group_id": connection.group_id,
+        "connection_data": {
+            "type": connection.data["type"],
+            "host": connection.data["host"],
+            "port": connection.data["port"],
+            "database": connection.data["database"],
+            "additional_params": connection.data["additional_params"],
+        },
+        "auth_data": {
+            "type": decrypted["type"],
+            "user": decrypted["user"],
+        },
+    }
+
+
+async def test_developer_plus_cannot_create_connection_with_type_mismatch(
+    client: AsyncClient,
+    group: MockGroup,
+    session: AsyncSession,
+    settings: Settings,
+    role_developer_plus: UserTestRoles,
+    event_loop,
+    request,
+):
+    # Arrange
+    user = group.get_member_of_role(role_developer_plus)
+
+    # Act
+    result = await client.post(
+        "v1/connections",
+        headers={"Authorization": f"Bearer {user.token}"},
+        json={
+            "group_id": group.id,
+            "name": "New connection",
+            "description": "",
+            "connection_data": {
+                "type": "postgres",
+                "host": "127.0.0.1",
+                "port": 5432,
+                "database_name": "postgres",
+            },
+            "auth_data": {
+                "type": "clickhouse",
+                "user": "user",
+                "password": "secret",
+            },
+        },
+    )
+
+    # Assert
+    assert result.json() == {
+        "error": {
+            "code": "invalid_request",
+            "message": "Invalid request",
+            "details": [
+                {
+                    "context": {},
+                    "input": {
+                        "group_id": group.id,
+                        "name": "New connection",
+                        "description": "",
+                        "connection_data": {
+                            "type": "postgres",
+                            "host": "127.0.0.1",
+                            "port": 5432,
+                            "database_name": "postgres",
+                        },
+                        "auth_data": {"type": "clickhouse", "user": "user", "password": "secret"},
+                    },
+                    "location": ["body"],
+                    "message": "Value error, Connection data and auth data must have same types",
+                    "code": "value_error",
+                },
+            ],
+        },
+    }