Skip to main content

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


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.3.2.tar.gz (372.3 kB view hashes)

Uploaded Source

Built Distribution

atriumsports_sdk-1.3.2-py3-none-any.whl (933.8 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page