⚡️ Speed up method StreamAvailabilityStrategy.check_availability by 12% in PR #45673 (async-job-salesforce/cdk-release)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #45673

If you approve this dependent PR, these changes will be merged into the original PR branch async-job-salesforce/cdk-release.

This PR will be automatically closed if the original PR is merged.

---

📄 StreamAvailabilityStrategy.check_availability() in airbyte-cdk/python/airbyte_cdk/sources/streams/concurrent/adapters.py

📈 Performance improved by 12% (0.12x faster)

⏱️ Runtime went down from 30.6 microseconds to 27.2 microseconds

Explanation and details

List of Optimizations.

Correctness verification

The new optimized code was tested for correctness. The results are listed below.

✅ 1 Passed − ⚙️ Existing Unit Tests

(click to show existing tests)

- sources/streams/concurrent/test_adapters.py

✅ 0 Passed − 🌀 Generated Regression Tests

(click to show generated tests)

# imports
# function to test
import logging
from abc import ABC, abstractmethod
from typing import Any, Optional
from unittest.mock import Mock

import pytest  # used for our unit tests
from airbyte_cdk.sources import Source
from airbyte_cdk.sources.streams import Stream
from airbyte_cdk.sources.streams.concurrent.adapters import \
    StreamAvailabilityStrategy
from airbyte_cdk.sources.streams.concurrent.availability_strategy import \
    AbstractAvailabilityStrategy

# unit tests

# Mock classes for testing
class MockStream(Stream):
    def __init__(self, check_availability=None, get_error_display_message=None):
        self._check_availability = check_availability
        self._get_error_display_message = get_error_display_message
        # Outputs were verified to be equal to the original implementation

    def check_availability(self, logger, source):
        if self._check_availability:
            return self._check_availability(logger, source)
        raise NotImplementedError
        # Outputs were verified to be equal to the original implementation

    def get_error_display_message(self, exception):
        if self._get_error_display_message:
            return self._get_error_display_message(exception)
        return None
        # Outputs were verified to be equal to the original implementation

class MockSource(Source):
    pass

class MockLogger(logging.Logger):
    def __init__(self):
        super().__init__("mock_logger")
        # Outputs were verified to be equal to the original implementation

# Basic Test Cases

🔘 (none found) − ⏪ Replay Tests

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Sep 19, 2024 6:05pm

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}