[DOP-22348] Add transformations for Transfers with dataframe row filtering (#184)

Author: IlyasDevelopment

docs/changelog/next_release/184.feature.rst

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

poetry.lock

@@ -131,6 +131,7 @@ pytest-asyncio = "^0.25.1"
 pytest-randomly = "^3.15.0"
 pytest-deadfixtures = "^2.2.1"
 pytest-mock = "^3.14.0"
+pytest-lazy-fixtures = "^1.1.1"
 onetl = {extras = ["spark", "s3", "hdfs"], version = "^0.12.0"}
 faker = "^33.3.0"
 coverage = "^7.6.1"

pyproject.toml

@@ -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/migrations/versions/2023-11-23_0007_create_transfer_table.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/models/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,21 @@ 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]
+
             return await self._update(
                 Transfer.id == transfer.id,
                 name=name or transfer.name,
@@ -179,6 +182,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 or transfer.transformations,
                 queue_id=new_queue_id or transfer.queue_id,
             )
         except IntegrityError as e:

syncmaster/db/repositories/transfer.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/dto/transfers.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/connections/oracle.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,9 +168,12 @@ 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
@@ -179,6 +192,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/__init__.py

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