Repository: apache/airflow
File: providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py (Python)

Code before:

    @cached_property
    def destination_hook(self) -> DbApiHook:
        return self.get_hook(conn_id=self.destination_conn_id, hook_params=self.destination_hook_params)

    def get_paginated_sql(self, offset: int) -> str:
        """Format the paginated SQL statement using the current format."""
        return self.paginated_sql_statement_clause.format(self.sql, self.page_size, offset)

    def render_template_fields(
        self,
        context: Context,
        jinja_env: jinja2.Environment | None = None,
    ) -> None:
        super().render_template_fields(context=context, jinja_env=jinja_env)

        # Make sure string are converted to integers
        if isinstance(self.page_size, str):
            self.page_size = int(self.page_size)
        commit_every = self.insert_args.get("commit_every")
        if isinstance(commit_every, str):
            self.insert_args["commit_every"] = int(commit_every)

    def _insert_rows(self, rows: list[Any], context: Context):
        if self._rows_processor:
            rows = self._rows_processor(rows, **context)  # type: ignore

        self.log.info("Inserting %d rows into %s", len(rows), self.destination_conn_id)

        self.destination_hook.insert_rows(table=self.destination_table, rows=rows, **self.insert_args)

    def execute(self, context: Context):
        if self.preoperator:
            self.log.info("Running preoperator")
            self.log.info(self.preoperator)
            self.destination_hook.run(self.preoperator)

        if self.page_size and isinstance(self.sql, str):
            self.defer(
                trigger=SQLExecuteQueryTrigger(
                    conn_id=self.source_conn_id,
                    hook_params=self.source_hook_params,
                    sql=self.get_paginated_sql(0),
                ),
                method_name=self.execute_complete.__name__,
            )
        else:
            if isinstance(self.sql, str):
                self.sql = [self.sql]

            self.log.info("Extracting data from %s", self.source_conn_id)

Review comment (performance):
Is there a specific rule ignored here? If so, I think it would be better to ignore only it, rather than everything, I'm guessing it is there for the rows parameter, which I think can be fixed by changing the return of the rows processor to be of type list[Any] rather than just Any.

In addition, I think it is a good idea to document the method a little better, as I had to read the code of the implementation to understand that it is applied on all rows in a batch rather than one by one, I think fixing the type hinting to
`Callable[[list[Any], Mapping[str, Any]], list[Any]]`
(Or even better, instead of the mapping use unpack for the airflow context type) should suffice, it would also remove the need for the type ignore.

Code after:

    @cached_property
    def source_hook(self) -> DbApiHook:
        return self.get_hook(conn_id=self.source_conn_id, hook_params=self.source_hook_params)

    @cached_property
    def destination_hook(self) -> DbApiHook:
        return self.get_hook(conn_id=self.destination_conn_id, hook_params=self.destination_hook_params)

    def get_paginated_sql(self, offset: int) -> str:
        """Format the paginated SQL statement using the current format."""
        return self.paginated_sql_statement_clause.format(self.sql, self.page_size, offset)

    def render_template_fields(
        self,
        context: Context,
        jinja_env: jinja2.Environment | None = None,
    ) -> None:
        super().render_template_fields(context=context, jinja_env=jinja_env)

        # Make sure string are converted to integers
        if isinstance(self.page_size, str):
            self.page_size = int(self.page_size)
        commit_every = self.insert_args.get("commit_every")
        if isinstance(commit_every, str):
            self.insert_args["commit_every"] = int(commit_every)

    def _insert_rows(self, rows: list[Any], context: Context):
        if self._rows_processor:
            rows = self._rows_processor(rows, **context)

        self.log.info("Inserting %d rows into %s", len(rows), self.destination_conn_id)

        self.destination_hook.insert_rows(table=self.destination_table, rows=rows, **self.insert_args)

    def execute(self, context: Context):
        if self.preoperator:
            self.log.info("Running preoperator")
            self.log.info(self.preoperator)
            self.destination_hook.run(self.preoperator)

        if self.page_size and isinstance(self.sql, str):
            self.defer(
                trigger=SQLExecuteQueryTrigger(
                    conn_id=self.source_conn_id,
                    hook_params=self.source_hook_params,
                    sql=self.get_paginated_sql(0),
                ),
                method_name=self.execute_complete.__name__,
            )
        else: