[DOP-22126] Add incremental strategy to API

Ilyas Gasanov · Ilyas Gasanov · commit ad204d168776 · 2025-02-14T14:19:00.000+03:00
diff --git a/syncmaster/schemas/v1/connection_types.py b/syncmaster/schemas/v1/connection_types.py
@@ -15,3 +15,21 @@
 FTPS_TYPE = Literal["ftps"]
 WEBDAV_TYPE = Literal["webdav"]
 SAMBA_TYPE = Literal["samba"]
+
+CONNECTION_TYPES = Literal[
+    "oracle",
+    "postgres",
+    "clickhouse",
+    "hive",
+    "mssql",
+    "mysql",
+    "s3",
+    "hdfs",
+    "sftp",
+    "ftp",
+    "ftps",
+    "webdav",
+    "samba",
+]
+FILE_CONNECTION_TYPES = Literal["s3", "hdfs", "sftp", "ftp", "ftps", "webdav", "samba"]
+DB_CONNECTION_TYPES = Literal["oracle", "postgres", "clickhouse", "hive", "mssql", "mysql"]
diff --git a/syncmaster/schemas/v1/transfers/__init__.py b/syncmaster/schemas/v1/transfers/__init__.py
@@ -2,12 +2,14 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
-from typing import Annotated
+from typing import Annotated, Literal, get_args
 
-from pydantic import BaseModel, Field, field_validator, model_validator
+from pydantic import BaseModel, Field, model_validator
 
+from syncmaster.schemas.v1.connection_types import FILE_CONNECTION_TYPES
 from syncmaster.schemas.v1.connections.connection import ReadConnectionSchema
 from syncmaster.schemas.v1.page import PageSchema
+from syncmaster.schemas.v1.transfer_types import INCREMENTAL_TYPE
 from syncmaster.schemas.v1.transfers.db import (
     ClickhouseReadTransferSourceAndTarget,
     HiveReadTransferSourceAndTarget,
@@ -246,6 +248,19 @@ def validate_scheduling(cls, values):
             raise ValueError("If transfer must be scheduled than set schedule param")
         return values
 
+    @model_validator(mode="after")
+    def validate_increment_by(cls, values):
+        if not isinstance(values.strategy_params, IncrementalStrategy):
+            return values
+
+        source_type = values.source_params.type
+        increment_by = values.strategy_params.increment_by
+
+        if source_type in get_args(FILE_CONNECTION_TYPES) and increment_by != "modified_since":
+            raise ValueError("Field 'increment_by' must be equal to 'modified_since' for file source types")
+
+        return values
+
 
 class UpdateTransferSchema(BaseModel):
     source_connection_id: int | None = None
diff --git a/syncmaster/schemas/v1/transfers/strategy.py b/syncmaster/schemas/v1/transfers/strategy.py
@@ -13,3 +13,4 @@ class FullStrategy(BaseModel):
 
 class IncrementalStrategy(BaseModel):
     type: INCREMENTAL_TYPE
+    increment_by: str
diff --git a/syncmaster/server/api/v1/connections.py b/syncmaster/server/api/v1/connections.py
@@ -17,16 +17,7 @@
 )
 from syncmaster.exceptions.credentials import AuthDataNotFoundError
 from syncmaster.exceptions.group import GroupNotFoundError
-from syncmaster.schemas.v1.connection_types import (
-    CLICKHOUSE_TYPE,
-    HDFS_TYPE,
-    HIVE_TYPE,
-    MSSQL_TYPE,
-    MYSQL_TYPE,
-    ORACLE_TYPE,
-    POSTGRES_TYPE,
-    S3_TYPE,
-)
+from syncmaster.schemas.v1.connection_types import CONNECTION_TYPES
 from syncmaster.schemas.v1.connections.connection import (
     ConnectionCopySchema,
     ConnectionPageSchema,
@@ -41,8 +32,6 @@
 
 router = APIRouter(tags=["Connections"], responses=get_error_responses())
 
-CONNECTION_TYPES = ORACLE_TYPE, POSTGRES_TYPE, CLICKHOUSE_TYPE, HIVE_TYPE, MSSQL_TYPE, MYSQL_TYPE, S3_TYPE, HDFS_TYPE
-
 
 @router.get("/connections")
 async def read_connections(
@@ -157,7 +146,7 @@ async def create_connection(
 
 @router.get("/connections/known_types", dependencies=[Depends(get_user(is_active=True))])
 async def read_connection_types() -> list[str]:
-    return [get_args(type_)[0] for type_ in CONNECTION_TYPES]
+    return list(get_args(CONNECTION_TYPES))
 
 
 @router.get("/connections/{connection_id}")
diff --git a/tests/test_unit/test_connections/test_read_connection_types.py b/tests/test_unit/test_connections/test_read_connection_types.py
@@ -3,7 +3,7 @@
 import pytest
 from httpx import AsyncClient
 
-from syncmaster.server.api.v1.connections import CONNECTION_TYPES
+from syncmaster.schemas.v1.connection_types import CONNECTION_TYPES
 from tests.mocks import MockUser
 
 pytestmark = [pytest.mark.asyncio, pytest.mark.server]
@@ -31,4 +31,4 @@ async def test_groupless_user_can_read_connection_types(client: AsyncClient, sim
     )
     # Assert
     assert result.status_code == 200
-    assert set(result.json()) == {get_args(type)[0] for type in CONNECTION_TYPES}
+    assert set(result.json()) == set(get_args(CONNECTION_TYPES))
diff --git a/tests/test_unit/test_transfers/test_create_transfer.py b/tests/test_unit/test_transfers/test_create_transfer.py
@@ -48,7 +48,7 @@ async def test_developer_plus_can_create_transfer(
                     "type": "csv",
                 },
             },
-            "strategy_params": {"type": "full"},
+            "strategy_params": {"type": "incremental", "increment_by": "modified_since"},
             "transformations": [
                 {
                     "type": "dataframe_rows_filter",
@@ -394,19 +394,6 @@ async def test_superuser_can_create_transfer(
                             "message": "Value error, If transfer must be scheduled than set schedule param",
                             "code": "value_error",
                             "context": {},
-                            "input": {
-                                "description": "",
-                                "group_id": 1,
-                                "is_scheduled": True,
-                                "name": "new test transfer",
-                                "queue_id": 1,
-                                "schedule": None,
-                                "source_connection_id": 1,
-                                "source_params": {"table_name": "source_table", "type": "postgres"},
-                                "strategy_params": {"type": "full"},
-                                "target_connection_id": 2,
-                                "target_params": {"table_name": "target_table", "type": "postgres"},
-                            },
                         },
                     ],
                 },
@@ -439,6 +426,37 @@ async def test_superuser_can_create_transfer(
                 },
             },
         ),
+        (
+            {
+                "source_params": {
+                    "type": "ftp",
+                    "directory_path": "/source_path",
+                    "file_format": {
+                        "type": "csv",
+                    },
+                },
+                "strategy_params": {
+                    "type": "incremental",
+                    "increment_by": "unknown",
+                },
+            },
+            {
+                "error": {
+                    "code": "invalid_request",
+                    "message": "Invalid request",
+                    "details": [
+                        {
+                            "location": ["body"],
+                            "message": (
+                                "Value error, Field 'increment_by' must be equal to 'modified_since' for file source types"
+                            ),
+                            "code": "value_error",
+                            "context": {},
+                        },
+                    ],
+                },
+            },
+        ),
         (
             {
                 "source_params": {
@@ -759,7 +777,9 @@ async def test_check_fields_validation_on_create_transfer(
     # Assert
     assert result.status_code == 422
 
-    if new_data == {"schedule": None}:
+    if (new_data == {"schedule": None}) or (
+        "strategy_params" in new_data and new_data["strategy_params"].get("increment_by") == "unknown"
+    ):
         error_json["error"]["details"][0]["input"] = transfer_data
 
     assert result.json() == error_json
diff --git a/tests/test_unit/test_transfers/test_file_transfers/test_read_transfer.py b/tests/test_unit/test_transfers/test_file_transfers/test_read_transfer.py
@@ -104,6 +104,10 @@
             "target_params": {
                 "file_name_template": "{run_created_at}_{index}.{extension}",
             },
+            "strategy_params": {
+                "type": "incremental",
+                "increment_by": "modified_since",
+            },
         },
     ],
 )
diff --git a/tests/test_unit/test_transfers/test_file_transfers/test_update_transfer.py b/tests/test_unit/test_transfers/test_file_transfers/test_update_transfer.py
@@ -145,54 +145,44 @@ async def test_developer_plus_can_update_s3_transfer(
 ):
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_developer_plus)
-    transformations = [
-        {
-            "type": "dataframe_rows_filter",
-            "filters": [
-                {
-                    "type": "is_not_null",
-                    "field": "col2",
-                },
-            ],
+    updated_fields = {
+        "source_params": {
+            "type": "ftp",
+            "directory_path": "/some/new/test/directory",
+            "file_format": create_transfer_data["source_and_target_params"]["file_format"],
+            "options": {"some": "option"},
         },
-    ]
+        "target_params": {
+            "type": "ftp",
+            "directory_path": "/some/new/test/directory",
+            "file_format": create_transfer_data["source_and_target_params"]["file_format"],
+            "file_name_template": "{index}.{extension}",
+            "options": {"some": "option"},
+        },
+        "strategy_params": {
+            "type": "incremental",
+            "increment_by": "modified_since",
+        },
+        "transformations": [
+            {
+                "type": "dataframe_rows_filter",
+                "filters": [
+                    {
+                        "type": "is_not_null",
+                        "field": "col2",
+                    },
+                ],
+            },
+        ],
+    }
 
     # Act
     result = await client.patch(
         f"v1/transfers/{group_transfer.id}",
         headers={"Authorization": f"Bearer {user.token}"},
-        json={
-            "source_params": {
-                "type": "ftp",
-                "directory_path": "/some/new/test/directory",
-                "file_format": create_transfer_data["source_and_target_params"]["file_format"],
-                "options": {"some": "option"},
-            },
-            "target_params": {
-                "type": "ftp",
-                "directory_path": "/some/new/test/directory",
-                "file_format": create_transfer_data["source_and_target_params"]["file_format"],
-                "file_name_template": "{index}.{extension}",
-                "options": {"some": "option"},
-            },
-            "transformations": transformations,
-        },
+        json=updated_fields,
     )
 
-    # Pre-Assert
-    source_params = group_transfer.source_params.copy()
-    source_params.update(
-        {
-            "directory_path": "/some/new/test/directory",
-            "file_format": create_transfer_data["source_and_target_params"]["file_format"],
-            "options": {"some": "option"},
-        },
-    )
-    target_params = {
-        **source_params,
-        "file_name_template": "{index}.{extension}",
-    }
-
     # Assert
     assert result.status_code == 200
     assert result.json() == {
@@ -204,9 +194,9 @@ async def test_developer_plus_can_update_s3_transfer(
         "is_scheduled": group_transfer.is_scheduled,
         "source_connection_id": group_transfer.source_connection_id,
         "target_connection_id": group_transfer.target_connection_id,
-        "source_params": source_params,
-        "target_params": target_params,
-        "strategy_params": group_transfer.strategy_params,
-        "transformations": transformations,
+        "source_params": updated_fields["source_params"],
+        "target_params": updated_fields["target_params"],
+        "strategy_params": updated_fields["strategy_params"],
+        "transformations": updated_fields["transformations"],
         "queue_id": group_transfer.transfer.queue_id,
     }
diff --git a/tests/test_unit/test_transfers/test_update_transfer.py b/tests/test_unit/test_transfers/test_update_transfer.py
@@ -14,28 +14,36 @@ async def test_developer_plus_can_update_transfer(
 ):
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_developer_plus)
+    updated_fields = {
+        "name": "New transfer name",
+        "is_scheduled": False,
+        "strategy_params": {
+            "type": "incremental",
+            "increment_by": "updated_at",
+        },
+    }
 
     # Act
     result = await client.patch(
         f"v1/transfers/{group_transfer.id}",
         headers={"Authorization": f"Bearer {user.token}"},
-        json={"name": "New transfer name", "is_scheduled": False},
+        json=updated_fields,
     )
 
     # Assert
     assert result.status_code == 200
     assert result.json() == {
         "id": group_transfer.id,
         "group_id": group_transfer.group_id,
-        "name": "New transfer name",
+        "name": updated_fields["name"],
         "description": group_transfer.description,
         "schedule": group_transfer.schedule,
-        "is_scheduled": False,
+        "is_scheduled": updated_fields["is_scheduled"],
         "source_connection_id": group_transfer.source_connection_id,
         "target_connection_id": group_transfer.target_connection_id,
         "source_params": group_transfer.source_params,
         "target_params": group_transfer.target_params,
-        "strategy_params": group_transfer.strategy_params,
+        "strategy_params": updated_fields["strategy_params"],
         "transformations": group_transfer.transformations,
         "queue_id": group_transfer.transfer.queue_id,
     }

Original file line number	Diff line number	Diff line change
`@@ -13,3 +13,4 @@ class FullStrategy(BaseModel):`
`13`	`13`
`14`	`14`	`class IncrementalStrategy(BaseModel):`
`15`	`15`	`type: INCREMENTAL_TYPE`
	`16`	`+ increment_by: str`