Add ODS_lookup.py and SQL_connections.py to the repository

codonPython/ODS_lookup.py

@@ -0,0 +1,97 @@
+import requests
+from typing import Dict, Iterable, Callable, List, Optional
+import pandas as pd
+import numpy as np
+
+
+def query_api(code: str) -> Dict:
+    """Query the ODS (organisation data service) API for a single org code
+    and return the full JSON result. Full API docs can be found here:
+    https://digital.nhs.uk/services/organisation-data-service/guidance-for-developers/organisation-endpoint
+
+    Parameters
+    ----------
+    code : str
+        3 character organization code.
+
+    Returns
+    ----------
+    dict
+        The data returned from the API.
+
+    Examples
+    ---------
+    >>> result = query_api("X26")
+    >>> result["Organisation"]["Name"]
+    'NHS DIGITAL'
+    >>> result["Organisation"]["GeoLoc"]["Location"]["AddrLn1"]
+    '1 TREVELYAN SQUARE'
+    """
+    if not isinstance(code, str):
+        raise ValueError(f"ODS code must be a string, received {type(code)}")
+
+    response = requests.get(
+        f"https://directory.spineservices.nhs.uk/ORD/2-0-0/organisations/{code}"
+    ).json()
+    if "errorCode" in response:
+        error_code = response["errorCode"]
+        error_text = response["errorText"]
+        raise ValueError(
+            f"API query failed with code {error_code} and text '{error_text}'."
+        )
+    return response
+
+
+def get_addresses(codes: Iterable[str]) -> pd.DataFrame:
+    """Query the ODS (organisation data service) API for a series of
+    org codes and return a data frame containing names and addresses.
+    Invalid codes will cause a message to be printed but will
+    otherwise be ignored, as an incomplete merge table is more
+    useful than no table at all.
+
+    Parameters
+    ----------
+    codes : list, ndarray or pd.Series
+        3 character organization codes to retrieve information for.
+
+    Returns
+    ----------
+    DataFrame
+        Address information for the given org codes.
+
+    Examples
+    ---------
+    >>> result = get_addresses(pd.Series(["X26"]))
+    >>> result.reindex(columns=sorted(result.columns))
+              Org_AddrLn1 Org_Code Org_Country     Org_Name Org_PostCode Org_Town
+    0  1 TREVELYAN SQUARE      X26     ENGLAND  NHS Digital      LS1 6AE    LEEDS
+    """
+
+    # Internal helper function to take the full result of a query
+    # and extract the relevant fields
+    def extract_data(api_result: Dict, code: str) -> Dict[str, str]:
+        org_info = api_result["Organisation"]
+        org_name = org_info["Name"]
+        org_address = org_info["GeoLoc"]["Location"]
+        result = {
+            "Org_Code": code,
+            "Org_Name": org_name.title().replace("Nhs", "NHS"),
+            **{f"Org_{k}": v for k, v in org_address.items() if k != "UPRN"},
+        }
+        return result
+
+    # Remove duplicate values
+    to_query = set(codes)
+    if np.nan in to_query:
+        # 'NaN' is actually a valid code but we don't want it for null values
+        to_query.remove(np.nan)
+
+    result = []
+    for code in to_query:
+        try:
+            api_result = query_api(code)
+            result.append(extract_data(api_result, code))
+        except ValueError as e:
+            print(f"No result for ODS code {code}. {e}")
+            continue
+    return pd.DataFrame(result)

codonPython/SQL_connections.py

@@ -0,0 +1,33 @@
+''' Author(s): Sam Hollings
+Desc: this module contains SQL_alchemy engines to connect to commonly used databases'''
+
+from sqlalchemy import create_engine
+
+
+def conn_dss():
+    '''Returns sqlalchemy Engine to connect to the DSS 2008 server (DMEDSS) DSS_CORPORATE database '''
+    engine = create_engine('mssql+pyodbc://DMEDSS/DSS_CORPORATE?driver=SQL+Server')
+    return engine
+
+
+def conn_dss2016uat():
+    '''Returns sqlalchemy Engine to connect to the DSS 2016 server (UAT) (DSSUAT) DSS_CORPORATE database '''
+    conn = create_engine('mssql+pyodbc://DSSUAT/DSS_CORPORATE?driver=SQL+Server')
+    return conn
+
+
+def conn_dummy(path=r''):
+    '''connect to the sqlite3 database in memory, or at specified path
+    parameters
+    ----------
+    path : string
+        The location and file in which the database for conn_dummy will be stored. Default is memory (RAM)
+    '''
+
+    conn_string = 'sqlite://'
+    if path != '':
+        path = '/' + path
+
+    conn = create_engine(r'{0}{1}'.format(conn_string, path))
+
+    return conn

codonPython/tests/ODS_test.py

@@ -0,0 +1,27 @@
+import pytest
+import numpy as np
+from codonPython import ODS_lookup
+
+
+def test_successful_query():
+    NHSD_code = "X26"
+    result = ODS_lookup.query_api(NHSD_code)
+    assert result["Organisation"]["Name"] == "NHS DIGITAL"
+
+
+def test_unsuccessful_query():
+    invalid_code = "ASDF"
+    with pytest.raises(ValueError):
+        ODS_lookup.query_api(invalid_code)
+
+
+def test_wrong_type():
+    invalid_code = 0
+    with pytest.raises(ValueError):
+        ODS_lookup.query_api(invalid_code)
+
+
+def test_unsuccessful_address_query():
+    invalid_code = ["ASDF", np.nan, None]
+    result = ODS_lookup.get_addresses(invalid_code)
+    assert result.empty

codonPython/tests/SQL_connections_test.py

@@ -0,0 +1,15 @@
+'''test script for SQL_connections
+- test the connections can run a dummy script (SELECT 1 as [Code], 'test' as [Name])'''
+import pandas as pd
+import pytest
+import codonPython.SQL_connections as conn
+
+
+@pytest.mark.parametrize("connection",
+                         [conn.conn_dummy(),
+                          conn.conn_dummy('test.db')
+                          ])
+def test_select1(connection):
+    result = pd.read_sql("""SELECT 1 as [Code], 'Test' as [Name]""", connection).iloc[0, 0]
+    expected = pd.DataFrame([{'Code': 1, 'Name': 'Test'}]).iloc[0, 0]
+    assert result == expected