airbyte_cdk.sources.streams.concurrent.availability_strategy 

 1#
 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 3#
 4
 5from abc import ABC, abstractmethod
 6from typing import Optional
 7
 8
 9class StreamAvailability:
10    @classmethod
11    def available(cls) -> "StreamAvailability":
12        return cls(True)
13
14    @classmethod
15    def unavailable(cls, reason: str) -> "StreamAvailability":
16        return cls(False, reason)
17
18    def __init__(self, available: bool, reason: Optional[str] = None) -> None:
19        self._available = available
20        self._reason = reason
21
22        if not available:
23            assert reason, "A reason needs to be provided if the stream is not available"
24
25    @property
26    def is_available(self) -> bool:
27        """
28        :return: True if the stream is available. False if the stream is not
29        """
30        return self._available
31
32    @property
33    def reason(self) -> Optional[str]:
34        """
35        :return: A message describing why the stream is not available. If the stream is available, this should return None.
36        """
37        return self._reason
class StreamAvailability: 
10class StreamAvailability:
11    @classmethod
12    def available(cls) -> "StreamAvailability":
13        return cls(True)
14
15    @classmethod
16    def unavailable(cls, reason: str) -> "StreamAvailability":
17        return cls(False, reason)
18
19    def __init__(self, available: bool, reason: Optional[str] = None) -> None:
20        self._available = available
21        self._reason = reason
22
23        if not available:
24            assert reason, "A reason needs to be provided if the stream is not available"
25
26    @property
27    def is_available(self) -> bool:
28        """
29        :return: True if the stream is available. False if the stream is not
30        """
31        return self._available
32
33    @property
34    def reason(self) -> Optional[str]:
35        """
36        :return: A message describing why the stream is not available. If the stream is available, this should return None.
37        """
38        return self._reason
StreamAvailability(available: bool, reason: Optional[str] = None) 
19    def __init__(self, available: bool, reason: Optional[str] = None) -> None:
20        self._available = available
21        self._reason = reason
22
23        if not available:
24            assert reason, "A reason needs to be provided if the stream is not available"
@classmethod
def available( cls) -> StreamAvailability: 
11    @classmethod
12    def available(cls) -> "StreamAvailability":
13        return cls(True)
@classmethod
def unavailable( cls, reason: str) -> StreamAvailability: 
15    @classmethod
16    def unavailable(cls, reason: str) -> "StreamAvailability":
17        return cls(False, reason)
is_available: bool 
26    @property
27    def is_available(self) -> bool:
28        """
29        :return: True if the stream is available. False if the stream is not
30        """
31        return self._available
Returns

True if the stream is available. False if the stream is not

reason: Optional[str] 
33    @property
34    def reason(self) -> Optional[str]:
35        """
36        :return: A message describing why the stream is not available. If the stream is available, this should return None.
37        """
38        return self._reason
Returns

A message describing why the stream is not available. If the stream is available, this should return None.