Please can this be deletedI have the files as my organisation has not agreedbelow. What I'm looking for this codeis 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 uploadedappreciated
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}')
Please can this be deleted as my organisation has not agreed for this code to be uploaded
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}')
I have the filesPlease can this be deleted as below. What I'm lookingmy organisation has not agreed for is some feedback with regardsthis code 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()?uploaded
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}')
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}')
Please can this be deleted as my organisation has not agreed for this code to be uploaded
Please deleteI have the files as not been agreedbelow. 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}')
Please delete as not been agreed
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}')
Loading
lang-py