⚡️ Speed up method SlackDataSource.admin_teams_settings_set_description by 46%

[codeflash-ai]

📄 46% (0.46x) speedup for SlackDataSource.admin_teams_settings_set_description in backend/python/app/sources/external/slack/slack.py

⏱️ Runtime : 10.5 milliseconds → 7.21 milliseconds (best of 178 runs)

📝 Explanation and details

The optimized code achieves a 46% runtime improvement and 61.8% throughput increase through three key optimizations:

1. Method Reference Caching
The biggest performance gain comes from caching the Slack method reference in self._admin_set_desc_method. The original code performed expensive hasattr() and getattr() lookups on every call (504,918ns total time in line profiler). The optimized version caches this lookup, reducing repeated attribute resolution from ~851ns per hit to ~223ns per hit in the hot path.

2. Efficient Dictionary Construction
Instead of conditionally building kwargs_api with multiple if statements and assignments, the optimized code constructs it directly with known required parameters:

kwargs_api = {
    'team_id': team_id, 
    'description': description
}

This eliminates 4 conditional checks per call and reduces the construction overhead.

3. Optimized Response Data Extraction
In _handle_slack_response(), replaced hasattr(response, 'data') with getattr(response, 'data', None), reducing attribute lookup overhead. The line profiler shows this optimization cut the response handling time from 4.1ms to 2.6ms (35% improvement in this function alone).

Performance Impact by Test Type:

• High-concurrency tests (50+ concurrent calls) benefit most from method caching, as the cached reference is reused across all concurrent executions
• Throughput tests see significant gains due to reduced per-call overhead - the optimizations are most effective when the same method is called repeatedly
• Error handling tests maintain the same comprehensive error categorization while benefiting from the faster response processing

The optimizations are particularly effective for this Slack API wrapper because it's designed for high-frequency usage patterns where the same method is called multiple times, making caching highly beneficial.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 627 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 the function/class under test
# Copy-pasted from the provided code (as per instructions)
import logging
from typing import Any, Dict

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


class MockSlackMethod:
    """Simulates the SlackClient.admin_teams_settings_setDescription method."""
    def __init__(self, response_data=None, raise_exc=None):
        self.response_data = response_data
        self.raise_exc = raise_exc

    def __call__(self, **kwargs):
        if self.raise_exc:
            raise self.raise_exc
        # Simulate Slack API response structure
        return type('MockResponse', (), {'data': self.response_data})

class MockWebClient:
    """Simulates a Slack WebClient object."""
    def __init__(self, method=None):
        if method:
            self.admin_teams_settings_setDescription = method

class MockSlackClient:
    """Simulates the SlackClient wrapper."""
    def __init__(self, web_client):
        self.web_client = web_client

    def get_web_client(self):
        return self.web_client

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

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_basic_success():
    """
    Test basic success scenario with valid team_id and description.
    """
    # Simulate Slack API response with 'ok': True
    response_data = {'ok': True, 'team_id': 'T123', 'description': 'A workspace description'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='A workspace description')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_basic_failure():
    """
    Test basic failure scenario where Slack API returns ok: False.
    """
    response_data = {'ok': False, 'error': 'invalid_auth'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_empty_response():
    """
    Test case where Slack API returns None (empty response).
    """
    method = MockSlackMethod(response_data=None)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    # The response object will have .data = None, so _handle_slack_response should handle this
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_with_additional_kwargs():
    """
    Test passing additional kwargs to the Slack API method.
    """
    response_data = {'ok': True, 'team_id': 'T123', 'description': 'desc', 'extra': 'value'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc', extra='value')

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_missing_method_alias():
    """
    Test when Slack client does not have the required method alias.
    """
    web_client = MockWebClient(method=None)  # No admin_teams_settings_setDescription attribute
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_exception_handling_invalid_auth():
    """
    Test error handling for invalid_auth error.
    """
    method = MockSlackMethod(raise_exc=Exception("invalid_auth"))
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_exception_handling_missing_scope():
    """
    Test error handling for missing_scope error.
    """
    method = MockSlackMethod(raise_exc=Exception("missing_scope"))
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_exception_handling_channel_not_found():
    """
    Test error handling for channel_not_found error.
    """
    method = MockSlackMethod(raise_exc=Exception("channel_not_found"))
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='T123', description='desc')

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_concurrent_execution():
    """
    Test concurrent execution of multiple async calls.
    """
    response_data = {'ok': True, 'team_id': 'T123', 'description': 'desc'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)

    async def call_func(idx):
        # Each call should succeed independently
        return await data_source.admin_teams_settings_set_description(team_id=f'T{idx}', description=f'desc{idx}')

    # Run 5 concurrent calls
    results = await asyncio.gather(*(call_func(i) for i in range(5)))
    for i, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_edge_empty_team_id_and_description():
    """
    Test edge case with empty strings for team_id and description.
    """
    response_data = {'ok': True, 'team_id': '', 'description': ''}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)
    result = await data_source.admin_teams_settings_set_description(team_id='', description='')

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_large_scale_concurrent():
    """
    Test large scale concurrent execution with 50 calls.
    """
    response_data = {'ok': True, 'team_id': 'T123', 'description': 'desc'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)

    async def call_func(idx):
        return await data_source.admin_teams_settings_set_description(team_id=f'T{idx}', description=f'desc{idx}')

    results = await asyncio.gather(*(call_func(i) for i in range(50)))
    for result in results:
        pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_small_load():
    """
    Throughput test: small load (10 requests).
    """
    response_data = {'ok': True, 'team_id': 'Tsmall', 'description': 'desc'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)

    async def call_func(idx):
        return await data_source.admin_teams_settings_set_description(team_id=f'T{idx}', description=f'desc{idx}')

    results = await asyncio.gather(*(call_func(i) for i in range(10)))
    for result in results:
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_medium_load():
    """
    Throughput test: medium load (100 requests).
    """
    response_data = {'ok': True, 'team_id': 'Tmed', 'description': 'desc'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)

    async def call_func(idx):
        return await data_source.admin_teams_settings_set_description(team_id=f'T{idx}', description=f'desc{idx}')

    results = await asyncio.gather(*(call_func(i) for i in range(100)))
    for result in results:
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_high_load():
    """
    Throughput test: high load (200 requests).
    """
    response_data = {'ok': True, 'team_id': 'Thigh', 'description': 'desc'}
    method = MockSlackMethod(response_data=response_data)
    web_client = MockWebClient(method=method)
    slack_client = MockSlackClient(web_client)
    data_source = SlackDataSource(slack_client)

    async def call_func(idx):
        return await data_source.admin_teams_settings_set_description(team_id=f'T{idx}', description=f'desc{idx}')

    results = await asyncio.gather(*(call_func(i) for i in range(200)))
    for result in results:
        pass
# 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 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

# --- Mocks and minimal stubs ---

class SlackResponse:
    """Minimal stub for SlackResponse returned by SlackDataSource methods."""
    def __init__(self, success, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error

    def __eq__(self, other):
        if not isinstance(other, SlackResponse):
            return False
        return self.success == other.success and self.data == other.data and self.error == other.error

class DummySlackClient:
    """A dummy Slack client with controllable behavior for testing."""

    def __init__(self):
        self.call_args = []
        self.should_raise = None
        self.response_data = None
        self.has_method = True

    def admin_teams_settings_setDescription(self, **kwargs):
        """Simulate the Slack API method."""
        self.call_args.append(kwargs)
        if self.should_raise:
            raise self.should_raise
        # Simulate returning an object with a .data attribute
        if isinstance(self.response_data, Exception):
            raise self.response_data
        if self.response_data is not None:
            # If it's a dict, wrap in a dummy object with .data
            return SimpleNamespace(data=self.response_data)
        # Default: return ok=True
        return SimpleNamespace(data={"ok": True, "team_id": kwargs.get("team_id"), "description": kwargs.get("description")})

    # Used to simulate the method being missing
    def remove_method(self):
        self.has_method = False
        delattr(self, "admin_teams_settings_setDescription")

    def get_web_client(self):
        """Return self as the web client."""
        if self.has_method:
            return self
        # Simulate missing method
        class NoMethodClient:
            pass
        return NoMethodClient()

class DummySlackClientDictResponse(DummySlackClient):
    """Returns a dict-like object, not an object with .data."""
    def admin_teams_settings_setDescription(self, **kwargs):
        self.call_args.append(kwargs)
        return {"ok": True, "team_id": kwargs.get("team_id"), "description": kwargs.get("description")}

# --- Begin tests ---

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_basic_success():
    """Test basic successful call returns expected SlackResponse with ok=True."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    team_id = "T123"
    description = "A cool workspace"
    resp = await ds.admin_teams_settings_set_description(team_id=team_id, description=description)

@pytest.mark.asyncio

async def test_admin_teams_settings_set_description_basic_with_extra_kwargs():
    """Test function passes extra kwargs to the underlying method."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    extra = {"foo": "bar", "baz": 42}
    await ds.admin_teams_settings_set_description(team_id="T1", description="desc", **extra)

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_handles_dict_like_response():
    """Test function handles dict-like response (not object with .data)."""
    dummy_client = DummySlackClientDictResponse()
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_handles_raw_response():
    """Test function handles response with no .data and not dict-like."""
    class WeirdResponse:
        def __str__(self):
            return "<weird>"
    class DummyClient(DummySlackClient):
        def admin_teams_settings_setDescription(self, **kwargs):
            return WeirdResponse()
    ds = SlackDataSource(DummyClient())
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_empty_response():
    """Test function handles empty response (None)."""
    class DummyClient(DummySlackClient):
        def admin_teams_settings_setDescription(self, **kwargs):
            return None
    ds = SlackDataSource(DummyClient())
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_from_api():
    """Test function handles API error (raises Exception)."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = ValueError("invalid_auth")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_token_type():
    """Test function handles not_allowed_token_type error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("not_allowed_token_type: must use xoxb token")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_missing_scope():
    """Test function handles missing_scope error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("missing_scope: admin.teams:write")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_account_inactive():
    """Test function handles account_inactive error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("account_inactive")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_token_revoked():
    """Test function handles token_revoked error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("token_revoked")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_channel_not_found():
    """Test function handles channel_not_found error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("channel_not_found")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_not_in_channel():
    """Test function handles not_in_channel error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("not_in_channel")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_error_generic():
    """Test function handles generic error message."""
    dummy_client = DummySlackClient()
    dummy_client.should_raise = RuntimeError("some_other_error")
    ds = SlackDataSource(dummy_client)
    resp = await ds.admin_teams_settings_set_description(team_id="T1", description="desc")

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_concurrent_calls():
    """Test concurrent execution of the function."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    team_ids = [f"T{i}" for i in range(5)]
    descriptions = [f"desc-{i}" for i in range(5)]
    # Run 5 concurrent calls
    results = await asyncio.gather(*[
        ds.admin_teams_settings_set_description(team_id=team_ids[i], description=descriptions[i])
        for i in range(5)
    ])
    for i, resp in enumerate(results):
        pass

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_many_concurrent_calls():
    """Test the function under a moderate concurrent load (50 calls)."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    n = 50
    # Prepare unique team_id/description pairs
    calls = [
        ds.admin_teams_settings_set_description(
            team_id=f"T{i}", description=f"desc-{i}"
        ) for i in range(n)
    ]
    results = await asyncio.gather(*calls)
    # All should succeed and return the correct data
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_large_payload():
    """Test the function with a large description string."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    large_desc = "A" * 1000  # 1000 characters, still reasonable
    resp = await ds.admin_teams_settings_set_description(team_id="Tbig", description=large_desc)

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_small_load():
    """Throughput: test function performance with 10 concurrent calls."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    n = 10
    calls = [
        ds.admin_teams_settings_set_description(team_id=f"T{i}", description=f"desc-{i}")
        for i in range(n)
    ]
    results = await asyncio.gather(*calls)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_medium_load():
    """Throughput: test function performance with 100 concurrent calls."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    n = 100
    calls = [
        ds.admin_teams_settings_set_description(team_id=f"T{i}", description=f"desc-{i}")
        for i in range(n)
    ]
    results = await asyncio.gather(*calls)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_error_mix():
    """Throughput: test function with a mix of successful and error responses."""
    class MixedDummyClient(DummySlackClient):
        def admin_teams_settings_setDescription(self, **kwargs):
            if kwargs["team_id"].endswith("9"):
                raise RuntimeError("token_revoked")
            return super().admin_teams_settings_setDescription(**kwargs)
    dummy_client = MixedDummyClient()
    ds = SlackDataSource(dummy_client)
    n = 20
    calls = [
        ds.admin_teams_settings_set_description(team_id=f"T{i}", description=f"desc-{i}")
        for i in range(n)
    ]
    results = await asyncio.gather(*calls)
    for i, resp in enumerate(results):
        if str(i).endswith("9"):
            pass
        else:
            pass

@pytest.mark.asyncio
async def test_admin_teams_settings_set_description_throughput_large_description():
    """Throughput: test function with large descriptions in concurrent calls."""
    dummy_client = DummySlackClient()
    ds = SlackDataSource(dummy_client)
    n = 20
    desc = "X" * 500
    calls = [
        ds.admin_teams_settings_set_description(team_id=f"T{i}", description=desc)
        for i in range(n)
    ]
    results = await asyncio.gather(*calls)
    for resp in 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_set_description-mh6agj3y and push.