⚡️ Speed up method SlackDataSource.admin_teams_settings_info by 36%

[codeflash-ai]

📄 36% (0.36x) speedup for SlackDataSource.admin_teams_settings_info in backend/python/app/sources/external/slack/slack.py

⏱️ Runtime : 3.10 milliseconds → 2.28 milliseconds (best of 247 runs)

📝 Explanation and details

The optimized code achieves a 36% runtime improvement and 59.4% throughput increase through several targeted micro-optimizations:

1. Streamlined response data extraction

• Uses getattr(response, 'data', None) directly instead of hasattr() + attribute access, eliminating redundant attribute lookups
• Restructured the data extraction flow to handle the most common case (responses with .data attribute) first

2. Optimized error handling with dictionary lookup

• Replaced sequential if/elif chains with a single dictionary containing all error mappings
• Uses a single loop with in operator for substring matching, reducing from O(n) separate string searches to O(1) dictionary access + O(k) loop where k=7 error types

3. Reduced attribute access overhead

• In admin_teams_settings_info(), stores getattr(self.client, 'admin_teams_settings_info', None) in a variable to avoid calling getattr twice
• Eliminates redundant hasattr() + callable(getattr()) pattern

4. Smarter dictionary initialization

• Conditionally creates kwargs_api dictionary only when team_id is not None, avoiding unnecessary empty dict creation and assignment

5. Improved success/error logic

• Changes default success value from True to data.get("ok", True) and restructures conditionals to reduce branching in the common success path
• Combines multiple condition checks into single statements where possible

These optimizations are particularly effective for:

• High-throughput scenarios (200+ concurrent requests) where the reduced attribute lookups compound significantly
• Success-heavy workloads where most Slack API calls succeed, benefiting from the streamlined success path
• Error-prone environments where the O(1) error lookup provides consistent performance regardless of error type

The changes maintain full functional compatibility while reducing CPU overhead through more efficient Python object access patterns.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 643 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
import logging
from types import SimpleNamespace
from typing import Any, Dict

import pytest  # used for our unit tests
from app.sources.external.slack.slack import SlackDataSource


# Minimal stub for WebClient with dynamic method injection
class WebClientStub:
    def __init__(self):
        self._methods = {}

    def add_method(self, name, func):
        setattr(self, name, func)
        self._methods[name] = func

    def get_web_client(self):
        return self

# Minimal stub for SlackClient
class SlackClient:
    def __init__(self, web_client):
        self._web_client = web_client

    def get_web_client(self):
        return self._web_client

# --- Unit tests for SlackDataSource.admin_teams_settings_info ---

@pytest.mark.asyncio
async def test_admin_teams_settings_info_basic_success():
    """Test basic successful response with 'ok': True and data"""
    # Prepare a stub WebClient with the required method
    def admin_teams_settings_info(team_id, **kwargs):
        # Simulate a Slack API response object with 'data' attribute
        return SimpleNamespace(data={'ok': True, 'team_id': team_id, 'settings': {'foo': 'bar'}})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    # Call the async function
    resp = await ds.admin_teams_settings_info(team_id="T12345")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_basic_error_response():
    """Test response with 'ok': False and error message"""
    def admin_teams_settings_info(team_id, **kwargs):
        return SimpleNamespace(data={'ok': False, 'error': 'team_not_found'})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T_NOT_FOUND")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_missing_method():
    """Test when Slack client does not have the required method"""
    web_client = WebClientStub()  # No method added
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T12345")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_none_response():
    """Test when the Slack API method returns None"""
    def admin_teams_settings_info(team_id, **kwargs):
        return None
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T12345")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_dict_like_response():
    """Test when the Slack API method returns a dict-like object"""
    class DictLike(dict):
        def __getattr__(self, item):
            return self[item]
    def admin_teams_settings_info(team_id, **kwargs):
        return DictLike({'ok': True, 'team_id': team_id, 'settings': {'foo': 'bar'}})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T12345")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_error_field_only():
    """Test when the Slack API returns only an 'error' field (no 'ok')"""
    def admin_teams_settings_info(team_id, **kwargs):
        return SimpleNamespace(data={'error': 'invalid_auth'})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T12345")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_exception_handling():
    """Test when the Slack API method raises an exception"""
    def admin_teams_settings_info(team_id, **kwargs):
        raise RuntimeError("not_allowed_token_type: Only bot tokens allowed")
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T12345")

@pytest.mark.asyncio

async def test_admin_teams_settings_info_kwargs_passed():
    """Test that extra kwargs are passed through to the Slack API method"""
    captured = {}
    def admin_teams_settings_info(team_id, **kwargs):
        captured['team_id'] = team_id
        captured.update(kwargs)
        return SimpleNamespace(data={'ok': True, 'team_id': team_id, 'extra': kwargs})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    resp = await ds.admin_teams_settings_info(team_id="T12345", foo="bar", baz=123)

@pytest.mark.asyncio
async def test_admin_teams_settings_info_concurrent_execution():
    """Test concurrent execution of the async function"""
    def admin_teams_settings_info(team_id, **kwargs):
        # Simulate a unique response per team_id
        return SimpleNamespace(data={'ok': True, 'team_id': team_id})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    team_ids = [f"T{i:05d}" for i in range(10)]
    coros = [ds.admin_teams_settings_info(team_id=tid) for tid in team_ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_info_large_scale_concurrent():
    """Test large scale concurrent execution (within reasonable bounds)"""
    def admin_teams_settings_info(team_id, **kwargs):
        return SimpleNamespace(data={'ok': True, 'team_id': team_id})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    team_ids = [f"T{i:05d}" for i in range(50)]
    coros = [ds.admin_teams_settings_info(team_id=tid) for tid in team_ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass

# --- Throughput tests ---

@pytest.mark.asyncio
async def test_admin_teams_settings_info_throughput_small_load():
    """Throughput test: small load of 5 concurrent requests"""
    def admin_teams_settings_info(team_id, **kwargs):
        return SimpleNamespace(data={'ok': True, 'team_id': team_id})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    team_ids = [f"T{i:05d}" for i in range(5)]
    coros = [ds.admin_teams_settings_info(team_id=tid) for tid in team_ids]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_admin_teams_settings_info_throughput_medium_load():
    """Throughput test: medium load of 20 concurrent requests"""
    def admin_teams_settings_info(team_id, **kwargs):
        return SimpleNamespace(data={'ok': True, 'team_id': team_id})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    team_ids = [f"T{i:05d}" for i in range(20)]
    coros = [ds.admin_teams_settings_info(team_id=tid) for tid in team_ids]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_admin_teams_settings_info_throughput_high_load():
    """Throughput test: high load of 100 concurrent requests (performance, but bounded)"""
    def admin_teams_settings_info(team_id, **kwargs):
        return SimpleNamespace(data={'ok': True, 'team_id': team_id})
    web_client = WebClientStub()
    web_client.add_method('admin_teams_settings_info', admin_teams_settings_info)
    slack_client = SlackClient(web_client)
    ds = SlackDataSource(slack_client)

    team_ids = [f"T{i:05d}" for i in range(100)]
    coros = [ds.admin_teams_settings_info(team_id=tid) for tid in team_ids]
    results = await asyncio.gather(*coros)
    # Check that all team_ids are present in the results
    result_ids = set(resp.data['team_id'] for resp in results)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import asyncio  # used to run async functions
# Import the function to test (copied exactly from the prompt)
import logging
from typing import Any, Dict

import pytest  # used for our unit tests
from app.sources.external.slack.slack import SlackDataSource


# Mocks and stubs for SlackClient and SlackResponse
class SlackResponse:
    def __init__(self, success, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error

class MockSlackClient:
    """Mock Slack client to simulate Slack WebClient behavior."""

    def __init__(self, behavior=None):
        """
        behavior: dict mapping method names to callables or values
        """
        self.behavior = behavior or {}

    def get_web_client(self):
        return self

    # The method under test
    def admin_teams_settings_info(self, **kwargs):
        # Simulate various behaviors based on input
        if self.behavior and "admin_teams_settings_info" in self.behavior:
            handler = self.behavior["admin_teams_settings_info"]
            if callable(handler):
                return handler(**kwargs)
            else:
                return handler
        # Default: simulate a successful Slack API response
        team_id = kwargs.get("team_id")
        if team_id == "fail":
            # Simulate Slack error response
            return {"ok": False, "error": "team_not_found"}
        elif team_id == "exception":
            # Simulate raising an exception
            raise RuntimeError("Simulated Slack API exception")
        elif team_id is None:
            # Simulate error for missing team_id
            return {"ok": False, "error": "missing_team_id"}
        # Simulate a successful response
        return {"ok": True, "team_id": team_id, "settings": {"name": f"Team {team_id}", "is_active": True}}

# ------------------ UNIT TESTS ------------------

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_info_basic_success():
    """Test basic successful Slack API call with valid team_id."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="T123")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_basic_error_response():
    """Test error response from Slack API when team_id is 'fail'."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="fail")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_missing_team_id():
    """Test error response when team_id is None."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id=None)

@pytest.mark.asyncio
async def test_admin_teams_settings_info_with_additional_kwargs():
    """Test passing additional kwargs to the Slack API method."""
    def handler(**kwargs):
        # Simulate returning all kwargs in the response
        return {"ok": True, "team_id": kwargs["team_id"], "extra": kwargs.get("foo", "bar")}
    client = MockSlackClient(behavior={"admin_teams_settings_info": handler})
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="T999", foo="baz")

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_info_client_missing_method():
    """Test when Slack client is missing the required method alias."""
    class NoMethodClient(MockSlackClient):
        def __getattr__(self, item):
            if item == "admin_teams_settings_info":
                raise AttributeError("Method not implemented")
            return super().__getattribute__(item)
    client = NoMethodClient()
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="T123")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_exception_handling():
    """Test Slack API exception handling in async context."""
    def handler(**kwargs):
        raise RuntimeError("invalid_auth")
    client = MockSlackClient(behavior={"admin_teams_settings_info": handler})
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="T123")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_concurrent_calls():
    """Test concurrent execution of multiple admin_teams_settings_info calls."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    team_ids = [f"T{i}" for i in range(5)]
    results = await asyncio.gather(*[ds.admin_teams_settings_info(team_id=tid) for tid in team_ids])
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_info_edge_empty_response():
    """Test handling of empty response from Slack API."""
    def handler(**kwargs):
        return None  # Simulate empty response
    client = MockSlackClient(behavior={"admin_teams_settings_info": handler})
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="T123")

@pytest.mark.asyncio
async def test_admin_teams_settings_info_edge_non_dict_response():
    """Test handling of non-dict response from Slack API."""
    def handler(**kwargs):
        return "unexpected string"
    client = MockSlackClient(behavior={"admin_teams_settings_info": handler})
    ds = SlackDataSource(client)
    result = await ds.admin_teams_settings_info(team_id="T123")

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_info_large_scale_concurrency():
    """Test large scale concurrent calls (up to 100 calls)."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    team_ids = [f"T{i}" for i in range(100)]
    results = await asyncio.gather(*[ds.admin_teams_settings_info(team_id=tid) for tid in team_ids])
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_info_large_scale_error_mix():
    """Test large scale concurrent calls with mixed success and error."""
    def handler(**kwargs):
        tid = kwargs.get("team_id")
        if tid.endswith("5"):
            return {"ok": False, "error": "team_not_found"}
        return {"ok": True, "team_id": tid}
    client = MockSlackClient(behavior={"admin_teams_settings_info": handler})
    ds = SlackDataSource(client)
    team_ids = [f"T{i}" for i in range(50)]
    results = await asyncio.gather(*[ds.admin_teams_settings_info(team_id=tid) for tid in team_ids])
    for idx, result in enumerate(results):
        if team_ids[idx].endswith("5"):
            pass
        else:
            pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_info_throughput_small_load():
    """Throughput test: small load (10 concurrent calls)."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    team_ids = [f"T{i}" for i in range(10)]
    results = await asyncio.gather(*[ds.admin_teams_settings_info(team_id=tid) for tid in team_ids])
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_info_throughput_medium_load():
    """Throughput test: medium load (50 concurrent calls)."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    team_ids = [f"T{i}" for i in range(50)]
    results = await asyncio.gather(*[ds.admin_teams_settings_info(team_id=tid) for tid in team_ids])
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_info_throughput_high_load():
    """Throughput test: high load (200 concurrent calls)."""
    client = MockSlackClient()
    ds = SlackDataSource(client)
    team_ids = [f"T{i}" for i in range(200)]
    results = await asyncio.gather(*[ds.admin_teams_settings_info(team_id=tid) for tid in team_ids])
    for idx, result in enumerate(results):
        pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-SlackDataSource.admin_teams_settings_info-mh69j6m3 and push.