Skip to content

prefect_fivetran.credentials

Module containing credentials for interacting with Fivetran

FivetranCredentials

Credentials block for credential use across Fivetran tasks and flows.

Attributes:

Name Type Description
api_key SecretStr

Fivetran API KEY
api_secret SecretStr

Fivetran API SECRET

Examples:

Load stored Fivetran credentials: 

from prefect_fivetran import FivetranCredentials

fivetran_credentials = FivetranCredentials.load("BLOCK_NAME")

Use FivetranCredentials instance to trigger a Fivetran sync: 

from prefect_fivetran import FivetranCredentials

credentials = FivetranCredentials(api_key="my_api_key", api_secret="my_api_secret")
async with fivetran_credentials.get_fivetran() as fivetran:
    fivetran.sync(connector_id="my_connector_id")

Source code in prefect_fivetran/credentials.py 
 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
49
50
51
52
53
54
class FivetranCredentials(Block):
    """
    Credentials block for credential use across Fivetran tasks and flows.

    Attributes:
        api_key: [Fivetran API KEY](
            https://fivetran.com/docs/rest-api/faq/access-rest-api)
        api_secret: Fivetran API SECRET

    Examples:
        Load stored Fivetran credentials:
        ```python
        from prefect_fivetran import FivetranCredentials

        fivetran_credentials = FivetranCredentials.load("BLOCK_NAME")
        ```

        Use FivetranCredentials instance to trigger a Fivetran sync:
        ```python
        from prefect_fivetran import FivetranCredentials

        credentials = FivetranCredentials(api_key="my_api_key", api_secret="my_api_secret")
        async with fivetran_credentials.get_fivetran() as fivetran:
            fivetran.sync(connector_id="my_connector_id")
        ```
    """  # noqa

    _block_type_name = "Fivetran Credentials"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/4lfv3KREIOQk1kUM74a66W/433c00edcbde38404a9c2252a721d8d1/fivetran_logo_icon_170149.png?h=250"  # noqa

    api_key: SecretStr = Field(
        default=..., title="API Key", description="API key from the Fivetran console."
    )
    api_secret: SecretStr = Field(
        default=...,
        title="API Secret",
        description="API secret from the Fivetran console.",
    )

    def get_fivetran(self):
        """
        Returns api_key and api_secret for Fivetran object
        """
        return FivetranClient(
            api_key=self.api_key.get_secret_value(),
            api_secret=self.api_secret.get_secret_value(),
        )

get_fivetran

Returns api_key and api_secret for Fivetran object

Source code in prefect_fivetran/credentials.py 
47
48
49
50
51
52
53
54
def get_fivetran(self):
    """
    Returns api_key and api_secret for Fivetran object
    """
    return FivetranClient(
        api_key=self.api_key.get_secret_value(),
        api_secret=self.api_secret.get_secret_value(),
    )