requests.exceptions.ConnectionError: HTTPSConnectionPool(host='<http://login.microsoftonline.com|login.microsoftonline.com>', port=443): Max retries exceeded with url: /b5953ddd-dd80-4110-904b-e503716f0caf/oauth2/v2.0/token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7b99afdc5ed0>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution'))

prepare_request

RetriableAPIError

GET /documents

GET /document/<doc_id>

RESTStream

discover_streams

_custom_initialization

def discover_streams(self) -> list[streams.GoogleSearchConsoleStream]:
        """Return a list of discovered streams.

        Returns:
            A list of discovered streams.
        """
        self._custom_initialization()
        return [
            streams.PerformanceReportPage(self, service=self.service),
            streams.PerformanceReportDate(self, service=self.service),
            streams.PerformanceReportCountry(self, service=self.service),
            streams.PerformanceReportQuery(self, service=self.service),
            streams.PerformanceReportDevice(self, service=self.service),
        ]

_custom_initialization

Tap

discover_streams

run_id

Tap.discover_streams

Stream.get_records

Tap.sync_all

@final

setup

get_records

class EventsStream(EgnyteStream):
    """Define custom stream."""

    name = "events"
    path = "/pubapi/v2/audit/stream"
    replication_key = "eventDate"
    records_jsonpath = "$.events[*]"
    is_sorted = True
    page = 0

    def get_url_params(
        self,
        context: Optional[dict],
        next_page_token: Optional[Any],
    ) -> dict[str, Any]:
        params: dict = {}
        if next_page_token:
            params["nextCursor"] = next_page_token
        else:
            params["startDate"] = self.start_time
        return params
    
    def parse_response(self, response: requests.Response | None):
        if not response:
            raise RetriableAPIError("No response from API")

        try:
            data = response.json()
        except AttributeError as e:
            <http://logging.info|logging.info>(f"{e} with response {response} and {response.text}")
            return

        events = data.get("events", [])

        for event in events:
            if not event:
                continue

            for key in ["loginDate", "logoutDate"]:
                if event.get(key) == 0:
                    event.pop(key, None)

            for key in ["loginDate", "logoutDate", "date"]:
                if key in event:
                    dt = datetime(1970, 1, 1, tzinfo=timezone.utc) + \
                        timedelta(milliseconds=event[key])
                    event["eventDate"] = dt.isoformat().replace("+00:00", "Z")
                    del event[key]
                    break

        if events:
            events.sort(key=lambda x: x.get("eventDate", ""))
        self.page = self.page + 1
        <http://logging.info|logging.info>(f'Page: {self.page} {events}')
        yield from events or ([{}] if data.get("moreEvents") else [])

    def post_process(
        self,
        row: dict,
        context: Context | None = None,
    ) -> dict | None:
        if not row:
            return None

        if 'id' not in row or row['id'] is None:
            row['id'] = str(uuid.uuid4())

        return super().post_process(row, context)

    def get_new_paginator(self) -> EgnyteEventsPaginator:
        return EgnyteEventsPaginator()

'e5deb04e-75ba-4894-8709-001c5f4295d5', 'auditSource': 'FILE_AUDIT', 'eventDate': '2024-12-19T07:14:18.591000Z'}]
2024-12-19 07:44:24,675 | INFO     | target-elasticsearch.events | Starting batch {'batch_id': '720b0104-adee-4855-b3dc-420fe0033a2a', 'batch_start_time': datetime.datetime(2024, 12, 19, 7, 44, 24, 675328, tzinfo=datetime.timezone.utc)} for dev_events_raw_egnyte_000
2024-12-19 07:44:27,219 | INFO     | root                 | Page: 2 [{'sourcePath': '/Shared/Projects/2022/22-0055 Vicksburg National Cemetery Burial Excavation and Stablization/Lab/FORDISC Results/Burial 19_Fordisc Results_updated.docx', 'targetPath': 'N/A', 'user': 'Brittany McClain ( <mailto:bmcclain@owest.com|bmcclain@owest.com> )', 'userId': '1121', 'action': 'Read', 'access': 'Desktop App', 'ipAddress': '174.26.41.229', 'actionInfo': '', 'auditSource': 'FILE_AUDIT', 'eventDate': '2024-12-19T07:09:36Z'},

SQLStream

replication_key

client.py

class TestStream(SQLStream):
    connector_class = TestConnector

class AStream(TestStream):
    name = "a"
    replication_key = "updated_at"

class BStream(TestStream):
    name = "b"
    replication_key = "updated_at"

tap.py

KNOWN_STREAMS = {
  "a": stream.AStream,
  "b": streams.BStream,
}


...


    def discover_streams(self):
        for stream in super().discover_streams():
            if stream_class := KNOWN_STREAMS.get(stream.name):
                stream = stream_class(
                    tap=self,
                    catalog_entry=stream.catalog_entry,
                    connector=self.tap_connector,
                )

            yield stream

2025-02-04T18:44:21.577497Z [info     ] 2025-02-04 13:44:21,576 | INFO     | tap-litmos.lt_UserDetails | Pagination stopped after 0 pages because no records were found in the last response

class LitmosPaginator(BaseAPIPaginator):

    def __init__(self, *args, **kwargs):
        super().__init__(None, *args, **kwargs)

    def has_more(self, response):
        return self.get_next(response) < date.today()

    def get_next(self, response):
        params = dict(parse_qsl(urlparse(response.request.url).query))
        return datetime.strptime(params["to"], OUTPUT_DATE_FORMAT).date() + timedelta(seconds=1)

get_records

tap

stream.get_records()

request_records())

RESTStream

- name: tap-oracle
    variant: s7clarke10
    pip_url: git+<https://github.com/s7clarke10/pipelinewise-tap-oracle.git>
    config:
      default_replication_method: LOG_BASED
      filter_schemas: IFSAPP
      filter_tables:
      - IFSAPP-ABC_CLASS_TAB
      host: xxxx
      port: 1521
      service_name: xxxx
      user: ifsapp
    metadata:
      IFSAPP-ABC_CLASS_TAB:
        replication-method: INCREMENTAL
        replication-key: ROWVERSION

tap-growthbook

meltano run

uv run

get_records

get_records

payload

payload.xyz

payload.xyz

request_records

replication_key_value

test

SegmentsStream

get_child_context(record)

{"segment_id": record["id"]}

SegmentMembersStream

schema

"segment_id"

replication_key

parse_response(response)

python
    def parse_response(self, response):
        for identifier in response.json().get("identifiers", []):
            yield {
                "member_id": identifier["id"],
                "cio_id":    identifier.get("cio_id"),
                "email":     identifier.get("email"),
            }

segment_id

parse_response

segment_id

context

python
    def post_process(self, row, context):
        row["segment_id"] = context["segment_id"]
        return row

post_process