[HMA] Remove ISignalTypeConfigStore from HMA (#1724)

hasher-matcher-actioner/src/OpenMediaMatch/background_tasks/fetcher.py

@@ -14,7 +14,8 @@
 
 from OpenMediaMatch.background_tasks.development import get_apscheduler
 from OpenMediaMatch.persistence import get_storage
-from OpenMediaMatch.storage.interface import ISignalExchangeStore, SignalTypeConfig
+from OpenMediaMatch.storage.interface import ISignalExchangeStore
+from threatexchange.cli.storage.interfaces import SignalTypeConfig
 from OpenMediaMatch.utils.time_utils import duration_to_human_str
 
 logger = logging.getLogger(__name__)

hasher-matcher-actioner/src/OpenMediaMatch/storage/interface.py

@@ -21,7 +21,7 @@
 import time
 
 import flask
-
+from threatexchange.cli.storage.interfaces import ISignalTypeConfigStore
 from threatexchange.utils import dataclass_json
 from threatexchange.content_type.content_base import ContentType
 from threatexchange.signal_type.signal_base import SignalType
@@ -60,70 +60,6 @@ def get_content_type_configs(self) -> t.Mapping[str, ContentTypeConfig]:
         """
 
 
-@dataclass
-class SignalTypeConfig:
-    """
-    Holder for SignalType configuration
-    """
-
-    # Signal types that are not enabled should not be used in hashing/matching
-    enabled_ratio: float
-    signal_type: t.Type[SignalType]
-
-    @property
-    def enabled(self) -> bool:
-        # TODO do a coin flip here, but also refactor this to do seeding
-        return self.enabled_ratio >= 0.0
-
-
-class ISignalTypeConfigStore(metaclass=abc.ABCMeta):
-    """Interface for accessing SignalType configuration"""
-
-    @abc.abstractmethod
-    def get_signal_type_configs(self) -> t.Mapping[str, SignalTypeConfig]:
-        """Return all installed signal types."""
-
-    @abc.abstractmethod
-    def _create_or_update_signal_type_override(
-        self, signal_type: str, enabled_ratio: float
-    ) -> None:
-        """Create or update database entry for a signal type, setting a new value."""
-
-    @t.final
-    def create_or_update_signal_type_override(
-        self, signal_type: str, enabled_ratio: float
-    ) -> None:
-        """Update enabled ratio of an installed signal type."""
-        installed_signal_types = self.get_signal_type_configs()
-        if signal_type not in installed_signal_types:
-            raise ValueError(f"Unknown signal type {signal_type}")
-        if not (0.0 <= enabled_ratio <= 1.0):
-            raise ValueError(
-                f"Invalid enabled ratio {enabled_ratio}. Must be in the range 0.0-1.0 inclusive."
-            )
-        self._create_or_update_signal_type_override(signal_type, enabled_ratio)
-
-    @t.final
-    def get_enabled_signal_types(self) -> t.Mapping[str, t.Type[SignalType]]:
-        """Helper shortcut for getting only enabled SignalTypes"""
-        return {
-            k: v.signal_type
-            for k, v in self.get_signal_type_configs().items()
-            if v.enabled
-        }
-
-    @t.final
-    def get_enabled_signal_types_for_content_type(
-        self, content_type: t.Type[ContentType]
-    ) -> t.Mapping[str, t.Type[SignalType]]:
-        """Helper shortcut for getting enabled types for a piece of content"""
-        return {
-            k: v.signal_type
-            for k, v in self.get_signal_type_configs().items()
-            if v.enabled and content_type in v.signal_type.get_content_types()
-        }
-
-
 @dataclass
 class SignalTypeIndexBuildCheckpoint:
     """

hasher-matcher-actioner/src/OpenMediaMatch/storage/mocked.py

@@ -21,7 +21,7 @@
 )
 
 from OpenMediaMatch.storage import interface
-from OpenMediaMatch.storage.interface import SignalTypeConfig
+from threatexchange.cli.storage.interfaces import SignalTypeConfig
 
 
 class MockedUnifiedStore(interface.IUnifiedStore):
@@ -49,7 +49,7 @@ def get_content_type_configs(self) -> t.Mapping[str, interface.ContentTypeConfig
     def get_signal_type_configs(self) -> t.Mapping[str, SignalTypeConfig]:
         # Needed to bamboozle mypy into working
         s_types: t.Sequence[t.Type[SignalType]] = (PdqSignal, VideoMD5Signal)
-        return {s.get_name(): interface.SignalTypeConfig(1.0, s) for s in s_types}
+        return {s.get_name(): SignalTypeConfig(1.0, s) for s in s_types}
 
     def _create_or_update_signal_type_override(
         self, signal_type: str, enabled_ratio: float

hasher-matcher-actioner/src/OpenMediaMatch/storage/postgres/impl.py

@@ -37,6 +37,7 @@
 )
 
 from OpenMediaMatch.storage import interface
+from threatexchange.cli.storage.interfaces import SignalTypeConfig
 from OpenMediaMatch.storage.postgres import database, flask_utils
 
 
@@ -140,12 +141,12 @@ def exchange_api_config_update(
         sesh.add(config)
         sesh.commit()
 
-    def get_signal_type_configs(self) -> t.Mapping[str, interface.SignalTypeConfig]:
+    def get_signal_type_configs(self) -> t.Mapping[str, SignalTypeConfig]:
         # If a signal is installed, then it is enabled by default. But it may be disabled by an
         # override in the database.
         signal_type_overrides = self._query_signal_type_overrides()
         return {
-            name: interface.SignalTypeConfig(
+            name: SignalTypeConfig(
                 signal_type_overrides.get(name, 1.0),
                 st,
             )

python-threatexchange/threatexchange/cli/storage/interfaces.py

@@ -34,27 +34,6 @@
 from threatexchange.signal_type.signal_base import SignalType
 
 
-@dataclass
-class ContentTypeConfig:
-    """
-    Holder for ContentType configuration.
-    """
-
-    # Content types that are not enabled should not be used in hashing/matching
-    enabled: bool
-    content_type: t.Type[ContentType]
-
-
-class IContentTypeConfigStore(metaclass=abc.ABCMeta):
-    """Interface for accessing ContentType configuration"""
-
-    @abc.abstractmethod
-    def get_content_type_configs(self) -> t.Mapping[str, ContentTypeConfig]:
-        """
-        Return all installed content types.
-        """
-
-
 @dataclass
 class SignalTypeConfig:
     """
@@ -117,3 +96,24 @@ def get_enabled_signal_types_for_content_type(
             for k, v in self.get_signal_type_configs().items()
             if v.enabled and content_type in v.signal_type.get_content_types()
         }
+
+
+@dataclass
+class ContentTypeConfig:
+    """
+    Holder for ContentType configuration.
+    """
+
+    # Content types that are not enabled should not be used in hashing/matching
+    enabled: bool
+    content_type: t.Type[ContentType]
+
+
+class IContentTypeConfigStore(metaclass=abc.ABCMeta):
+    """Interface for accessing ContentType configuration"""
+
+    @abc.abstractmethod
+    def get_content_type_configs(self) -> t.Mapping[str, ContentTypeConfig]:
+        """
+        Return all installed content types.
+        """