[DOP-21832] Move Connection type field to the root of the model

Author: Ilyas Gasanov

docs/changelog/next_release/169.breaking.rst

@@ -0,0 +1,2 @@
+- Moved the `type` field from within `connection_data` to the root level of the `Connection` model.
+- Decoupled `auth_data.type` from the connection type and linked it to the authentication type instead.

docs/design/entities.rst

@@ -52,15 +52,15 @@ Example:
         "group_id": "1",
         "name": "Beautiful name",
         "description": "What a great connection !",
+        "type": "postgres",
         "connection_data": {
-            "type": "postgres",
             "host": "127.0.0.1",
             "port": 5432,
             "database_name": "postgres",
             "additional_params": {},
         },
         "auth_data": {
-            "type": "postgres",
+            "type": "basic",
             "user": "user_name",
             "password": "password",
         }

syncmaster/backend/api/v1/connections.py

@@ -4,9 +4,10 @@
 from typing import get_args
 
 from fastapi import APIRouter, Depends, Query, status
+from pydantic import TypeAdapter
 
 from syncmaster.backend.services import UnitOfWork, get_user
-from syncmaster.db.models import Connection, Transfer, User
+from syncmaster.db.models import Connection, ConnectionType, Transfer, User
 from syncmaster.db.utils import Permission
 from syncmaster.errors.registration import get_error_responses
 from syncmaster.exceptions import ActionNotAllowedError
@@ -26,7 +27,6 @@
     ORACLE_TYPE,
     POSTGRES_TYPE,
     S3_TYPE,
-    ConnectionType,
 )
 from syncmaster.schemas.v1.connections.connection import (
     ConnectionCopySchema,
@@ -80,13 +80,16 @@ async def read_connections(
     if pagination.items:
         credentials = await unit_of_work.credentials.read_bulk([item.id for item in pagination.items])
         items = [
-            ReadConnectionSchema(
-                id=item.id,
-                group_id=item.group_id,
-                name=item.name,
-                description=item.description,
-                auth_data=credentials.get(item.id, None),
-                data=item.data,
+            TypeAdapter(ReadConnectionSchema).validate_python(
+                {
+                    "id": item.id,
+                    "group_id": item.group_id,
+                    "name": item.name,
+                    "description": item.description,
+                    "type": item.type,
+                    "data": item.data,
+                    "auth_data": credentials.get(item.id, None),
+                },
             )
             for item in pagination.items
         ]
@@ -126,6 +129,7 @@ async def create_connection(
     async with unit_of_work:
         connection = await unit_of_work.connection.create(
             name=connection_data.name,
+            type=connection_data.type,
             description=connection_data.description,
             group_id=connection_data.group_id,
             data=connection_data.data.dict(),
@@ -137,13 +141,16 @@ async def create_connection(
         )
 
     credentials = await unit_of_work.credentials.read(connection.id)
-    return ReadConnectionSchema(
-        id=connection.id,
-        group_id=connection.group_id,
-        name=connection.name,
-        description=connection.description,
-        data=connection.data,
-        auth_data=credentials,
+    return TypeAdapter(ReadConnectionSchema).validate_python(
+        {
+            "id": connection.id,
+            "group_id": connection.group_id,
+            "name": connection.name,
+            "description": connection.description,
+            "type": connection.type,
+            "data": connection.data,
+            "auth_data": credentials,
+        },
     )
 
 
@@ -172,13 +179,16 @@ async def read_connection(
     except AuthDataNotFoundError:
         credentials = None
 
-    return ReadConnectionSchema(
-        id=connection.id,
-        group_id=connection.group_id,
-        name=connection.name,
-        description=connection.description,
-        data=connection.data,
-        auth_data=credentials,
+    return TypeAdapter(ReadConnectionSchema).validate_python(
+        {
+            "id": connection.id,
+            "group_id": connection.group_id,
+            "name": connection.name,
+            "description": connection.description,
+            "type": connection.type,
+            "data": connection.data,
+            "auth_data": credentials,
+        },
     )
 
 
@@ -202,15 +212,15 @@ async def update_connection(
 
     async with unit_of_work:
         data = changes.data.dict(exclude={"auth_data"}) if changes.data else {}
-        if data.get("type", None) is not None:
-            source_connection: Connection = await unit_of_work.connection.read_by_id(connection_id=connection_id)
-            if data["type"] != source_connection.data["type"]:
-                linked_transfers: Sequence[Transfer] = await unit_of_work.transfer.list_by_connection_id(connection_id)
-                if linked_transfers:
-                    raise ConnectionTypeUpdateError
+        source_connection: Connection = await unit_of_work.connection.read_by_id(connection_id=connection_id)
+        if changes.type != source_connection.type:
+            linked_transfers: Sequence[Transfer] = await unit_of_work.transfer.list_by_connection_id(connection_id)
+            if linked_transfers:
+                raise ConnectionTypeUpdateError
         connection = await unit_of_work.connection.update(
             connection_id=connection_id,
             name=changes.name,
+            type=changes.type,
             description=changes.description,
             data=data,
         )
@@ -222,13 +232,16 @@ async def update_connection(
             )
 
     credentials = await unit_of_work.credentials.read(connection_id)
-    return ReadConnectionSchema(
-        id=connection.id,
-        group_id=connection.group_id,
-        name=connection.name,
-        description=connection.description,
-        data=connection.data,
-        auth_data=credentials,
+    return TypeAdapter(ReadConnectionSchema).validate_python(
+        {
+            "id": connection.id,
+            "group_id": connection.group_id,
+            "name": connection.name,
+            "description": connection.description,
+            "type": connection.type,
+            "data": connection.data,
+            "auth_data": credentials,
+        },
     )
 
 

syncmaster/backend/api/v1/runs.py

@@ -19,7 +19,7 @@
 from syncmaster.exceptions.base import ActionNotAllowedError
 from syncmaster.exceptions.run import CannotConnectToTaskQueueError
 from syncmaster.exceptions.transfer import TransferNotFoundError
-from syncmaster.schemas.v1.connections.connection import ReadAuthDataSchema
+from syncmaster.schemas.v1.connections.connection_base import ReadAuthDataSchema
 from syncmaster.schemas.v1.transfers.run import (
     CreateRunSchema,
     ReadRunSchema,

syncmaster/backend/api/v1/transfers.py

@@ -4,7 +4,7 @@
 from fastapi import APIRouter, Depends, Query, status
 
 from syncmaster.backend.services import UnitOfWork, get_user
-from syncmaster.db.models import User
+from syncmaster.db.models import ConnectionType, User
 from syncmaster.db.utils import Permission
 from syncmaster.errors.registration import get_error_responses
 from syncmaster.exceptions.base import ActionNotAllowedError
@@ -16,7 +16,6 @@
     DifferentTypeConnectionsAndParamsError,
     TransferNotFoundError,
 )
-from syncmaster.schemas.v1.connection_types import ConnectionType
 from syncmaster.schemas.v1.status import (
     StatusCopyTransferResponseSchema,
     StatusResponseSchema,
@@ -103,16 +102,16 @@ async def create_transfer(
     ):
         raise DifferentTransferAndConnectionsGroupsError
 
-    if target_connection.data["type"] != transfer_data.target_params.type:
+    if target_connection.type != transfer_data.target_params.type:
         raise DifferentTypeConnectionsAndParamsError(
-            connection_type=target_connection.data["type"],
+            connection_type=target_connection.type,
             conn="target",
             params_type=transfer_data.target_params.type,
         )
 
-    if source_connection.data["type"] != transfer_data.source_params.type:
+    if source_connection.type != transfer_data.source_params.type:
         raise DifferentTypeConnectionsAndParamsError(
-            connection_type=source_connection.data["type"],
+            connection_type=source_connection.type,
             conn="source",
             params_type=transfer_data.source_params.type,
         )
@@ -302,16 +301,16 @@ async def update_transfer(
     if queue.group_id != transfer.group_id:
         raise DifferentTransferAndQueueGroupError
 
-    if transfer_data.target_params and target_connection.data["type"] != transfer_data.target_params.type:
+    if transfer_data.target_params and target_connection.type != transfer_data.target_params.type:
         raise DifferentTypeConnectionsAndParamsError(
-            connection_type=target_connection.data["type"],
+            connection_type=target_connection.type,
             conn="target",
             params_type=transfer_data.target_params.type,
         )
 
-    if transfer_data.source_params and source_connection.data["type"] != transfer_data.source_params.type:
+    if transfer_data.source_params and source_connection.type != transfer_data.source_params.type:
         raise DifferentTypeConnectionsAndParamsError(
-            connection_type=source_connection.data["type"],
+            connection_type=source_connection.type,
             conn="source",
             params_type=transfer_data.source_params.type,
         )

syncmaster/db/migrations/versions/2023-11-23_0004_create_connection_table.py

@@ -30,6 +30,7 @@ def upgrade():
         "connection",
         sa.Column("id", sa.BigInteger(), nullable=False),
         sa.Column("group_id", sa.BigInteger(), nullable=False),
+        sa.Column("type", sa.String(length=128), nullable=False),
         sa.Column("name", sa.String(length=128), nullable=False),
         sa.Column("description", sa.String(length=512), nullable=False),
         sa.Column("data", sa.JSON(), nullable=False),

syncmaster/db/models/__init__.py

@@ -3,7 +3,7 @@
 from syncmaster.db.models.apscheduler_job import APSchedulerJob
 from syncmaster.db.models.auth_data import AuthData
 from syncmaster.db.models.base import Base
-from syncmaster.db.models.connection import Connection
+from syncmaster.db.models.connection import Connection, ConnectionType
 from syncmaster.db.models.group import Group, GroupMemberRole, UserGroup
 from syncmaster.db.models.queue import Queue
 from syncmaster.db.models.run import Run, RunType, Status

syncmaster/db/models/connection.py

@@ -2,20 +2,33 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
+from enum import StrEnum
 from typing import Any
 
 from sqlalchemy import JSON, Computed, Index, UniqueConstraint
 from sqlalchemy.dialects.postgresql import TSVECTOR
 from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship
+from sqlalchemy_utils import ChoiceType
 
 from syncmaster.db.mixins import DeletableMixin, ResourceMixin, TimestampMixin
 from syncmaster.db.models.base import Base
 from syncmaster.db.models.group import Group
 
 
+class ConnectionType(StrEnum):
+    POSTGRES = "postgres"
+    HIVE = "hive"
+    ORACLE = "oracle"
+    CLICKHOUSE = "clickhouse"
+    MSSQL = "mssql"
+    MYSQL = "mysql"
+    S3 = "s3"
+    HDFS = "hdfs"
+
+
 class Connection(Base, ResourceMixin, DeletableMixin, TimestampMixin):
+    type: Mapped[ConnectionType] = mapped_column(ChoiceType(ConnectionType), nullable=False)
     data: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default={})
-
     group: Mapped[Group] = relationship("Group")
 
     search_vector: Mapped[str] = mapped_column(

syncmaster/db/repositories/connection.py

@@ -41,7 +41,7 @@ async def paginate(
             stmt = self._construct_vector_search(stmt, combined_query)
 
         if connection_type is not None:
-            stmt = stmt.where(Connection.data.op("->>")("type").in_(connection_type))
+            stmt = stmt.where(Connection.type.in_(connection_type))
 
         return await self._paginate_scalar_result(
             query=stmt.order_by(Connection.name),
@@ -63,6 +63,7 @@ async def read_by_id(
     async def create(
         self,
         group_id: int,
+        type: str,
         name: str,
         description: str,
         data: dict[str, Any],
@@ -71,6 +72,7 @@ async def create(
             insert(Connection)
             .values(
                 group_id=group_id,
+                type=type,
                 name=name,
                 description=description,
                 data=data,
@@ -89,6 +91,7 @@ async def update(
         self,
         connection_id: int,
         name: str | None,
+        type: str | None,
         description: str | None,
         data: dict[str, Any],
     ) -> Connection:
@@ -100,6 +103,7 @@ async def update(
             return await self._update(
                 Connection.id == connection_id,
                 Connection.is_deleted.is_(False),
+                type=type or connection.type,
                 name=name or connection.name,
                 description=description or connection.description,
                 data=data,

syncmaster/db/repositories/transfer.py

@@ -70,7 +70,7 @@ async def paginate(
                 Transfer.source_connection_id == SourceConnection.id,
             )
             stmt = stmt.where(
-                SourceConnection.data.op("->>")("type").in_(source_connection_type),
+                SourceConnection.type.in_(source_connection_type),
             )
 
         if target_connection_type is not None:
@@ -79,7 +79,7 @@ async def paginate(
                 Transfer.target_connection_id == TargetConnection.id,
             )
             stmt = stmt.where(
-                TargetConnection.data.op("->>")("type").in_(target_connection_type),
+                TargetConnection.type.in_(target_connection_type),
             )
 
         return await self._paginate_scalar_result(