codekingpro/portable-devtools

1from typing import TYPE_CHECKING, Optional, Dict, Any
2from uuid import UUID
3import json
4
5if TYPE_CHECKING:
6    from chromadb.api import ServerAPI  # noqa: F401
7
8
9class AttachedFunction:
10    """Represents a function attached to a collection."""
11
12    def __init__(
13        self,
14        client: "ServerAPI",
15        id: UUID,
16        name: str,
17        function_name: str,
18        input_collection_id: UUID,
19        output_collection: str,
20        params: Optional[Dict[str, Any]],
21        tenant: str,
22        database: str,
23    ):
24        """Initialize an AttachedFunction.
25
26        Args:
27            client: The API client
28            id: Unique identifier for this attached function
29            name: Name of this attached function instance
30            function_name: The function name (e.g., "record_counter", "statistics")
31            input_collection_id: ID of the input collection
32            output_collection: Name of the output collection
33            params: Function-specific parameters
34            tenant: The tenant name
35            database: The database name
36        """
37        self._client = client
38        self._id = id
39        self._name = name
40        self._function_name = function_name
41        self._input_collection_id = input_collection_id
42        self._output_collection = output_collection
43        self._params = params
44        self._tenant = tenant
45        self._database = database
46
47    @property
48    def id(self) -> UUID:
49        """The unique identifier of this attached function."""
50        return self._id
51
52    @property
53    def name(self) -> str:
54        """The name of this attached function instance."""
55        return self._name
56
57    @property
58    def function_name(self) -> str:
59        """The function name."""
60        return self._function_name
61
62    @property
63    def input_collection_id(self) -> UUID:
64        """The ID of the input collection."""
65        return self._input_collection_id
66
67    @property
68    def output_collection(self) -> str:
69        """The name of the output collection."""
70        return self._output_collection
71
72    @property
73    def params(self) -> Optional[Dict[str, Any]]:
74        """The function parameters."""
75        return self._params
76
77    @staticmethod
78    def _normalize_params(params: Optional[Any]) -> Dict[str, Any]:
79        """Normalize params to a consistent dict format.
80
81        Handles None, empty strings, JSON strings, and dicts.
82        """
83        if params is None:
84            return {}
85        if isinstance(params, str):
86            try:
87                result = json.loads(params) if params else {}
88                return result if isinstance(result, dict) else {}
89            except json.JSONDecodeError:
90                return {}
91        if isinstance(params, dict):
92            return params
93        return {}
94
95    def __repr__(self) -> str:
96        return (
97            f"AttachedFunction(id={self._id}, name='{self._name}', "
98            f"function_name='{self._function_name}', "
99            f"input_collection_id={self._input_collection_id}, "
100            f"output_collection='{self._output_collection}')"
101        )
102
103    def __eq__(self, other: object) -> bool:
104        """Compare two AttachedFunction objects for equality."""
105        if not isinstance(other, AttachedFunction):
106            return False
107
108        # Normalize params: handle None, {}, and JSON strings
109        self_params = self._normalize_params(self._params)
110        other_params = self._normalize_params(other._params)
111
112        return (
113            self._id == other._id
114            and self._name == other._name
115            and self._function_name == other._function_name
116            and self._input_collection_id == other._input_collection_id
117            and self._output_collection == other._output_collection
118            and self_params == other_params
119            and self._tenant == other._tenant
120            and self._database == other._database
121        )
122
123    def __hash__(self) -> int:
124        """Return hash of the AttachedFunction."""
125        # Normalize params using the same logic as __eq__
126        normalized_params = self._normalize_params(self._params)
127        params_tuple = (
128            tuple(sorted(normalized_params.items())) if normalized_params else ()
129        )
130
131        return hash(
132            (
133                self._id,
134                self._name,
135                self._function_name,
136                self._input_collection_id,
137                self._output_collection,
138                params_tuple,
139                self._tenant,
140                self._database,
141            )
142        )
143