Skip to main content

Sites

Sites service provides API operations for sites data. site.id values are unique identifiers internal to Glassdome and are referenced by other APIs (e.g., PCF).

Code Examples​

Python SDK​

Link: https://pypi.org/project/glassdome-waypoint-sdk/

"""
Example: list sites.

Usage:
    export WAYPOINT_BASE_URL="https://waypoint.glassdome.dev"
    export WAYPOINT_API_KEY="your-api-key"
    python sites.py
"""

import os

from glassdome_waypoint_sdk import (
    WaypointClient,
    WaypointConfig,
)


def main():
    base_url = os.getenv("WAYPOINT_BASE_URL") or exit("WAYPOINT_BASE_URL is not set")
    api_key = os.getenv("WAYPOINT_API_KEY") or exit("WAYPOINT_API_KEY is not set")

    client = WaypointClient.from_api_key(WaypointConfig(base_url=base_url), api_key)

    print("Listing sites with page size 10...")
    # List sites with page size 10
    sites, next_token = client.site.list_sites(page_size=10)
    for site in sites:
        print(f"Site ID: {site.id}, Name: {site.name}")

    # List sites with next page token if available
    if next_token:
        print("Listing sites with next page token...")
        sites, _ = client.site.list_sites(page_token=next_token)
        for site in sites:
            print(f"Site ID: {site.id}, Name: {site.name}")


if __name__ == "__main__":
    main()

Airflow Provider​

Link: https://pypi.org/project/airflow-providers-glassdome-waypoint/

"""
Example: list sites.

This example assumes that the Glassdome Waypoint connection has already been created
with the connection ID "glassdome_waypoint_default".

You can create a connection with a different connection ID and use it for the hook.
"""

from datetime import timedelta

from airflow.sdk import dag, task


@task
def list_sites():
    import logging

    from glassdome_waypoint.hooks.waypoint import WaypointHook

    client = WaypointHook("glassdome_waypoint_default").get_client()

    logging.info("Listing sites with page size 10...")
    # List sites with page size 10
    sites, next_token = client.site.list_sites(page_size=10)
    for site in sites:
        logging.info(f"Site ID: {site.id}, Name: {site.name}")

    # List sites with next page token if available
    if next_token:
        logging.info("Listing sites with next page token...")
        sites, _ = client.site.list_sites(page_token=next_token)
        for site in sites:
            logging.info(f"Site ID: {site.id}, Name: {site.name}")


@dag(
    dag_id="example_sites",
    description="List sites in your company",
    schedule=None,
    dagrun_timeout=timedelta(minutes=1),
    tags=["examples"],
    default_args={
        "owner": "Glassdome",
        "retries": 0,
    },
)
def example_sites():
    list_sites()


dag = example_sites()

if __name__ == "__main__":
    dag.test()