Skip to content

prefect_census.credentials

Module containing credentials for interacting with Census.

Classes

CensusCredentials

Bases: CredentialsBlock

Credentials block for credential use across Census tasks and flows.

Attributes:

Name Type Description
api_key SecretStr

API key to authenticate with the Census API. Refer to the Census authentication docs for retrieving the API key.

Examples:

Load stored Census credentials: 

from prefect_census import CensusCredentials

census_creds = CensusCredentials.load("BLOCK_NAME")

Use CensusCredentials instance to trigger a sync run: 

import asyncio
from prefect import flow
from prefect_census import CensusCredentials

credentials = CensusCredentials(api_key="my_api_key")

@flow
async def trigger_sync_run_flow():
    async with credentials.get_client() as client:
        await client.trigger_sync_run(sync_id=42)

asyncio.run(trigger_sync_run_flow())

Load saved Census credentials within a flow: 

from prefect import flow

from prefect_census import CensusCredentials
from prefect_census.syncs import trigger_census_sync

@flow
def trigger_census_sync_run_flow():
    credentials = CensusCredentials.load("my-census-credentials")
    trigger_census_sync(credentials=credentials, sync_id=42)

trigger_census_sync_run_flow()

Source code in prefect_census/credentials.py 
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
class CensusCredentials(CredentialsBlock):
    """
    Credentials block for credential use across Census tasks and flows.

    Attributes:
        api_key: API key to authenticate with the Census
            API. Refer to the [Census authentication docs](
            https://docs.getcensus.com/basics/api#getting-api-access)
            for retrieving the API key.

    Examples:
        Load stored Census credentials:
        ```python
        from prefect_census import CensusCredentials

        census_creds = CensusCredentials.load("BLOCK_NAME")
        ```

        Use CensusCredentials instance to trigger a sync run:
        ```python
        import asyncio
        from prefect import flow
        from prefect_census import CensusCredentials

        credentials = CensusCredentials(api_key="my_api_key")

        @flow
        async def trigger_sync_run_flow():
            async with credentials.get_client() as client:
                await client.trigger_sync_run(sync_id=42)

        asyncio.run(trigger_sync_run_flow())
        ```

        Load saved Census credentials within a flow:
        ```python
        from prefect import flow

        from prefect_census import CensusCredentials
        from prefect_census.syncs import trigger_census_sync

        @flow
        def trigger_census_sync_run_flow():
            credentials = CensusCredentials.load("my-census-credentials")
            trigger_census_sync(credentials=credentials, sync_id=42)

        trigger_census_sync_run_flow()
        ```
    """

    _block_type_name = "Census Credentials"
    _documentation_url = "https://prefecthq.github.io/prefect-census/credentials/"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/b2f805555c778f37f861b67c4a861908f66a6e35-700x700.png"  # noqa

    api_key: SecretStr = Field(
        ..., title="API Key", description="API key to authenticate with the Census API."
    )

    def get_client(self) -> CensusClient:
        """
        Provides an authenticated client for working with the Census API.

        Returns:
            A authenticated Census API client
        """
        return CensusClient(api_key=self.api_key.get_secret_value())

Functions

get_client

Provides an authenticated client for working with the Census API.

Returns:

Type Description
CensusClient

A authenticated Census API client
Source code in prefect_census/credentials.py 
71
72
73
74
75
76
77
78
def get_client(self) -> CensusClient:
    """
    Provides an authenticated client for working with the Census API.

    Returns:
        A authenticated Census API client
    """
    return CensusClient(api_key=self.api_key.get_secret_value())