Repository: PrefectHQ/prefect-census (language: python)

Issue:
Add `CensusClient`
Create a class named `CensusClient` that handles the execution of API calls against the Census API. 

`CensusClient` should accept an API key on initialization and use that API key to authenticate all API requests made for the lifespan of the instance. 

The client should be capable of making asynchronous API requests via use of the `httpx` library.

Each API route that `CensusClient` supports should be exposed as a method with a name that is descriptive of the action being taken. A method should also exist that accepts an HTTP verb, URL, and payload to perform a generic HTTP call against the Census API.

To complete this issue, I recommend starting with just the generic method to make HTTP calls. Further methods can be built when needed for tasks and flows.

A concrete example of a client is the [`DbtCloudAdministrativeClient`](https://github.com/PrefectHQ/prefect-dbt/blob/main/prefect_dbt/cloud/clients.py) in `prefect-dbt`. That implementation can be improved upon, and I encourage you to improve upon it here!
Add `CensusCredentials` block
Create a block that is capable of storing a Census API key for use with tasks and flows. 

The API key should be stored as a `SecretStr` to prevent accidental logging of the API key. The block should have a method named `get_client` that returns an authenticated client object capable of communicating with the Census API.
Add `trigger_census_sync` task
Create a task that accepts a `CensusCredentials` block and the ID of a Census sync, and triggers a run of the specified sync.

The task should return the ID of the sync run that was triggered. If the sync trigger failed, the task should raise a `CensusSyncTriggerFailed` exception which inherits from `RuntimeException` with the message received from the API.

[Documentation for the Census API route to trigger syncs](https://docs.getcensus.com/basics/api/syncs#post-syncs-id-trigger).
Add `wait_census_sync_completion` flow
Create a flow that accepts a `CensusCredentials` block, a sync run ID, a poll interval defined in seconds, and a maximum amount of time to wait defined in seconds. The poll interval and max wait time should have reasonable default values.

The flow should use the `get_census_sync_run_info` task to check the status of the current sync run. 
- If the sync is still running, sleep for the defined poll interval, and then check the sync run status again.
- If the sync run is completed, return the final details of the sync run.
- If the sync run failed, raise a `CensusSyncRunFailed` exception which inherits from `RuntimeError` with the message returned from the API.
- If the sync run time has exceeded the specified max wait time, raise a `CensusSyncRunTimeout` exception which inherits from `RuntimeError` with a reasonable message for the exception.


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index de081ac..888321c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 ### Added
+- `trigger_census_sync` task - [#10](https://github.com/PrefectHQ/prefect-census/pull/10)
+- `get_census_sync_run_info` task - [#10](https://github.com/PrefectHQ/prefect-census/pull/10)
+- `CensusCredentials` block - [#10](https://github.com/PrefectHQ/prefect-census/pull/10)
+- `CensusClient` - [#10](https://github.com/PrefectHQ/prefect-census/pull/10)
+- `wait_census_sync_completion` flow - [#10](https://github.com/PrefectHQ/prefect-census/pull/10)
+- `trigger_census_sync_and_wait_for_completion` flow - [#10](https://github.com/PrefectHQ/prefect-census/pull/10)
 
 ### Changed
 
diff --git a/README.md b/README.md
index 2d074d6..59cacb9 100644
--- a/README.md
+++ b/README.md
@@ -41,27 +41,67 @@ pip install prefect-census
 Then, register to [view the block](https://orion-docs.prefect.io/ui/blocks/) on Prefect Cloud:
 
 ```bash
-prefect block register -m prefect_census.credentials
+prefect block register -m prefect_census
 ```
 
 Note, to use the `load` method on Blocks, you must already have a block document [saved through code](https://orion-docs.prefect.io/concepts/blocks/#saving-blocks) or [saved through the UI](https://orion-docs.prefect.io/ui/blocks/).
 
 ### Write and run a flow
+Trigger Census sync run and wait for completion:
 
 ```python
 from prefect import flow
-from prefect_census.tasks import (
-    goodbye_prefect_census,
-    hello_prefect_census,
-)
 
+from prefect_census import CensusCredentials
+from prefect_census.syncs import trigger_census_sync_run_and_wait_for_completion
 
 @flow
-def example_flow():
-    hello_prefect_census
-    goodbye_prefect_census
+def my_flow():
+    ...
+    creds = CensusCredentials(api_key="my_api_key")
+    run_result = trigger_census_sync_run_and_wait_for_completion(
+        credentials=creds,
+        sync_id=42
+    )
+    ...
+
+my_flow()
+```
+
+Get Census sync run info:
+```python
+from prefect import flow
+
+from prefect_census import CensusCredentials
+from prefect_census.runs import get_census_sync_run_info
+
+@flow
+def get_sync_run_info_flow():
+    credentials = CensusCredentials(api_key="my_api_key")
+
+    return get_census_sync_run_info(
+        credentials=credentials,
+        run_id=42
+    )
+
+get_sync_run_info_flow()
+```
+
+Call custom endpoint:
+```python
+from prefect import flow
+from prefect_census import CensusCredentials
+from prefect_census.client import CensusClient
+
+@flow
+def my_flow(sync_id):
+    creds_block = CensusCredentials(api_key="my_api_key")
+
+    client = CensusClient(api_key=creds_block.api_key.get_secret_value())
+    response = client.call_endpoint(http_method="GET", path=f"/syncs/{sync_id}")
+    return response
 
-example_flow()
+my_flow(42)
 ```
 
 ## Resources
diff --git a/docs/client.md b/docs/client.md
new file mode 100644
index 0000000..21fc333
--- /dev/null
+++ b/docs/client.md
@@ -0,0 +1,1 @@
+::: prefect_census.client
\ No newline at end of file
diff --git a/docs/credentials.md b/docs/credentials.md
new file mode 100644
index 0000000..e29c3e4
--- /dev/null
+++ b/docs/credentials.md
@@ -0,0 +1,1 @@
+::: prefect_census.credentials
\ No newline at end of file
diff --git a/docs/runs.md b/docs/runs.md
new file mode 100644
index 0000000..fa2771f
--- /dev/null
+++ b/docs/runs.md
@@ -0,0 +1,1 @@
+::: prefect_census.runs
\ No newline at end of file
diff --git a/docs/syncs.md b/docs/syncs.md
new file mode 100644
index 0000000..1033d96
--- /dev/null
+++ b/docs/syncs.md
@@ -0,0 +1,1 @@
+::: prefect_census.syncs
\ No newline at end of file
diff --git a/mkdocs.yml b/mkdocs.yml
index 5e9b984..5cffd2a 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -48,4 +48,9 @@ plugins:
 
 nav:
     - Home: index.md
+    - Client: client.md
+    - Credentials: credentials.md
+    - Runs: runs.md
+    - Syncs: syncs.md
+
 
diff --git a/prefect_census/__init__.py b/prefect_census/__init__.py
index 4cb4ffb..9fe1627 100644
--- a/prefect_census/__init__.py
+++ b/prefect_census/__init__.py
@@ -1,4 +1,7 @@
 from . import _version
-from .blocks import CensusBlock  # noqa
+from .client import CensusClient  # noqa
+from .credentials import CensusCredentials  # noqa
+from .runs import get_census_sync_run_info  # noqa
+from .syncs import trigger_census_sync  # noqa
 
 __version__ = _version.get_versions()["version"]
diff --git a/prefect_census/client.py b/prefect_census/client.py
new file mode 100644
index 0000000..d5cf2e1
--- /dev/null
+++ b/prefect_census/client.py
@@ -0,0 +1,131 @@
+"""Module containing client for interacting with the Census API"""
+from typing import Any, Dict, Optional
+
+import prefect
+from httpx import AsyncClient, Response
+from prefect.utilities.asyncutils import sync_compatible
+
+
+class CensusClient:
+    """
+    Client for interacting with the Census API.
+
+    Attributes:
+        api_key: API key to authenticate with the Census API.
+    """
+
+    def __init__(self, api_key: str):
+        self._closed = False
+        self._started = False
+
+        self.client = AsyncClient(
+            base_url="https://app.getcensus.com/api/v1",
+            headers={
+                "Authorization": f"Bearer {api_key}",
+                "user-agent": f"prefect-{prefect.__version__}",
+            },
+        )
+
+    @sync_compatible
+    async def call_endpoint(
+        self,
+        http_method: str,
+        path: str,
+        params: Optional[Dict[str, Any]] = None,
+        json: Optional[Dict[str, Any]] = None,
+    ) -> Response:
+        """
+        Call an endpoint in the Census API.
+
+        Args:
+            http_method: HTTP method to call on the endpoint.
+            path: The partial path for request (e.g. /syncs/42). Will be
+                appended onto the base URL as determined by the client configuration.
+            params: Query parameters to include in the request.
+            json: JSON serializable body to send in the request.
+
+        Returns:
+            The response from the Census API.
+
+        Example:
+            ```python
+            from prefect import flow
+            from prefect_census import CensusCredentials
+            from prefect_census.client import CensusClient
+
+            @flow
+            def my_flow(sync_id):
+                creds_block = CensusCredentials(api_key="my_api_key")
+
+                client = CensusClient(
+                    api_key=creds_block.api_key.get_secret_value()
+                )
+                response = client.call_endpoint(
+                    http_method="GET",
+                    path=f"/syncs/{sync_id}"
+                )
+                return response
+
+            my_flow(42)
+            ```
+        """
+
+        response = await self.client.request(
+            method=http_method, url=path, params=params, json=json
+        )
+        response.raise_for_status()
+        return response
+
+    @sync_compatible
+    async def get_run_info(self, run_id: int) -> Response:
+        """
+        Sends a request to the [get sync id info endpoint](https://docs.getcensus.com/basics/api/syncs#get-syncs-id)  # noqa
+
+        Args:
+            run_id: The ID of the sync run to get details for.
+
+        Returns:
+            The response from the Census API.
+        """
+        return await self.call_endpoint(
+            http_method="GET",
+            path=f"/sync_runs/{run_id}",
+        )
+
+    @sync_compatible
+    async def trigger_sync_run(
+        self, sync_id: int, force_full_sync: bool = False
+    ) -> Response:
+        """
+        Sends a request to the [trigger sync run endpoint](https://docs.getcensus.com/basics/api/sync-runs)  # noqa
+        to initiate a sync run.
+
+        Args:
+            sync_id: The ID of the sync to trigger.
+            force_full_sync: If the sync should perform a full sync.
+
+        Returns:
+            The response from the Census API.
+        """
+        return await self.call_endpoint(
+            http_method="POST",
+            path=f"/syncs/{sync_id}/trigger",
+            params={"force_full_sync": force_full_sync},
+        )
+
+    async def __aenter__(self):
+        """Async context manager entry method."""
+        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):
+        """Async context manager exit method."""
+        self._closed = True
+        await self.client.__aexit__()
diff --git a/prefect_census/credentials.py b/prefect_census/credentials.py
new file mode 100644
index 0000000..de4dc1b
--- /dev/null
+++ b/prefect_census/credentials.py
@@ -0,0 +1,70 @@
+"""Module containing credentials for interacting with Census."""
+from httpx import AsyncClient
+from prefect.blocks.core import Block
+from pydantic import Field, SecretStr
+
+from prefect_census.client import CensusClient
+
+
+class CensusCredentials(Block):
+    """
+    Credentials block for credential use across Census tasks and flows.
+
+    Attributes:
+        api_key: API key to authenticate with the Census
+            API. Refer to the [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"
+    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/3oznRx2UFkd2XyqNkEZpzB/4e0967a828aec5e2527cedadf8d24e8a/llmjpn8a0pgu8szjmnyi.webp?h=250"  # noqa
+
+    api_key: SecretStr = Field(
+        ..., title="API Key", description="API key to authenticate with the Census API."
+    )
+
+    def get_client(self) -> AsyncClient:
+        """
+        Returns a newly instantiated client for working with the Census API.
+        """
+        return CensusClient(api_key=self.api_key.get_secret_value())
diff --git a/prefect_census/runs.py b/prefect_census/runs.py
new file mode 100644
index 0000000..07bd7be
--- /dev/null
+++ b/prefect_census/runs.py
@@ -0,0 +1,198 @@
+"""Module containing tasks and flows for interacting with Census sync runs"""
+import asyncio
+from enum import Enum
+from typing import Any, Dict, Tuple
+
+from httpx import HTTPStatusError
+from prefect import flow, task
+from prefect.logging import get_run_logger
+
+from prefect_census.credentials import CensusCredentials
+from prefect_census.utils import extract_user_message
+
+
+class CensusSyncRunFailed(RuntimeError):
+    """Raised when unable to retrieve Census sync run"""
+
+    pass
+
+
+class CensusSyncRunTimeout(RuntimeError):
+    """
+    Raised when a triggered job run does not complete in the configured max
+    wait seconds
+    """
+
+
+class CensusGetSyncRunInfoFailed(RuntimeError):
+    """Used to idicate retrieve sync run info."""
+
+
+class CensusSyncRunCancelled(Exception):
+    """Raised when a triggered sync run is cancelled"""
+
+
+class CensusSyncRunStatus(Enum):
+    """Census sync statuses."""
+
+    CANCELLED = "cancelled"
+    WORKING = "working"
+    FAILED = "failed"
+    COMPLETED = "completed"
+
+    @classmethod
+    def is_terminal_status_code(cls, status_code: str) -> bool:
+        """
+        Returns True if a status code is terminal for a job run.
+        Returns False otherwise.
+        """
+        return status_code in [
+            cls.CANCELLED.value,
+            cls.FAILED.value,
+            cls.COMPLETED.value,
+        ]
+
+
+@task(
+    name="Get Census sync run details",
+    description=(
+        "Retrieves details of a Census sync run" "for the sync with the given sync_id."
+    ),
+    retries=3,
+    retry_delay_seconds=10,
+)
+async def get_census_sync_run_info(credentials: CensusCredentials, run_id: int):
+    """
+    A task to retrieve information a Census sync run.
+
+    Args:
+        credentials: Credentials for authenticating with Census.
+        run_id: The ID of the run of the sync to trigger.
+
+    Returns:
+        The run data returned by the Census API.
+        Keys:
+        - status
+        - data
+                - id
+                - label
+                - schedule_frequency
+                - schedule_day
+                - schedule_hour
+                - schedule_minute
+                - created_at
+                - updated_at
+                - operation
+                - paused
+                - status
+                - lead_union_insert_to
+                - field_behavior
+                - field_normalization
+                - mirror_strategy
+                - source_attributes
+                    - connection_id
+                    - object
+                    - type
+                    - id
+                    - table_catalog
+                    - table_schema
+                    - table_name
+                - destination_attributes
+                    - connection_id
+                    - object
+                - mappings
+                    - from
+                    - type
+                    - data
+                    - to
+                    - is_primary_identifier
+                    - generate_field
+                    - preserve_values
+                    - operation
+
+
+    Example:
+        Get Census sync run info:
+        ```python
+        from prefect import flow
+
+        from prefect_census import CensusCredentials
+        from prefect_census.runs import get_census_sync_run_info
+
+        @flow
+        def get_sync_run_info_flow():
+            credentials = CensusCredentials(api_key="my_api_key")
+
+            return get_census_sync_run_info(
+                credentials=credentials,
+                run_id=42
+            )
+
+        get_sync_run_info_flow()
+        ```
+    """  # noqa
+    try:
+        async with credentials.get_client() as client:
+            response = await client.get_run_info(run_id)
+    except HTTPStatusError as e:
+        raise CensusGetSyncRunInfoFailed(extract_user_message(e)) from e
+
+    return response.json()["data"]
+
+
+@flow(
+    name="Wait for Census sync run",
+    description="Waits for the Census sync run to finish running.",
+)
+async def wait_census_sync_completion(
+    run_id: int,
+    credentials: CensusCredentials,
+    max_wait_seconds: int = 60,
+    poll_frequency_seconds: int = 5,
+) -> Tuple[CensusSyncRunStatus, Dict[str, Any]]:
+    """
+    Wait for the given Census sync run to finish running.
+
+    Args:
+        run_id: The ID of the sync run to wait for.
+        credentials: Credentials for authenticating with Census.
+        max_wait_seconds: Maximum number of seconds to wait for sync to complete.
+        poll_frequency_seconds: Number of seconds to wait in between checks for
+            run completion.
+
+    Raises:
+        CensusSyncRunTimeout: When the elapsed wait time exceeds `max_wait_seconds`.
+
+    Returns:
+        run_status: An enum representing the final Census sync run status.
+        run_data: A dictionary containing information about the run after completion.
+    """
+    logger = get_run_logger()
+    seconds_waited_for_run_completion = 0
+    wait_for = []
+    while seconds_waited_for_run_completion <= max_wait_seconds:
+        run_data_future = await get_census_sync_run_info.submit(
+            credentials=credentials,
+            run_id=run_id,
+            wait_for=wait_for,
+        )
+        run_data = await run_data_future.result()
+        run_status = run_data.get("status")
+
+        if CensusSyncRunStatus.is_terminal_status_code(run_status):
+            return CensusSyncRunStatus(run_status), run_data
+
+        wait_for = [run_data_future]
+        logger.info(
+            "Census sync run with ID %i has status %s. Waiting for %i seconds.",
+            run_id,
+            CensusSyncRunStatus(run_status).name,
+            poll_frequency_seconds,
+        )
+        await asyncio.sleep(poll_frequency_seconds)
+        seconds_waited_for_run_completion += poll_frequency_seconds
+
+    raise CensusSyncRunTimeout(
+        f"Max wait time of {max_wait_seconds} seconds exceeded while waiting "
+        f"for sync run with ID {run_id}"
+    )
diff --git a/prefect_census/syncs.py b/prefect_census/syncs.py
new file mode 100644
index 0000000..fb37303
--- /dev/null
+++ b/prefect_census/syncs.py
@@ -0,0 +1,206 @@
+"""Module containing tasks and flows for interacting with Census syncs."""
+from typing import Any, Dict
+
+from httpx import HTTPStatusError
+from prefect import flow, task
+from prefect.logging import get_run_logger
+
+from prefect_census.credentials import CensusCredentials
+from prefect_census.runs import (
+    CensusSyncRunCancelled,
+    CensusSyncRunFailed,
+    CensusSyncRunStatus,
+    wait_census_sync_completion,
+)
+from prefect_census.utils import extract_user_message
+
+
+class CensusSyncTriggerFailed(RuntimeError):
+    """Used to indicate sync triggered."""
+
+
+@task(
+    name="Trigger Census sync run",
+    description="Triggers a Census sync run for the sync with the given sync_id.",
+    retries=3,
+    retry_delay_seconds=10,
+)
+async def trigger_census_sync(
+    credentials: CensusCredentials, sync_id: int
+) -> Dict[str, Any]:
+    """
+    A task to trigger a Census sync run.
+
+    Args:
+        credentials: Credentials for authenticating with Census.
+        sync_id: The ID of the sync to trigger.
+
+    Returns:
+        The run data returned from the Census API.
+        Keys:
+            - status
+            - data
+                - id
+                - sync_id
+                - source_record_count
+                - records_processed
+                - records_updated
+                - records_failed
+                - records_invalid
+                - created_at
+                - updated_at
+                - completed_at
+                - scheduled_execution_time
+                - error_code
+                - error_message
+                - error_detail
+                - status
+                - canceled
+                - full_sync
+                - sync_trigger_reason
+                - ui_tag
+                - ui_detail
+
+    Examples:
+        Trigger a Census sync run:
+        ```python
+        from prefect import flow
+
+        from prefect_census import CensusCredentials
+        from prefect_census.syncs import trigger_census_sync
+
+        @flow
+        def trigger_census_sync_flow():
+            credentials = CensusCredentials(api_key="my_api_key")
+            trigger_census_sync(credentials=credentials, sync_id=42)
+
+        trigger_census_sync_flow()
+        ```
+    """
+    logger = get_run_logger()
+
+    logger.info(f"Triggering Census sync run for sync with ID {sync_id}")
+    try:
+        async with credentials.get_client() as client:
+            response = await client.trigger_sync_run(sync_id=sync_id)
+    except HTTPStatusError as e:
+        raise CensusSyncTriggerFailed(extract_user_message(e)) from e
+
+    run_data = response.json()["data"]
+
+    if "id" in run_data:
+        logger.info(
+            f"Census sync run successfully triggered for sync with ID {id}. "
+            "You can view the status of this sync run at "
+            f"https://app.getcensus.com/sync/{id}/sync-history"
+        )
+
+    return run_data["sync_run_id"]
+
+
+@flow(
+    name="Trigger Census sync run and wait for completion",
+    description="Triggers a Census sync run and waits for the"
+    "triggered run to complete.",
+)
+async def trigger_census_sync_run_and_wait_for_completion(
+    credentials: CensusCredentials,
+    sync_id: int,
+    max_wait_seconds: int = 900,
+    poll_frequency_seconds: int = 10,
+) -> dict:
+    """
+    Flow that triggers a sync run and waits for the triggered run to complete.
+
+    Args:
+        credentials: Credentials for authenticating with Census.
+        sync_id: The ID of the sync to trigger.
+        max_wait_seconds: Maximum number of seconds to wait for sync to complete
+        poll_frequency_seconds: Number of seconds to wait in between checks for run completion.
+
+    Raises:
+        CensusSyncRunCancelled: The triggered Census sync run was cancelled.
+        CensusSyncRunFailed: The triggered Census sync run failed.
+        RuntimeError: The triggered Census sync run ended in an unexpected state.
+
+    Returns:
+        The run data returned by the Census API.
+        Keys:
+            - status
+            - data
+                sync_run_id
+
+    Examples:
+        Trigger a Census sync using CensusCredentials instance and wait
+        for completion as a standalone flow:
+        ```python
+        import asyncio
+
+        from prefect_census import CensusCredentials
+        from prefect_census.syncs import trigger_census_sync_run_and_wait_for_completion
+
+        asyncio.run(
+            trigger_census_sync_run_and_wait_for_completion(
+                credentials=CensusCredentials(
+                    api_key="my_api_key"
+                ),
+                sync_id=42
+            )
+        )
+        ```
+
+        Trigger a Census sync and wait for completion as a subflow:
+        ```python
+        from prefect import flow
+
+        from prefect_census import CensusCredentials
+        from prefect_census.syncs import trigger_census_sync_run_and_wait_for_completion
+
+        @flow
+        def my_flow():
+            ...
+            creds = CensusCredentials(api_key="my_api_key")
+            run_result = trigger_census_sync_run_and_wait_for_completion(
+                credentials=creds,
+                sync_id=42
+            )
+            ...
+
+        my_flow()
+        ```
+    """  # noqa
+    logger = get_run_logger()
+
+    triggered_run_data_future = await trigger_census_sync.submit(
+        credentials=credentials, sync_id=sync_id
+    )
+
+    run_id = await triggered_run_data_future.result()
+    if run_id is None:
+        raise RuntimeError("Unable to determine run ID for triggered sync.")
+
+    final_run_status, run_data = await wait_census_sync_completion(
+        run_id=run_id,
+        credentials=credentials,
+        max_wait_seconds=max_wait_seconds,
+        poll_frequency_seconds=poll_frequency_seconds,
+    )
+
+    if final_run_status == CensusSyncRunStatus.COMPLETED:
+        logger.info(
+            "Census sync run with ID %s completed successfully!",
+            run_id,
+        )
+        return run_data
+
+    elif final_run_status == CensusSyncRunStatus.CANCELLED:
+        raise CensusSyncRunCancelled(
+            f"Triggered sync run with ID {run_id} was cancelled."
+        )
+    elif final_run_status == CensusSyncRunStatus.FAILED:
+        raise CensusSyncRunFailed(f"Triggered sync run with ID: {run_id} failed.")
+    else:
+        raise RuntimeError(
+            f"Triggered sync run with ID: {run_id} ended with unexpected"
+            f"status {final_run_status}"
+        )
diff --git a/prefect_census/utils.py b/prefect_census/utils.py
new file mode 100644
index 0000000..51ff96f
--- /dev/null
+++ b/prefect_census/utils.py
@@ -0,0 +1,24 @@
+"""Utilities for common interactions with the Census API"""
+from json import JSONDecodeError
+from typing import Optional
+
+from httpx import HTTPStatusError
+
+
+def extract_user_message(error: HTTPStatusError) -> Optional[str]:
+    """
+    Extract user message from an error response from the Census API.
+
+    Args:
+        error: An HTTPStatusError raised by httpx
+
+    Returns:
+        The status from Census API response or None if a status cannot
+            be extracted.
+    """
+    try:
+        response_payload = error.response.json()
+        user_message = response_payload.get("status", None)
+        return user_message
+    except JSONDecodeError:
+        pass
diff --git a/requirements-dev.txt b/requirements-dev.txt
index dae3fd3..39fd95f 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -8,6 +8,7 @@ mkdocstrings[python]
 isort
 pre-commit
 pytest-asyncio
+respx
 mock; python_version < '3.8'
 mkdocs-gen-files
 interrogate


Test patch:
diff --git a/tests/test_runs.py b/tests/test_runs.py
new file mode 100644
index 0000000..bf053f4
--- /dev/null
+++ b/tests/test_runs.py
@@ -0,0 +1,28 @@
+import pytest
+from httpx import Response
+
+from prefect_census.credentials import CensusCredentials
+from prefect_census.runs import CensusGetSyncRunInfoFailed, get_census_sync_run_info
+
+
+@pytest.fixture
+def census_credentials():
+    return CensusCredentials(api_key="my_api_key")
+
+
+class TestGetCensusSyncRunInfo:
+    async def test_get_census_sync_run_info(self, respx_mock, census_credentials):
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/42",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(return_value=Response(200, json={"data": {"sync_run_id": 42}}))
+
+        await get_census_sync_run_info.fn(credentials=census_credentials, run_id=42)
+
+    async def test_get_nonexistent_run(self, respx_mock, census_credentials):
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/4",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(return_value=Response(404, json={"status": {"message": "Not found!"}}))
+        with pytest.raises(CensusGetSyncRunInfoFailed, match="Not found!"):
+            await get_census_sync_run_info.fn(credentials=census_credentials, run_id=4)
diff --git a/tests/test_syncs.py b/tests/test_syncs.py
new file mode 100644
index 0000000..ffe35cb
--- /dev/null
+++ b/tests/test_syncs.py
@@ -0,0 +1,261 @@
+import pytest
+from httpx import Response
+from prefect import flow
+
+from prefect_census.credentials import CensusCredentials
+from prefect_census.runs import (
+    CensusSyncRunCancelled,
+    CensusSyncRunFailed,
+    CensusSyncRunTimeout,
+)
+from prefect_census.syncs import (
+    CensusSyncTriggerFailed,
+    trigger_census_sync,
+    trigger_census_sync_run_and_wait_for_completion,
+)
+
+
+@pytest.fixture
+def census_credentials():
+    return CensusCredentials(api_key="my_api_key")
+
+
+class TestTriggerCensusSyncRun:
+    async def test_trigger_sync(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/45/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(return_value=Response(200, json={"data": {"sync_run_id": 45}}))
+
+        @flow
+        async def test_flow():
+            return await trigger_census_sync(credentials=census_credentials, sync_id=45)
+
+        result = await test_flow()
+        assert result == 45
+
+    async def test_trigger_nonexistent_sync(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/46/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(404, json={"status": {"user_message": "Not found!"}})
+        )
+
+        @flow
+        async def test_trigger_nonexistent_job():
+            await trigger_census_sync(census_credentials, sync_id=46)
+
+        with pytest.raises(CensusSyncTriggerFailed, match="Not found!"):
+            await test_trigger_nonexistent_job()
+
+
+class TestTriggerCensusSyncRunAndWaitForCompletion:
+    async def test_run_success(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "sync_run_id": 12345}})
+        )
+
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/12345",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(
+                200,
+                json={"data": {"id": 5, "sync_run_id": 12345, "status": "completed"}},
+            )
+        )
+
+        result = await trigger_census_sync_run_and_wait_for_completion(
+            credentials=census_credentials, sync_id=5
+        )
+
+        assert result == {"id": 5, "status": "completed", "sync_run_id": 12345}
+
+    def test_run_in_sync_flow(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "sync_run_id": 12345}})
+        )
+
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/12345",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(
+                200,
+                json={"data": {"id": 5, "sync_run_id": 12345, "status": "completed"}},
+            )
+        )
+
+        @flow
+        def test_sync_flow():
+            return trigger_census_sync_run_and_wait_for_completion(
+                credentials=census_credentials, sync_id=5
+            )
+
+        result = test_sync_flow()
+
+        assert result == {"id": 5, "status": "completed", "sync_run_id": 12345}
+
+    async def test_run_success_with_wait(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "sync_run_id": 12345}})
+        )
+
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/12345",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            side_effect=[
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={
+                        "data": {"id": 5, "sync_run_id": 12345, "status": "completed"}
+                    },
+                ),
+            ]
+        )
+
+        result = await trigger_census_sync_run_and_wait_for_completion(
+            credentials=census_credentials, sync_id=5, poll_frequency_seconds=2
+        )
+
+        assert result == {"id": 5, "status": "completed", "sync_run_id": 12345}
+
+    async def test_run_failure_with_wait(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "sync_run_id": 12345}})
+        )
+
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/12345",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            side_effect=[
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "failed"}},
+                ),
+            ]
+        )
+
+        with pytest.raises(CensusSyncRunFailed):
+            await trigger_census_sync_run_and_wait_for_completion(
+                credentials=census_credentials, sync_id=5, poll_frequency_seconds=2
+            )
+
+    async def test_run_failure_no_run_id(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "status": "completed"}})
+        )
+
+        with pytest.raises(KeyError, match="sync_run_id"):
+            await trigger_census_sync_run_and_wait_for_completion(
+                credentials=census_credentials, sync_id=5, poll_frequency_seconds=2
+            )
+
+    async def test_run_cancelled_with_wait(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "sync_run_id": 12345}})
+        )
+
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/12345",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            side_effect=[
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={
+                        "data": {"id": 5, "sync_run_id": 12345, "status": "cancelled"}
+                    },
+                ),
+            ]
+        )
+
+        with pytest.raises(CensusSyncRunCancelled):
+            await trigger_census_sync_run_and_wait_for_completion(
+                credentials=census_credentials, sync_id=5, poll_frequency_seconds=2
+            )
+
+    async def test_run_timed_out(self, respx_mock, census_credentials):
+        respx_mock.post(
+            "https://app.getcensus.com/api/v1/syncs/5/trigger",
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            return_value=Response(200, json={"data": {"id": 5, "sync_run_id": 12345}})
+        )
+
+        respx_mock.get(
+            "https://app.getcensus.com/api/v1/sync_runs/12345",  # noqa
+            headers={"Authorization": "Bearer my_api_key"},
+        ).mock(
+            side_effect=[
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+                Response(
+                    200,
+                    json={"data": {"id": 5, "sync_run_id": 12345, "status": "working"}},
+                ),
+            ]
+        )
+
+        with pytest.raises(CensusSyncRunTimeout):
+            await trigger_census_sync_run_and_wait_for_completion(
+                credentials=census_credentials,
+                sync_id=5,
+                poll_frequency_seconds=1,
+                max_wait_seconds=3,
+            )