⚡️ Speed up method StreamAvailabilityStrategy.check_availability by 10% 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 10% (0.10x faster)

⏱️ Runtime went down from 29.7 microseconds to 27.0 microseconds

Explanation and details

Certainly! To make this Python program faster, I'll focus on some areas for optimization. Since Python is an interpreted language, some practices favor readability over performance, but there are still ways to optimize such as.

1. Remove Redundant Checks: In the StreamAvailabilityStrategy class, there’s a check for the availability of the check_availability method, which could be moved out of the try-except block to lower the overhead.
2. Avoid Repeated Attribute Lookups: Cache attribute lookups to improve runtime.

Here's the optimized code.

Changes Made.

1. Moved the check for check_availability out of the try block: It reduces the overhead of entering and exiting the try block unnecessarily.
2. Used getattr with default value: This way, the check if stream_check_avail is None makes the code more readable and Pythonic.

These small but potent changes can improve the runtime efficiency and readability of your code. They help in minimizing redundant operations and improve overall performance.

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
import logging
from abc import ABC, abstractmethod
from typing import Optional

import pytest  # used for our unit tests
# function to test
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 MockStreamAvailable(Stream):
    def check_availability(self, logger, source):
        return True, None
        # Outputs were verified to be equal to the original implementation

class MockStreamUnavailable(Stream):
    def check_availability(self, logger, source):
        return False, "Service is down"
        # Outputs were verified to be equal to the original implementation

class MockStreamWithoutCheck(Stream):
    pass

class MockStreamWithException(Stream):
    def check_availability(self, logger, source):
        raise Exception("Generic error")
        # Outputs were verified to be equal to the original implementation

class MockStreamWithUserFriendlyException(Stream):
    def check_availability(self, logger, source):
        raise Exception("Specific error")
        # Outputs were verified to be equal to the original implementation
    
    def get_error_display_message(self, exception):
        return "User-friendly error message"
        # Outputs were verified to be equal to the original implementation

class MockStreamWithEmptyMessage(Stream):
    def check_availability(self, logger, source):
        return False, ""
        # Outputs were verified to be equal to the original implementation

class MockStreamWithNoneMessage(Stream):
    def check_availability(self, logger, source):
        return False, None
        # Outputs were verified to be equal to the original implementation

class MockStreamWithNonBooleanAvailability(Stream):
    def check_availability(self, logger, source):
        return "yes", "Non-boolean availability"
        # Outputs were verified to be equal to the original implementation

class MockStreamWithLatency(Stream):
    def check_availability(self, logger, source):
        import time
        time.sleep(5)  # Simulate latency
        return True, None
        # Outputs were verified to be equal to the original implementation

class MockStreamWithLogger(Stream):
    def check_availability(self, logger, source):
        logger.info("Checking availability")
        return True, None
        # Outputs were verified to be equal to the original implementation

class ComplexStream(Stream):
    def __init__(self, config):
        self.config = config
        # Outputs were verified to be equal to the original implementation

    def check_availability(self, logger, source):
        # Complex logic here
        return True, None
        # Outputs were verified to be equal to the original implementation

# 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 5:28pm

[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.}