[DOP-22348] Add transformations for Transfers with dataframe row filtering

Author: Ilyas Gasanov

docs/changelog/next_release/183.feature.rst

@@ -0,0 +1 @@
+Add transformations for **Transfers** with dataframe row filtering

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

@@ -44,6 +44,7 @@ def upgrade():
         sa.Column("strategy_params", sa.JSON(), nullable=False),
         sa.Column("source_params", sa.JSON(), nullable=False),
         sa.Column("target_params", sa.JSON(), nullable=False),
+        sa.Column("transformations", sa.JSON(), nullable=False),
         sa.Column("is_scheduled", sa.Boolean(), nullable=False),
         sa.Column("schedule", sa.String(length=32), nullable=False),
         sa.Column("queue_id", sa.BigInteger(), nullable=False),

syncmaster/db/models/transfer.py

@@ -46,6 +46,7 @@ class Transfer(
     strategy_params: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default={})
     source_params: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default={})
     target_params: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default={})
+    transformations: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False, default=list)
     is_scheduled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
     schedule: Mapped[str] = mapped_column(String(32), nullable=False, default="")
     queue_id: Mapped[int] = mapped_column(

syncmaster/db/repositories/transfer.py

@@ -115,6 +115,7 @@ async def create(
         source_params: dict[str, Any],
         target_params: dict[str, Any],
         strategy_params: dict[str, Any],
+        transformations: list[dict[str, Any]],
         queue_id: int,
         is_scheduled: bool,
         schedule: str | None,
@@ -130,6 +131,7 @@ async def create(
                 source_params=source_params,
                 target_params=target_params,
                 strategy_params=strategy_params,
+                transformations=transformations,
                 queue_id=queue_id,
                 is_scheduled=is_scheduled,
                 schedule=schedule or "",
@@ -154,20 +156,27 @@ async def update(
         source_params: dict[str, Any],
         target_params: dict[str, Any],
         strategy_params: dict[str, Any],
+        transformations: list[dict[str, Any]],
         is_scheduled: bool | None,
         schedule: str | None,
         new_queue_id: int | None,
     ) -> Transfer:
         try:
-            for key in transfer.source_params:
-                if key not in source_params or source_params[key] is None:
-                    source_params[key] = transfer.source_params[key]
-            for key in transfer.target_params:
-                if key not in target_params or target_params[key] is None:
-                    target_params[key] = transfer.target_params[key]
-            for key in transfer.strategy_params:
-                if key not in strategy_params or strategy_params[key] is None:
-                    strategy_params[key] = transfer.strategy_params[key]
+            for old, new in [
+                (transfer.source_params, source_params),
+                (transfer.target_params, target_params),
+                (transfer.strategy_params, strategy_params),
+            ]:
+                for key in old:
+                    if key not in new or new[key] is None:
+                        new[key] = old[key]
+
+            new_transformations = {d["type"]: d["filters"] for d in transformations}
+            old_transformations = {d["type"]: d["filters"] for d in transfer.transformations}
+            for t_type, t_filters in new_transformations.items():
+                old_transformations[t_type] = t_filters
+            transformations = [{"type": t, "filters": f} for t, f in old_transformations.items()]
+
             return await self._update(
                 Transfer.id == transfer.id,
                 name=name or transfer.name,
@@ -179,6 +188,7 @@ async def update(
                 target_connection_id=target_connection_id or transfer.target_connection_id,
                 source_params=source_params,
                 target_params=target_params,
+                transformations=transformations,
                 queue_id=new_queue_id or transfer.queue_id,
             )
         except IntegrityError as e:

syncmaster/dto/transfers.py

@@ -15,6 +15,7 @@ class TransferDTO:
 @dataclass
 class DBTransferDTO(TransferDTO):
     table_name: str
+    transformations: list[dict] | None = None
 
 
 @dataclass
@@ -23,6 +24,7 @@ class FileTransferDTO(TransferDTO):
     file_format: CSV | JSONLine | JSON | Excel | XML | ORC | Parquet
     options: dict
     df_schema: dict | None = None
+    transformations: list[dict] | None = None
 
     _format_parsers = {
         "csv": CSV,

syncmaster/schemas/v1/connections/oracle.py

@@ -24,7 +24,7 @@ class CreateOracleConnectionDataSchema(BaseModel):
     additional_params: dict = Field(default_factory=dict)
 
     @model_validator(mode="before")
-    def check_owner_id(cls, values):
+    def validate_connection_identifiers(cls, values):
         sid, service_name = values.get("sid"), values.get("service_name")
         if sid and service_name:
             raise ValueError("You must specify either sid or service_name but not both")
@@ -47,7 +47,7 @@ class UpdateOracleConnectionDataSchema(BaseModel):
     additional_params: dict | None = Field(default_factory=dict)
 
     @model_validator(mode="before")
-    def check_owner_id(cls, values):
+    def validate_connection_identifiers(cls, values):
         sid, service_name = values.get("sid"), values.get("service_name")
         if sid and service_name:
             raise ValueError("You must specify either sid or service_name but not both")

syncmaster/schemas/v1/transfers/__init__.py

@@ -2,7 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
-from pydantic import BaseModel, Field, model_validator
+from typing import Annotated
+
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 from syncmaster.schemas.v1.connections.connection import ReadConnectionSchema
 from syncmaster.schemas.v1.page import PageSchema
@@ -27,6 +29,9 @@
     S3ReadTransferTarget,
 )
 from syncmaster.schemas.v1.transfers.strategy import FullStrategy, IncrementalStrategy
+from syncmaster.schemas.v1.transfers.transformations.dataframe_rows_filter import (
+    DataframeRowsFilter,
+)
 from syncmaster.schemas.v1.types import NameConstr
 
 ReadTransferSchemaSource = (
@@ -97,6 +102,8 @@
     | None
 )
 
+TransformationSchema = DataframeRowsFilter
+
 
 class CopyTransferSchema(BaseModel):
     new_group_id: int
@@ -129,6 +136,9 @@ class ReadTransferSchema(BaseModel):
         ...,
         discriminator="type",
     )
+    transformations: list[Annotated[TransformationSchema, Field(..., discriminator="type")]] = Field(
+        default_factory=list,
+    )
 
     class Config:
         from_attributes = True
@@ -158,15 +168,27 @@ class CreateTransferSchema(BaseModel):
         discriminator="type",
         description="Incremental or archive download options",
     )
+    transformations: list[
+        Annotated[TransformationSchema, Field(None, discriminator="type", description="List of transformations")]
+    ] = Field(default_factory=list)
 
     @model_validator(mode="before")
-    def check_owner_id(cls, values):
+    def validate_scheduling(cls, values):
         is_scheduled, schedule = values.get("is_scheduled"), values.get("schedule")
         if is_scheduled and schedule is None:
             # TODO make checking cron string
             raise ValueError("If transfer must be scheduled than set schedule param")
         return values
 
+    @field_validator("transformations", mode="after")
+    def validate_transformations_uniqueness(cls, transformations):
+        if transformations:
+            types = [tr.type for tr in transformations]
+            duplicates = {t for t in types if types.count(t) > 1}
+            if duplicates:
+                raise ValueError(f"Duplicate 'type' values found in transformations: {' '.join(map(str, duplicates))}")
+        return transformations
+
 
 class UpdateTransferSchema(BaseModel):
     source_connection_id: int | None = None
@@ -179,6 +201,7 @@ class UpdateTransferSchema(BaseModel):
     source_params: UpdateTransferSchemaSource = Field(discriminator="type", default=None)
     target_params: UpdateTransferSchemaTarget = Field(discriminator="type", default=None)
     strategy_params: FullStrategy | IncrementalStrategy | None = Field(discriminator="type", default=None)
+    transformations: list[Annotated[TransformationSchema, Field(discriminator="type", default=None)]] = None
 
 
 class ReadFullTransferSchema(ReadTransferSchema):

syncmaster/schemas/v1/transfers/transformations/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0

syncmaster/schemas/v1/transfers/transformations/dataframe_rows_filter.py

@@ -0,0 +1,86 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field
+
+from syncmaster.schemas.v1.transformation_types import DATAFRAME_ROWS_FILTER
+
+
+class BaseRowFilter(BaseModel):
+    field: str
+
+
+class EqualFilter(BaseRowFilter):
+    type: Literal["equal"]
+    value: str | None = None
+
+
+class NotEqualFilter(BaseRowFilter):
+    type: Literal["not_equal"]
+    value: str | None = None
+
+
+class GreaterThanFilter(BaseRowFilter):
+    type: Literal["greater_than"]
+    value: str
+
+
+class GreaterOrEqualFilter(BaseRowFilter):
+    type: Literal["greater_or_equal"]
+    value: str
+
+
+class LessThanFilter(BaseRowFilter):
+    type: Literal["less_than"]
+    value: str
+
+
+class LessOrEqualFilter(BaseRowFilter):
+    type: Literal["less_or_equal"]
+    value: str
+
+
+class LikeFilter(BaseRowFilter):
+    type: Literal["like"]
+    value: str
+
+
+class ILikeFilter(BaseRowFilter):
+    type: Literal["ilike"]
+    value: str
+
+
+class NotLikeFilter(BaseRowFilter):
+    type: Literal["not_like"]
+    value: str
+
+
+class NotILikeFilter(BaseRowFilter):
+    type: Literal["not_ilike"]
+    value: str
+
+
+class RegexpFilter(BaseRowFilter):
+    type: Literal["regexp"]
+    value: str
+
+
+RowFilter = (
+    EqualFilter
+    | NotEqualFilter
+    | GreaterThanFilter
+    | GreaterOrEqualFilter
+    | LessThanFilter
+    | LessOrEqualFilter
+    | LikeFilter
+    | ILikeFilter
+    | NotLikeFilter
+    | NotILikeFilter
+    | RegexpFilter
+)
+
+
+class DataframeRowsFilter(BaseModel):
+    type: DATAFRAME_ROWS_FILTER
+    filters: list[Annotated[RowFilter, Field(..., discriminator="type")]] = Field(default_factory=list)

syncmaster/schemas/v1/transformation_types.py

@@ -0,0 +1,5 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from typing import Literal
+
+DATAFRAME_ROWS_FILTER = Literal["dataframe_rows_filter"]