Subscribing to Data Events with the CFBD GraphQL API

Over the weekend, I introduced the brand new and experimental CFBD GraphQL API. I already broke down a lot of the advantages of utilizing GraphQL, which includese extra dynamic querying and granular management over the information. One profit is so massive that it deserves its personal submit, GraphQL Subscriptions.

Subscriptions do precisely what they are saying. They let you subscribe to information updates. If you happen to’re a Patreon subscriber, chances are you’ll already be conversant in the reside endpoints within the CFBD REST API (e.g. /scoreboard). Whereas these endpoints current reside information, additionally they require you, the consumer, to implement some form of polling mechanism to re-trigger the endpoint on a cycle. And what’s extra, the information returned by the endpoint might or might not have modified. It is as much as the consumer to determine if it has.

In GraphQL, nevertheless, subscriptions are event-based. You specify a GraphQL question as a subscription and, as a substitute of polling the information supply repeatedly, the question auto-triggers every time that information has truly up to date. As an alternative of constructing a bunch of calls, you specify one operation after which the information is pushed on to your code each time it adjustments within the CFBD database.

Subscriptions are fairly easy. Let’s take a daily GraphQL question, one which queries betting traces from a selected sportsbook for all future video games:

question bettingQuery {
sport(
the place: {
standing: { _eq: “scheduled” }
traces: { supplier: { title: { _eq: “Bovada” } } }
_or: [
{ homeClassification: { _eq: “fbs” } }
{ awayClassification: { _eq: “fbs” } }
]
}
) {
homeTeam
awayTeam
traces(the place: { supplier: { title: { _eq: “Bovada” } } }) {
unfold
overUnder
supplier {
title
}
}
}
}

Fairly normal question, proper? If we wished, we might name this question often, parsing the response to see if any of the information has modified. A lot less complicated could be turning it right into a subscription:

subscription bettingSubscription {
sport(
the place: {
standing: { _eq: “scheduled” }
traces: { supplier: { title: { _eq: “Bovada” } } }
_or: [
{ homeClassification: { _eq: “fbs” } }
{ awayClassification: { _eq: “fbs” } }
]
}
) {
homeTeam
awayTeam
traces(the place: { supplier: { title: { _eq: “Bovada” } } }) {
unfold
overUnder
supplier {
title
}
}
}
}

That was easy! The one change I made was altering the question operation to a subscription operation (I additionally modified the arbitrary title of bettingSubscription). Now, each time the information returned by this question adjustments in CFBD, I’ll get an replace pushed on to me. No extra polling time and again. No extra attempting to determine if something has truly modified.

If you wish to get pushed an replace each time a sport’s standing adjustments to “accomplished” so you understand that it is time to pull play or field rating information, you are able to do that. If you wish to be alerted as above when a sportsbook unfold has modified, you are able to do that. Need to be pushed an replace when recruiting information adjustments? Now you can do this, too.

Making a Subscription in Python

One necessary factor to notice, Insomnia doesn’t assist GraphQL subscriptions. Nevertheless, I nonetheless suggest at all times designing all your GraphQL operations Insomnia since you’ll be able to reap the benefits of its autocomplete and interactive GraphQL docs. You’d simply construct the subscription as a question after which change it to a subscription when placing it into your Python code.

We’ll be working with three PyPI packages: gql, asyncio, and backoff. So be certain that to have all of those put in in your atmosphere.

We’ll stroll by two completely different examples. Right here is the primary instance and it is fairly easy:

from gql import Consumer, gql
from gql.transport.websockets import WebsocketsTransport

transport = WebsocketsTransport(
url=”wss://graphql.collegefootballdata.com/v1/graphql”,
headers={ “Authorization”: “Bearer YOUR_API_KEY”}
)

consumer = Consumer(
transport=transport,
fetch_schema_from_transport=True,
)

question = gql(”’
subscription bettingSubscription {
sport(
the place: {
standing: { _eq: “scheduled” }
traces: { supplier: { title: { _eq: “Bovada” } } }
_or: [
{ homeClassification: { _eq: “fbs” } }
{ awayClassification: { _eq: “fbs” } }
]
}
) {
homeTeam
awayTeam
traces(the place: { supplier: { title: { _eq: “Bovada” } } }) {
unfold
overUnder
supplier {
title
}
}
}
}
”’)

for end in consumer.subscribe(question):
# put your logic right here
print(end result)

Let’s stroll by what this code is doing. On line 4, we’re making a WebsocketsTransport. You will notice that is completely different than what we did within the earlier submit for making GraphQL queries. If you happen to bear in mind, queries and mutations are simply HTTP POST requests. If you happen to take a look at line 5, we’re as a substitute utilizing a wss:// protocol. As an alternative of constructing an HTTP request, we’re working over a WebSocket. Not like the HTTP protocol, WebSockets set up a persistent connection that enable for two-way communication. That is how GraphQL subscriptions are doable. A persistent connection is opened over a WebSocket. The consumer submits the subscription to the GraphQL server after which the GraphQL server pushes a communication out to the consumer each time there may be an replace related to that subscription.

On line 6, be sure you change YOUR_API_KEY with the identical API key you employ to entry the CFBD REST API.

Beginning at line 14, we construct out a GraphQL operation that might be submitted to the GraphQL server as a subscription. This is identical subscription we outlined in the beginning of this submit which subscribes to updates to the spreads and totals from a selected sportsbook (Bovada) for upcoming video games.

On line 39, we start looping by subscription updates. The GraphQL server will return an preliminary information set pertaining to the subscription question. At any time when there are updates to the information set, extra outcomes will seem within the loop and our code will act upon it. Within the instance above, we’re merely printing the outcomes to the console, however that is the place you’ll put the logic that you simply wish to be executed each time there’s a information replace, reminiscent of pushing the up to date information to your personal information retailer.

I discussed that we might be strolling by two completely different examples. There may be one potential problem with the instance above: WebSocket connections, whereas extremely helpful, will be very brittle. The persistent connection will be interrupted for any variety of causes: community outage in your finish, community outage on the GraphQL server’s finish, the GraphQL server happening briefly for upkeep, and so forth.

Fortunately, there are methods to handle this. That is the place we might be utilizing the asyncio and backoff packages. Let’s begin with some imports:

import asyncio
import backoff

from gql import Consumer, gql
from gql.transport.websockets import WebsocketsTransport

Subsequent, we’re going to extract the GraphQL operation into its personal async perform. We are going to take a session as a parameter, which might be used to subscribe to a WebSocket session we’ll create later. That is principally a replica and paste from the earlier instance

async def subscribe(session):
question = gql(”’
subscription bettingSubscription {
sport(
the place: {
standing: { _eq: “scheduled” }
traces: { supplier: { title: { _eq: “Bovada” } } }
_or: [
{ homeClassification: { _eq: “fbs” } }
{ awayClassification: { _eq: “fbs” } }
]
}
) {
homeTeam
awayTeam
traces {
unfold
overUnder
supplier {
title
}
}
}
}
”’)

async for end in session.subscribe(question):
# put your logic right here
print(end result)

We are going to now create one other perform for managing the WebSocket connection and calling our subscription perform:

@backoff.on_exception(backoff.expo, Exception, max_time=60)
async def graphql_connection():
transport = WebsocketsTransport(
url=”wss://graphql.collegefootballdata.com/v1/graphql”,
headers={ “Authorization”: “Bearer YOUR_API_KEY”}
)

consumer = Consumer(
transport=transport,
fetch_schema_from_transport=True,
)

async with consumer as session:
job = asyncio.create_task(subscribe(session))

await asyncio.collect(job)

The backoff module is used on line 1. This establishes some retry logic with an exponential backoff. In different phrases, if the WebSocket connection will get interrupted for any purpose, it would retry this methodology time and again with an exponential enhance within the wait interval in between retries.

Beginning on line 3, we have now some extra code copy and pasted from the earlier instance. Make sure you enter your CFBD API key in on line 5.

The final 4 traces cope with calling the subscription methodology utilizing the WebSocket session that was established on the earlier traces. What’s fascinating is that we’re calling the subscribe methodology inside a job. We might reap the benefits of this to name a number of subscriptions without delay if we had a number of. This could allow all of them to share the identical WebSocket connection. The modified code would look just like this:

def subscribe1(session):
# GraphQL subscription right here

def subscribe2(session):
# GraphQL subscription right here

def subscribe3(session):
# GraphQL subscription right here

def subscribe4(session):
# GraphQL subscription right here

consumer = Consumer(
transport=transport,
fetch_schema_from_transport=True,
)

async with consumer as session:
task1 = asyncio.create_task(subscribe1(session))
task2 = asyncio.create_task(subscribe2(session))
task3 = asyncio.create_task(subscribe3(session))
task4 = asyncio.create_task(subscribe4(session))

await asyncio.collect(task1, task2, task3, task4)

This modification has 4 completely different subscriptions to trace, every encapsulated by its personal perform.

The very last thing we have to do is name the graphql_connection perform and that is the place the asyncio bundle comes into play:

asyncio.run(graphql_connection())

Placing every thing collectively, your last code ought to look just like this:

import asyncio
import backoff

from gql import Consumer, gql
from gql.transport.websockets import WebsocketsTransport

async for end in session.subscribe(question):
# put your logic right here
print(end result)

consumer = Consumer(
transport=transport,
fetch_schema_from_transport=True,
)

async with consumer as session:
job = asyncio.create_task(subscribe(session))

await asyncio.collect(job)

asyncio.run(graphql_connection())

Conclusion

GraphQL subscriptions are an incredible and environment friendly mechanism for subscribing to information updates. Whether or not you need to reduce in your API calls or be extra environment friendly along with your code, they’re an incredible possibility. They’re additionally an incredible possibility if you have to know when information updates. The experimental CFBD GraphQL API is accessible to Patreon subscribers at Tier 3. Be a part of at present if you want to test it out. Additionally, try my earlier submit to see extra examples of what the GraphQL API can do for you. As at all times, let me know what you suppose!

Source link