Repository: narwhals-dev/narwhals
Pull request #2218 (2025-03-15 14:38:06)

chore: Deprecate `native_namespace` in favour of `backend` in `scan_csv`
## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [x] 🐳 Other

## Related issues

- Related issue #1888

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below



Patch:
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index 360132e8..bc72bb46 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -13,8 +13,6 @@ from narwhals._compliant import EagerExpr
 from narwhals._expression_parsing import ExprKind
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
 from narwhals._expression_parsing import is_scalar_like
-from narwhals.dependencies import get_numpy
-from narwhals.dependencies import is_numpy_array
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.utils import Implementation
 from narwhals.utils import generate_temporary_column_name
@@ -25,7 +23,6 @@ if TYPE_CHECKING:
 
     from narwhals._arrow.dataframe import ArrowDataFrame
     from narwhals._arrow.namespace import ArrowNamespace
-    from narwhals.dtypes import DType
     from narwhals.utils import Version
     from narwhals.utils import _FullContext
 
@@ -203,44 +200,6 @@ class ArrowExpr(EagerExpr["ArrowDataFrame", ArrowSeries]):
             version=self._version,
         )
 
-    def map_batches(
-        self: Self,
-        function: Callable[[Any], Any],
-        return_dtype: DType | type[DType] | None,
-    ) -> Self:
-        def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            input_series_list = self._call(df)
-            output_names = [input_series.name for input_series in input_series_list]
-            result = [function(series) for series in input_series_list]
-
-            if is_numpy_array(result[0]):
-                result = [
-                    df.__narwhals_namespace__()
-                    ._create_compliant_series(array)
-                    .alias(output_name)
-                    for array, output_name in zip(result, output_names)
-                ]
-            elif (np := get_numpy()) is not None and np.isscalar(result[0]):
-                result = [
-                    df.__narwhals_namespace__()
-                    ._create_compliant_series([array])
-                    .alias(output_name)
-                    for array, output_name in zip(result, output_names)
-                ]
-            if return_dtype is not None:
-                result = [series.cast(return_dtype) for series in result]
-            return result
-
-        return self.__class__(
-            func,
-            depth=self._depth + 1,
-            function_name=self._function_name + "->map_batches",
-            evaluate_output_names=self._evaluate_output_names,
-            alias_output_names=self._alias_output_names,
-            backend_version=self._backend_version,
-            version=self._version,
-        )
-
     def cum_count(self: Self, *, reverse: bool) -> Self:
         return self._reuse_series("cum_count", reverse=reverse)
 
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index 34ad9ea7..d7561c32 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -51,9 +51,6 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr]):
     def _series(self) -> type[ArrowSeries]:
         return ArrowSeries
 
-    def _create_compliant_series(self: Self, value: Any) -> ArrowSeries:
-        return self._series._from_iterable(value, name="", context=self)
-
     # --- not in spec ---
     def __init__(
         self: Self, *, backend_version: tuple[int, ...], version: Version
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index dbb359b9..45171f13 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -27,6 +27,7 @@ from narwhals._arrow.utils import native_to_narwhals_dtype
 from narwhals._arrow.utils import nulls_like
 from narwhals._arrow.utils import pad_series
 from narwhals._compliant import EagerSeries
+from narwhals.dependencies import is_numpy_array_1d
 from narwhals.exceptions import InvalidOperationError
 from narwhals.utils import Implementation
 from narwhals.utils import generate_temporary_column_name
@@ -52,6 +53,7 @@ if TYPE_CHECKING:
     from narwhals._arrow.typing import _AsPyType
     from narwhals._arrow.typing import _BasicDataType
     from narwhals.dtypes import DType
+    from narwhals.typing import Into1DArray
     from narwhals.typing import _1DArray
     from narwhals.typing import _2DArray
     from narwhals.utils import Version
@@ -156,6 +158,12 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
             value = value.as_py()
         return super()._from_scalar(value)
 
+    @classmethod
+    def from_numpy(cls, data: Into1DArray, /, *, context: _FullContext) -> Self:
+        return cls._from_iterable(
+            data if is_numpy_array_1d(data) else [data], name="", context=context
+        )
+
     def __narwhals_namespace__(self: Self) -> ArrowNamespace:
         from narwhals._arrow.namespace import ArrowNamespace
 
@@ -437,7 +445,7 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
     def __array__(self: Self, dtype: Any = None, *, copy: bool | None = None) -> _1DArray:
         return self.native.__array__(dtype=dtype, copy=copy)
 
-    def to_numpy(self: Self) -> _1DArray:
+    def to_numpy(self: Self, dtype: Any = None, *, copy: bool | None = None) -> _1DArray:
         return self.native.to_numpy()
 
     def alias(self: Self, name: str) -> Self:
diff --git a/narwhals/_compliant/expr.py b/narwhals/_compliant/expr.py
index ee56b07c..03e2343e 100644
--- a/narwhals/_compliant/expr.py
+++ b/narwhals/_compliant/expr.py
@@ -29,6 +29,8 @@ from narwhals._compliant.typing import EagerExprT
 from narwhals._compliant.typing import EagerSeriesT
 from narwhals._compliant.typing import NativeExprT_co
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
+from narwhals.dependencies import get_numpy
+from narwhals.dependencies import is_numpy_array
 from narwhals.dtypes import DType
 from narwhals.utils import _ExprNamespace
 from narwhals.utils import deprecated
@@ -760,6 +762,38 @@ class EagerExpr(
             ddof=ddof,
         )
 
+    def map_batches(
+        self: Self,
+        function: Callable[[Any], Any],
+        return_dtype: DType | type[DType] | None,
+    ) -> Self:
+        def func(df: EagerDataFrameT) -> Sequence[EagerSeriesT]:
+            input_series_list = self(df)
+            output_names = [input_series.name for input_series in input_series_list]
+            result = [function(series) for series in input_series_list]
+            if is_numpy_array(result[0]) or (
+                (np := get_numpy()) is not None and np.isscalar(result[0])
+            ):
+                from_numpy = partial(
+                    self.__narwhals_namespace__()._series.from_numpy, context=self
+                )
+                result = [
+                    from_numpy(array).alias(output_name)
+                    for array, output_name in zip(result, output_names)
+                ]
+            if return_dtype is not None:
+                result = [series.cast(return_dtype) for series in result]
+            return result
+
+        return self._from_callable(
+            func,
+            depth=self._depth + 1,
+            function_name=self._function_name + "->map_batches",
+            evaluate_output_names=self._evaluate_output_names,
+            alias_output_names=self._alias_output_names,
+            context=self,
+        )
+
     @property
     def cat(self) -> EagerExprCatNamespace[Self]:
         return EagerExprCatNamespace(self)
diff --git a/narwhals/_compliant/namespace.py b/narwhals/_compliant/namespace.py
index 742f0274..f5449ec4 100644
--- a/narwhals/_compliant/namespace.py
+++ b/narwhals/_compliant/namespace.py
@@ -13,7 +13,6 @@ from narwhals._compliant.typing import CompliantFrameT
 from narwhals._compliant.typing import EagerDataFrameT
 from narwhals._compliant.typing import EagerExprT
 from narwhals._compliant.typing import EagerSeriesT_co
-from narwhals.utils import deprecated
 from narwhals.utils import exclude_column_names
 from narwhals.utils import get_column_names
 from narwhals.utils import passthrough_column_names
@@ -85,13 +84,3 @@ class EagerNamespace(
 ):
     @property
     def _series(self) -> type[EagerSeriesT_co]: ...
-
-    @deprecated(
-        "Internally used for `numpy.ndarray` -> `CompliantSeries`\n"
-        "Also referenced in untyped `nw.dataframe.DataFrame._extract_compliant`\n"
-        "See Also:\n"
-        "  - https://github.com/narwhals-dev/narwhals/pull/2149#discussion_r1986283345\n"
-        "  - https://github.com/narwhals-dev/narwhals/issues/2116\n"
-        "  - https://github.com/narwhals-dev/narwhals/pull/2169"
-    )
-    def _create_compliant_series(self, value: Any) -> EagerSeriesT_co: ...
diff --git a/narwhals/_compliant/series.py b/narwhals/_compliant/series.py
index e6be3770..85f4b0a8 100644
--- a/narwhals/_compliant/series.py
+++ b/narwhals/_compliant/series.py
@@ -6,6 +6,8 @@ from typing import Iterable
 from typing import Protocol
 from typing import TypeVar
 
+from narwhals._translate import NumpyConvertible
+
 if TYPE_CHECKING:
     from typing_extensions import Self
 
@@ -14,7 +16,9 @@ if TYPE_CHECKING:
     from narwhals._compliant.namespace import CompliantNamespace  # noqa: F401
     from narwhals._compliant.namespace import EagerNamespace
     from narwhals.dtypes import DType
+    from narwhals.typing import Into1DArray
     from narwhals.typing import NativeSeries
+    from narwhals.typing import _1DArray  # noqa: F401
     from narwhals.utils import Implementation
     from narwhals.utils import Version
     from narwhals.utils import _FullContext
@@ -24,7 +28,7 @@ __all__ = ["CompliantSeries", "EagerSeries"]
 NativeSeriesT_co = TypeVar("NativeSeriesT_co", bound="NativeSeries", covariant=True)
 
 
-class CompliantSeries(Protocol):
+class CompliantSeries(NumpyConvertible["_1DArray", "Into1DArray"], Protocol):
     @property
     def dtype(self) -> DType: ...
     @property
@@ -36,6 +40,8 @@ class CompliantSeries(Protocol):
     def __narwhals_namespace__(self) -> Any: ...  # CompliantNamespace[Any, Self]: ...
     def _from_native_series(self, series: Any) -> Self: ...
     def _to_expr(self) -> Any: ...  # CompliantExpr[Any, Self]: ...
+    @classmethod
+    def from_numpy(cls, data: Into1DArray, /, *, context: _FullContext) -> Self: ...
 
 
 class EagerSeries(CompliantSeries, Protocol[NativeSeriesT_co]):
@@ -60,3 +66,5 @@ class EagerSeries(CompliantSeries, Protocol[NativeSeriesT_co]):
 
     def _to_expr(self) -> EagerExpr[Any, Any]:
         return self.__narwhals_namespace__()._expr._from_series(self)  # type: ignore[no-any-return]
+
+    def cast(self, dtype: DType | type[DType]) -> Self: ...
diff --git a/narwhals/_compliant/typing.py b/narwhals/_compliant/typing.py
index ba970e4d..6b0fc0dc 100644
--- a/narwhals/_compliant/typing.py
+++ b/narwhals/_compliant/typing.py
@@ -15,6 +15,7 @@ if TYPE_CHECKING:
     from narwhals._compliant.expr import CompliantExpr
     from narwhals._compliant.expr import EagerExpr
     from narwhals._compliant.expr import NativeExpr
+    from narwhals._compliant.namespace import EagerNamespace
     from narwhals._compliant.series import CompliantSeries
     from narwhals._compliant.series import EagerSeries
 
@@ -48,5 +49,9 @@ EagerDataFrameT = TypeVar("EagerDataFrameT", bound="EagerDataFrame[Any]")
 EagerSeriesT = TypeVar("EagerSeriesT", bound="EagerSeries[Any]")
 EagerSeriesT_co = TypeVar("EagerSeriesT_co", bound="EagerSeries[Any]", covariant=True)
 EagerExprT = TypeVar("EagerExprT", bound="EagerExpr[Any, Any]")
+EagerNamespaceAny: TypeAlias = (
+    "EagerNamespace[EagerDataFrame[Any], EagerSeries[Any], EagerExpr[Any, Any]]"
+)
+
 AliasNames: TypeAlias = Callable[[Sequence[str]], Sequence[str]]
 AliasName: TypeAlias = Callable[[str], str]
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index 8ba9e776..41945232 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -12,6 +12,7 @@ from typing import Callable
 from typing import Literal
 from typing import Sequence
 from typing import TypeVar
+from typing import cast
 
 from narwhals.dependencies import is_narwhals_series
 from narwhals.dependencies import is_numpy_array
@@ -27,6 +28,7 @@ if TYPE_CHECKING:
     from narwhals._compliant import CompliantExprT
     from narwhals._compliant import CompliantFrameT
     from narwhals._compliant import CompliantNamespace
+    from narwhals._compliant.typing import EagerNamespaceAny
     from narwhals.expr import Expr
     from narwhals.typing import CompliantDataFrame
     from narwhals.typing import CompliantLazyFrame
@@ -100,7 +102,8 @@ def extract_compliant(
     if is_narwhals_series(other):
         return other._compliant_series._to_expr()
     if is_numpy_array(other):
-        return plx._create_compliant_series(other)._to_expr()  # type: ignore[attr-defined]
+        ns = cast("EagerNamespaceAny", plx)
+        return ns._series.from_numpy(other, context=ns)._to_expr()
     return other
 
 
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index 68b05fe2..190d3ce2 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -16,7 +16,6 @@ from narwhals._pandas_like.series import PandasLikeSeries
 from narwhals._pandas_like.utils import align_series_full_broadcast
 from narwhals._pandas_like.utils import check_column_names_are_unique
 from narwhals._pandas_like.utils import convert_str_slice_to_int_slice
-from narwhals._pandas_like.utils import create_compliant_series
 from narwhals._pandas_like.utils import extract_dataframe_comparand
 from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
@@ -433,16 +432,14 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         return scale_bytes(sz, unit=unit)
 
     def with_row_index(self: Self, name: str) -> Self:
-        row_index = create_compliant_series(
-            range(len(self._native_frame)),
-            index=self._native_frame.index,
-            implementation=self._implementation,
-            backend_version=self._backend_version,
-            version=self._version,
+        frame = self._native_frame
+        namespace = self.__narwhals_namespace__()
+        row_index = namespace._series._from_iterable(
+            range(len(frame)), name="", context=self, index=frame.index
         ).alias(name)
         return self._from_native_frame(
             horizontal_concat(
-                [row_index._native_series, self._native_frame],
+                [row_index.native, frame],
                 implementation=self._implementation,
                 backend_version=self._backend_version,
             )
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index bc186267..104a4efc 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -13,8 +13,6 @@ from narwhals._expression_parsing import evaluate_output_names_and_aliases
 from narwhals._expression_parsing import is_elementary_expression
 from narwhals._pandas_like.group_by import AGGREGATIONS_TO_PANDAS_EQUIVALENT
 from narwhals._pandas_like.series import PandasLikeSeries
-from narwhals.dependencies import get_numpy
-from narwhals.dependencies import is_numpy_array
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.utils import generate_temporary_column_name
 
@@ -23,7 +21,6 @@ if TYPE_CHECKING:
 
     from narwhals._pandas_like.dataframe import PandasLikeDataFrame
     from narwhals._pandas_like.namespace import PandasLikeNamespace
-    from narwhals.dtypes import DType
     from narwhals.utils import Implementation
     from narwhals.utils import Version
     from narwhals.utils import _FullContext
@@ -299,39 +296,6 @@ class PandasLikeExpr(EagerExpr["PandasLikeDataFrame", PandasLikeSeries]):
             version=self._version,
         )
 
-    def map_batches(
-        self: Self,
-        function: Callable[[Any], Any],
-        return_dtype: DType | type[DType] | None,
-    ) -> Self:
-        def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            input_series_list = self._call(df)
-            output_names = [input_series.name for input_series in input_series_list]
-            result = [function(series) for series in input_series_list]
-            if is_numpy_array(result[0]) or (
-                (np := get_numpy()) is not None and np.isscalar(result[0])
-            ):
-                result = [
-                    df.__narwhals_namespace__()
-                    ._create_compliant_series(array)
-                    .alias(output_name)
-                    for array, output_name in zip(result, output_names)
-                ]
-            if return_dtype is not None:
-                result = [series.cast(return_dtype) for series in result]
-            return result
-
-        return self.__class__(
-            func,
-            depth=self._depth + 1,
-            function_name=self._function_name + "->map_batches",
-            evaluate_output_names=self._evaluate_output_names,
-            alias_output_names=self._alias_output_names,
-            implementation=self._implementation,
-            backend_version=self._backend_version,
-            version=self._version,
-        )
-
     def cum_count(self: Self, *, reverse: bool) -> Self:
         return self._reuse_series("cum_count", call_kwargs={"reverse": reverse})
 
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index 33f2543e..ec96167d 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -17,7 +17,6 @@ from narwhals._pandas_like.expr import PandasLikeExpr
 from narwhals._pandas_like.selectors import PandasSelectorNamespace
 from narwhals._pandas_like.series import PandasLikeSeries
 from narwhals._pandas_like.utils import align_series_full_broadcast
-from narwhals._pandas_like.utils import create_compliant_series
 from narwhals._pandas_like.utils import diagonal_concat
 from narwhals._pandas_like.utils import extract_dataframe_comparand
 from narwhals._pandas_like.utils import horizontal_concat
@@ -61,14 +60,6 @@ class PandasLikeNamespace(
         self._backend_version = backend_version
         self._version = version
 
-    def _create_compliant_series(self: Self, value: Any) -> PandasLikeSeries:
-        return create_compliant_series(
-            value,
-            implementation=self._implementation,
-            backend_version=self._backend_version,
-            version=self._version,
-        )
-
     # --- selection ---
     def lit(self: Self, value: Any, dtype: DType | None) -> PandasLikeExpr:
         def _lit_pandas_series(df: PandasLikeDataFrame) -> PandasLikeSeries:
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index 41c61983..a62abaff 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -26,6 +26,7 @@ from narwhals._pandas_like.utils import object_native_to_narwhals_dtype
 from narwhals._pandas_like.utils import rename
 from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals._pandas_like.utils import set_index
+from narwhals.dependencies import is_numpy_array_1d
 from narwhals.dependencies import is_numpy_scalar
 from narwhals.exceptions import InvalidOperationError
 from narwhals.utils import Implementation
@@ -45,6 +46,7 @@ if TYPE_CHECKING:
     from narwhals._pandas_like.dataframe import PandasLikeDataFrame
     from narwhals._pandas_like.namespace import PandasLikeNamespace
     from narwhals.dtypes import DType
+    from narwhals.typing import Into1DArray
     from narwhals.typing import _1DArray
     from narwhals.typing import _AnyDArray
     from narwhals.utils import Version
@@ -191,6 +193,17 @@ class PandasLikeSeries(EagerSeries[Any]):
             version=context._version,
         )
 
+    @classmethod
+    def from_numpy(cls, data: Into1DArray, /, *, context: _FullContext) -> Self:
+        implementation = context._implementation
+        arr = data if is_numpy_array_1d(data) else [data]
+        return cls(
+            implementation.to_native_namespace().Series(arr, name=""),
+            implementation=implementation,
+            backend_version=context._backend_version,
+            version=context._version,
+        )
+
     def __len__(self: Self) -> int:
         return len(self.native)
 
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index a442c640..0fb7b352 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -147,31 +147,6 @@ def extract_dataframe_comparand(
     return other._native_series
 
 
-def create_compliant_series(
-    iterable: Any,
-    index: Any = None,
-    *,
-    implementation: Implementation,
-    backend_version: tuple[int, ...],
-    version: Version,
-) -> PandasLikeSeries:
-    from narwhals._pandas_like.series import PandasLikeSeries
-
-    if implementation in PANDAS_LIKE_IMPLEMENTATION:
-        series = implementation.to_native_namespace().Series(
-            iterable, index=index, name=""
-        )
-        return PandasLikeSeries(
-            series,
-            implementation=implementation,
-            backend_version=backend_version,
-            version=version,
-        )
-    else:  # pragma: no cover
-        msg = f"Expected pandas-like implementation ({PANDAS_LIKE_IMPLEMENTATION}), found {implementation}"
-        raise TypeError(msg)
-
-
 def horizontal_concat(
     dfs: list[Any], *, implementation: Implementation, backend_version: tuple[int, ...]
 ) -> Any:
diff --git a/narwhals/_polars/namespace.py b/narwhals/_polars/namespace.py
index 2bcf62f9..20d191f9 100644
--- a/narwhals/_polars/namespace.py
+++ b/narwhals/_polars/namespace.py
@@ -55,11 +55,6 @@ class PolarsNamespace:
     def _series(self) -> type[PolarsSeries]:
         return PolarsSeries
 
-    def _create_compliant_series(self, value: Any) -> PolarsSeries:
-        return self._series(
-            pl.Series(value), backend_version=self._backend_version, version=self._version
-        )
-
     def nth(self: Self, *indices: int) -> PolarsExpr:
         if self._backend_version < (1, 0, 0):
             msg = "`nth` is only supported for Polars>=1.0.0. Please use `col` for columns selection instead."
diff --git a/narwhals/_polars/series.py b/narwhals/_polars/series.py
index 830f562f..e5f06dda 100644
--- a/narwhals/_polars/series.py
+++ b/narwhals/_polars/series.py
@@ -13,6 +13,7 @@ from narwhals._polars.utils import extract_args_kwargs
 from narwhals._polars.utils import extract_native
 from narwhals._polars.utils import narwhals_to_native_dtype
 from narwhals._polars.utils import native_to_narwhals_dtype
+from narwhals.dependencies import is_numpy_array_1d
 from narwhals.utils import Implementation
 from narwhals.utils import validate_backend_version
 
@@ -26,8 +27,10 @@ if TYPE_CHECKING:
     from narwhals._polars.expr import PolarsExpr
     from narwhals._polars.namespace import PolarsNamespace
     from narwhals.dtypes import DType
+    from narwhals.typing import Into1DArray
     from narwhals.typing import _1DArray
     from narwhals.utils import Version
+    from narwhals.utils import _FullContext
 
     T = TypeVar("T")
 
@@ -71,6 +74,14 @@ class PolarsSeries:
             self._native_series, backend_version=self._backend_version, version=version
         )
 
+    @classmethod
+    def from_numpy(cls, data: Into1DArray, /, *, context: _FullContext) -> Self:
+        return cls(
+            pl.Series(data if is_numpy_array_1d(data) else [data]),
+            backend_version=context._backend_version,
+            version=context._version,
+        )
+
     def _from_native_series(self: Self, series: pl.Series) -> Self:
         return self.__class__(
             series, backend_version=self._backend_version, version=self._version
@@ -163,6 +174,9 @@ class PolarsSeries:
             raise NotImplementedError(msg)
         return self._from_native_series(ser.replace_strict(old, new, return_dtype=dtype))
 
+    def to_numpy(self, dtype: Any = None, *, copy: bool | None = None) -> _1DArray:
+        return self.__array__(dtype, copy=copy)
+
     def __array__(self: Self, dtype: Any, *, copy: bool | None) -> _1DArray:
         if self._backend_version < (0, 20, 29):
             return self._native_series.__array__(dtype=dtype)
diff --git a/narwhals/_translate.py b/narwhals/_translate.py
new file mode 100644
index 00000000..870eadc6
--- /dev/null
+++ b/narwhals/_translate.py
@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import Protocol
+
+if TYPE_CHECKING:
+    from typing_extensions import Self
+    from typing_extensions import TypeVar
+
+
+else:  # pragma: no cover
+    import sys
+    from importlib.util import find_spec
+
+    if sys.version_info >= (3, 13):
+        from typing import TypeVar
+    elif find_spec("typing_extensions"):
+        from typing_extensions import TypeVar
+    else:
+        from typing import TypeVar as _TypeVar
+
+        def TypeVar(  # noqa: ANN202, N802
+            name: str,
+            *constraints: Any,
+            bound: Any | None = None,
+            covariant: bool = False,
+            contravariant: bool = False,
+            **kwds: Any,  # noqa: ARG001
+        ):
+            return _TypeVar(
+                name,
+                *constraints,
+                bound=bound,
+                covariant=covariant,
+                contravariant=contravariant,
+            )
+
+
+ToNumpyT_co = TypeVar("ToNumpyT_co", covariant=True)
+FromNumpyDT_contra = TypeVar(
+    "FromNumpyDT_contra", contravariant=True, default=ToNumpyT_co
+)
+FromNumpyT_contra = TypeVar("FromNumpyT_contra", contravariant=True)
+
+
+class ToNumpy(Protocol[ToNumpyT_co]):
+    def to_numpy(self, *args: Any, **kwds: Any) -> ToNumpyT_co: ...
+
+
+class FromNumpy(Protocol[FromNumpyT_contra]):
+    @classmethod
+    def from_numpy(cls, data: FromNumpyT_contra, *args: Any, **kwds: Any) -> Self: ...
+
+
+class NumpyConvertible(
+    ToNumpy[ToNumpyT_co],
+    FromNumpy[FromNumpyDT_contra],
+    Protocol[ToNumpyT_co, FromNumpyDT_contra],
+):
+    def to_numpy(self, dtype: Any, *, copy: bool | None) -> ToNumpyT_co: ...
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index 294607f4..9b7ddc7c 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -50,6 +50,7 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._compliant import IntoCompliantExpr
+    from narwhals._compliant.typing import EagerNamespaceAny
     from narwhals.group_by import GroupBy
     from narwhals.group_by import LazyGroupBy
     from narwhals.series import Series
@@ -422,7 +423,7 @@ class DataFrame(BaseFrame[DataFrameT]):
         from narwhals.expr import Expr
         from narwhals.series import Series
 
-        plx = self.__narwhals_namespace__()
+        plx: EagerNamespaceAny = self.__narwhals_namespace__()
         if isinstance(arg, BaseFrame):
             return arg._compliant_frame
         if isinstance(arg, Series):
@@ -440,7 +441,7 @@ class DataFrame(BaseFrame[DataFrameT]):
             )
             raise TypeError(msg)
         if is_numpy_array(arg):
-            return plx._create_compliant_series(arg)._to_expr()
+            return plx._series.from_numpy(arg, context=plx)._to_expr()
         raise InvalidIntoExprError.from_invalid_type(type(arg))
 
     @property
diff --git a/narwhals/functions.py b/narwhals/functions.py
index cb0ccf32..98809976 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -840,8 +840,13 @@ def _read_csv_impl(
     return from_native(native_frame, eager_only=True)
 
 
+@deprecate_native_namespace(warn_version="1.31.0", required=True)
 def scan_csv(
-    source: str, *, native_namespace: ModuleType, **kwargs: Any
+    source: str,
+    *,
+    backend: ModuleType | Implementation | str | None = None,
+    native_namespace: ModuleType | None = None,  # noqa: ARG001
+    **kwargs: Any,
 ) -> LazyFrame[Any]:
     """Lazily read from a CSV file.
 
@@ -850,10 +855,22 @@ def scan_csv(
 
     Arguments:
         source: Path to a file.
+        backend: The eager backend for DataFrame creation.
+            `backend` can be specified in various ways:
+
+            - As `Implementation.<BACKEND>` with `BACKEND` being `PANDAS`, `PYARROW`,
+                `POLARS`, `MODIN` or `CUDF`.
+            - As a string: `"pandas"`, `"pyarrow"`, `"polars"`, `"modin"` or `"cudf"`.
+            - Directly as a module `pandas`, `pyarrow`, `polars`, `modin` or `cudf`.
         native_namespace: The native library to use for DataFrame creation.
+
+            **Deprecated** (v1.31.0):
+                Please use `backend` instead. Note that `native_namespace` is still available
+                (and won't emit a deprecation warning) if you use `narwhals.stable.v1`,
+                see [perfect backwards compatibility policy](../backcompat.md/).
         kwargs: Extra keyword arguments which are passed to the native CSV reader.
             For example, you could use
-            `nw.scan_csv('file.csv', native_namespace=pd, engine='pyarrow')`.
+            `nw.scan_csv('file.csv', backend=pd, engine='pyarrow')`.
 
     Returns:
         LazyFrame.
@@ -862,7 +879,7 @@ def scan_csv(
         >>> import duckdb
         >>> import narwhals as nw
         >>>
-        >>> nw.scan_csv("file.csv", native_namespace=duckdb).to_native()  # doctest:+SKIP
+        >>> nw.scan_csv("file.csv", backend="duckdb").to_native()  # doctest:+SKIP
         ┌─────────┬───────┐
         │    a    │   b   │
         │ varchar │ int32 │
@@ -872,13 +889,15 @@ def scan_csv(
         │ z       │     3 │
         └─────────┴───────┘
     """
-    return _scan_csv_impl(source, native_namespace=native_namespace, **kwargs)
+    backend = cast("ModuleType | Implementation | str", backend)
+    return _scan_csv_impl(source, backend=backend, **kwargs)
 
 
 def _scan_csv_impl(
-    source: str, *, native_namespace: ModuleType, **kwargs: Any
+    source: str, *, backend: ModuleType | Implementation | str, **kwargs: Any
 ) -> LazyFrame[Any]:
-    implementation = Implementation.from_native_namespace(native_namespace)
+    implementation = Implementation.from_backend(backend)
+    native_namespace = implementation.to_native_namespace()
     native_frame: NativeFrame | NativeLazyFrame
     if implementation is Implementation.POLARS:
         native_frame = native_namespace.scan_csv(source, **kwargs)
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 3ba0c88f..d5436e21 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -2359,8 +2359,13 @@ def read_csv(
     )
 
 
+@deprecate_native_namespace(required=True)
 def scan_csv(
-    source: str, *, native_namespace: ModuleType, **kwargs: Any
+    source: str,
+    *,
+    backend: ModuleType | Implementation | str | None = None,
+    native_namespace: ModuleType | None = None,  # noqa: ARG001
+    **kwargs: Any,
 ) -> LazyFrame[Any]:
     """Lazily read from a CSV file.
 
@@ -2369,16 +2374,29 @@ def scan_csv(
 
     Arguments:
         source: Path to a file.
+        backend: The eager backend for DataFrame creation.
+            `backend` can be specified in various ways:
+
+            - As `Implementation.<BACKEND>` with `BACKEND` being `PANDAS`, `PYARROW`,
+                `POLARS`, `MODIN` or `CUDF`.
+            - As a string: `"pandas"`, `"pyarrow"`, `"polars"`, `"modin"` or `"cudf"`.
+            - Directly as a module `pandas`, `pyarrow`, `polars`, `modin` or `cudf`.
         native_namespace: The native library to use for DataFrame creation.
+
+            **Deprecated** (v1.31.0):
+                Please use `backend` instead. Note that `native_namespace` is still available
+                (and won't emit a deprecation warning) if you use `narwhals.stable.v1`,
+                see [perfect backwards compatibility policy](../backcompat.md/).
         kwargs: Extra keyword arguments which are passed to the native CSV reader.
             For example, you could use
-            `nw.scan_csv('file.csv', native_namespace=pd, engine='pyarrow')`.
+            `nw.scan_csv('file.csv', backend=pd, engine='pyarrow')`.
 
     Returns:
         LazyFrame.
     """
+    backend = cast("ModuleType | Implementation | str", backend)
     return _stableify(  # type: ignore[no-any-return]
-        _scan_csv_impl(source, native_namespace=native_namespace, **kwargs)
+        _scan_csv_impl(source, backend=backend, **kwargs)
     )
 
 
diff --git a/narwhals/typing.py b/narwhals/typing.py
index 0209583d..33d4712d 100644
--- a/narwhals/typing.py
+++ b/narwhals/typing.py
@@ -209,9 +209,12 @@ TimeUnit: TypeAlias = Literal["ns", "us", "ms", "s"]
 
 _ShapeT = TypeVar("_ShapeT", bound="tuple[int, ...]")
 _NDArray: TypeAlias = "np.ndarray[_ShapeT, Any]"
-_1DArray: TypeAlias = "_NDArray[tuple[int]]"  # noqa: PYI042, PYI047
+_1DArray: TypeAlias = "_NDArray[tuple[int]]"  # noqa: PYI042
 _2DArray: TypeAlias = "_NDArray[tuple[int, int]]"  # noqa: PYI042, PYI047
 _AnyDArray: TypeAlias = "_NDArray[tuple[int, ...]]"  # noqa: PYI047
+_NumpyScalar: TypeAlias = "np.generic[Any]"
+Into1DArray: TypeAlias = "_1DArray | _NumpyScalar"
+"""A 1-dimensional `numpy.ndarray` or scalar that can be converted into one."""
 
 
 class DTypes:


Test patch:
diff --git a/tests/read_scan_test.py b/tests/read_scan_test.py
index e6dc82f0..79cb426c 100644
--- a/tests/read_scan_test.py
+++ b/tests/read_scan_test.py
@@ -77,8 +77,8 @@ def test_scan_csv(
     filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
     df_pl.write_csv(filepath)
     df = nw.from_native(constructor(data))
-    native_namespace = nw.get_native_namespace(df)
-    result = nw.scan_csv(filepath, native_namespace=native_namespace)
+    backend = nw.get_native_namespace(df)
+    result = nw.scan_csv(filepath, backend=backend)
     assert_equal_data(result, data)
     assert isinstance(result, nw.LazyFrame)
 
@@ -94,8 +94,8 @@ def test_scan_csv_v1(
     filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
     df_pl.write_csv(filepath)
     df = nw_v1.from_native(constructor(data))
-    native_namespace = nw_v1.get_native_namespace(df)
-    result = nw_v1.scan_csv(filepath, native_namespace=native_namespace)
+    backend = nw_v1.get_native_namespace(df)
+    result = nw_v1.scan_csv(filepath, backend=backend)
     assert_equal_data(result, data)
     assert isinstance(result, nw_v1.LazyFrame)
 
@@ -105,7 +105,7 @@ def test_scan_csv_kwargs(tmpdir: pytest.TempdirFactory) -> None:
     df_pl = pl.DataFrame(data)
     filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
     df_pl.write_csv(filepath)
-    result = nw.scan_csv(filepath, native_namespace=pd, engine="pyarrow")
+    result = nw.scan_csv(filepath, backend=pd, engine="pyarrow")
     assert_equal_data(result, data)