Get future events from Pinnacle API

Question

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

Questions:

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.

Code:

auth.py

"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')
 }

database.py

"Functionality for interacting with the database."
import pymysql
from contextlib import contextmanager
SERVER = 'localhost'
USER = 'root'
PASSWORD = ''
DATABASE = 'bets'
@contextmanager
def get_connection():
 "Creates database connection."
 connection = pymysql.connect(host=SERVER, user=USER, password=PASSWORD, db=DATABASE) 
 try:
 yield connection
 finally:
 connection.close()
 
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,
 away))
 con.commit()
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

get_fixtures.py

"""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']
USERNAME = ""
PASSWORD = ""
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 = "https://api.pinnacle.com/v1/fixtures?sportId=29&leagueIds=" + ','.join(leagues)
 try:
 response = requests.get(url, headers=HEADERS)
 except requests.ConnectionError:
 print(f"{datetime.datetime.now()} No Internet connection")
 return None
 except requests.HTTPError:
 print(f"{datetime.datetime.now()} An HTTP error occured.")
 return None
 if response.text == '':
 print(f"{datetime.datetime.now()} 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']):
 notify_new_fixture(fixture)
 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)
 save_fixtures(fixtures)
 print(f"{datetime.datetime.now()}")
 time.sleep(delay)
def notify_new_fixture(fixture):
 """ Prints a notification about a new fixture. """
 print(f"{datetime.datetime.now()} {fixture['id']} {fixture['home']} - {fixture['away']}") 
 
if __name__ == '__main__':
 update_fixtures(LEAGUES, DELAY)

Answer 1

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'
PASSWORD = ''
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!

@contextmanager
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

This:

url = "https://api.pinnacle.com/v1/fixtures?sportId=29&leagueIds=" + ','.join(leagues)

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

https://requests.readthedocs.io/en/master/user/quickstart/#passing-parameters-in-urls

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

• Konstantin Kostanzhoglo: Thanks for the review! Could you provide an example what can be done to avoid storing password as hard-coded source global?
• Konstantin Kostanzhoglo: Regarding your last suggestion: response.json != json.loads(response.text), it outputs <bound method Response.json of <Response [200]>>
• Reinderien: Yep; it's a function, it needs to be json()
• Reinderien: Re. an example what can be done to avoid storing password as hard-coded source global, I do not have a good answer, so I have asked: stackoverflow.com/questions/62600585/…
• Konstantin Kostanzhoglo: Have you already deleted it?