Python module for integration to Atrium Sports APIs
Project description
Atrium Sports API SDK
Python module to make use of the Atrium Sports Datacore API
Datacore REST API
from atriumsports import AtriumSports
atrium = AtriumSports(
{
"sport": "basketball",
"credential_id": "XXXXX",
"credential_secret": "YYYY",
"organizations": ["b1a23"],
}
)
datacore = atrium.client("datacore")
response = datacore.get("/o/b1a23/competitions", limit=500)
for data in response.data():
print(data)
or using openapi client
GET endpoints
from pprint import pprint
from atriumsports import AtriumSports
from atriumsports.datacore.openapi import CompetitionsApi
atrium = AtriumSports(
{
"sport": "basketball",
"credential_id": "XXXXX",
"credential_secret": "YYYY",
"organizations": ["b1a23"],
}
)
datacore = atrium.client("datacore")
# prepare api client with access token and connection pool
with datacore as api_client:
# create api instance object for handling input and output of chosen endpoint
api_instance = CompetitionsApi(api_client)
response = api_instance.competition_list(sport="basketball", organization_id="b1a23")
pprint(response)
POST and PUT endpoints
from atriumsports import AtriumSports
from atriumsports.datacore.openapi import EntitiesApi
from atriumsports.datacore.openapi import EntityPostBody
from atriumsports.datacore.openapi import EntityPutBody
atrium = AtriumSports(
{
"sport": "basketball",
"credential_id": "XXXXX",
"credential_secret": "YYYY",
"organizations": ["b1a23"],
}
)
datacore = atrium.client("datacore")
with datacore as api_client:
api_instance = EntitiesApi(api_client)
response = api_instance.entity_insert(
sport="basketball",
organization_id="b1a23",
entity_post_body=EntityPostBody(
name_full_local="Test",
status="INACTIVE",
),
)
entity_id = response.data[0].entity_id
response = api_instance.entity_update(
sport="basketball",
organization_id="b1a23",
entity_id=entity_id,
entity_put_body=EntityPutBody(
status="ACTIVE",
),
)
assert response.data[0].status == "ACTIVE"
Response body
Response body is a pydantic object containing deserialized response data.
Example:
from atriumsports import AtriumSports
from atriumsports.datacore.openapi import LeaguesApi
atrium = AtriumSports(
{
"sport": "basketball",
"credential_id": "XXXXX",
"credential_secret": "YYYY",
"organizations": ["b1a23"],
}
)
datacore = atrium.client("datacore")
with datacore as api_client:
api_instance = LeaguesApi(api_client)
# throws error if response body doesn't pass validation
response = api_instance.league_list("b1a23", "basketball")
print(response.data)
print(response.data[0].region_type)
Datacore Streaming API
import time
from atriumsports import AtriumSports
atrium = AtriumSports(
{
"sport": "basketball",
"credential_id": "XXXXX",
"credential_secret": "YYYY",
"environment": "sandpit",
}
)
stream_api = atrium.client("datacore-stream")
def on_connect_callback_function(client):
"""example callback when connected"""
print("connected")
def on_read_callback_function(client, topic, message):
"""example callback when message read"""
print("{}: {}".format(topic, message))
connected = stream_api.connect(
{
"fixture_id": "f71dfdd6-51f1-11ea-8889-22953e2ee7e2", # fixture_id
"scopes": ["write:stream_events", "read:stream_events"], # Scopes
"on_read": on_read_callback_function,
"on_connect": on_connect_callback_function,
}
)
if not connected:
print(stream_api.error())
else:
stream_api.publish(
"write:stream_events",
{
"type": "event",
"data": {
"eventClass": "sport",
"eventId": "c2404cc0-9f75-11e8-98d0-529269fb1459",
"entityId": "c24048a6-9f75-11e8-98d0-529269fb1459",
"personId": "c2405b2a-9f75-11e8-98d0-529269fb1459",
"eventType": "2pt",
"subType": "jumpshot",
"clock": "PT08:23",
"shotClock": "PT12.3",
"periodId": 2,
"success": True,
"timestamp": "2018-08-14T16:45:34.34",
"clientId": "c2408302-9f75-11e8-98d0-529269fb1459",
"clientType": "TestApi:1.1.2",
},
},
qos=1, # QoS level 0=at most once, 1=at least once, 2=exactly once
)
time.sleep(40)
stream_api.disconnect()
For the available apis and models please check modules under atriumsports.datacore.openapi
.
Using your IDE, navigate to this module to see all details of the generated API classes.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
atriumsports_sdk-1.7.0b1.tar.gz
(427.7 kB
view hashes)
Built Distribution
Close
Hashes for atriumsports_sdk-1.7.0b1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | b268dd34682d25ca72c84dbd1d2abb995e92727ea31f98200ad0633f344200e8 |
|
MD5 | edbfdb7ace6f14b79a6a00547249da05 |
|
BLAKE2b-256 | c45285b92c98ef969cecbe09239d5843ecc91ba43eda2e5b1fb5417cb6c2ab32 |