Remove attrs dependency.

Author: tobiasraabe

docs_src/how_to_guides/the_data_catalog.py

@@ -1,11 +1,11 @@
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 import cloudpickle
-from attrs import define
 
 
-@define
+@dataclass
 class PickleNode:
     """A node for pickle files.
 

pyproject.toml

@@ -21,7 +21,6 @@ classifiers = [
 ]
 dynamic = ["version"]
 dependencies = [
-    "attrs>=21.3.0",
     "click>=8.1.8,!=8.2.0",
     "click-default-group>=1.2.4",
     "networkx>=2.4.0",

src/_pytask/cache.py

@@ -5,16 +5,15 @@
 import functools
 import hashlib
 import inspect
+from dataclasses import dataclass
+from dataclasses import field
 from inspect import FullArgSpec
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import ParamSpec
 from typing import Protocol
 from typing import TypeVar
 
-from attrs import define
-from attrs import field
-
 from _pytask._hashlib import hash_value
 
 if TYPE_CHECKING:
@@ -35,17 +34,17 @@ class HasCache(Protocol):
     cache: Cache
 
 
-@define
+@dataclass
 class CacheInfo:
     hits: int = 0
     misses: int = 0
 
 
-@define
+@dataclass
 class Cache:
-    _cache: dict[str, Any] = field(factory=dict)
-    _sentinel: Any = field(factory=object)
-    cache_info: CacheInfo = field(factory=CacheInfo)
+    _cache: dict[str, Any] = field(default_factory=dict)
+    _sentinel: Any = field(default_factory=object)
+    cache_info: CacheInfo = field(default_factory=CacheInfo)
 
     def memoize(self, func: Callable[P, R]) -> Memoized[P, R]:
         func_module = getattr(func, "__module__", "")

src/_pytask/clean.py

@@ -6,11 +6,11 @@
 import itertools
 import shutil
 import sys
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 from typing import Any
 
 import click
-from attrs import define
 
 from _pytask.click import ColoredCommand
 from _pytask.click import EnumChoice
@@ -243,7 +243,7 @@ def _find_all_unknown_paths(
     )
 
 
-@define(repr=False)
+@dataclass(repr=False)
 class _RecursivePathNode:
     """A class for a path to a file or directory which recursively instantiates itself.
 

src/_pytask/coiled_utils.py

@@ -1,18 +1,17 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 from typing import Any
 
-from attrs import define
-
 if TYPE_CHECKING:
     from collections.abc import Callable
 
 try:
     from coiled.function import Function
 except ImportError:
 
-    @define
+    @dataclass
     class Function:
         cluster_kwargs: dict[str, Any]
         environ: dict[str, Any]

src/_pytask/collect_utils.py

@@ -3,13 +3,12 @@
 from __future__ import annotations
 
 import inspect
+from dataclasses import replace
 from typing import TYPE_CHECKING
 from typing import Annotated
 from typing import Any
 from typing import get_origin
 
-import attrs
-
 from _pytask._inspect import get_annotations
 from _pytask.exceptions import NodeNotCollectedError
 from _pytask.models import NodeInfo
@@ -308,7 +307,7 @@ def collect_dependency(
         # If a node is a dependency and its value is not set, the node is a product in
         # another task and the value will be set there. Thus, we wrap the original node
         # in another node to retrieve the value after it is set.
-        new_node = attrs.evolve(node, value=node)
+        new_node = replace(node, value=node)
         node_info = node_info._replace(value=new_node)
 
     collected_node = session.hook.pytask_collect_node(

src/_pytask/dag_utils.py

@@ -3,11 +3,11 @@
 from __future__ import annotations
 
 import itertools
+from dataclasses import dataclass
+from dataclasses import field
 from typing import TYPE_CHECKING
 
 import networkx as nx
-from attrs import define
-from attrs import field
 
 from _pytask.mark_utils import has_mark
 
@@ -61,7 +61,7 @@ def node_and_neighbors(dag: nx.DiGraph, node: str) -> Iterable[str]:
     return itertools.chain(dag.predecessors(node), [node], dag.successors(node))
 
 
-@define
+@dataclass
 class TopologicalSorter:
     """The topological sorter class.
 
@@ -78,9 +78,9 @@ class TopologicalSorter:
     """
 
     dag: nx.DiGraph
-    priorities: dict[str, int] = field(factory=dict)
-    _nodes_processing: set[str] = field(factory=set)
-    _nodes_done: set[str] = field(factory=set)
+    priorities: dict[str, int] = field(default_factory=dict)
+    _nodes_processing: set[str] = field(default_factory=set)
+    _nodes_done: set[str] = field(default_factory=set)
 
     @classmethod
     def from_dag(cls, dag: nx.DiGraph) -> TopologicalSorter:

src/_pytask/data_catalog.py

@@ -20,7 +20,6 @@
 from _pytask.exceptions import NodeNotCollectedError
 from _pytask.models import NodeInfo
 from _pytask.node_protocols import PNode
-from _pytask.node_protocols import PPathNode
 from _pytask.node_protocols import PProvisionalNode
 from _pytask.node_protocols import warn_about_upcoming_attributes_field_on_nodes
 from _pytask.nodes import PickleNode
@@ -39,6 +38,14 @@ def _get_parent_path_of_data_catalog_module(stacklevel: int = 2) -> Path:
     return Path.cwd()
 
 
+def _is_path_node_type(node_type: type[Any]) -> bool:
+    """Return True if the class looks like a path-based node."""
+    for cls in node_type.__mro__:
+        if "path" in getattr(cls, "__annotations__", {}):
+            return True
+    return False
+
+
 @dataclass(kw_only=True)
 class DataCatalog:
     """A data catalog.
@@ -115,7 +122,7 @@ def add(self, name: str, node: PNode | PProvisionalNode | Any = None) -> None:
 
         if node is None:
             filename = hashlib.sha256(name.encode()).hexdigest()
-            if isinstance(self.default_node, PPathNode):
+            if _is_path_node_type(self.default_node):
                 assert self.path is not None
                 self._entries[name] = self.default_node(
                     name=name, path=self.path / f"{filename}.pkl"

src/_pytask/explain.py

@@ -2,12 +2,12 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from dataclasses import field
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Literal
 
-from attrs import define
-from attrs import field
 from rich.text import Text
 
 from _pytask.console import console
@@ -34,14 +34,14 @@
 ]
 
 
-@define
+@dataclass
 class ChangeReason:
     """Represents a reason why a node changed."""
 
     node_name: str
     node_type: NodeType
     reason: ReasonType
-    details: dict[str, Any] = field(factory=dict)
+    details: dict[str, Any] = field(default_factory=dict)
     verbose: int = 1
 
     def __rich_console__(
@@ -71,11 +71,11 @@ def __rich_console__(
             yield Text(f"  • {self.node_name}: {self.reason}")
 
 
-@define
+@dataclass
 class TaskExplanation:
     """Represents the explanation for why a task needs to be executed."""
 
-    reasons: list[ChangeReason] = field(factory=list)
+    reasons: list[ChangeReason] = field(default_factory=list)
     task: PTask | None = None
     outcome: TaskOutcome | None = None
     verbose: int = 1

src/_pytask/live.py

@@ -3,13 +3,12 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from dataclasses import field
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import NamedTuple
 
 import click
-from attrs import define
-from attrs import field
 from rich.box import ROUNDED
 from rich.errors import LiveError
 from rich.live import Live
@@ -85,7 +84,7 @@ def pytask_execute(session: Session) -> Generator[None, None, None]:
     return (yield)
 
 
-@define(eq=False)
+@dataclass(eq=False)
 class LiveManager:
     """A class for live displays during a session.
 
@@ -106,7 +105,9 @@ class LiveManager:
     """
 
     _live: Live = field(
-        factory=lambda: Live(renderable=None, console=console, auto_refresh=False)
+        default_factory=lambda: Live(
+            renderable=None, console=console, auto_refresh=False
+        )
     )
 
     def start(self) -> None:
@@ -157,7 +158,7 @@ class _ReportEntry(NamedTuple):
     task: PTask
 
 
-@define(eq=False, kw_only=True)
+@dataclass(eq=False, kw_only=True)
 class LiveExecution:
     """A class for managing the table displaying task progress during the execution."""
 
@@ -168,8 +169,8 @@ class LiveExecution:
     initial_status: TaskExecutionStatus = TaskExecutionStatus.RUNNING
     sort_final_table: bool = False
     n_tasks: int | str = "x"
-    _reports: list[_ReportEntry] = field(factory=list)
-    _running_tasks: dict[str, _TaskEntry] = field(factory=dict)
+    _reports: list[_ReportEntry] = field(default_factory=list)
+    _running_tasks: dict[str, _TaskEntry] = field(default_factory=dict)
 
     @hookimpl(wrapper=True)
     def pytask_execute_build(self) -> Generator[None, None, None]:
@@ -306,7 +307,7 @@ def update_report(self, new_report: ExecutionReport) -> None:
         self._update_table()
 
 
-@define(eq=False, kw_only=True)
+@dataclass(eq=False, kw_only=True)
 class LiveCollection:
     """A class for managing the live status during the collection."""