Repository: narwhals-dev/narwhals
Pull request #2324 (2025-04-01 16:20:45)

feat: Adds private `Namespace` class
## What type of PR is this? (check all applicable)

- [x] 💾 Refactor
- [x] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [x] ✅ Test
- [x] 🐳 Other

## Related issues

- Mentioned in (#2283)
- Related
  - #2315
  - #2304
  - #2116

## 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
Had this idea nagging away at me while working on (#2116).

### Super high-level
This PR adds a `Namespace` class, that can be created via either:

```py
Namespace.from_backend(...)
Namespace.from_native_object(...)
```

Supporting these new `@classmethod`s is a new `@property` for `Implementation`:

```py
Implementation._backend_version
```

### Notes
I've tried to keep things isolated from the rest of `narwhals` - for now - but **the goal would be replacing similar logic** from all of:
- `functions.py`
- `translate.py`
- `utils.py`

> [!IMPORTANT]
> Not planning to make this a public class

## Example
This *further* simplifies (#2283) to a one-liner, with the added bonus of typing from a `backend` string

![image](https://github.com/user-attachments/assets/0918e2a7-418d-4e5c-8626-667703208fe7)



dangotbanned: (https://github.com/narwhals-dev/narwhals/pull/2356#discussion_r2032967492)

@FBruzzesi this PR has some interesting typing stuff as well 😅
dangotbanned: > thanks @dangotbanned
> 

thanks @MarcoGorelli!

Just going to make the replacements now in `utils` and `functions` - where this'll be used first

- https://github.com/narwhals-dev/narwhals/pull/2324#discussion_r2023222391
- https://github.com/narwhals-dev/narwhals/pull/2283
dangotbanned: @MarcoGorelli after (https://github.com/narwhals-dev/narwhals/pull/2324#commits-pushed-03a08ef) I'm gonna call it here, too easy to get carried away 😅 

Hopefully you get an idea for how much of `from_native` this can simplify

Patch:
diff --git a/narwhals/_compliant/typing.py b/narwhals/_compliant/typing.py
index 906d1757..30788b32 100644
--- a/narwhals/_compliant/typing.py
+++ b/narwhals/_compliant/typing.py
@@ -93,6 +93,11 @@ CompliantLazyFrameT = TypeVar("CompliantLazyFrameT", bound=CompliantLazyFrameAny
 CompliantLazyFrameT_co = TypeVar(
     "CompliantLazyFrameT_co", bound=CompliantLazyFrameAny, covariant=True
 )
+CompliantNamespaceT = TypeVar("CompliantNamespaceT", bound=CompliantNamespaceAny)
+CompliantNamespaceT_co = TypeVar(
+    "CompliantNamespaceT_co", bound=CompliantNamespaceAny, covariant=True
+)
+
 IntoCompliantExpr: TypeAlias = "CompliantExpr[CompliantFrameT, CompliantSeriesOrNativeExprT_co] | CompliantSeriesOrNativeExprT_co"
 
 DepthTrackingExprT = TypeVar("DepthTrackingExprT", bound=DepthTrackingExprAny)
diff --git a/narwhals/_namespace.py b/narwhals/_namespace.py
new file mode 100644
index 00000000..0e3931bf
--- /dev/null
+++ b/narwhals/_namespace.py
@@ -0,0 +1,354 @@
+"""Narwhals-level equivalent of `CompliantNamespace`."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import Callable
+from typing import Generic
+from typing import Literal
+from typing import Protocol
+from typing import TypeVar
+from typing import overload
+
+from narwhals._compliant.typing import CompliantNamespaceAny
+from narwhals._compliant.typing import CompliantNamespaceT_co
+from narwhals.dependencies import get_cudf
+from narwhals.dependencies import get_modin
+from narwhals.dependencies import get_pandas
+from narwhals.dependencies import get_polars
+from narwhals.dependencies import get_pyarrow
+from narwhals.dependencies import is_dask_dataframe
+from narwhals.dependencies import is_duckdb_relation
+from narwhals.dependencies import is_pyspark_dataframe
+from narwhals.dependencies import is_sqlframe_dataframe
+from narwhals.utils import Implementation
+from narwhals.utils import Version
+
+if TYPE_CHECKING:
+    from types import ModuleType
+    from typing import ClassVar
+
+    import duckdb
+    import pandas as pd
+    import polars as pl
+    import pyarrow as pa
+    import pyspark.sql as pyspark_sql
+    from typing_extensions import TypeAlias
+    from typing_extensions import TypeIs
+
+    from narwhals._arrow.namespace import ArrowNamespace
+    from narwhals._dask.namespace import DaskNamespace
+    from narwhals._duckdb.namespace import DuckDBNamespace
+    from narwhals._pandas_like.namespace import PandasLikeNamespace
+    from narwhals._polars.namespace import PolarsNamespace
+    from narwhals._spark_like.dataframe import SQLFrameDataFrame
+    from narwhals._spark_like.namespace import SparkLikeNamespace
+    from narwhals.typing import DataFrameLike
+    from narwhals.typing import NativeFrame
+    from narwhals.typing import NativeLazyFrame
+    from narwhals.typing import NativeSeries
+
+    T = TypeVar("T")
+
+    _Guard: TypeAlias = "Callable[[Any], TypeIs[T]]"
+
+    _Polars: TypeAlias = Literal["polars"]
+    _Arrow: TypeAlias = Literal["pyarrow"]
+    _Dask: TypeAlias = Literal["dask"]
+    _DuckDB: TypeAlias = Literal["duckdb"]
+    _PandasLike: TypeAlias = Literal["pandas", "cudf", "modin"]
+    _SparkLike: TypeAlias = Literal["pyspark", "sqlframe"]
+    _EagerOnly: TypeAlias = "_PandasLike | _Arrow"
+    _EagerAllowed: TypeAlias = "_Polars | _EagerOnly"
+    _LazyOnly: TypeAlias = "_SparkLike | _Dask | _DuckDB"
+    _LazyAllowed: TypeAlias = "_Polars | _LazyOnly"
+
+    Polars: TypeAlias = Literal[_Polars, Implementation.POLARS]
+    Arrow: TypeAlias = Literal[_Arrow, Implementation.PYARROW]
+    Dask: TypeAlias = Literal[_Dask, Implementation.DASK]
+    DuckDB: TypeAlias = Literal[_DuckDB, Implementation.DUCKDB]
+    PandasLike: TypeAlias = Literal[
+        _PandasLike, Implementation.PANDAS, Implementation.CUDF, Implementation.MODIN
+    ]
+    SparkLike: TypeAlias = Literal[
+        _SparkLike, Implementation.PYSPARK, Implementation.SQLFRAME
+    ]
+    EagerOnly: TypeAlias = "PandasLike | Arrow"
+    EagerAllowed: TypeAlias = "EagerOnly | Polars"
+    LazyOnly: TypeAlias = "SparkLike | Dask | DuckDB"
+    LazyAllowed: TypeAlias = "LazyOnly | Polars"
+
+    BackendName: TypeAlias = "_EagerAllowed | _LazyAllowed"
+    IntoBackend: TypeAlias = "BackendName | Implementation | ModuleType"
+
+    EagerAllowedNamespace: TypeAlias = "Namespace[PandasLikeNamespace] | Namespace[ArrowNamespace] | Namespace[PolarsNamespace]"
+    EagerAllowedImplementation: TypeAlias = Literal[
+        Implementation.PANDAS,
+        Implementation.CUDF,
+        Implementation.MODIN,
+        Implementation.PYARROW,
+        Implementation.POLARS,
+    ]
+
+    class _NativeDask(Protocol):
+        _partition_type: type[pd.DataFrame]
+
+    class _NativeCuDF(Protocol):
+        def to_pylibcudf(self, *args: Any, **kwds: Any) -> Any: ...
+
+    class _ModinDataFrame(Protocol):
+        _pandas_class: type[pd.DataFrame]
+
+    class _ModinSeries(Protocol):
+        _pandas_class: type[pd.Series[Any]]
+
+    _NativePolars: TypeAlias = "pl.DataFrame | pl.LazyFrame | pl.Series"
+    _NativeArrow: TypeAlias = "pa.Table | pa.ChunkedArray[Any]"
+    _NativeDuckDB: TypeAlias = "duckdb.DuckDBPyRelation"
+    _NativePandas: TypeAlias = "pd.DataFrame | pd.Series[Any]"
+    _NativeModin: TypeAlias = "_ModinDataFrame | _ModinSeries"
+    _NativePandasLike: TypeAlias = "_NativePandas | _NativeCuDF | _NativeModin"
+    _NativeSQLFrame: TypeAlias = "SQLFrameDataFrame"
+    _NativePySpark: TypeAlias = "pyspark_sql.DataFrame"
+    _NativeSparkLike: TypeAlias = "_NativeSQLFrame | _NativePySpark"
+
+    NativeKnown: TypeAlias = "_NativePolars | _NativeArrow | _NativePandasLike | _NativeSparkLike | _NativeDuckDB | _NativeDask"
+    NativeUnknown: TypeAlias = (
+        "NativeFrame | NativeSeries | NativeLazyFrame | DataFrameLike"
+    )
+    NativeAny: TypeAlias = "NativeKnown | NativeUnknown"
+
+__all__ = ["Namespace"]
+
+
+class Namespace(Generic[CompliantNamespaceT_co]):
+    _compliant_namespace: CompliantNamespaceT_co
+    _version: ClassVar[Version] = Version.MAIN
+
+    def __init__(self, namespace: CompliantNamespaceT_co, /) -> None:
+        self._compliant_namespace = namespace
+
+    def __init_subclass__(cls, *args: Any, version: Version, **kwds: Any) -> None:
+        super().__init_subclass__(*args, **kwds)
+
+        if isinstance(version, Version):
+            cls._version = version
+        else:
+            msg = f"Expected {Version} but got {type(version).__name__!r}"
+            raise TypeError(msg)
+
+    def __repr__(self) -> str:
+        return f"Namespace[{type(self.compliant).__name__}]"
+
+    @property
+    def compliant(self) -> CompliantNamespaceT_co:
+        return self._compliant_namespace
+
+    @property
+    def implementation(self) -> Implementation:
+        return self.compliant._implementation
+
+    @property
+    def version(self) -> Version:
+        return self._version
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: PandasLike, /) -> Namespace[PandasLikeNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: Polars, /) -> Namespace[PolarsNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: Arrow, /) -> Namespace[ArrowNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: SparkLike, /) -> Namespace[SparkLikeNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: DuckDB, /) -> Namespace[DuckDBNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: Dask, /) -> Namespace[DaskNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: EagerAllowed, /) -> EagerAllowedNamespace: ...
+
+    @overload
+    @classmethod
+    def from_backend(cls, backend: ModuleType, /) -> Namespace[CompliantNamespaceAny]: ...
+
+    @classmethod
+    def from_backend(
+        cls: type[Namespace[Any]], backend: IntoBackend, /
+    ) -> Namespace[Any]:
+        """Instantiate from native namespace module, string, or Implementation.
+
+        Arguments:
+            backend: native namespace module, string, or Implementation.
+
+        Returns:
+            Namespace.
+
+        Examples:
+            >>> from narwhals._namespace import Namespace
+            >>> Namespace.from_backend("polars")
+            Namespace[PolarsNamespace]
+        """
+        impl = Implementation.from_backend(backend)
+        backend_version = impl._backend_version()
+        version = cls._version
+        ns: CompliantNamespaceAny
+        if impl.is_pandas_like():
+            from narwhals._pandas_like.namespace import PandasLikeNamespace
+
+            ns = PandasLikeNamespace(
+                implementation=impl, backend_version=backend_version, version=version
+            )
+
+        elif impl.is_polars():
+            from narwhals._polars.namespace import PolarsNamespace
+
+            ns = PolarsNamespace(backend_version=backend_version, version=version)
+        elif impl.is_pyarrow():
+            from narwhals._arrow.namespace import ArrowNamespace
+
+            ns = ArrowNamespace(backend_version=backend_version, version=version)
+        elif impl.is_spark_like():
+            from narwhals._spark_like.namespace import SparkLikeNamespace
+
+            ns = SparkLikeNamespace(
+                implementation=impl, backend_version=backend_version, version=version
+            )
+        elif impl.is_duckdb():
+            from narwhals._duckdb.namespace import DuckDBNamespace
+
+            ns = DuckDBNamespace(backend_version=backend_version, version=version)
+        elif impl.is_dask():
+            from narwhals._dask.namespace import DaskNamespace
+
+            ns = DaskNamespace(backend_version=backend_version, version=version)
+        else:
+            msg = "Not supported Implementation"  # pragma: no cover
+            raise AssertionError(msg)
+        return cls(ns)
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativePolars, /
+    ) -> Namespace[PolarsNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativePandasLike, /
+    ) -> Namespace[PandasLikeNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(cls, native: _NativeArrow, /) -> Namespace[ArrowNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativeSparkLike, /
+    ) -> Namespace[SparkLikeNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativeDuckDB, /
+    ) -> Namespace[DuckDBNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(cls, native: _NativeDask, /) -> Namespace[DaskNamespace]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: NativeUnknown, /
+    ) -> Namespace[CompliantNamespaceAny]: ...
+
+    @classmethod
+    def from_native_object(
+        cls: type[Namespace[Any]], native: NativeAny, /
+    ) -> Namespace[Any]:
+        if is_native_polars(native):
+            return cls.from_backend(Implementation.POLARS)
+        elif is_native_pandas(native):
+            return cls.from_backend(Implementation.PANDAS)
+        elif is_native_arrow(native):
+            return cls.from_backend(Implementation.PYARROW)
+        elif is_native_spark_like(native):
+            return cls.from_backend(
+                Implementation.SQLFRAME
+                if is_native_sqlframe(native)
+                else Implementation.PYSPARK
+            )
+        elif is_native_dask(native):
+            return cls.from_backend(Implementation.DASK)  # pragma: no cover
+        elif is_native_duckdb(native):
+            return cls.from_backend(Implementation.DUCKDB)
+        elif is_native_cudf(native):  # pragma: no cover
+            return cls.from_backend(Implementation.CUDF)
+        elif is_native_modin(native):  # pragma: no cover
+            return cls.from_backend(Implementation.MODIN)
+        else:
+            msg = f"Unsupported type: {type(native).__qualname__!r}"
+            raise TypeError(msg)
+
+
+def is_native_polars(obj: Any) -> TypeIs[_NativePolars]:
+    return (pl := get_polars()) is not None and isinstance(
+        obj, (pl.DataFrame, pl.Series, pl.LazyFrame)
+    )
+
+
+def is_native_arrow(obj: Any) -> TypeIs[_NativeArrow]:
+    return (pa := get_pyarrow()) is not None and isinstance(
+        obj, (pa.Table, pa.ChunkedArray)
+    )
+
+
+def is_native_dask(obj: Any) -> TypeIs[_NativeDask]:
+    return is_dask_dataframe(obj)
+
+
+is_native_duckdb: _Guard[_NativeDuckDB] = is_duckdb_relation
+is_native_sqlframe: _Guard[_NativeSQLFrame] = is_sqlframe_dataframe
+is_native_pyspark: _Guard[_NativePySpark] = is_pyspark_dataframe
+
+
+def is_native_pandas(obj: Any) -> TypeIs[_NativePandas]:
+    return (pd := get_pandas()) is not None and isinstance(obj, (pd.DataFrame, pd.Series))
+
+
+def is_native_modin(obj: Any) -> TypeIs[_NativeModin]:
+    return (mpd := get_modin()) is not None and isinstance(
+        obj, (mpd.DataFrame, mpd.Series)
+    )  # pragma: no cover
+
+
+def is_native_cudf(obj: Any) -> TypeIs[_NativeCuDF]:
+    return (cudf := get_cudf()) is not None and isinstance(
+        obj, (cudf.DataFrame, cudf.Series)
+    )  # pragma: no cover
+
+
+def is_native_pandas_like(obj: Any) -> TypeIs[_NativePandasLike]:
+    return (
+        is_native_pandas(obj) or is_native_cudf(obj) or is_native_modin(obj)
+    )  # pragma: no cover
+
+
+def is_native_spark_like(obj: Any) -> TypeIs[_NativeSparkLike]:
+    return is_native_pyspark(obj) or is_native_sqlframe(obj)
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index a1b104d4..f0c0eecc 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -9,6 +9,7 @@ from typing import Iterator
 from typing import Mapping
 from typing import Sequence
 
+from narwhals._namespace import is_native_spark_like
 from narwhals._spark_like.utils import evaluate_exprs
 from narwhals._spark_like.utils import import_functions
 from narwhals._spark_like.utils import import_native_dtypes
@@ -22,7 +23,6 @@ from narwhals.utils import check_column_exists
 from narwhals.utils import find_stacklevel
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
-from narwhals.utils import is_spark_like_dataframe
 from narwhals.utils import not_implemented
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
@@ -99,7 +99,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame["SparkLikeExpr", "SQLFrameDataFrame"
 
     @staticmethod
     def _is_native(obj: SQLFrameDataFrame | Any) -> TypeIs[SQLFrameDataFrame]:
-        return is_spark_like_dataframe(obj)
+        return is_native_spark_like(obj)
 
     @classmethod
     def from_native(cls, data: SQLFrameDataFrame, /, *, context: _FullContext) -> Self:
diff --git a/narwhals/functions.py b/narwhals/functions.py
index 0ba84ab8..7e6dca99 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -34,9 +34,9 @@ from narwhals.translate import from_native
 from narwhals.translate import to_native
 from narwhals.utils import Implementation
 from narwhals.utils import Version
-from narwhals.utils import _into_compliant_namespace
 from narwhals.utils import deprecate_native_namespace
 from narwhals.utils import flatten
+from narwhals.utils import import_namespace
 from narwhals.utils import is_compliant_expr
 from narwhals.utils import is_eager_allowed
 from narwhals.utils import is_sequence_but_not_str
@@ -241,7 +241,7 @@ def _new_series_impl(
 ) -> Series[Any]:
     implementation = Implementation.from_backend(backend)
     if is_eager_allowed(implementation):
-        ns = _into_compliant_namespace(implementation, version)
+        ns = import_namespace(version).from_backend(implementation).compliant
         series = ns._series.from_iterable(values, name=name, context=ns, dtype=dtype)
         return from_native(series, series_only=True)
     elif implementation is Implementation.DASK:  # pragma: no cover
@@ -327,7 +327,7 @@ def _from_dict_impl(
         data, backend = _from_dict_no_backend(data)
     implementation = Implementation.from_backend(backend)
     if is_eager_allowed(implementation):
-        ns = _into_compliant_namespace(implementation, version)
+        ns = import_namespace(version).from_backend(implementation).compliant
         frame = ns._dataframe.from_dict(data, schema=schema, context=ns)
         return from_native(frame, eager_only=True)
     elif implementation is Implementation.UNKNOWN:  # pragma: no cover
@@ -443,7 +443,7 @@ def _from_numpy_impl(
         raise TypeError(msg)
     implementation = Implementation.from_backend(backend)
     if is_eager_allowed(implementation):
-        ns = _into_compliant_namespace(implementation, version)
+        ns = import_namespace(version).from_backend(implementation).compliant
         frame = ns.from_numpy(data, schema)
         return from_native(frame, eager_only=True)
     else:  # pragma: no cover
@@ -531,7 +531,7 @@ def _from_arrow_impl(
         raise TypeError(msg)
     implementation = Implementation.from_backend(backend)
     if is_eager_allowed(implementation):
-        ns = _into_compliant_namespace(implementation, version)
+        ns = import_namespace(version).from_backend(implementation).compliant
         frame = ns._dataframe.from_arrow(data, context=ns)
         return from_native(frame, eager_only=True)
     else:  # pragma: no cover
diff --git a/narwhals/stable/v1/_namespace.py b/narwhals/stable/v1/_namespace.py
new file mode 100644
index 00000000..fbfa36f6
--- /dev/null
+++ b/narwhals/stable/v1/_namespace.py
@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+from narwhals._compliant.typing import CompliantNamespaceT_co
+from narwhals._namespace import Namespace as NwNamespace
+from narwhals.utils import Version
+
+__all__ = ["Namespace"]
+
+
+class Namespace(NwNamespace[CompliantNamespaceT_co], version=Version.V1): ...
diff --git a/narwhals/translate.py b/narwhals/translate.py
index 89544cbf..c0b59e96 100644
--- a/narwhals/translate.py
+++ b/narwhals/translate.py
@@ -11,6 +11,9 @@ from typing import Literal
 from typing import TypeVar
 from typing import overload
 
+from narwhals._namespace import is_native_arrow
+from narwhals._namespace import is_native_polars
+from narwhals._namespace import is_native_spark_like
 from narwhals.dependencies import get_cudf
 from narwhals.dependencies import get_cupy
 from narwhals.dependencies import get_dask
@@ -20,7 +23,6 @@ from narwhals.dependencies import get_numpy
 from narwhals.dependencies import get_pandas
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import get_pyarrow
-from narwhals.dependencies import get_pyspark
 from narwhals.dependencies import is_cudf_dataframe
 from narwhals.dependencies import is_cudf_series
 from narwhals.dependencies import is_dask_dataframe
@@ -35,10 +37,8 @@ from narwhals.dependencies import is_polars_lazyframe
 from narwhals.dependencies import is_polars_series
 from narwhals.dependencies import is_pyarrow_chunked_array
 from narwhals.dependencies import is_pyarrow_table
-from narwhals.dependencies import is_pyspark_dataframe
 from narwhals.utils import Version
-from narwhals.utils import is_native_polars
-from narwhals.utils import is_spark_like_dataframe
+from narwhals.utils import import_namespace
 
 if TYPE_CHECKING:
     from narwhals.dataframe import DataFrame
@@ -580,39 +580,25 @@ def _from_native_impl(  # noqa: PLR0915
         )
 
     # PyArrow
-    elif is_pyarrow_table(native_object):
-        from narwhals._arrow.dataframe import ArrowDataFrame
-
-        pa = get_pyarrow()
-        if series_only:
+    elif is_native_arrow(native_object):
+        if series_only and is_pyarrow_table(native_object):
             if not pass_through:
-                msg = "Cannot only use `series_only` with arrow table"
+                msg = f"Cannot only use `series_only` with {type(native_object).__qualname__}"
                 raise TypeError(msg)
             return native_object
-        return DataFrame(
-            ArrowDataFrame(
-                native_object,
-                backend_version=parse_version(pa),
-                version=version,
-                validate_column_names=True,
-            ),
-            level="full",
-        )
-    elif is_pyarrow_chunked_array(native_object):
-        from narwhals._arrow.series import ArrowSeries
-
-        pa = get_pyarrow()
-        if not allow_series:
+        if not allow_series and not is_pyarrow_table(native_object):
             if not pass_through:
                 msg = "Please set `allow_series=True` or `series_only=True`"
                 raise TypeError(msg)
             return native_object
-        return Series(
-            ArrowSeries(
-                native_object, backend_version=parse_version(pa), name="", version=version
-            ),
-            level="full",
+        pa_compliant = (
+            import_namespace(version)
+            .from_native_object(native_object)
+            .compliant.from_native(native_object)
         )
+        if is_compliant_dataframe(pa_compliant):
+            return DataFrame(pa_compliant, level="full")
+        return Series(pa_compliant, level="full")
 
     # Dask
     elif is_dask_dataframe(native_object):
@@ -637,26 +623,19 @@ def _from_native_impl(  # noqa: PLR0915
 
     # DuckDB
     elif is_duckdb_relation(native_object):
-        from narwhals._duckdb.namespace import DuckDBNamespace
-
         if eager_only or series_only:  # pragma: no cover
             if not pass_through:
-                msg = (
-                    "Cannot only use `series_only=True` or `eager_only=False` "
-                    "with DuckDBPyRelation"
-                )
-            else:
-                return native_object
-            raise TypeError(msg)
-        import duckdb  # ignore-banned-import
-
-        duckdb_ns = DuckDBNamespace(
-            backend_version=parse_version(duckdb), version=version
+                msg = "Cannot only use `series_only=True` or `eager_only=False` with DuckDBPyRelation"
+                raise TypeError(msg)
+            return native_object
+        duckdb_compliant = (
+            import_namespace(version)
+            .from_native_object(native_object)
+            .compliant.from_native(native_object)
         )
-        duckdb_ldf = duckdb_ns.from_native(native_object)
         if version is Version.V1:
-            return DataFrame(duckdb_ldf, level="interchange")
-        return LazyFrame(duckdb_ldf, level="lazy")
+            return DataFrame(duckdb_compliant, level="interchange")
+        return LazyFrame(duckdb_compliant, level="lazy")
 
     # Ibis
     elif is_ibis_table(native_object):  # pragma: no cover
@@ -688,31 +667,15 @@ def _from_native_impl(  # noqa: PLR0915
         )
 
     # PySpark
-    elif is_spark_like_dataframe(native_object):  # pragma: no cover
-        from narwhals._spark_like.namespace import SparkLikeNamespace
-
-        if is_pyspark_dataframe(native_object):
-            name = "pyspark"
-            backend_version = parse_version(get_pyspark())
-            implementation = Implementation.PYSPARK
-        else:
-            name = "SQLFrame"
-            import sqlframe._version
-
-            backend_version = parse_version(sqlframe._version)
-            implementation = Implementation.SQLFRAME
+    elif is_native_spark_like(native_object):  # pragma: no cover
+        ns_spark = import_namespace(version).from_native_object(native_object)
         if series_only:
-            msg = f"Cannot only use `series_only` with {name} DataFrame"
+            msg = f"Cannot only use `series_only` with {ns_spark.implementation._alias} DataFrame"
             raise TypeError(msg)
         if eager_only or eager_or_interchange_only:
-            msg = f"Cannot only use `eager_only` or `eager_or_interchange_only` with {name} DataFrame"
+            msg = f"Cannot only use `eager_only` or `eager_or_interchange_only` with {ns_spark.implementation._alias} DataFrame"
             raise TypeError(msg)
-        spark_ns = SparkLikeNamespace(
-            backend_version=backend_version,
-            version=version,
-            implementation=implementation,
-        )
-        return LazyFrame(spark_ns.from_native(native_object), level="lazy")
+        return LazyFrame(ns_spark.compliant.from_native(native_object), level="lazy")
 
     # Interchange protocol
     elif _supports_dataframe_interchange(native_object):
diff --git a/narwhals/utils.py b/narwhals/utils.py
index 666755c9..b48e5d43 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -24,6 +24,7 @@ from typing import overload
 from warnings import warn
 
 from narwhals.dependencies import get_cudf
+from narwhals.dependencies import get_dask
 from narwhals.dependencies import get_dask_dataframe
 from narwhals.dependencies import get_duckdb
 from narwhals.dependencies import get_ibis
@@ -31,6 +32,7 @@ from narwhals.dependencies import get_modin
 from narwhals.dependencies import get_pandas
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import get_pyarrow
+from narwhals.dependencies import get_pyspark
 from narwhals.dependencies import get_pyspark_sql
 from narwhals.dependencies import get_sqlframe
 from narwhals.dependencies import is_cudf_series
@@ -41,8 +43,6 @@ from narwhals.dependencies import is_pandas_like_series
 from narwhals.dependencies import is_pandas_series
 from narwhals.dependencies import is_polars_series
 from narwhals.dependencies import is_pyarrow_chunked_array
-from narwhals.dependencies import is_pyspark_dataframe
-from narwhals.dependencies import is_sqlframe_dataframe
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.exceptions import DuplicateError
 from narwhals.exceptions import InvalidOperationError
@@ -52,9 +52,7 @@ if TYPE_CHECKING:
     from typing import AbstractSet as Set
 
     import pandas as pd
-    import polars as pl
     import pyarrow as pa
-    import pyspark.sql as pyspark_sql
     from typing_extensions import Concatenate
     from typing_extensions import LiteralString
     from typing_extensions import ParamSpec
@@ -62,22 +60,16 @@ if TYPE_CHECKING:
     from typing_extensions import TypeAlias
     from typing_extensions import TypeIs
 
-    from narwhals._arrow.namespace import ArrowNamespace
     from narwhals._compliant import CompliantExpr
     from narwhals._compliant import CompliantExprT
     from narwhals._compliant import CompliantFrameT
-    from narwhals._compliant import CompliantNamespace
     from narwhals._compliant import CompliantSeriesOrNativeExprT_co
     from narwhals._compliant import CompliantSeriesT
     from narwhals._compliant import NativeFrameT_co
     from narwhals._compliant import NativeSeriesT_co
     from narwhals._compliant.typing import EvalNames
-    from narwhals._dask.namespace import DaskNamespace
-    from narwhals._duckdb.namespace import DuckDBNamespace
-    from narwhals._pandas_like.namespace import PandasLikeNamespace
-    from narwhals._polars.namespace import PolarsNamespace
-    from narwhals._spark_like.dataframe import SQLFrameDataFrame
-    from narwhals._spark_like.namespace import SparkLikeNamespace
+    from narwhals._namespace import EagerAllowedImplementation
+    from narwhals._namespace import Namespace
     from narwhals._translate import ArrowStreamExportable
     from narwhals._translate import IntoArrowTable
     from narwhals.dataframe import DataFrame
@@ -97,8 +89,6 @@ if TYPE_CHECKING:
     FrameOrSeriesT = TypeVar(
         "FrameOrSeriesT", bound=Union[LazyFrame[Any], DataFrame[Any], Series[Any]]
     )
-    _SparkLikeDataFrame: TypeAlias = "SQLFrameDataFrame | pyspark_sql.DataFrame"
-    _NativePolars: TypeAlias = "pl.DataFrame | pl.LazyFrame | pl.Series"
     _T = TypeVar("_T")
     _T1 = TypeVar("_T1")
     _T2 = TypeVar("_T2")
@@ -107,19 +97,6 @@ if TYPE_CHECKING:
     P = ParamSpec("P")
     R = TypeVar("R")
 
-    _PandasLike: TypeAlias = (
-        "Literal[Implementation.PANDAS, Implementation.CUDF, Implementation.MODIN]"
-    )
-    _Arrow: TypeAlias = "Literal[Implementation.PYARROW]"
-    _Polars: TypeAlias = "Literal[Implementation.POLARS]"
-    _SparkLike: TypeAlias = "Literal[Implementation.PYSPARK, Implementation.SQLFRAME]"
-    _Dask: TypeAlias = "Literal[Implementation.DASK]"
-    _DuckDB: TypeAlias = "Literal[Implementation.DUCKDB]"
-    _EagerOnly: TypeAlias = "_PandasLike | _Arrow"
-    _EagerAllowed: TypeAlias = "_Polars | _EagerOnly"
-    _LazyOnly: TypeAlias = "_SparkLike | _Dask | _DuckDB"
-    _LazyAllowed: TypeAlias = "_Polars | _LazyOnly"
-
     class _SupportsVersion(Protocol):
         __version__: str
 
@@ -563,6 +540,25 @@ class Implementation(Enum):
         }
         return mapping[self]
 
+    def _backend_version(self) -> tuple[int, ...]:
+        native = self.to_native_namespace()
+        into_version: Any
+        if self not in {
+            Implementation.PYSPARK,
+            Implementation.DASK,
+            Implementation.SQLFRAME,
+        }:
+            into_version = native
+        elif self is Implementation.PYSPARK:
+            into_version = get_pyspark()  # pragma: no cover
+        elif self is Implementation.DASK:
+            into_version = get_dask()
+        else:
+            import sqlframe._version
+
+            into_version = sqlframe._version
+        return parse_version(into_version)
+
 
 MIN_VERSIONS: dict[Implementation, tuple[int, ...]] = {
     Implementation.PANDAS: (0, 25, 3),
@@ -578,65 +574,6 @@ MIN_VERSIONS: dict[Implementation, tuple[int, ...]] = {
 }
 
 
-@overload
-def _into_compliant_namespace(
-    impl: _PandasLike, version: Version, /
-) -> PandasLikeNamespace: ...
-@overload
-def _into_compliant_namespace(impl: _Polars, version: Version, /) -> PolarsNamespace: ...
-@overload
-def _into_compliant_namespace(impl: _Arrow, version: Version, /) -> ArrowNamespace: ...
-@overload
-def _into_compliant_namespace(
-    impl: _SparkLike, version: Version, /
-) -> SparkLikeNamespace: ...
-@overload
-def _into_compliant_namespace(impl: _DuckDB, version: Version, /) -> DuckDBNamespace: ...
-@overload
-def _into_compliant_namespace(impl: _Dask, version: Version, /) -> DaskNamespace: ...
-@overload
-def _into_compliant_namespace(
-    impl: _EagerAllowed, version: Version, /
-) -> PandasLikeNamespace | PolarsNamespace | ArrowNamespace: ...
-def _into_compliant_namespace(
-    impl: Implementation, version: Version, /
-) -> CompliantNamespace[Any, Any]:
-    native = impl.to_native_namespace()
-    into_version = native if not impl.is_sqlframe() else native._version
-    backend_version = parse_version(into_version)
-    if impl.is_pandas_like():
-        from narwhals._pandas_like.namespace import PandasLikeNamespace
-
-        return PandasLikeNamespace(
-            implementation=impl, backend_version=backend_version, version=version
-        )
-    elif impl.is_polars():
-        from narwhals._polars.namespace import PolarsNamespace
-
-        return PolarsNamespace(backend_version=backend_version, version=version)
-    elif impl.is_pyarrow():
-        from narwhals._arrow.namespace import ArrowNamespace
-
-        return ArrowNamespace(backend_version=backend_version, version=version)
-    elif impl.is_spark_like():  # pragma: no cover
-        from narwhals._spark_like.namespace import SparkLikeNamespace
-
-        return SparkLikeNamespace(
-            implementation=impl, backend_version=backend_version, version=version
-        )
-    elif impl.is_duckdb():  # pragma: no cover
-        from narwhals._duckdb.namespace import DuckDBNamespace
-
-        return DuckDBNamespace(backend_version=backend_version, version=version)
-    elif impl.is_dask():  # pragma: no cover
-        from narwhals._dask.namespace import DaskNamespace
-
-        return DaskNamespace(backend_version=backend_version, version=version)
-    else:
-        msg = "Not supported Implementation"  # pragma: no cover
-        raise AssertionError(msg)
-
-
 def validate_backend_version(
     implementation: Implementation, backend_version: tuple[int, ...]
 ) -> None:
@@ -663,6 +600,34 @@ def import_dtypes_module(version: Version) -> DTypes:
         raise AssertionError(msg)
 
 
+def _into_version(obj: Version | _StoresVersion, /) -> Version:
+    if isinstance(obj, Version):
+        return obj
+    elif _hasattr_static(obj, "_version"):
+        return obj._version
+    msg = f"Expected {Version} but got {type(obj).__name__!r}"
+    raise TypeError(msg)
+
+
+def import_namespace(version: Version | _StoresVersion, /) -> type[Namespace[Any]]:
+    version = _into_version(version)
+    if version is Version.MAIN:
+        from narwhals._namespace import Namespace
+
+        return Namespace
+    elif version is Version.V1:
+        from narwhals.stable.v1._namespace import Namespace
+
+        return Namespace
+    else:  # pragma: no cover
+        msg = (
+            "Congratulations, you have entered unreachable code.\n"
+            "Please report an issue at https://github.com/narwhals-dev/narwhals/issues.\n"
+            f"Version: {version}"
+        )
+        raise AssertionError(msg)
+
+
 def remove_prefix(text: str, prefix: str) -> str:  # pragma: no cover
     if text.startswith(prefix):
         return text[len(prefix) :]
@@ -1558,16 +1523,6 @@ def _hasattr_static(obj: Any, attr: str) -> bool:
     return getattr_static(obj, attr, sentinel) is not sentinel
 
 
-def is_spark_like_dataframe(obj: Any) -> TypeIs[_SparkLikeDataFrame]:
-    return is_sqlframe_dataframe(obj) or is_pyspark_dataframe(obj)
-
-
-def is_native_polars(obj: Any) -> TypeIs[_NativePolars]:
-    return (pl := get_polars()) is not None and isinstance(
-        obj, (pl.DataFrame, pl.Series, pl.LazyFrame)
-    )
-
-
 def is_compliant_dataframe(
     obj: CompliantDataFrame[CompliantSeriesT, CompliantExprT, NativeFrameT_co] | Any,
 ) -> TypeIs[CompliantDataFrame[CompliantSeriesT, CompliantExprT, NativeFrameT_co]]:
@@ -1592,7 +1547,7 @@ def is_compliant_expr(
     return hasattr(obj, "__narwhals_expr__")
 
 
-def is_eager_allowed(obj: Implementation) -> TypeIs[_EagerAllowed]:
+def is_eager_allowed(obj: Implementation) -> TypeIs[EagerAllowedImplementation]:
     return obj in {
         Implementation.PANDAS,
         Implementation.MODIN,
@@ -1602,16 +1557,6 @@ def is_eager_allowed(obj: Implementation) -> TypeIs[_EagerAllowed]:
     }
 
 
-def is_lazy_allowed(obj: Implementation) -> TypeIs[_LazyAllowed]:  # pragma: no cover
-    return obj in {
-        Implementation.POLARS,
-        Implementation.PYSPARK,
-        Implementation.SQLFRAME,
-        Implementation.DASK,
-        Implementation.DUCKDB,
-    }
-
-
 def has_native_namespace(obj: Any) -> TypeIs[SupportsNativeNamespace]:
     return hasattr(obj, "__native_namespace__")
 


Test patch:
diff --git a/tests/namespace_test.py b/tests/namespace_test.py
new file mode 100644
index 00000000..70923b63
--- /dev/null
+++ b/tests/namespace_test.py
@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import re
+from collections import deque
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import Callable
+from typing import Iterable
+from typing import Sequence
+from typing import cast
+
+import pytest
+
+import narwhals as nw
+from narwhals._namespace import Namespace
+from narwhals.utils import Version
+from narwhals.utils import import_namespace
+
+if TYPE_CHECKING:
+    from typing_extensions import TypeAlias
+    from typing_extensions import assert_type
+
+    from narwhals._arrow.namespace import ArrowNamespace  # noqa: F401
+    from narwhals._namespace import BackendName
+    from narwhals._namespace import _EagerAllowed
+    from narwhals._pandas_like.namespace import PandasLikeNamespace  # noqa: F401
+    from narwhals._polars.namespace import PolarsNamespace  # noqa: F401
+    from narwhals.typing import _2DArray
+    from tests.utils import Constructor
+
+IntoIterable: TypeAlias = Callable[[Sequence[Any]], Iterable[Any]]
+
+_EAGER_ALLOWED = "polars", "pandas", "pyarrow", "modin", "cudf"
+_LAZY_ONLY = "dask", "duckdb", "pyspark", "sqlframe"
+_LAZY_ALLOWED = ("polars", *_LAZY_ONLY)
+_BACKENDS = (*_EAGER_ALLOWED, *_LAZY_ONLY)
+
+eager_allowed = pytest.mark.parametrize("backend", _EAGER_ALLOWED)
+lazy_allowed = pytest.mark.parametrize("backend", _LAZY_ALLOWED)
+backends = pytest.mark.parametrize("backend", _BACKENDS)
+
+
+@backends
+def test_namespace_from_backend_name(backend: BackendName) -> None:
+    pytest.importorskip(backend)
+    namespace = Namespace.from_backend(backend)
+    assert namespace.implementation.name.lower() == backend
+    assert namespace.version is Version.MAIN
+
+
+def test_namespace_from_native_object(constructor: Constructor) -> None:
+    data = {"a": [1, 2, 3], "b": [4, 5, 6]}
+    frame = constructor(data)
+    namespace = Namespace.from_native_object(frame)
+    nw_frame = nw.from_native(frame)
+    assert namespace.implementation == nw_frame.implementation
+
+
+def test_namespace_from_native_object_invalid() -> None:
+    data = {"a": [1, 2, 3], "b": [4, 5, 6]}
+    with pytest.raises(TypeError, match=r"dict"):
+        Namespace.from_native_object(data)  # pyright: ignore[reportCallIssue, reportArgumentType]
+
+
+@eager_allowed
+def test_namespace_from_backend_typing(backend: _EagerAllowed) -> None:
+    pytest.importorskip(backend)
+    namespace = Namespace.from_backend(backend)
+    if TYPE_CHECKING:
+        assert_type(
+            namespace,
+            "Namespace[PolarsNamespace] | Namespace[PandasLikeNamespace] | Namespace[ArrowNamespace]",
+        )
+    assert repr(namespace) in {
+        "Namespace[PolarsNamespace]",
+        "Namespace[PandasLikeNamespace]",
+        "Namespace[ArrowNamespace]",
+    }
+
+
+@pytest.mark.parametrize("into_iter", [list, tuple, deque, iter])
+@eager_allowed
+def test_namespace_series_from_iterable(
+    backend: _EagerAllowed, into_iter: IntoIterable
+) -> None:
+    pytest.importorskip(backend)
+    data = 1, 2, 3
+    namespace = Namespace.from_backend(backend)
+    compliant = namespace.compliant
+    iterable = into_iter(data)
+    compliant_series = compliant._series.from_iterable(
+        iterable, context=compliant, name="hello"
+    )
+    series = nw.from_native(compliant_series, series_only=True)
+    assert isinstance(series, nw.Series)
+    assert series.to_list() == list(data)
+
+
+def test_namespace_from_numpy_polars() -> None:
+    pytest.importorskip("polars")
+    pytest.importorskip("numpy")
+    import numpy as np
+    import polars as pl
+
+    arr: _2DArray = cast("_2DArray", np.array([[5, 2, 0, 1], [1, 4, 7, 8], [1, 2, 3, 9]]))
+    columns = "a", "b", "c"
+    frame = Namespace.from_backend("polars").compliant.from_numpy(arr, columns).native
+    if TYPE_CHECKING:
+        assert_type(frame, pl.DataFrame)
+
+    assert isinstance(frame, pl.DataFrame)
+    assert frame.columns == list(columns)
+
+
+def test_import_namespace() -> None:
+    pytest.importorskip("pandas")
+    pytest.importorskip("polars")
+    import pandas as pd
+
+    data = {"a": [1, 2, 3]}
+    native = pd.DataFrame(data)
+    ns = import_namespace(Version.V1).from_native_object(native)
+    assert ns.version is Version.V1
+    assert ns.implementation.is_pandas()
+
+    ns = import_namespace(Version.MAIN).from_native_object(native)
+    assert ns.version is Version.MAIN
+    assert ns.implementation.is_pandas()
+
+    compliant = ns.compliant._dataframe.from_dict(data, context=ns.compliant, schema=None)
+    ns_pl = import_namespace(ns.compliant).from_native_object(compliant.to_polars())
+    assert ns_pl.version is Version.MAIN
+    assert ns_pl.implementation.is_polars()
+
+    with pytest.raises(TypeError):
+        import_namespace(data)  # type: ignore[arg-type]
+
+
+def test_namespace_init_subclass() -> None:
+    pytest.importorskip("polars")
+
+    class NamespaceV1Any(Namespace[Any], version=Version.V1): ...
+
+    ns_any = NamespaceV1Any.from_backend("polars")
+    assert ns_any.version is Version.V1
+
+    class NamespaceV1NoTypeVar(Namespace, version=Version.V1): ...  # type: ignore[type-arg]
+
+    ns_no_type_var = NamespaceV1NoTypeVar.from_backend("polars")
+    assert ns_no_type_var.version is Version.V1
+
+    with pytest.raises(
+        TypeError, match=re.compile(r"missing.+required.+argument.+version")
+    ):
+
+        class NamespaceNoVersion(Namespace): ...  # type: ignore[call-arg, type-arg]
+
+    with pytest.raises(TypeError, match=re.compile(r"Expected.+Version.+but got.+str")):
+
+        class NamespaceBadVersion(Namespace, version="invalid version"): ...  # type: ignore[arg-type, type-arg]