Python: token request, SQL user fetch and API calls

Question

I have the files as below. What I'm looking for is some feedback with regards to security, reliability, maintenance, scaling and performance.

What is going on?

• Make a request to get a token to access some APIs.
• Use SQL to get a list of users.
• For each user, need to call 3/4 APIs and store the information in a MSSQL DB.

Some things to note

• Documentation needs to be better
• I wanted to put all SQL in the relevant file but I wasn't sure a connection open and closing so many times would be a good idea. Not sure if there's a workaround?
• Security - I put confidential information into environment variables. The SQL is also paramaterised. Not sure what else I can focus on
• Performance - I tried using generators where I could. Not sure if it is needed in all instances
• Maintenance - tried to make the ability to add new endpoints as easy as possible
• Readability - I think I did well. Could do better with better comments. Tried to keep to PEP8 as much as possible
• I've not used classes as wasn't sure this was the right for it. Should I have done?
• Scalability - any feedback on this would be great

So yeah, any feedback would be appreciated

eds.py

"""
Backup solution to the APIs going down.
Request a token.
With that we get the patient ID by passing through the NHS number using demographics endpoint.
We then request medication, allergies and contraindictions and record it into SQL
"""
import os
import json
from sqlite3 import InterfaceError
import requests
import pyodbc
import nhs_sql # CUSTOM
import graphnet # CUSTOM
DB_SERVER = os.environ.get('DB_SERVER')
DB_NAME = os.environ.get('DB_NAME')
DB_USER = os.environ.get('DB_USER')
DB_PASS = os.environ.get('DB_PASS')
DB_DRIVER = os.environ.get('DB_DRIVER')
GRAPHNET_API = os.environ.get('GRAPHNET_API')
GRAPHNET_KEY = os.environ.get('GRAPHNET_KEY')
GRAPHNET_SECRET = os.environ.get('GRAPHNET_SECRET')
GRAPHNET_CLIENT_ID = os.environ.get('GRAPHNET_CLIENT_ID')
def get_data():
 """
 Loop through all NHS numbers
 Get the patient ID for the patient
 """
 cnxn = pyodbc.connect(f'DRIVER={DB_DRIVER};SERVER={DB_SERVER};DATABASE={DB_NAME};UID={DB_USER};PWD={DB_PASS}')
 cursor = cnxn.cursor()
 endpoints = {}
 for endpoint in graphnet.get_endpoints('eds'):
 endpoints[endpoint['pretty']] = endpoint
 headers = graphnet.get_headers()
 for nhs_number in nhs_sql.get_inpatients():
 print(f'\n[+] NHS: {nhs_number}')
 demographics_json = graphnet.get_demographics(nhs_number)
 if demographics_json is not None:
 cursor.execute('EXEC spSearch ?,?',(nhs_number,json.dumps(demographics_json)))
 cnxn.commit()
 if demographics_json['MatchingPatients'] is not None:
 patient_id = demographics_json['MatchingPatients'][0]['Details']['DetailId']
 opted_out = demographics_json['MatchingPatients'][0]['Details']['OptedOut']
 if opted_out is False:
 for key,value in endpoints.items():
 print(f' - Endpoint: {key}')
 api_url = value['api_url'].format(GRAPHNET_API,patient_id)
 stored_procedure = f'EXEC {value["stored_procedure"]} ?,?'
 endpoint_request = requests.get(
 api_url,
 headers=headers
 )
 endpoint_json = endpoint_request.json()
 if endpoint_json is not None:
 print(' - DB Hit')
 cursor.execute(stored_procedure,(nhs_number,json.dumps(endpoint_json)))
 cnxn.commit()
if __name__ == '__main__':
 try:
 get_data()
 except KeyboardInterrupt:
 print('\n\n033円[33m[*] Warning: cancelled by user')
 except Exception as e:
 print(f'\n\n033円[31m[*] Exception: {e}')

nhs_sql.py

"""
Useful SQL connections
"""
import os
import pyodbc
DB_SERVER = os.environ.get('DB_SERVER')
DB_NAME = os.environ.get('DB_NAME')
DB_USER = os.environ.get('DB_USER')
DB_PASS = os.environ.get('DB_PASS')
DB_DRIVER = os.environ.get('DB_DRIVER')
def get_inpatients():
 """
 Getting all the inpatients currently sitting in A&E
 """
 cnxn = pyodbc.connect(f'DRIVER={DB_DRIVER};SERVER={DB_SERVER};DATABASE={DB_NAME};UID={DB_USER};PWD={DB_PASS}')
 cursor = cnxn.cursor()
 for row in cursor.execute('EXEC spGetInpatients'):
 yield row[0]
def main():
 """
 Hello world
 """
 print("[+] Hello world")
if __name__ == '__main__':
 main()

graphnet.py

"""
Getting headers, tokens and hitting endpoints
"""
import os
import urllib.parse
import requests
import endpoints
GRAPHNET_AUTH = os.environ.get('GRAPHNET_AUTH')
GRAPHNET_API = os.environ.get('GRAPHNET_API')
GRAPHNET_KEY = os.environ.get('GRAPHNET_KEY')
GRAPHNET_SECRET = os.environ.get('GRAPHNET_SECRET')
GRAPHNET_CLIENT_ID = os.environ.get('GRAPHNET_CLIENT_ID')
def get_headers():
 """
 Get the headers for each API call.
 These change regularly for reasons unknown so best do it in one place
 """
 headers = {
 'Content-Type': 'application/x-www-form-urlencoded',
 'Ocp-Apim-Subscription-Key': GRAPHNET_KEY,
 'Authorization': f'Bearer {get_token()}',
 }
 return headers
def get_endpoints(key = 'everything'):
 """
 Returning the endpoints based on the param that is passed
 Defaults to 'everything' if we need to bring back, well, everythng
 """
 for endpoint in endpoints.config.values():
 if key in endpoint['keys']:
 yield endpoint
def get_token():
 """
 Gets the token from graphnet
 """
 headers = {
 'Content-Type': 'application/x-www-form-urlencoded'
 }
 params = {
 'grant_type':'client_credentials',
 'client_id':GRAPHNET_CLIENT_ID,
 'client_secret':GRAPHNET_SECRET
 }
 token_response = requests.post(GRAPHNET_AUTH, data=params, headers=headers)
 json_response = token_response.json()
 return json_response['access_token']
def get_demographics(nhs):
 """
 Get demographics
 """
 headers = {
 'Content-Type': 'application/x-www-form-urlencoded',
 'Ocp-Apim-Subscription-Key': GRAPHNET_KEY,
 'Authorization': f'Bearer {get_token()}',
 }
 params = urllib.parse.urlencode({
 'AllTenancies':1,
 'NhsNumber':nhs
 })
 demographics = requests.post(
 f'{GRAPHNET_API}/demographics/patients/search',
 data=params,
 headers=headers
 )
 return demographics.json()
def main():
 """
 Hello World
 """
 print("[+] Hello world")
if __name__ == '__main__':
 main()

endpoints.py

"""
All of the endpoints
"""
config = {
 'allergies': {
 'pretty': 'Allergies',
 'api_url':'{}/allergies/patients/{}/allergies-summary',
 'stored_procedure': 'spAllergies',
 'keys': ['eds','everything']
 },
 'contraindictions': {
 'pretty': 'Contraindictions',
 'api_url':'{}/gp/patients/{}/gp-contraindications',
 'stored_procedure': 'spContraindications',
 'keys': ['eds','everything']
 },
 'meds_repeat': {
 'pretty': 'Medications - Repeat',
 'api_url':'{}/gp/patients/{}/gp-medications/repeat',
 'stored_procedure': 'spMedicationRepeat',
 'keys': ['eds','everything']
 },
 'meds_issued': {
 'pretty': 'Medications - Issued',
 'api_url':'{}/gp/patients/{}/gp-medications/issued',
 'stored_procedure': 'spMedicationIssued',
 'keys': ['eds','everything']
 }
}

Have read the answer given. Is this something reasonable for the requests.get() and requests.post()?

try:
 with requests.post(
 f'{GRAPHNET_API}/demographics/patients/search',
 data=params,
 headers=headers
 ) as demographics:
 demographics.raise_for_status()
 return demographics.json()
except requests.exceptions.HTTPError as errh:
 print(f'Http Error: {errh}')
except requests.exceptions.ConnectionError as errc:
 print(f'Error Connecting: {errc}')
except requests.exceptions.Timeout as errt:
 print(f'Timeout Error: {errt}')
except requests.exceptions.RequestException as err:
 print(f'Argh. Something Else: {err}')

• Peilonrayz: Hi pee2pee. Please do not invalidate users answers by editing code mentioned in answers. Please see what you may and may not do after receiving answers . If you have a DMCA request please follow the the terms of service's instructions.

Answer 1

Are you sure you want to use environ.get()? Those keys seem mandatory, so environ[] should be used instead.

It seems minor, but I would hesitate to leave DB_PASS as a visible global. Consider moving all of those global constants into a method that accepts no parameters and produces a connection string. Also, do this in one place, instead of at the top of multiple files.

get_data is going to spin up a new database connection every time, and those are expensive. However, it seems like pooling is enabled by default so the reality might not be so bad. You should do a sanity check on this setting.

cnxn is a kind of hostile abbreviation for connection, and conn would be more legible.

This:

endpoints = {}
for endpoint in graphnet.get_endpoints('eds'):
 endpoints[endpoint['pretty']] = endpoint

can be replaced by a dictionary comprehension:

 endpoints = {
 endpoint['pretty']
 for endpoint in graphnet.get_endpoints('eds')
 }

demographics_json['MatchingPatients'][0]['Details'] should receive a temporary variable.

if opted_out is False should be if not opted_out, so long as there are no nullity shenanigans.

When calling requests.get(), always wrap the result in a with and call raise_for_status.

Delete your hello world main()s.

Since you're doing nothing but membership checks on endpoint['keys'], replace that with a set literal {}. Also: don't keep your config as a dictionary of untyped dictionaries; instead keep it as a dictionary of well-typed classes (dataclasses or named tuples).

• pee2pee: Thank you for taking the time to do this. Instead of the hello_world main(s), what should I do instead?
• Reinderien: Nothing? Just delete them. Not every file will be an entry point.
• pee2pee: I did delete them but get an error: "Undefined variable 'main'"
• Reinderien: Classes offer better types and confidence in correctness, particularly if they're immutable (named tuple, or frozen attribute on dataclass). Also more self-documenting code.
• Reinderien: <3 you're quite welcome! come back to code review for more advice on new code, or even on this code (in a new question) once it's gone through one refactoring pass.