Problem
My program makes API requests to Pinnacle Sports to retrieve future events and saves them to the database.
Questions:
- 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.
- Do I handle possible request errors the right way? Some problems that come to my mind: no Internet connection,
HTTPerror, empty response . - Should I get rid of the raw loop in
save_fixturesand to introduce a function for dealing with each league separately? - 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 likeloggingor 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)
Solution
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().