Skip to content

prefect_fivetran.clients

Clients for interacting with the Fivetran API

FivetranClient

Client for interacting with the Fivetran API.

Parameters:

Name Type Description Default
api_key str

API key to authenticate with the Fivetran API.

 required
api_secret str

API secret to authenticate with the Fivetran API.

 required
Source code in prefect_fivetran/clients.py 
 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
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class FivetranClient:
    """
    Client for interacting with the Fivetran API.
    Args:
        api_key: API key to authenticate with the Fivetran API.
        api_secret: API secret to authenticate with the Fivetran API.
    """

    def __init__(
        self,
        api_key: str,
        api_secret: str,
    ):
        if not api_key:
            raise ValueError("Value for parameter `api_key` must be provided.")
        if not api_secret:
            raise ValueError("Value for parameter `api_secret` must be provided.")

        self._closed = False
        self._started = False

        self.client = AsyncClient(
            headers={"user-agent": f"prefect-{prefect.__version__}"},
        )

        self.client.hooks = {
            "response": lambda r, *args, **kwargs: r.raise_for_status()
        }
        self.client.auth = (api_key, api_secret)

    def parse_timestamp(self, api_time: str):
        """
        Returns either the pendulum-parsed actual timestamp or
        a very out-of-date timestamp if not set
        """
        return (
            pendulum.parse(api_time)
            if api_time is not None
            else pendulum.from_timestamp(-1)
        )

    async def get_connector(
        self,
        connector_id: str,
    ) -> Dict:
        """
        Retrieve Fivetran connector to details.
        Args:
            connector_id: ID of the Fivetran connector with which to interact.
        Returns:
            Dict containing the details of a Fivetran connector
        """
        URL_CONNECTOR: str = "https://api.fivetran.com/v1/connectors/{}".format(
            connector_id
        )

        return (await self.client.get(URL_CONNECTOR)).json()

    async def patch_connector(
        self,
        connector_id: str,
        data: Dict,
    ) -> Dict:
        """
        Alter Fivetran connector metadata.
        Args:
            connector_id: ID of the Fivetran connector with which to interact.
            data: Key Value pair of connector metadata to be changed.
        Returns:
            Dict containing the details of a Fivetran connector
        """
        URL_CONNECTOR: str = "https://api.fivetran.com/v1/connectors/{}".format(
            connector_id
        )

        return (
            await self.client.patch(
                URL_CONNECTOR,
                json=data,
                headers={"Content-Type": "application/json;version=2"},
            )
        ).json()

    async def force_connector(
        self,
        connector_id: str,
    ) -> str:
        """
        Start a Fivetran data sync
        Args:
            connector_id: ID of the Fivetran connector with which to interact.
        Returns:
            The timestamp of the end of the connector's last run, or now if it
            has not yet run.
        """

        return (
            await self.client.post(
                "https://api.fivetran.com/v1/connectors/" + connector_id + "/force"
            )
        ).json()

    async def __aenter__(self):
        if self._closed:
            raise RuntimeError(
                "The client cannot be started again after it has been closed."
            )
        if self._started:
            raise RuntimeError("The client cannot be started more than once.")

        self._started = True

        return self

    async def __aexit__(self, *exc):
        self._closed = True
        await self.client.__aexit__()

force_connector async

Start a Fivetran data sync

Parameters:

Name Type Description Default
connector_id str

ID of the Fivetran connector with which to interact.

 required

Returns:

Type Description
str

The timestamp of the end of the connector's last run, or now if it
str

has not yet run.
Source code in prefect_fivetran/clients.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def force_connector(
    self,
    connector_id: str,
) -> str:
    """
    Start a Fivetran data sync
    Args:
        connector_id: ID of the Fivetran connector with which to interact.
    Returns:
        The timestamp of the end of the connector's last run, or now if it
        has not yet run.
    """

    return (
        await self.client.post(
            "https://api.fivetran.com/v1/connectors/" + connector_id + "/force"
        )
    ).json()

get_connector async

Retrieve Fivetran connector to details.

Parameters:

Name Type Description Default
connector_id str

ID of the Fivetran connector with which to interact.

 required

Returns:

Type Description
Dict

Dict containing the details of a Fivetran connector
Source code in prefect_fivetran/clients.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
async def get_connector(
    self,
    connector_id: str,
) -> Dict:
    """
    Retrieve Fivetran connector to details.
    Args:
        connector_id: ID of the Fivetran connector with which to interact.
    Returns:
        Dict containing the details of a Fivetran connector
    """
    URL_CONNECTOR: str = "https://api.fivetran.com/v1/connectors/{}".format(
        connector_id
    )

    return (await self.client.get(URL_CONNECTOR)).json()

parse_timestamp

Returns either the pendulum-parsed actual timestamp or a very out-of-date timestamp if not set

Source code in prefect_fivetran/clients.py 
40
41
42
43
44
45
46
47
48
49
def parse_timestamp(self, api_time: str):
    """
    Returns either the pendulum-parsed actual timestamp or
    a very out-of-date timestamp if not set
    """
    return (
        pendulum.parse(api_time)
        if api_time is not None
        else pendulum.from_timestamp(-1)
    )

patch_connector async

Alter Fivetran connector metadata.

Parameters:

Name Type Description Default
connector_id str

ID of the Fivetran connector with which to interact.

 required
data Dict

Key Value pair of connector metadata to be changed.

 required

Returns:

Type Description
Dict

Dict containing the details of a Fivetran connector
Source code in prefect_fivetran/clients.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def patch_connector(
    self,
    connector_id: str,
    data: Dict,
) -> Dict:
    """
    Alter Fivetran connector metadata.
    Args:
        connector_id: ID of the Fivetran connector with which to interact.
        data: Key Value pair of connector metadata to be changed.
    Returns:
        Dict containing the details of a Fivetran connector
    """
    URL_CONNECTOR: str = "https://api.fivetran.com/v1/connectors/{}".format(
        connector_id
    )

    return (
        await self.client.patch(
            URL_CONNECTOR,
            json=data,
            headers={"Content-Type": "application/json;version=2"},
        )
    ).json()