"""Base class for all feeds.""" from abc import ABC from stream.feed.base import BaseFeed from ..utils import datadog __all__ = ["BaseExtendedFeed"] class BaseExtendedFeed(BaseFeed, ABC): """Base class for both sync and async feeds.""" __trace_methods: tuple[str, ...] = ( "remove_activity", "get", "follow", "unfollow", "followers", "following", "update_activity_to_targets", ) def __init__(self, *args, **kwargs): """Patch Feed methods to add datadog traces.""" super().__init__(*args, **kwargs) datadog.wrap_methods(self, self.__trace_methods, name_prefix="stream.feed")