3
\$\begingroup\$

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)
\$\endgroup\$
3
\$\begingroup\$

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().

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thanks for the review! Could you provide an example what can be done to avoid storing password as hard-coded source global? \$\endgroup\$ – Konstantin Kostanzhoglo Jun 26 at 13:25
  • \$\begingroup\$ Regarding your last suggestion: response.json != json.loads(response.text), it outputs <bound method Response.json of <Response [200]>> \$\endgroup\$ – Konstantin Kostanzhoglo Jun 26 at 16:12
  • \$\begingroup\$ Yep; it's a function, it needs to be json() \$\endgroup\$ – Reinderien Jun 26 at 16:28
  • \$\begingroup\$ 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/… \$\endgroup\$ – Reinderien Jun 26 at 18:15
  • \$\begingroup\$ Have you already deleted it? \$\endgroup\$ – Konstantin Kostanzhoglo Jun 26 at 18:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.