[DOP-22126] Add incremental strategy to API (#202)

Author: IlyasDevelopment

docs/changelog/next_release/202.feature.rst

@@ -0,0 +1 @@
+Add `increment_by` field to `strategy_params`

syncmaster/schemas/v1/connection_types.py

@@ -15,3 +15,21 @@
 FTPS_TYPE = Literal["ftps"]
 WEBDAV_TYPE = Literal["webdav"]
 SAMBA_TYPE = Literal["samba"]
+
+CONNECTION_TYPES = [
+    "oracle",
+    "postgres",
+    "clickhouse",
+    "hive",
+    "mssql",
+    "mysql",
+    "s3",
+    "hdfs",
+    "sftp",
+    "ftp",
+    "ftps",
+    "webdav",
+    "samba",
+]
+FILE_CONNECTION_TYPES = ["s3", "hdfs", "sftp", "ftp", "ftps", "webdav", "samba"]
+DB_CONNECTION_TYPES = ["oracle", "postgres", "clickhouse", "hive", "mssql", "mysql"]

syncmaster/schemas/v1/transfers/__init__.py

@@ -4,8 +4,9 @@
 
 from typing import Annotated
 
-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.transfers.db import (
@@ -246,6 +247,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 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

syncmaster/schemas/v1/transfers/strategy.py

@@ -13,3 +13,4 @@ class FullStrategy(BaseModel):
 
 class IncrementalStrategy(BaseModel):
     type: INCREMENTAL_TYPE
+    increment_by: str

syncmaster/server/api/v1/connections.py

@@ -1,7 +1,6 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
 from collections.abc import Sequence
-from typing import get_args
 
 from fastapi import APIRouter, Depends, Query, status
 from pydantic import TypeAdapter
@@ -17,16 +16,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 +31,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 +145,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 CONNECTION_TYPES
 
 
 @router.get("/connections/{connection_id}")

tests/test_unit/test_connections/test_read_connection_types.py

@@ -1,9 +1,7 @@
-from typing import get_args
-
 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 +29,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(CONNECTION_TYPES)

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",
@@ -325,6 +325,7 @@ async def test_superuser_can_create_transfer(
     }
 
 
+# TODO: refactor annotations & fixtures
 @pytest.mark.parametrize(
     argnames=["new_data", "error_json"],
     argvalues=(
@@ -394,19 +395,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 +427,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 +778,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

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",
+            },
         },
     ],
 )

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,
     }

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,
     }