Get future events from Pinnacle API

Posted on


My program makes API requests to Pinnacle Sports to retrieve future events and saves them to the database.


  1. Do I manage the database resources correctly? It takes about 3 seconds to check whether there are duplicate entries so that apparently it is very inefficient. The previous approach was to have global constants with the database connection and cursor and it worked much faster, though probably not that safe.
  2. Do I handle possible request errors the right way? Some problems that come to my mind: no Internet connection, HTTP error, empty response .
  3. Should I get rid of the raw loop in save_fixtures and to introduce a function for dealing with each league separately?
  4. How can I track the status of a script? At the moment I output everything in the console, though maybe there are more convenient ways for doing that like logging or something.


"Creates signature and headers for interacting with Pinnacle API"

import base64

def create_signature(username, password):
    "Given username and password creates base64 encoded signature username:password"

    return base64.b64encode(f'{username}:{password}'.encode('utf-8'))

def create_headers(signature):
    "Given a signature creates required headers for interacting with Pinnacle API"

    return  {
        'Content-length' : '0',
        'Content-type'   : 'application/json',
        'Authorization'  : "Basic " + signature.decode('utf-8')

"Functionality for interacting with the database."

import pymysql
from contextlib import contextmanager

SERVER = 'localhost'
USER = 'root'
DATABASE = 'bets'

def get_connection():
    "Creates database connection."
    connection = pymysql.connect(host=SERVER, user=USER, password=PASSWORD, db=DATABASE)    
        yield connection

def record_fixture(league_id, fixture):
    "Records given fixture to the database."
    with get_connection() as con:
        with con.cursor() as cursor:  
            event_id = fixture['id']
            starts = fixture['starts'][0:10] # Example: 2019-08-22
            home = fixture['home']
            away = fixture['away']
            sql = "INSERT INTO fixture (event_id, league_id, match_date, 
                    home_team, away_team) VALUES (%s, %s, %s, %s, %s)"
            cursor.execute(sql, (event_id, league_id, starts, home,

def is_duplicated_entry(event_id):
    "Returns True if an entry with given event_id already exists"
    with get_connection() as con:
        with con.cursor() as cursor:                
            cursor = con.cursor()
            sql = "SELECT * from fixture WHERE event_id = %s"
            result = cursor.execute(sql, event_id)
            return result != 0

"""Obtains fixture list from Pinnacle API for the given list of leagues
and records them to the database."""

import json
import datetime
import time
import requests
import auth
import database

LEAGUES = ['1980', '5487', '2436', '5488', '2196', '5490', '1842', '5874',
           '2627', '2630', '5452', '6263', '5938']
SIGNATURE = auth.create_signature(USERNAME, PASSWORD)
HEADERS = auth.create_headers(SIGNATURE)
DELAY = 60

def get_fixtures(leagues):
    "Gets fixtures list for the given list of leagues."

    url = "" + ','.join(leagues)
        response = requests.get(url, headers=HEADERS)
    except requests.ConnectionError:
        print(f"{} No Internet connection")
        return None
    except requests.HTTPError:
        print(f"{} An HTTP error occured.")
        return None
    if response.text == '':
        print(f"{} There are no fixtures available")
        return None
    fixtures = json.loads(response.text)
    return fixtures

def save_fixtures(fixtures):
    "Records fixtures to the database and notifies about the new fixtures."

    if not fixtures is None:
        for league in fixtures['league']:
            for fixture in league['events']:
                if not database.is_duplicated_entry(fixture['id']):
                    database.record_fixture(league['id'], fixture)

def update_fixtures(leagues, delay=DELAY):
    Every DELAY seconds retrieves fixture list for the given leagues
    and records them to the database.
    while True:
        fixtures = get_fixtures(leagues)

def notify_new_fixture(fixture):
    """ Prints a notification about a new fixture. """
    print(f"{} {fixture['id']} {fixture['home']} - {fixture['away']}")    

if __name__ == '__main__':
    update_fixtures(LEAGUES, DELAY)


Type hints

def create_signature(username, password):

can be

def create_signature(username: str, password: str) -> bytes:

Quote style consistency

Pick single or double:

'Authorization'  : "Basic "

Connection string security

Please (please) do not store a password as a hard-coded source global:

SERVER = 'localhost'
USER = 'root'
DATABASE = 'bets'

for so many reasons. At least the password – and often the entire connection string – are externalized and encrypted, one way or another. I had assumed that there are high-quality wallet libraries that can make this safe for you, but have struggled to find one, so I have asked on Security.

A correct context manager!

def get_connection():

Thank you! Too bad pymysql didn’t bundle this…

String continuation

        sql = "INSERT INTO fixture (event_id, league_id, match_date, 
                home_team, away_team) VALUES (%s, %s, %s, %s, %s)"

I find more legible as

sql = (
    "INSERT INTO fixture (event_id, league_id, match_date, "
    "home_team, away_team) VALUES (%s, %s, %s, %s, %s)"

with a bonus being that there won’t be stray whitespace from your indentation.

Requests sugar


url = "" + ','.join(leagues)

should not bake in its query params. Pass those as a dict to requests like this:

Also, do not call json.loads(response.text); just use response.json().

Leave a Reply

Your email address will not be published. Required fields are marked *