sadpandajoe commented on code in PR #42089:
URL: https://github.com/apache/superset/pull/42089#discussion_r3694559947
##########
superset/utils/slack.py:
##########
@@ -194,65 +401,226 @@ def _get_channels(
raise
-def get_channels_with_search(
+def _filter_slack_channels(
+ channels: list[SlackChannel],
+ *,
+ search_string: str,
+ types: Optional[list[SlackChannelTypes]],
+ exact_match: bool,
+) -> list[SlackChannel]:
+ """Filter a complete Slack channel listing by type and target."""
+ if types and len(types) != len(SlackChannelTypes):
+ conditions: list[Callable[[SlackChannel], bool]] = []
+ if SlackChannelTypes.PUBLIC in types:
+ conditions.append(lambda channel: not channel["is_private"])
+ if SlackChannelTypes.PRIVATE in types:
+ conditions.append(lambda channel: channel["is_private"])
+
+ channels = [
+ channel for channel in channels if any(cond(channel) for cond in
conditions)
+ ]
+
+ if not search_string:
+ return channels
+
+ search_array = recipients_string_to_list(search_string)
+ return [
+ channel
+ for channel in channels
+ if any(
+ (
+ search.casefold() == channel["name"].casefold()
+ or search.casefold() == channel["id"].casefold()
+ if exact_match
+ else (
+ search.casefold() in channel["name"].casefold()
+ or search.casefold() in channel["id"].casefold()
+ )
+ )
+ for search in search_array
+ )
+ ]
+
+
+def _get_channels_with_search(
search_string: str = "",
types: Optional[list[SlackChannelTypes]] = None,
exact_match: bool = False,
force: bool = False,
-) -> list[SlackChannelSchema]:
+ cache: bool = True,
+ *,
+ return_cache_status: bool = False,
+) -> tuple[list[SlackChannel], bool]:
"""
The slack api is paginated but does not include search, so we need to fetch
all channels and filter them ourselves
This will search by slack name or id
"""
+ used_cache = False
+ cache_timeout = app.config["SLACK_CACHE_TIMEOUT"]
+ cache_enabled = cache and cache_timeout != CACHE_DISABLED_TIMEOUT
try:
- channels = get_channels(
- force=force,
- cache_timeout=app.config["SLACK_CACHE_TIMEOUT"],
- )
+ if return_cache_status and cache_enabled and not force:
+ channels, used_cache = _get_channels_with_cache_status()
+ else:
+ channels = get_channels(
+ force=force,
+ cache=cache_enabled,
+ cache_timeout=cache_timeout,
+ )
except SlackApiError as ex:
- # Check if it's a rate limit error
- status_code = getattr(ex.response, "status_code", None)
- if status_code == 429:
- raise SupersetException(
- f"Slack API rate limit exceeded: {ex}. "
- "For large workspaces, consider increasing "
- "SLACK_API_RATE_LIMIT_RETRY_COUNT"
- ) from ex
- raise SupersetException(f"Failed to list channels: {ex}") from ex
- except SlackClientError as ex:
- raise SupersetException(f"Failed to list channels: {ex}") from ex
+ error_code = get_slack_api_error_code(ex)
+ error_class = (
+ SlackChannelListingError
+ if is_transient_slack_api_error(ex, error_code)
+ else SlackChannelListingClientError
+ )
+ message = f"Failed to list channels: {ex}"
+ if get_slack_api_status_code(ex) == 429:
+ message = (
+ f"Slack API rate limit exceeded: {ex}. For large workspaces, "
+ "consider increasing SLACK_API_RATE_LIMIT_RETRY_COUNT"
+ )
+ raise error_class(message) from ex
+ except SLACK_TRANSIENT_TRANSPORT_ERRORS as ex:
+ error_class = (
+ SlackChannelListingError
+ if is_transient_slack_transport_error(ex)
+ else SlackChannelListingClientError
+ )
+ raise error_class(f"Failed to list channels: {ex}") from ex
+ except (SlackSDKClientError, SlackClientError) as ex:
+ raise SlackChannelListingClientError(f"Failed to list channels: {ex}")
from ex
+
+ channels = _filter_slack_channels(
+ channels,
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ )
+ return channels, used_cache
- if types and not len(types) == len(SlackChannelTypes):
- conditions: list[Callable[[SlackChannelSchema], bool]] = []
- if SlackChannelTypes.PUBLIC in types:
- conditions.append(lambda channel: not channel["is_private"])
- if SlackChannelTypes.PRIVATE in types:
- conditions.append(lambda channel: channel["is_private"])
- channels = [
- channel for channel in channels if any(cond(channel) for cond in
conditions)
- ]
+def get_channels_with_search(
+ search_string: str = "",
+ types: Optional[list[SlackChannelTypes]] = None,
+ exact_match: bool = False,
+ force: bool = False,
+ cache: bool = True,
+) -> list[SlackChannel]:
+ """Fetch and filter Slack channels without exposing cache provenance."""
+ channels, _ = _get_channels_with_search(
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ force=force,
+ cache=cache,
+ )
+ return channels
- # The search string can be multiple channels separated by commas
- if search_string:
- search_array = recipients_string_to_list(search_string)
- channels = [
- channel
- for channel in channels
- if any(
- (
- search.lower() == channel["name"].lower()
- or search.lower() == channel["id"].lower()
- if exact_match
- else (
- search.lower() in channel["name"].lower()
- or search.lower() in channel["id"].lower()
- )
- )
- for search in search_array
+
+def get_channels_with_search_and_cache_status(
+ search_string: str = "",
+ types: Optional[list[SlackChannelTypes]] = None,
+ exact_match: bool = False,
+) -> tuple[list[SlackChannel], bool]:
+ """Fetch filtered Slack channels and report whether the listing was
cached."""
+ return _get_channels_with_search(
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ return_cache_status=True,
+ )
+
+
+def refresh_cached_slack_channels_with_search(
+ search_string: str = "",
+ types: Optional[list[SlackChannelTypes]] = None,
+ exact_match: bool = False,
+) -> list[SlackChannel]:
+ """Refresh stale channels with a best-effort cache-backend cooldown.
+
+ External cache backends record a cooldown only after the refreshed listing
+ is stored successfully. Disabled and metastore-backed caches use an
uncached
+ request without a cooldown because metastore writes commit the report
+ transaction. Concurrent workers can still refresh in parallel when the
+ backend cannot provide transaction-safe coordination.
+ """
+ team_id = get_team_id()
+ cache_key = _get_slack_channels_cache_key(team_id)
+ cooldown_key = f"{cache_key}_refresh_cooldown"
+ cache_timeout = app.config["SLACK_CACHE_TIMEOUT"]
+
+ if (
+ _slack_channel_cache_uses_report_session()
+ or cache_timeout == CACHE_DISABLED_TIMEOUT
+ ):
+ return get_channels_with_search(
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ force=True,
+ cache=False,
+ )
+
+ try:
+ refresh_is_recent = cache_manager.cache.get(cooldown_key) is not None
+ except Exception: # pylint: disable=broad-exception-caught
+ refresh_is_recent = False
+ logger.warning(
+ "Could not read Slack channel refresh cooldown; refreshing from
Slack",
+ exc_info=True,
+ )
+
+ if refresh_is_recent:
+ channels, _ = _get_channels_with_cache_status()
+ return _filter_slack_channels(
+ channels,
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ )
+
+ refreshed_channels = get_channels_with_search(
Review Comment:
Fixed in `bdbdb4fcae` — refresh ownership is claimed atomically with the
cache backend’s `add`. Claim losers read the shared cache directly and never
enumerate Slack, including on an empty or unreadable cache; the owner releases
its claim after a failed listing or cache write. The owner/loser and failure
paths are covered by unit tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
