Skip to content

Subscriptions

API endpoint for fetching different types of subscriptions.

subscription_routers()

Retrieve all active or provisioning router subscriptions.

Source code in gso/api/v1/subscriptions.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@router.get(
    "/routers",
    status_code=status.HTTP_200_OK,
    response_model=list[SubscriptionDomainModelSchema],
)
def subscription_routers() -> list[dict[str, Any]]:
    """Retrieve all active or provisioning router subscriptions."""
    subscriptions = []
    routers = get_router_subscriptions(lifecycles=[SubscriptionLifecycle.ACTIVE, SubscriptionLifecycle.PROVISIONING])
    for r in routers:
        subscription = SubscriptionModel.from_subscription(r["subscription_id"])
        extended_model = build_extended_domain_model(subscription)
        subscriptions.append(extended_model)

    return subscriptions

subscription_dashboard_devices()

Retrieve FQDN for all dashboard devices that are monitored.

Source code in gso/api/v1/subscriptions.py 
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
@router.get(
    "/dashboard_devices",
    status_code=status.HTTP_200_OK,
    response_class=Response,
    responses={
        200: {
            "content": {"text/plain": {}},
            "description": "Return a flat file of FQDNs.",
        }
    },
)
def subscription_dashboard_devices() -> Response:
    """Retrieve FQDN for all dashboard devices that are monitored."""
    fqdns = []
    dashboard_devices = get_subscriptions(
        product_types=[ProductType.ROUTER, ProductType.SUPER_POP_SWITCH, ProductType.OFFICE_ROUTER],
        lifecycles=[SubscriptionLifecycle.ACTIVE, SubscriptionLifecycle.PROVISIONING],
    )
    for device in dashboard_devices:
        subscription = SubscriptionModel.from_subscription(device["subscription_id"])
        extended_model = build_extended_domain_model(subscription)
        if extended_model["product"]["product_type"] == ProductType.ROUTER:
            fqdns.append(extended_model["router"]["router_fqdn"])
        elif extended_model["product"]["product_type"] == ProductType.SUPER_POP_SWITCH:
            fqdns.append(extended_model["super_pop_switch"]["super_pop_switch_fqdn"])
        elif extended_model["product"]["product_type"] == ProductType.OFFICE_ROUTER:
            fqdns.append(extended_model["office_router"]["office_router_fqdn"])

    fqdn_flat_file = "\n".join(fqdns)
    return Response(content=fqdn_flat_file, media_type="text/plain")

subscription_edge_ports(edge_port_name=Query(None, description='Filter EdgePorts by their `edge_port_name` field'), node__router_fqdn=Query(None, description='Filter EdgePorts by router FQDN'))

List all EdgePort subscriptions, optionally filtering by edge_port_name or node_router_fqdn.

Source code in gso/api/v1/subscriptions.py 
 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
@router.get(
    "/edge_ports",
    status_code=status.HTTP_200_OK,
    response_model=list[SubscriptionDomainModelSchema],
)
def subscription_edge_ports(
    edge_port_name: str | None = Query(None, description="Filter EdgePorts by their `edge_port_name` field"),
    node__router_fqdn: str | None = Query(None, description="Filter EdgePorts by router FQDN"),
) -> list[dict[str, Any]]:
    """List all EdgePort subscriptions, optionally filtering by `edge_port_name` or `node_router_fqdn`."""
    # Build up filter clauses
    filters: list[Any] = []
    if edge_port_name:
        filters.append({"edge_port_name": edge_port_name})

    edge_ports = get_subscriptions(
        product_types=[ProductType.EDGE_PORT],
        instance_filters=filters,
    )

    subscriptions = []
    for ep in edge_ports:
        subscription = SubscriptionModel.from_subscription(ep["subscription_id"])
        extended_model = build_extended_domain_model(subscription)
        if node__router_fqdn:
            if extended_model["edge_port"]["node"]["router_fqdn"] == node__router_fqdn:
                subscriptions.append(extended_model)
        else:
            subscriptions.append(extended_model)

    return subscriptions