refactor: Make ArrowDataFrame compliant

Author: dangotbanned

narwhals/_arrow/dataframe.py

@@ -5,6 +5,7 @@
 from typing import Any
 from typing import Iterator
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 from typing import cast
 from typing import overload
@@ -26,6 +27,7 @@
 from narwhals.utils import check_column_names_are_unique
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import is_sequence_but_not_str
+from narwhals.utils import not_implemented
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
 from narwhals.utils import scale_bytes
@@ -449,29 +451,17 @@ def join(
             ),
         )
 
-    def join_asof(
-        self: Self,
-        other: Self,
-        *,
-        left_on: str | None,
-        right_on: str | None,
-        by_left: list[str] | None,
-        by_right: list[str] | None,
-        strategy: Literal["backward", "forward", "nearest"],
-        suffix: str,
-    ) -> Self:
-        msg = "join_asof is not yet supported on PyArrow tables"  # pragma: no cover
-        raise NotImplementedError(msg)
+    join_asof = not_implemented()
 
-    def drop(self: Self, columns: list[str], strict: bool) -> Self:  # noqa: FBT001
+    def drop(self: Self, columns: Sequence[str], strict: bool) -> Self:  # noqa: FBT001
         to_drop = parse_columns_to_drop(
             compliant_frame=self, columns=columns, strict=strict
         )
         return self._from_native_frame(
             self._native_frame.drop(to_drop), validate_column_names=False
         )
 
-    def drop_nulls(self: ArrowDataFrame, subset: list[str] | None) -> ArrowDataFrame:
+    def drop_nulls(self: ArrowDataFrame, subset: Sequence[str] | None) -> ArrowDataFrame:
         if subset is None:
             return self._from_native_frame(
                 self._native_frame.drop_null(), validate_column_names=False
@@ -667,9 +657,7 @@ def collect(
         msg = f"Unsupported `backend` value: {backend}"  # pragma: no cover
         raise AssertionError(msg)  # pragma: no cover
 
-    def clone(self: Self) -> Self:
-        msg = "clone is not yet supported on PyArrow tables"
-        raise NotImplementedError(msg)
+    clone = not_implemented()
 
     def item(self: Self, row: int | None, column: int | str | None) -> Any:
         from narwhals._arrow.series import maybe_extract_py_scalar
@@ -695,7 +683,7 @@ def item(self: Self, row: int | None, column: int | str | None) -> Any:
             self._native_frame[_col][row], return_py_scalar=True
         )
 
-    def rename(self: Self, mapping: dict[str, str]) -> Self:
+    def rename(self: Self, mapping: Mapping[str, str]) -> Self:
         df = self._native_frame
         new_cols = [mapping.get(c, c) for c in df.column_names]
         return self._from_native_frame(df.rename_columns(new_cols))
@@ -746,7 +734,7 @@ def is_unique(self: Self) -> ArrowSeries:
 
     def unique(
         self: ArrowDataFrame,
-        subset: list[str] | None,
+        subset: Sequence[str] | None,
         *,
         keep: Literal["any", "first", "last", "none"],
         maintain_order: bool | None = None,
@@ -757,7 +745,7 @@ def unique(
 
         df = self._native_frame
         check_column_exists(self.columns, subset)
-        subset = subset or self.columns
+        subset = list(subset or self.columns)
 
         if keep in {"any", "first", "last"}:
             agg_func_map = {"any": "min", "first": "min", "last": "max"}
@@ -808,18 +796,15 @@ def sample(
 
     def unpivot(
         self: Self,
-        on: list[str] | None,
-        index: list[str] | None,
+        on: Sequence[str] | None,
+        index: Sequence[str] | None,
         variable_name: str,
         value_name: str,
     ) -> Self:
         native_frame = self._native_frame
         n_rows = len(self)
-
-        index_: list[str] = [] if index is None else index
-        on_: list[str] = (
-            [c for c in self.columns if c not in index_] if on is None else on
-        )
+        index_ = [] if index is None else index
+        on_ = [c for c in self.columns if c not in index_] if on is None else on
         concat = (
             partial(pa.concat_tables, promote_options="permissive")
             if self._backend_version >= (14, 0, 0)