Python queries

Data can also be accessed via Python. The access is different for the Root Types query and subscription (see GraphiQL IDE. Use the following example scripts to get data from the productive deployment of DISRUPT.

query example

import requests
import json

# For access to non public sources you can generate the Basic Auth Header with 
# Username and Password here https://www.debugbear.com/basic-auth-header-generator
# For acces to public sources, just leave it as it is
basic_auth = "Basic yourBase64encodedusernameandpassword"

# GraphiQL url
url = "https://city.app.sdk-cloud.de/api/graphql"

# Query
body = """
query entities {
  entities(source: 1, last: 10) {
    edges {
      node {
        id
        time
        class {
          id
          name
        }
        trajectory {
          time
          coordinateLongLat
        }
      }
    }
  }
}
"""

# Request
response = requests.post(
    url=url, 
    json={"query": body}, 
    headers={"Authorization": basic_auth})
print("response status code: ", response.status_code)

# Print response
if response.status_code == 200:
    data = json.loads(response.content)
    print(json.dumps(data, indent = 4))

subscription example

import asyncio
from gql import Client, gql
from gql.transport.websockets import WebsocketsTransport

# For access to non public sources you can generate the Basic Auth Header with 
# Username and Password here https://www.debugbear.com/basic-auth-header-generator
# For acces to public sources, just leave it as it is
basic_auth = "Basic yourBase64encodedusernameandpassword"

# Websocket configuration
transport = WebsocketsTransport(
    url = "wss://city.app.sdk-cloud.de/api/graphql/ws", 
    subprotocols = ["graphql-ws"],
    headers = {"Authorization": basic_auth}
  )

# Query
subscription_query = gql("""
  subscription subscription{
    measurements(source:1 intervalMs:100) {
      trackingId
      long
      lat
      time
    }
  }
""")

# Subscribe and print
async def subscribe():
    async with Client(transport=transport) as session:
        async for result in session.subscribe(subscription_query):
            print("New Data Received:", result)

# Run the subscription
asyncio.run(subscribe())