tap-exact/tap_exact/tap.py at 8513ad7e5e6e6e70074cdc7f7c3475eecc73e563 · TicketSwap/tap-exact · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
"""Exact tap class."""

from __future__ import annotations

from singer_sdk import Tap
from singer_sdk.typing import (
    ArrayType,
    DateTimeType,
    PropertiesList,
    Property,
    StringType,
)

from tap_exact import streams


class TapExact(Tap):
    """Exact tap class."""

    name = "tap-exact"

    config_jsonschema = PropertiesList(
        Property("start_date", DateTimeType),
        Property("client_id", StringType, secret=True, required=True),
        Property("client_secret", StringType, secret=True, required=True),
        Property("tokens_s3_bucket", StringType, required=True),
        Property("tokens_s3_key", StringType, required=True),
        Property("divisions", ArrayType(StringType), required=True),
    ).to_dict()

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

        Returns:
            A list of discovered streams.
        """
        return [
            streams.GLAccountsStream(self),
            streams.GLClassificationsStream(self),
            streams.GLAccountClassificationMappingsStream(self),
            streams.TransactionLinesStream(self),
            streams.AssetsStream(self),
            streams.DeletedStream(self),
        ]


if __name__ == "__main__":
    TapExact.cli()