Repository: cancervariants/therapy-normalization (language: python)

Issue:
Issue 62 idfix
pending confirmation that duplicates are no longer being created


Discussion / hints:

Fix patch:
diff --git a/cli.py b/cli.py
index 4398899f..b5432641 100644
--- a/cli.py
+++ b/cli.py
@@ -1,7 +1,8 @@
 """This module provides a CLI util to make updates to normalizer database."""
 import click
 from botocore.exceptions import ClientError
-from therapy.etl import ChEMBL, Wikidata, DrugBank, NCIt, ChemIDplus, RxNorm
+from therapy.etl import ChEMBL, Wikidata, DrugBank, NCIt, ChemIDplus, RxNorm, \
+    Merge
 from therapy.schemas import SourceName
 from timeit import default_timer as timer
 from therapy.database import Database
@@ -10,6 +11,10 @@
 from os import environ
 
 
+MERGE_SOURCES = {SourceName.WIKIDATA.value, SourceName.RXNORM.value,
+                 SourceName.NCIT.value, SourceName.CHEMIDPLUS.value}
+
+
 class CLI:
     """Class for updating the normalizer database via Click"""
 
@@ -91,6 +96,7 @@ def _check_norm_srcs_match(self, sources, normalizers):
 
     def _update_normalizers(self, normalizers, sources, db):
         """Update selected normalizer sources."""
+        processed_ids = []
         for n in normalizers:
             click.echo(f"\nDeleting {n}...")
             start_delete = timer()
@@ -101,13 +107,24 @@ def _update_normalizers(self, normalizers, sources, db):
                        f"{delete_time:.5f} seconds.\n")
             click.echo(f"Loading {n}...")
             start_load = timer()
-            sources[n](database=db)
+            source = sources[n](database=db)
+            source_ids = source.perform_etl()
+            if source.__class__.__name__ in MERGE_SOURCES:
+                processed_ids += source_ids
             end_load = timer()
             load_time = end_load - start_load
             click.echo(f"Loaded {n} in {load_time:.5f} seconds.")
             click.echo(f"Total time for {n}: "
                        f"{(delete_time + load_time):.5f} seconds.")
 
+        if processed_ids:
+            click.echo("Generating merged concepts...")
+            start = timer()
+            merge = Merge(db)
+            merge.create_merged_concepts(processed_ids)
+            end = timer()
+            click.echo(f"Generated merged concepts in {end - start:.5f} seconds.")  # noqa: E501
+
     def _delete_data(self, source, database):
         # Delete source's metadata
         try:
diff --git a/dynamodb_revisions/add_merge.py b/dynamodb_revisions/add_merge.py
new file mode 100644
index 00000000..2ea74774
--- /dev/null
+++ b/dynamodb_revisions/add_merge.py
@@ -0,0 +1,53 @@
+"""Add merged records and references to database."""
+from therapy.database import Database
+from therapy.etl.merge import Merge
+from timeit import default_timer as timer
+from therapy.schemas import SourceName
+from os import environ
+import click
+import sys
+
+
+def main():
+    """Execute scan on DB to gather concept IDs and add merged concepts."""
+    start = timer()
+    db = Database()
+
+    print("Compiling concept IDs...")
+    concept_ids = []
+    done = False
+    start_key = None
+    args = dict()
+    while not done:
+        if start_key:
+            args['ExclusiveStartKey'] = start_key
+        response = db.therapies.scan(**args)
+        items = response['Items']
+        for item in items:
+            if item['label_and_type'].endswith('##identity'):
+                if item['src_name'] != SourceName.CHEMBL.value and \
+                        item['src_name'] != SourceName.DRUGBANK.value:
+                    concept_ids.append(item['concept_id'])
+
+        start_key = response.get('LastEvaluatedKey', None)
+        done = start_key is None
+
+    print("Concept ID scan successful.")
+    print(f"{len(concept_ids)} concept IDs gathered.")
+
+    merge = Merge(db)
+
+    merge.create_merged_concepts(concept_ids)
+    end = timer()
+    print(f"Merge generation successful, time: {end - start}")
+
+
+if __name__ == '__main__':
+    if 'THERAPY_NORM_DB_URL' not in environ.keys():
+        if click.confirm("Are you sure you want to update"
+                         " the production database?", default=False):
+            click.echo("Updating production db...")
+        else:
+            click.echo("Exiting.")
+            sys.exit()
+    main()
diff --git a/main.py b/main.py
index 1987ea8a..55bf3166 100644
--- a/main.py
+++ b/main.py
@@ -1,13 +1,13 @@
 """Main application for FastAPI"""
-from therapy.query import Normalizer, InvalidParameterException
-from therapy.schemas import Service
+from therapy.query import QueryHandler, InvalidParameterException
+from therapy.schemas import Service, NormalizationService
 from fastapi import FastAPI, HTTPException, Query
 from fastapi.openapi.utils import get_openapi
 import html
 from typing import Optional
 
 
-normalizer = Normalizer()
+query_handler = QueryHandler()
 app = FastAPI(docs_url='/therapy', openapi_url='/therapy/openapi.json')
 
 
@@ -37,32 +37,33 @@ def custom_openapi():
 app.openapi = custom_openapi
 
 # endpoint description text
-read_query_summary = """Given query, provide highest matches from
-                     aggregated sources."""
-response_description = "A response to a validly-formed query."
-q_descr = "Therapy to normalize."
-keyed_descr = """Optional. If true, return response as key-value pairs of
-              sources to source matches. False by default"""
-incl_descr = """Optional. Comma-separated list of source names to include in
-             response. Will exclude all other sources. Will return HTTP
-             status code 422: Unprocessable Entity if both 'incl' and
-             'excl' parameters are given."""
-excl_descr = """Optional. Comma-separated list of source names to exclude in
-             response. Will include all other sources. Will return HTTP
-             status code 422: Unprocessable Entity if both 'incl' and
-             'excl' parameters are given."""
+get_matches_summary = ("Given query, provide highest matches from "
+                       "each source.")
+response_descr = "A response to a validly-formed query."
+q_descr = "Therapy to search."
+keyed_descr = ("If true, return response as key-value pairs of "
+               "sources to source matches. False by default")
+incl_descr = ("Comma-separated list of source names to include in "
+              "response. Will exclude all other sources. Will return HTTP "
+              "status code 422: Unprocessable Entity if both 'incl' and "
+              "'excl' parameters are given.")
+excl_descr = ("Comma-separated list of source names to exclude in "
+              "response. Will include all other sources. Will return HTTP "
+              "status code 422: Unprocessable Entity if both 'incl' and"
+              "'excl' parameters are given.")
 
 
 @app.get("/therapy/search",
-         summary=read_query_summary,
+         summary=get_matches_summary,
          operation_id="getQueryResponse",
-         response_description=response_description,
+         response_description=response_descr,
          response_model=Service)
-def read_query(q: str = Query(..., description=q_descr),
-               keyed: Optional[bool] = Query(False, description=keyed_descr),
-               incl: Optional[str] = Query('', description=incl_descr),
-               excl: Optional[str] = Query('', description=excl_descr)):
-    """Return strongest match concepts to query string provided by user.
+def search(q: str = Query(..., description=q_descr),
+           keyed: Optional[bool] = Query(False, description=keyed_descr),
+           incl: Optional[str] = Query('', description=incl_descr),
+           excl: Optional[str] = Query('', description=excl_descr)):
+    """For each source, return strongest-match concepts for query string
+    provided by user.
 
     :param q: therapy search term
     :param keyed: if true, response is structured as key/value pair of
@@ -75,9 +76,31 @@ def read_query(q: str = Query(..., description=q_descr),
     :returns: JSON response with matched records and source metadata
     """
     try:
-        resp = normalizer.normalize(html.unescape(q), keyed=keyed, incl=incl,
-                                    excl=excl)
+        response = query_handler.search_sources(html.unescape(q), keyed=keyed,
+                                                incl=incl, excl=excl)
     except InvalidParameterException as e:
         raise HTTPException(status_code=422, detail=str(e))
+    return response
 
-    return resp
+
+merged_matches_summary = "Given query, provide merged normalized record."
+merged_response_descr = "A response to a validly-formed query."
+merged_q_descr = "Therapy to normalize."
+
+
+@app.get("/therapy/normalize",
+         summary=merged_matches_summary,
+         operation_id="getQuerymergedResponse",
+         response_description=merged_response_descr,
+         response_model=NormalizationService)
+def normalize(q: str = Query(..., description=merged_q_descr)):
+    """Return merged strongest-match concept for query string provided by
+    user.
+
+    :param q: therapy search term
+    """
+    try:
+        response = query_handler.search_groups(q)
+    except InvalidParameterException as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    return response
diff --git a/profiler.py b/profiler.py
deleted file mode 100644
index 139f545b..00000000
--- a/profiler.py
+++ /dev/null
@@ -1,78 +0,0 @@
-"""This tool is used for profiling query return speed.
-Call from command line with optional # of trials argument; otherwise,
-will run 12 trials by default.
-"""
-from therapy.query import Normalizer
-import random
-from timeit import default_timer as timer
-from sys import argv
-
-
-QUERY_TERM_LIST = [
-    # labels
-    'ACETAMINOPHEN',
-    '2,4-DICHLOROPHENOXYBUTYRIC ACID',
-    # labels case insensitive
-    'DaPANSUTRILE',
-    'TmC-647055',
-    # concept ID
-    'chembl:CHEMBL1200807',
-    'chembl:CHEMBL55895',
-    # concept ID namespace case insensitive
-    'wikidata:Q28852422',
-    'chembl:CHEMBL1201274',
-    # aliases
-    'Adagen',
-    '4-Hydroxy3-Methoxy Cinnamaldehyde',
-    # aliases case insensitive
-    'tcMDC-133999',
-    '3-BROMO-Benzoic Acid',
-    # trade name
-    'BETAMETHASONE VALERATE',
-    'M.V.I.-12 ADULT',
-    # trade name case insensitive
-    'THIOtHIXENE',
-    'COLY-mYCIN S',
-    # nonexistant
-    'alksdfkjlwer',
-    'roulkwernduel',
-    # blank query
-    '',
-]
-
-
-def run_trials(n=12):
-    """Run n trials"""
-    normalizer = Normalizer()
-    """Run n trials on a miscellaneous slate of queries"""
-    queries_per_trial = []
-    print(f"Running {n} trials")
-    for i in range(n):
-        to_randomize = QUERY_TERM_LIST[:]
-        random.shuffle(to_randomize)
-        queries_per_trial.append(to_randomize)
-    times = []
-    for i, queries in enumerate(queries_per_trial):
-        start = timer()
-        for query in queries:
-            normalizer.normalize(query)
-        end = timer()
-        duration = end - start
-        print(f"Trial {i + 1}: {duration:.5f} seconds")
-        times.append(duration)
-    avg = sum(times) / len(times)
-    max_val = max(times)
-    min_val = min(times)
-    print(f"Average: {avg:.5f} seconds")
-    print(f"maximum: {max_val:.5f} seconds")
-    print(f"minimum: {min_val:.5f} seconds")
-
-
-if __name__ == '__main__':
-    if len(argv) > 1:
-        try:
-            run_trials(int(argv[1]))
-        except TypeError:
-            raise Exception("Cannot parse {argv[1]} as an integer")
-    else:
-        run_trials()
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 71402dea..1d43dc9c 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -2,12 +2,9 @@
 # These requirements were autogenerated by pipenv
 # To regenerate from the project's Pipfile, run:
 #
-#    pipenv lock --requirements --dev
+#    pipenv lock --requirements --dev-only
 #
 
-# Note: in pipenv 2020.x, "--dev" changed to emit both default and development
-# requirements. To emit only development requirements, pass "--dev-only".
-
 -i https://pypi.org/simple
 -e .
 anyio==2.0.2; python_full_version >= '3.6.2'
@@ -19,121 +16,91 @@ attrs==20.3.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2,
 babel==2.9.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 backcall==0.2.0
 backports-datetime-fromisoformat==1.0.0
-beautifulsoup4==4.9.3
-bleach==3.2.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
-boto3==1.16.54
-botocore==1.19.54
-bs4==0.0.1
+bleach==3.2.2; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
 certifi==2020.12.5
 cffi==1.14.4
 cfgv==3.2.0; python_full_version >= '3.6.1'
 chardet==4.0.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
 civicpy==1.1.2
 click==7.1.2
-colorama==0.4.4; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
 coverage==5.3.1
 coveralls==3.0.0
-cython==0.29.21
 decorator==4.4.2
 defusedxml==0.6.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
 deprecation==2.1.0
 distlib==0.3.1
 docopt==0.6.2
-docutils==0.16; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
 entrypoints==0.3; python_version >= '2.7'
-fastapi==0.63.0
 filelock==3.0.12
 flake8-docstrings==1.5.0
 flake8==3.8.4
-h11==0.12.0; python_version >= '3.6'
-httptools==0.1.1
-identify==1.5.12; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
+identify==1.5.13; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 idna==2.10; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 iniconfig==1.1.1
 ipykernel==5.4.3
 ipython-genutils==0.2.0
 ipython==7.19.0; python_version >= '3.7'
-isodate==0.6.0
 jedi==0.18.0; python_version >= '3.6'
 jinja2==2.11.2; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
-jmespath==0.10.0; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2, 3.3'
 json5==0.9.5
 jsonschema==3.2.0
 jupyter-client==6.1.11; python_version >= '3.5'
 jupyter-core==4.7.0; python_version >= '3.6'
-jupyter-server==1.2.1; python_version >= '3.6'
+jupyter-server==1.2.2; python_version >= '3.6'
 jupyterlab-pygments==0.1.2
 jupyterlab-server==2.1.2; python_version >= '3.6'
-jupyterlab==3.0.4
-keyring==21.8.0; python_version >= '3.6'
-lxml==4.6.2
+jupyterlab==3.0.5
 markupsafe==1.1.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 mccabe==0.6.1
 mistune==0.8.4
 nbclassic==0.2.6; python_version >= '3.6'
 nbclient==0.5.1; python_version >= '3.6'
 nbconvert==6.0.7; python_version >= '3.6'
-nbformat==5.1.1; python_version >= '3.5'
+nbformat==5.1.2; python_version >= '3.5'
 nest-asyncio==1.4.3; python_version >= '3.5'
 networkx==2.5; python_version >= '3.6'
 nodeenv==1.5.0
 notebook==6.2.0; python_version >= '3.5'
 numpy==1.19.5; python_version >= '3.6'
 obonet==0.2.6; python_version >= '3'
-owlready2==0.26
 packaging==20.8; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
-pandas==1.2.0; python_full_version >= '3.7.1'
+pandas==1.2.1; python_full_version >= '3.7.1'
 pandocfilters==1.4.3
 parso==0.8.1; python_version >= '3.6'
 pexpect==4.8.0; sys_platform != 'win32'
 pickleshare==0.7.5
-pkginfo==1.6.1
 pluggy==0.13.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 pre-commit==2.9.3
 prometheus-client==0.9.0
-prompt-toolkit==3.0.10; python_full_version >= '3.6.1'
+prompt-toolkit==3.0.14; python_full_version >= '3.6.1'
 ptyprocess==0.7.0
 py==1.10.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 pycodestyle==2.6.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 pycparser==2.20; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
-pydantic==1.7.3
 pydocstyle==5.1.1; python_version >= '3.5'
 pyflakes==2.2.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 pygments==2.7.4; python_version >= '3.5'
-pyparsing==2.4.7; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2, 3.3'
+pyparsing==2.4.7; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2'
 pyrsistent==0.17.3; python_version >= '3.5'
 pysam==0.16.0.1
-pytest-cov==2.10.1
-pytest==6.2.1
-python-dateutil==2.8.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
+pytest-cov==2.11.1
+pytest==6.2.2
+python-dateutil==2.8.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2'
 pytz==2020.5
-pyyaml==5.3.1
-pyzmq==21.0.0; python_version >= '3.6'
-rdflib==5.0.0
-readme-renderer==28.0
-requests-toolbelt==0.9.1
+pyyaml==5.4.1
+pyzmq==21.0.1; python_version >= '3.6'
 requests==2.25.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
-rfc3986==1.4.0
-s3transfer==0.3.4
 send2trash==1.5.0
-six==1.15.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
+six==1.15.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2'
 sniffio==1.2.0; python_version >= '3.5'
-snowballstemmer==2.0.0
-soupsieve==2.1; python_version >= '3.0'
-starlette==0.13.6; python_version >= '3.6'
+snowballstemmer==2.1.0
 terminado==0.9.2; python_version >= '3.6'
 testpath==0.4.4
-toml==0.10.2; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2, 3.3'
+toml==0.10.2; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2'
 tornado==6.1; python_version >= '3.5'
-tqdm==4.56.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 traitlets==5.0.5; python_version >= '3.7'
-twine==3.3.0
-typing-extensions==3.7.4.3
 urllib3==1.26.2; python_version != '3.4'
-uvicorn==0.13.3
-uvloop==0.14.0
 vcfpy==0.13.3
-virtualenv==20.3.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
+virtualenv==20.4.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 wcwidth==0.2.5
 webencodings==0.5.1
-websockets==8.1
diff --git a/requirements.txt b/requirements.txt
index 71ee71ea..b378edcc 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,9 +9,9 @@
 appnope==0.1.2; sys_platform == 'darwin' and platform_system == 'Darwin'
 backcall==0.2.0
 beautifulsoup4==4.9.3
-bleach==3.2.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
-boto3==1.16.54
-botocore==1.19.54
+bleach==3.2.2; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
+boto3==1.16.59
+botocore==1.19.59
 bs4==0.0.1
 certifi==2020.12.5
 chardet==4.0.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
@@ -29,31 +29,32 @@ ipython-genutils==0.2.0
 ipython==7.19.0; python_version >= '3.7'
 isodate==0.6.0
 jedi==0.18.0; python_version >= '3.6'
-jmespath==0.10.0; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2, 3.3'
+jmespath==0.10.0; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2'
 jupyter-client==6.1.11; python_version >= '3.5'
 jupyter-core==4.7.0; python_version >= '3.6'
-keyring==21.8.0; python_version >= '3.6'
+keyring==22.0.0; python_version >= '3.6'
 lxml==4.6.2
 owlready2==0.26
 packaging==20.8; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 parso==0.8.1; python_version >= '3.6'
 pexpect==4.8.0; sys_platform != 'win32'
 pickleshare==0.7.5
-pkginfo==1.6.1
-prompt-toolkit==3.0.10; python_full_version >= '3.6.1'
+pkginfo==1.7.0
+prompt-toolkit==3.0.14; python_full_version >= '3.6.1'
 ptyprocess==0.7.0
 pydantic==1.7.3
 pygments==2.7.4; python_version >= '3.5'
-pyparsing==2.4.7; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2, 3.3'
-python-dateutil==2.8.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
-pyzmq==21.0.0; python_version >= '3.6'
+pyparsing==2.4.7; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2'
+python-dateutil==2.8.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2'
+pyyaml==5.4.1
+pyzmq==21.0.1; python_version >= '3.6'
 rdflib==5.0.0
 readme-renderer==28.0
 requests-toolbelt==0.9.1
 requests==2.25.1; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
 rfc3986==1.4.0
 s3transfer==0.3.4
-six==1.15.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
+six==1.15.0; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2'
 soupsieve==2.1; python_version >= '3.0'
 starlette==0.13.6; python_version >= '3.6'
 tornado==6.1; python_version >= '3.5'
diff --git a/therapy/database.py b/therapy/database.py
index 582fb662..9b7b8776 100644
--- a/therapy/database.py
+++ b/therapy/database.py
@@ -1,44 +1,67 @@
 """This module creates the database."""
 import boto3
+from therapy.schemas import NamespacePrefix, SourceName
+from botocore.exceptions import ClientError
+from boto3.dynamodb.conditions import Key
 from os import environ
+from typing import Optional, Dict, List, Any
+import logging
+
+logger = logging.getLogger('therapy')
+logger.setLevel(logging.DEBUG)
+
+
+# should this be located in a shared module
+PREFIX_LOOKUP = {v.value: SourceName[k].value
+                 for k, v in NamespacePrefix.__members__.items()
+                 if k in SourceName.__members__.keys()}
 
 
 class Database:
     """The database class."""
 
-    def __init__(self, db_url='', region_name='us-east-2'):
-        """Initialize Database class."""
+    def __init__(self, db_url: str = '', region_name: str = 'us-east-2'):
+        """Initialize Database class.
+
+        :param str db_url: database endpoint URL to connect to
+        :param str region_name: AWS region name to use
+        """
         if db_url:
-            self.ddb = boto3.resource('dynamodb', region_name=region_name,
-                                      endpoint_url=db_url)
-            self.ddb_client = boto3.client('dynamodb',
-                                           region_name=region_name,
-                                           endpoint_url=db_url)
+            boto_params = {
+                'region_name': region_name,
+                'endpoint_url': db_url
+            }
         elif 'THERAPY_NORM_DB_URL' in environ.keys():
-            db_url = environ['THERAPY_NORM_DB_URL']
-            self.ddb = boto3.resource('dynamodb', region_name=region_name,
-                                      endpoint_url=db_url)
-            self.ddb_client = boto3.client('dynamodb',
-                                           region_name=region_name,
-                                           endpoint_url=db_url)
+            boto_params = {
+                'region_name': region_name,
+                'endpoint_url': environ['THERAPY_NORM_DB_URL']
+            }
         else:
-            self.ddb = boto3.resource('dynamodb', region_name=region_name)
-            self.ddb_client = boto3.client('dynamodb',
-                                           region_name=region_name)
+            boto_params = {
+                'region_name': region_name
+            }
+        self.dynamodb = boto3.resource('dynamodb', **boto_params)
+        self.dynamodb_client = boto3.client('dynamodb', **boto_params)
+
+        # create tables if nonexistant if not connecting to remote database
         if db_url or 'THERAPY_NORM_DB_URL' in environ.keys():
-            existing_tables = self.ddb_client.list_tables()['TableNames']
+            existing_tables = self.dynamodb_client.list_tables()['TableNames']
             self.create_therapies_table(existing_tables)
             self.create_meta_data_table(existing_tables)
 
-        self.therapies = self.ddb.Table('therapy_concepts')
-        self.metadata = self.ddb.Table('therapy_metadata')
+        self.therapies = self.dynamodb.Table('therapy_concepts')
+        self.metadata = self.dynamodb.Table('therapy_metadata')
+        self.batch = self.therapies.batch_writer()
         self.cached_sources = {}
 
-    def create_therapies_table(self, existing_tables):
-        """Create Therapies table if not exists."""
+    def create_therapies_table(self, existing_tables: List):
+        """Create Therapies table if it doesn't already exist.
+
+        :param List existing_tables: list of existing table names
+        """
         table_name = 'therapy_concepts'
         if table_name not in existing_tables:
-            self.ddb.create_table(
+            self.dynamodb.create_table(
                 TableName=table_name,
                 KeySchema=[
                     {
@@ -89,11 +112,14 @@ def create_therapies_table(self, existing_tables):
                 }
             )
 
-    def create_meta_data_table(self, existing_tables):
-        """Create MetaData table if not exists."""
+    def create_meta_data_table(self, existing_tables: List):
+        """Create MetaData table if not exists.
+
+        :param List existing_tables: list of existing table names
+        """
         table_name = 'therapy_metadata'
         if table_name not in existing_tables:
-            self.ddb.create_table(
+            self.dynamodb.create_table(
                 TableName=table_name,
                 KeySchema=[
                     {
@@ -112,3 +138,112 @@ def create_meta_data_table(self, existing_tables):
                     'WriteCapacityUnits': 10
                 }
             )
+
+    def get_record_by_id(self, concept_id: str,
+                         case_sensitive: bool = True,
+                         merge: bool = False) -> Optional[Dict]:
+        """Fetch record corresponding to provided concept ID
+
+        :param str concept_id: concept ID for therapy record
+        :param bool case_sensitive: if true, performs exact lookup, which is
+            more efficient. Otherwise, performs filter operation, which
+            doesn't require correct casing.
+        :param bool merge: if true, look for merged concept
+        :return: complete therapy record, if match is found; None otherwise
+        """
+        try:
+            if merge:
+                pk = f'{concept_id.lower()}##merger'
+            else:
+                pk = f'{concept_id.lower()}##identity'
+            if case_sensitive:
+                match = self.therapies.get_item(Key={
+                    'label_and_type': pk,
+                    'concept_id': concept_id
+                })
+                return match['Item']
+            else:
+                exp = Key('label_and_type').eq(pk)
+                response = self.therapies.query(KeyConditionExpression=exp)
+                return response['Items'][0]
+        except ClientError as e:
+            logger.error(e.response['Error']['Message'])
+            return None
+        except KeyError:  # record doesn't exist
+            return None
+        except IndexError:  # record doesn't exist
+            return None
+
+    def get_records_by_type(self, query: str,
+                            match_type: str) -> List[Dict]:
+        """Retrieve records for given query and match type.
+
+        :param query: string to match against
+        :param str match_type: type of match to look for. Should be one
+            of "alias", "trade_name", "label", or "rx_brand" (use
+            `get_record_by_id` for concept ID lookup)
+        :return: list of matching records. Empty if lookup fails.
+        """
+        pk = f'{query}##{match_type.lower()}'
+        filter_exp = Key('label_and_type').eq(pk)
+        try:
+            matches = self.therapies.query(KeyConditionExpression=filter_exp)
+            return matches.get('Items', None)
+        except ClientError as e:
+            logger.error(e.response['Error']['Message'])
+            return []
+
+    def get_merged_record(self, merge_ref) -> Optional[Dict]:
+        """Fetch merged record from given reference.
+
+        :param str merge_ref: key for merged record, formated as a string
+            of grouped concept IDs separated by vertical bars, ending with
+            `##merger`. Must be correctly-cased.
+        :return: complete merged record if lookup successful, None otherwise
+        """
+        try:
+            match = self.therapies.get_item(Key={
+                'label_and_type': merge_ref.lower(),
+                'concept_id': merge_ref
+            })
+            return match['Item']
+        except ClientError as e:
+            logger.error(e.response['Error']['Message'])
+            return None
+        except KeyError:
+            return None
+
+    def add_record(self, record: Dict, record_type: str):
+        """Add new record to database.
+
+        :param Dict record: record (of any type) to upload. Must include
+            `concept_id` key. If record is of the `identity` type, the
+            concept_id must be correctly-cased.
+        :param str record_type: one of 'identity', 'label', 'alias',
+            'trade_name', or 'merger'.
+        """
+        if record_type in ('label', 'trade_name', 'alias', 'identity'):
+            prefix = record['concept_id'].split(':')[0]
+            record['src_name'] = PREFIX_LOOKUP[prefix]
+        record['label_and_type'] = f'{record["concept_id"].lower()}##{record_type}'  # noqa: E501
+        self.batch.put_item(Item=record)
+
+    def update_record(self, concept_id: str, field: str, new_value: Any):
+        """Update the field of an individual record to a new value.
+
+        :param str concept_id: record to update
+        :param str field: name of field to update
+        :parm str new_value: new value
+        """
+        key = {
+            'label_and_type': f'{concept_id.lower()}##identity',
+            'concept_id': concept_id
+        }
+        update_expression = f"set {field}=:r"
+        update_values = {':r': new_value}
+        try:
+            self.therapies.update_item(Key=key,
+                                       UpdateExpression=update_expression,
+                                       ExpressionAttributeValues=update_values)
+        except ClientError as e:
+            logger.error(e.response['Error']['Message'])
diff --git a/therapy/etl/__init__.py b/therapy/etl/__init__.py
index 121a845f..1351824c 100644
--- a/therapy/etl/__init__.py
+++ b/therapy/etl/__init__.py
@@ -5,3 +5,4 @@
 from .drugbank import DrugBank  # noqa: F401
 from .chemidplus import ChemIDplus  # noqa: F401
 from .rxnorm import RxNorm  # noqa: F401
+from .merge import Merge  # noqa: F401
diff --git a/therapy/etl/base.py b/therapy/etl/base.py
index 5114e3fd..86d6e43d 100644
--- a/therapy/etl/base.py
+++ b/therapy/etl/base.py
@@ -2,6 +2,7 @@
 from abc import ABC, abstractmethod
 from therapy.schemas import NamespacePrefix
 from therapy.database import Database
+from typing import List
 
 # prefixes for translating ID namespaces
 IDENTIFIER_PREFIXES = {
@@ -19,10 +20,22 @@
 class Base(ABC):
     """The ETL base class."""
 
-    def __init__(self, database: Database, *args, **kwargs):
+    def __init__(self, database: Database):
         """Extract from sources."""
         self.database = database
-        self._load_data(*args, **kwargs)
+        self._added_ids = []
+
+    @abstractmethod
+    def perform_etl(self) -> List[str]:
+        """Public-facing method to begin ETL procedures on given data.
+
+        Returned concept IDs can be passed to Merge method for computing
+        merged concepts.
+
+        :return: list of concept IDs which were successfully processed and
+            uploaded.
+        """
+        raise NotImplementedError
 
     @abstractmethod
     def _extract_data(self, *args, **kwargs):
@@ -33,5 +46,5 @@ def _transform_data(self, *args, **kwargs):
         raise NotImplementedError
 
     @abstractmethod
-    def _add_meta(self, *args, **kwargs):
+    def _load_meta(self, *args, **kwargs):
         raise NotImplementedError
diff --git a/therapy/etl/chembl.py b/therapy/etl/chembl.py
index 0b9078b3..750d2729 100644
--- a/therapy/etl/chembl.py
+++ b/therapy/etl/chembl.py
@@ -1,10 +1,12 @@
 """This module defines the ChEMBL ETL methods."""
 from .base import Base
 from therapy import PROJECT_ROOT
+from therapy.database import Database
+from therapy.schemas import SourceName, NamespacePrefix, ApprovalStatus, Meta
+from typing import List
 from ftplib import FTP
 import logging
 import tarfile
-from therapy.schemas import SourceName, NamespacePrefix, ApprovalStatus, Meta
 import sqlite3
 import os
 import shutil
@@ -16,6 +18,31 @@
 class ChEMBL(Base):
     """ETL the ChEMBL source into therapy.db."""
 
+    def __init__(self,
+                 database: Database,
+                 data_path=PROJECT_ROOT / 'data' / 'chembl' / 'chembl_27.db'):
+        """Initialize CHEMBl ETL instance.
+
+        :param Path data_path: path to CHEMBl source SQLite3 database file.
+        """
+        self.database = database
+        self._data_path = data_path
+        self._added_ids = []
+
+    def perform_etl(self) -> List[str]:
+        """Public-facing method to initiate ETL procedures on given data.
+
+        :return: List of concept IDs which were successfully processed and
+            uploaded.
+        """
+        self._load_meta()
+        self._extract_data()
+        self._transform_data()
+        self._load_json()
+        self._conn.commit()
+        self._conn.close()
+        return self._added_ids
+
     def _extract_data(self, *args, **kwargs):
         """Extract data from the ChEMBL source."""
         logger.info('Extracting chembl_27.db...')
@@ -56,15 +83,6 @@ def _transform_data(self, *args, **kwargs):
         self._cursor.execute("DROP TABLE DictionarySynonyms;")
         self._cursor.execute("DROP TABLE TradeNames;")
 
-    def _load_data(self, *args, **kwargs):
-        """Load the ChEMBL source into database."""
-        self._add_meta()
-        self._extract_data()
-        self._transform_data()
-        self._load_json()
-        self._conn.commit()
-        self._conn.close()
-
     @staticmethod
     def _download_chembl_27(filepath):
         """Download the ChEMBL source."""
@@ -220,6 +238,7 @@ def _load_therapy(self, record, batch):
         record['label_and_type'] = \
             f"{record['concept_id'].lower()}##identity"
         batch.put_item(Item=record)
+        self._added_ids.append(record['concept_id'])
 
     def _load_label(self, record, batch):
         """Load label record into DynamoDB."""
@@ -274,7 +293,7 @@ def _load_trade_name(self, record, batch):
 
             record['trade_names'] = list(set(record['trade_names']))
 
-    def _add_meta(self, *args, **kwargs):
+    def _load_meta(self, *args, **kwargs):
         """Add ChEMBL metadata."""
         metadata = Meta(data_license='CC BY-SA 3.0',
                         data_license_url='https://creativecommons.org/licenses/by-sa/3.0/',  # noqa: E501
diff --git a/therapy/etl/chemidplus.py b/therapy/etl/chemidplus.py
index 779517a5..1388a150 100644
--- a/therapy/etl/chemidplus.py
+++ b/therapy/etl/chemidplus.py
@@ -3,6 +3,7 @@
 Courtesy of the U.S. National Library of Medicine.
 """
 from .base import Base
+from typing import List, Dict
 from therapy import PROJECT_ROOT
 from therapy.database import Database
 from therapy.schemas import Drug, NamespacePrefix, Meta, SourceName, \
@@ -11,7 +12,6 @@
 from ftplib import FTP
 import xml.etree.ElementTree as ET
 import logging
-from typing import Dict
 from boto3.dynamodb.table import BatchWriter
 import re
 
@@ -45,15 +45,22 @@ def __init__(self,
         it's unnecessary to provide `src_dir` and `src_fname` args.
         """
         self.database = database
+        self._data_path = data_path
         self._src_server = src_server
         self._src_dir_path = src_dir_path
         self._src_fname = src_fname
-        self._added_ids = set()
+        self._added_ids = []
 
-        # perform ETL
-        self._extract_data(data_path)
-        self._add_meta()
+    def perform_etl(self) -> List[str]:
+        """Public-facing method to initiate ETL procedures on given data.
+
+        :return: List of concept IDs which were successfully processed and
+            uploaded.
+        """
+        self._extract_data()
+        self._load_meta()
         self._transform_data()
+        return self._added_ids
 
     def _download_data(self, data_path: Path):
         """Download source data from default location."""
@@ -74,21 +81,22 @@ def _download_data(self, data_path: Path):
         outfile_path.rename(data_path / f'chemidplus_{version}.xml')
         logger.info('Finished downloading ChemIDplus data')
 
-    def _extract_data(self, data_path: Path):
+    def _extract_data(self):
         """Acquire ChemIDplus dataset.
 
         :arg pathlib.Path data_path: directory containing source data
         """
-        data_path.mkdir(exist_ok=True, parents=True)
-        dir_files = list(data_path.iterdir())
+        self._data_path.mkdir(exist_ok=True, parents=True)
+        dir_files = list(self._data_path.iterdir())
 
         if len(dir_files) == 0:
-            file = self._get_file(data_path)
+            file = self._get_file(self._data_path)
         else:
             file = sorted([f for f in dir_files
                            if f.name.startswith('chemidplus')])
             if not file:
-                file = self._get_file(data_path)
+                file = self._get_file(self._data_path)
+
         self._data_src = file[-1]
         self._version = self._data_src.stem.split('_')[1]
 
@@ -173,9 +181,9 @@ def _load_record(self, batch: BatchWriter, record: Dict):
         record['src_name'] = SourceName.CHEMIDPLUS.value
         record['label_and_type'] = f'{concept_id_l}##identity'
         batch.put_item(Item=record)
-        self._added_ids.add(record['concept_id'])
+        self._added_ids.append(record['concept_id'])
 
-    def _add_meta(self):
+    def _load_meta(self):
         """Add source metadata."""
         meta = Meta(data_license="custom",
                     data_license_url="https://www.nlm.nih.gov/databases/download/terms_and_conditions.html",  # noqa: E501
diff --git a/therapy/etl/drugbank.py b/therapy/etl/drugbank.py
index 2930f6d7..9122d3d9 100644
--- a/therapy/etl/drugbank.py
+++ b/therapy/etl/drugbank.py
@@ -1,10 +1,12 @@
 """This module defines the DrugBank ETL methods."""
-from therapy.etl.base import Base
 from therapy import PROJECT_ROOT
-import logging
 from therapy.schemas import SourceName, NamespacePrefix, ApprovalStatus, Meta
-from therapy.etl.base import IDENTIFIER_PREFIXES
+from therapy.etl.base import IDENTIFIER_PREFIXES, Base
+from therapy.database import Database
+import logging
 from lxml import etree
+from typing import List
+from pathlib import Path
 import requests
 from requests.auth import HTTPBasicAuth
 from os import environ
@@ -38,24 +40,31 @@ class DrugBank(Base):
     """ETL the DrugBank source into therapy.db."""
 
     def __init__(self,
-                 database,
-                 version='5.1.7',
-                 data_url='https://go.drugbank.com/releases/5-1-7/'
-                          'downloads/all-full-database',
-                 data_dir=PROJECT_ROOT / 'data' / 'drugbank'
-                 ):
-        """Initialize DrugBank class.
-
-        :param Database database: DynamoDB object
-        :param str version: DrugBank version to use
-        :param str data_url: URL to DrugBank data file
-        :param Path data_dir: Path to DrugBank data directory
+                 database: Database,
+                 data_path: Path = PROJECT_ROOT / 'data' / 'drugbank',
+                 data_url: str = 'https://go.drugbank.com/releases/5-1-7/'
+                                 'downloads/all-full-database',
+                 version: str = '5.1.7'):
+        """Initialize ETL class instance.
+
+        :param Path data_path: directory containing source data
         """
         self._database = database
-        self._version = version
+        self._data_path = data_path
         self._data_url = data_url
-        self._data_dir = data_dir
-        self._load_data()
+        self._version = version
+        self._added_ids = []
+
+    def perform_etl(self) -> List[str]:
+        """Public-facing method to initiate ETL procedures on given data.
+
+        :return: List of concept IDs which were successfully processed and
+            uploaded.
+        """
+        self._extract_data()
+        self._load_meta()
+        self._transform_data()
+        return self._added_ids
 
     def _download_data(self):
         """Download DrugBank database XML file.
@@ -92,17 +101,14 @@ def _download_data(self):
                            "and DRUGBANK_PWD.")
         logger.info("Successfully downloaded DrugBank file.")
 
-    def _extract_data(self, *args, **kwargs):
+    def _extract_data(self):
         """Extract data from the DrugBank source."""
         logger.info("Extracting DrugBank file...")
-        if 'data_path' in kwargs:
-            self._data_src = kwargs['data_path']
-        else:
-            self._data_dir.mkdir(exist_ok=True, parents=True)
-            file_path = self._data_dir / f"drugbank_{self._version}.xml"
-            if not file_path.exists():
-                self._download_data()
-            self._data_src = file_path
+        self._data_path.mkdir(exist_ok=True, parents=True)
+        file_path = self._data_path / f"drugbank_{self._version}.xml"
+        if not file_path.exists():
+            self._download_data()
+        self._data_src = file_path
         logger.info(f"Extracted {self._data_src}.")
 
     def _transform_data(self):
@@ -172,12 +178,6 @@ def _transform_data(self):
                     self._load_trade_names(params['trade_names'],
                                            params['concept_id'], batch)
 
-    def _load_data(self, *args, **kwargs):
-        """Load the DrugBank source into normalized database."""
-        self._extract_data()
-        self._add_meta()
-        self._transform_data()
-
     def _load_therapy(self, batch, params):
         """Filter out trade names and aliases that exceed 20 and add item to
         therapies table.
@@ -193,6 +193,7 @@ def _load_therapy(self, batch, params):
                         {a.casefold() for a in params[label_type]}) > 20:
                     del params[label_type]
         batch.put_item(Item=params)
+        self._added_ids.append(params['concept_id'])
 
     def _load_drugbank_id(self, element, params):
         """Load drugbank id as concept id or alias."""
@@ -306,7 +307,7 @@ def _load_trade_names(self, trade_names, concept_id, batch):
             }
             batch.put_item(Item=trade_name)
 
-    def _add_meta(self):
+    def _load_meta(self):
         """Add DrugBank metadata."""
         meta = Meta(data_license='CC BY-NC 4.0',
                     data_license_url='https://creativecommons.org/licenses/by-nc/4.0/legalcode',  # noqa: E501
diff --git a/therapy/etl/merge.py b/therapy/etl/merge.py
new file mode 100644
index 00000000..cd5790fa
--- /dev/null
+++ b/therapy/etl/merge.py
@@ -0,0 +1,188 @@
+"""Create concept groups and merged records."""
+from therapy.schemas import SourceName, NamespacePrefix
+from therapy.database import Database
+from typing import Set, Dict
+import logging
+from timeit import default_timer as timer
+
+logger = logging.getLogger('therapy')
+logger.setLevel(logging.DEBUG)
+
+DISALLOWED_SOURCES = {NamespacePrefix.DRUGBANK.value,
+                      NamespacePrefix.CHEMBL.value}
+
+
+class Merge:
+    """Handles record merging."""
+
+    def __init__(self, database: Database):
+        """Initialize Merge instance.
+
+        :param Database database: db instance to use for record retrieval
+            and creation.
+        """
+        self._database = database
+        self._groups = {}  # dict keying concept IDs to group Sets
+
+    def create_merged_concepts(self, record_ids: Set[str]):
+        """Create concept groups, generate merged concept records, and
+        update database.
+
+        :param Set[str] record_ids: concept identifiers from which groups
+            should be generated. Should *not* include any records from
+            excluded sources.
+        """
+        logger.info('Generating record ID sets...')
+        start = timer()
+        for record_id in record_ids:
+            new_group = self._create_record_id_set(record_id)
+            if new_group:
+                for concept_id in new_group:
+                    self._groups[concept_id] = new_group
+        end = timer()
+        logger.debug(f'Built record ID sets in {end - start} seconds')
+
+        logger.info('Creating merged records and updating database...')
+        uploaded_ids = set()
+        start = timer()
+        for record_id, group in self._groups.items():
+            if record_id in uploaded_ids:
+                continue
+            merged_record = self._generate_merged_record(group)
+
+            # add group merger item to DB
+            self._database.add_record(merged_record, 'merger')
+
+            # add updated references
+            for concept_id in group:
+                if not self._database.get_record_by_id(concept_id, False):
+                    logger.error(f"Updating nonexistent record: {concept_id} "
+                                 f"for {merged_record['label_and_type']}")
+                else:
+                    merge_ref = merged_record['concept_id'].lower()
+                    self._database.update_record(concept_id, 'merge_ref',
+                                                 merge_ref)
+            uploaded_ids |= group
+        logger.info('Merged concept generation successful.')
+        end = timer()
+        logger.debug(f'Generated and added concepts in {end - start} seconds')
+
+    def _get_other_ids(self, record: Dict) -> Set[str]:
+        """Extract references to entries in other sources from a record.
+        Crucially, filter to allowed sources only.
+
+        :param Dict record: record to process
+        :return: Set of other_identifier values
+        :rtype: Set
+        """
+        other_ids = set()
+        for other_id in record.get('other_identifiers', []):
+            if other_id.split(':')[0] not in DISALLOWED_SOURCES:
+                other_ids.add(other_id)
+        return other_ids
+
+    def _create_record_id_set(self, record_id: str,
+                              observed_id_set: Set = set()) -> Set[str]:
+        """Create concept ID group for an individual record ID.
+
+        :param str record_id: concept ID for record to build group from
+        :return: set of related identifiers pertaining to a common concept.
+        """
+        if record_id in self._groups:
+            return self._groups[record_id]
+        else:
+            db_record = self._database.get_record_by_id(record_id)
+            if not db_record:
+                # attempt RxNorm brand lookup
+                brand_lookup = self._database.get_records_by_type(record_id,
+                                                                  'rx_brand')
+                if len(brand_lookup) == 1:
+                    lookup_id = brand_lookup[0]['concept_id']
+                    db_record = self._database.get_record_by_id(lookup_id,
+                                                                False)
+                    if db_record:
+                        record_id = db_record['concept_id']
+                        return self._create_record_id_set(record_id,
+                                                          observed_id_set)
+                    else:
+                        return observed_id_set - {record_id}
+
+                else:
+                    logger.warning(f"Record ID set creator could not resolve "
+                                   f"lookup for {record_id} in ID set: "
+                                   f"{observed_id_set}")
+                    return observed_id_set - {record_id}
+
+            local_id_set = self._get_other_ids(db_record)
+            if not local_id_set:
+                return observed_id_set | {db_record['concept_id']}
+            merged_id_set = {record_id} | observed_id_set
+            for local_record_id in local_id_set - observed_id_set:
+                merged_id_set |= self._create_record_id_set(local_record_id,
+                                                            merged_id_set)
+            return merged_id_set
+
+    def _generate_merged_record(self, record_id_set: Set[str]) -> Dict:
+        """Generate merged record from provided concept ID group.
+        Where attributes are sets, they should be merged, and where they are
+        scalars, assign from the highest-priority source where that attribute
+        is non-null.
+
+        Priority is RxNorm > NCIt > ChemIDplus > Wikidata. ChEMBL and DrugBank
+        identifiers should not be passed in record_id_set.
+
+        :param Set record_id_set: group of concept IDs
+        :return: completed merged drug object to be stored in DB
+        """
+        records = []
+        for record_id in record_id_set:
+            record = self._database.get_record_by_id(record_id)
+            if record:
+                records.append(record)
+            else:
+                logger.error(f"Merge record generator could not retrieve "
+                             f"record for {record_id} in {record_id_set}")
+
+        def record_order(record):
+            """Provide priority values of concepts for sort function."""
+            src = record['src_name']
+            if src == SourceName.RXNORM.value:
+                source_rank = 1
+            elif src == SourceName.NCIT.value:
+                source_rank = 2
+            elif src == SourceName.CHEMIDPLUS.value:
+                source_rank = 5
+            elif src == SourceName.WIKIDATA.value:
+                source_rank = 6
+            else:
+                raise Exception(f"Prohibited source: {src} in concept_id "
+                                f"{record['concept_id']}")
+            return (source_rank, record['concept_id'])
+        records.sort(key=record_order)
+
+        merged_attrs = {'aliases': set(), 'concept_id': None,
+                        'trade_names': set(), 'xrefs': set()}
+        set_fields = ['aliases', 'trade_names', 'xrefs']
+        for record in records:
+            for field in set_fields:
+                if field in record:
+                    merged_attrs[field] |= set(record[field])
+            if not merged_attrs['concept_id']:
+                merged_attrs['concept_id'] = record['concept_id']
+            else:
+                merged_attrs['concept_id'] += f"|{record['concept_id']}"
+
+            if 'label' not in merged_attrs \
+                    and 'label' in record and record['label']:
+                merged_attrs['label'] = record.get('label')
+
+        for field in set_fields:
+            field_value = merged_attrs[field]
+            if field_value:
+                merged_attrs[field] = list(field_value)
+            else:
+                del merged_attrs[field]
+
+        merged_attrs['label_and_type'] = \
+            f'{merged_attrs["concept_id"].lower()}##merger'
+        return merged_attrs
diff --git a/therapy/etl/ncit.py b/therapy/etl/ncit.py
index 2636886a..10eb33b8 100644
--- a/therapy/etl/ncit.py
+++ b/therapy/etl/ncit.py
@@ -6,7 +6,8 @@
 import logging
 import owlready2 as owl
 from owlready2.entity import ThingClass
-from typing import Set
+from typing import List, Set
+from pathlib import Path
 import requests
 import zipfile
 from os import remove, rename
@@ -28,87 +29,93 @@ def __init__(self,
                  database: Database,
                  src_dir: str = "https://evs.nci.nih.gov/ftp1/NCI_Thesaurus/archive/20.09d_Release/",  # noqa F401
                  src_fname: str = "Thesaurus_20.09d.OWL.zip",
-                 *args,
-                 **kwargs):
+                 data_path: Path = PROJECT_ROOT / 'data' / 'ncit',
+                 chemidplus_path: Path = PROJECT_ROOT / 'data' / 'chemidplus'):
         """Override base class init method. Call ETL methods.
 
-        Args:
-            src_dir: URL of source directory
-            src_fname: filename for source file within source directory URL
+        :param therapy.database.Database database: app database instance
+        :param str src_dir: URL of remote directory containing source input
+        :param str src_fname: filename for source file within source directory
+        :param pathlib.Path data_path: path to local NCIt data directory
         """
         self.database = database
         self._SRC_DIR = src_dir
         self._SRC_FNAME = src_fname
+        self._data_path = data_path
+        self._chemidplus_path = chemidplus_path
+        self._added_ids = []
+
+    def perform_etl(self) -> List[str]:
+        """Public-facing method to initiate ETL procedures on given data.
+
+        :return: List of concept IDs which were successfully processed and
+            uploaded.
+        """
         self._extract_data()
+        self._load_meta()
         self._transform_data()
+        return self._added_ids
 
     def _download_data(self):
-        """Download NCI thesaurus source file for loading into normalizer"""
+        """Download NCI thesaurus source file for loading into normalizer."""
         logger.info('Downloading NCI Thesaurus...')
         url = self._SRC_DIR + self._SRC_FNAME
-        out_dir = PROJECT_ROOT / 'data' / 'ncit'
-        zip_path = out_dir / 'ncit.zip'
+        zip_path = self._data_path / 'ncit.zip'
         response = requests.get(url, stream=True)
         handle = open(zip_path, "wb")
         for chunk in response.iter_content(chunk_size=512):
             if chunk:
                 handle.write(chunk)
         with zipfile.ZipFile(zip_path, 'r') as zip_ref:
-            zip_ref.extractall(out_dir)
+            zip_ref.extractall(self._data_path)
         remove(zip_path)
         version = self._SRC_DIR.split('/')[-2].split('_')[0]
-        rename(out_dir / 'Thesaurus.owl', out_dir / f'ncit_{version}.owl')
+        rename(self._data_path / 'Thesaurus.owl', self._data_path / f'ncit_{version}.owl')  # noqa: E501
         logger.info('Finished downloading NCI Thesaurus')
 
-    def _extract_data(self, *args, **kwargs):
-        """Get NCIt source file"""
-        if 'data_path' in kwargs:
-            self._data_src = kwargs['data_path']
-        else:
-            data_dir = PROJECT_ROOT / 'data' / 'ncit'
-            data_dir.mkdir(exist_ok=True, parents=True)
-            dir_files = list(data_dir.iterdir())
-            if len(dir_files) == 0:
-                self._download_data()
-                dir_files = list(data_dir.iterdir())
-            self._data_src = sorted(dir_files)[-1]
-        self._version = self._data_src.stem.split('_')[1]
-
-    def get_desc_nodes(self, node: ThingClass,
-                       uq_nodes: Set[ThingClass]) -> Set[ThingClass]:
+    def _extract_data(self):
+        """Get NCIt source file."""
+        self._data_path.mkdir(exist_ok=True, parents=True)
+        dir_files = list(self._data_path.iterdir())
+        if len(dir_files) == 0:
+            self._download_data()
+            dir_files = list(self._data_path.iterdir())
+        self._data_file = sorted(dir_files)[-1]
+        self._version = self._data_file.stem.split('_')[1]
+
+    def _get_desc_nodes(self, node: ThingClass,
+                        uq_nodes: Set[ThingClass]) -> Set[ThingClass]:
         """Create set of unique subclasses of node parameter.
         Should be originally called on ncit:C1909: Pharmacologic Substance.
 
-        Args:
-            node: concept node to either retrieve descendants of, or to
-                normalize and add to DB
-            uq_nodes: set of unique class nodes found so far from recursive
-                tree exploration
-
-        Returns:
-            the uq_nodes set, updated with any class nodes found from
-                recursive exploration of this branch of the class tree
+        :param owlready2.entity.ThingClass node: concept node to either
+            retrieve descendants of, or to normalize and add to DB
+        :param Set[owlready2.entity.ThingClass] uq_nodes: set of unique class
+            nodes found so far from recursive tree exploration
+        :return: the uq_nodes set, updated with any class nodes found from
+            recursive exploration of this branch of the class tree
+        :rtype: Set[owlready2.entity.ThingClass]
         """
         children = node.descendants()
         if children:
             for child_node in children:
                 if child_node is not node:
                     uq_nodes.add(child_node)
-                    self.get_desc_nodes(child_node, uq_nodes)
+                    self._get_desc_nodes(child_node, uq_nodes)
         return uq_nodes
 
-    def get_typed_nodes(self, uq_nodes: Set[ThingClass],
-                        ncit: owl.namespace.Ontology) -> Set[ThingClass]:
+    def _get_typed_nodes(self, uq_nodes: Set[ThingClass],
+                         ncit: owl.namespace.Ontology) -> Set[ThingClass]:
         """Get all nodes with semantic_type Pharmacologic Substance
 
-        Args:
-            uq_nodes: set of unique class nodes found so far.
-            ncit: owlready2 Ontology instance for NCI Thesaurus
-
-        Returns:
-            uq_nodes, with the addition of all classes found to have
-                semantic_type Pharmacologic Substance and not of type
-                Retired_Concept
+        :param Set[owlready2.entity.ThingClass] uq_nodes: set of unique class
+            nodes found so far.
+        :param owl.namespace.Ontology ncit: owlready2 Ontology instance for
+            NCI Thesaurus.
+        :return: uq_nodes, with the addition of all classes found to have
+            semantic_type Pharmacologic Substance and not of type
+            Retired_Concept
+        :rtype: Set[owlready2.entity.ThingClass]
         """
         graph = owl.default_world.as_rdflib_graph()
 
@@ -133,14 +140,13 @@ def get_typed_nodes(self, uq_nodes: Set[ThingClass],
             uq_nodes.add(class_object)
         return uq_nodes
 
-    def _transform_data(self, *args, **kwargs):
+    def _transform_data(self):
         """Get data from file and construct objects for loading"""
-        ncit = owl.get_ontology(self._data_src.absolute().as_uri())
+        ncit = owl.get_ontology(self._data_file.absolute().as_uri())
         ncit.load()
-        self._add_meta()
         uq_nodes = set()
-        uq_nodes = self.get_desc_nodes(ncit.C1909, uq_nodes)
-        uq_nodes = self.get_typed_nodes(uq_nodes, ncit)
+        uq_nodes = self._get_desc_nodes(ncit.C1909, uq_nodes)
+        uq_nodes = self._get_typed_nodes(uq_nodes, ncit)
         with self.database.therapies.batch_writer() as batch:
             for node in uq_nodes:
                 concept_id = f"{NamespacePrefix.NCIT.value}:{node.name}"
@@ -178,7 +184,7 @@ def _transform_data(self, *args, **kwargs):
                                   xrefs=xrefs)
                 self._load_therapy(therapy, batch)
 
-    def _add_meta(self):
+    def _load_meta(self):
         """Load metadata"""
         metadata = Meta(data_license="CC BY 4.0",
                         data_license_url="https://creativecommons.org/licenses/by/4.0/legalcode",  # noqa F401
@@ -197,13 +203,13 @@ def _add_meta(self):
     def _load_therapy(self, therapy: Therapy, batch):
         """Load individual therapy into dynamodb table
 
-        Args:
-            therapy: complete Therapy instance
-            batch: dynamoDB batch_writer context manager object
+        :param therapy.schemas.Therapy therapy: complete Therapy instance
+        :param batch: dynamoDB batch_writer context manager object
         """
         item = therapy.dict()
         concept_id_lower = item['concept_id'].lower()
-        if len({a.casefold(): a for a in item['aliases']}) > 20 or not item['aliases']:  # noqa: E501
+        if len({a.casefold(): a for a in item['aliases']}) > 20 \
+                or not item['aliases']:
             del item['aliases']
         else:
             if 'aliases' in item:
@@ -224,10 +230,11 @@ def _load_therapy(self, therapy: Therapy, batch):
                 'src_name': SourceName.NCIT.value
             })
         else:
-            del therapy['label']
+            del therapy.label
         item['label_and_type'] = f"{concept_id_lower}##identity"
         item['src_name'] = SourceName.NCIT.value
         for element in ['other_identifiers', 'xrefs']:
             if not item[element]:
                 del item[element]
         batch.put_item(Item=item)
+        self._added_ids.append(item['concept_id'])
diff --git a/therapy/etl/rxnorm.py b/therapy/etl/rxnorm.py
index c3591f4f..4fbe6c22 100644
--- a/therapy/etl/rxnorm.py
+++ b/therapy/etl/rxnorm.py
@@ -14,6 +14,7 @@
 import datetime
 import logging
 import botocore
+from typing import List
 from os import environ, remove
 import subprocess
 import shutil
@@ -56,7 +57,18 @@ def __init__(self,
         self._xref_srcs = {src for src in NamespacePrefix.__members__}
         self._data_url = data_url
         self.database = database
-        self._load_data()
+        self._added_ids = []
+
+    def perform_etl(self) -> List[str]:
+        """Load the RxNorm source into database.
+
+        :return: List of concept IDs which were successfully processed and
+            uploaded.
+        """
+        self._extract_data()
+        self._load_meta()
+        self._transform_data()
+        return self._added_ids
 
     def _extract_data(self, *args, **kwargs):
         """Extract data from the RxNorm source."""
@@ -244,6 +256,7 @@ def _load_therapy(self, params, batch):
         params['label_and_type'] = f"{params['concept_id'].lower()}##identity"
         try:
             batch.put_item(Item=params)
+            self._added_ids.append(params['concept_id'])
         except botocore.exceptions.ClientError:
             if (len((params['label_and_type']).encode('utf-8')) >= 2048) or \
                     (len((params['concept_id']).encode('utf-8')) >= 1024):
@@ -441,7 +454,7 @@ def _add_other_ids_xrefs(self, params, row):
             else:
                 logger.info(f"{other_id_xref} not in NameSpacePrefix.")
 
-    def _add_meta(self):
+    def _load_meta(self):
         """Add RxNorm metadata."""
         meta = Meta(data_license='UMLS Metathesaurus',
                     data_license_url='https://www.nlm.nih.gov/research/umls/'
@@ -457,9 +470,3 @@ def _add_meta(self):
         params = dict(meta)
         params['src_name'] = SourceName.RXNORM.value
         self.database.metadata.put_item(Item=params)
-
-    def _load_data(self):
-        """Load the RxNorm source into database."""
-        self._extract_data()
-        self._add_meta()
-        self._transform_data()
diff --git a/therapy/etl/wikidata.py b/therapy/etl/wikidata.py
index 4418d814..0ebda81e 100644
--- a/therapy/etl/wikidata.py
+++ b/therapy/etl/wikidata.py
@@ -1,12 +1,13 @@
 """This module defines the Wikidata ETL methods."""
 from .base import Base, IDENTIFIER_PREFIXES
 from therapy import PROJECT_ROOT
-import json
 from therapy.schemas import SourceName, NamespacePrefix, \
     SourceIDAfterNamespace, Meta
 from therapy.database import Database
+import json
 import logging
-from typing import Dict
+from typing import Dict, List
+from pathlib import Path
 from wikibaseintegrator import wbi_core
 import datetime
 
@@ -50,39 +51,51 @@
 class Wikidata(Base):
     """Extract, transform, and load the Wikidata source into therapy.db."""
 
-    def __init__(self, database: Database, *args, **kwargs):
-        """Initialize wikidata ETL class"""
+    def __init__(self,
+                 database: Database,
+                 data_path: Path = PROJECT_ROOT / 'data' / 'wikidata'):
+        """Initialize wikidata ETL class.
+
+        :param therapy.database.Database: DB instance to use
+        :param pathlib.Path data_path: path to wikidata data directory
+        """
         self.database = database
-        self._extract_data(*args, **kwargs)
-        self._add_meta()
+        self._data_path = data_path
+        self._added_ids = []
+
+    def perform_etl(self) -> List[str]:
+        """Public-facing method to initiate ETL procedures on given data.
+
+        :return: List of concept IDs which were successfully processed and
+            uploaded.
+        """
+        self._extract_data()
+        self._load_meta()
         self._transform_data()
+        return self._added_ids
 
-    def _extract_data(self, *args, **kwargs):
+    def _extract_data(self):
         """Extract data from the Wikidata source."""
-        logger.info('Extracting Wikidata...')
-        if 'data_path' in kwargs:
-            self._data_src = kwargs['data_path']
-        else:
-            wd_dir = PROJECT_ROOT / 'data' / 'wikidata'
-            wd_dir.mkdir(exist_ok=True, parents=True)
-
-            data = (wbi_core.ItemEngine.execute_sparql_query(
-                SPARQL_QUERY))['results']['bindings']
-
-            transformed_data = list()
-            for item in data:
-                params = dict()
-                for attr in item:
-                    params[attr] = item[attr]['value']
-                transformed_data.append(params)
-
-            self._version = datetime.datetime.today().strftime('%Y%m%d')
-            with open(f"{wd_dir}/wikidata_{self._version}.json", 'w+') as f:
-                json.dump(transformed_data, f)
-            self._data_src = sorted(list(wd_dir.iterdir()))[-1]
+        self._data_path.mkdir(exist_ok=True, parents=True)
+
+        data = (wbi_core.ItemEngine.execute_sparql_query(
+            SPARQL_QUERY))['results']['bindings']
+
+        transformed_data = list()
+        for item in data:
+            params = dict()
+            for attr in item:
+                params[attr] = item[attr]['value']
+            transformed_data.append(params)
+
+        self._version = datetime.datetime.today().strftime('%Y%m%d')
+        with open(f"{self._data_path}/wikidata_{self._version}.json",
+                  'w+') as f:
+            json.dump(transformed_data, f)
+        self._data_src = sorted(list(self._data_path.iterdir()))[-1]
         logger.info('Successfully extracted Wikidata.')
 
-    def _add_meta(self):
+    def _load_meta(self):
         """Add Wikidata metadata."""
         metadata = Meta(src_name=SourceName.WIKIDATA.value,
                         data_license='CC0 1.0',
@@ -100,11 +113,7 @@ def _add_meta(self):
         self.database.metadata.put_item(Item=params)
 
     def _transform_data(self):
-        """Transform the Wikidata source data.
-        Currently, gather all items in memory and then batch-load into
-        DynamoDB. Worth considering whether adding record directly and then
-        issuing update statements to append additional aliases would be better.
-        """
+        """Transform the Wikidata source data."""
         with open(self._data_src, 'r') as f:
             records = json.load(f)
 
@@ -120,8 +129,8 @@ def _transform_data(self):
                     item['concept_id'] = concept_id
                     item['src_name'] = SourceName.WIKIDATA.value
 
-                    other_ids = []
-                    xrefs = []
+                    other_ids = list()
+                    xrefs = list()
                     for key in IDENTIFIER_PREFIXES.keys():
                         if key in record.keys():
                             other_id = record[key]
@@ -159,11 +168,11 @@ def _transform_data(self):
                 self._load_therapy(item, batch)
 
     def _load_therapy(self, item: Dict, batch):
-        """Load individual therapy record into DynamoDB
-        Args:
-            item: dict containing, at minimum, label_and_type and concept_id
-                keys.
-            batch: boto3 batch writer
+        """Load individual therapy record into database.
+
+        :param Dict item: containing, at minimum, label_and_type and concept_id
+            keys.
+        :param batch: boto3 batch writer
         """
         if 'aliases' in item:
             item['aliases'] = list(set(item['aliases']))
@@ -172,6 +181,7 @@ def _load_therapy(self, item: Dict, batch):
                 del item['aliases']
 
         batch.put_item(Item=item)
+        self._added_ids.append(item['concept_id'])
         concept_id_lower = item['concept_id'].lower()
 
         if 'aliases' in item.keys():
@@ -191,14 +201,3 @@ def _load_therapy(self, item: Dict, batch):
                 'concept_id': concept_id_lower,
                 'src_name': SourceName.WIKIDATA.value
             })
-
-    def _sqlite_str(self, string):
-        """Sanitizes string to use as value in SQL statement.
-        Some wikidata entries include items with single quotes,
-        like wikidata:Q80863 alias: 5'-(Tetrahydrogen triphosphate) Adenosine
-        """
-        if string == "NULL":
-            return "NULL"
-        else:
-            sanitized = string.replace("'", "''")
-            return f"{sanitized}"
diff --git a/therapy/query.py b/therapy/query.py
index 81ed55bd..5f32bb6f 100644
--- a/therapy/query.py
+++ b/therapy/query.py
@@ -33,7 +33,7 @@ def __init__(self, message):
         super().__init__(message)
 
 
-class Normalizer:
+class QueryHandler:
     """Class for normalizer management. Stores reference to database instance
     and normalizes query input.
     """
@@ -41,12 +41,12 @@ class Normalizer:
     def __init__(self, db_url: str = '', db_region: str = 'us-east-2'):
         """Initialize Normalizer instance.
 
-        :param db_url: URL to database source.
-        :param db_region: AWS default region.
+        :param str db_url: URL to database source.
+        :param str db_region: AWS default region.
         """
         self.db = Database(db_url=db_url, region_name=db_region)
 
-    def emit_warnings(self, query_str) -> Optional[Dict]:
+    def _emit_warnings(self, query_str) -> Optional[Dict]:
         """Emit warnings if query contains non breaking space characters.
 
         :param str query_str: query string
@@ -63,7 +63,7 @@ def emit_warnings(self, query_str) -> Optional[Dict]:
             )
         return warnings
 
-    def fetch_meta(self, src_name: str) -> Meta:
+    def _fetch_meta(self, src_name: str) -> Meta:
         """Fetch metadata for src_name.
 
         :param str src_name: name of source to get metadata for
@@ -80,12 +80,12 @@ def fetch_meta(self, src_name: str) -> Meta:
                 self.db.cached_sources[src_name] = response
                 return response
             except ClientError as e:
-                print(e.response['Error']['Message'])
+                logger.error(e.response['Error']['Message'])
 
-    def add_record(self,
-                   response: Dict[str, Dict],
-                   item: Dict,
-                   match_type: MatchType) -> (Dict, str):
+    def _add_record(self,
+                    response: Dict[str, Dict],
+                    item: Dict,
+                    match_type: str) -> (Dict, str):
         """Add individual record (i.e. Item in DynamoDB) to response object
 
         :param Dict[str, Dict] response: in-progress response object to return
@@ -109,19 +109,19 @@ def add_record(self,
             pass
         elif matches[src_name] is None:
             matches[src_name] = {
-                'match_type': match_type,
+                'match_type': MatchType[match_type.upper()],
                 'records': [drug],
-                'meta_': self.fetch_meta(src_name)
+                'meta_': self._fetch_meta(src_name)
             }
-        elif matches[src_name]['match_type'].value == match_type.value:
+        elif matches[src_name]['match_type'] == MatchType[match_type.upper()]:
             matches[src_name]['records'].append(drug)
 
         return (response, src_name)
 
-    def fetch_records(self,
-                      response: Dict[str, Dict],
-                      concept_ids: List[str],
-                      match_type: MatchType) -> (Dict, Set):
+    def _fetch_records(self,
+                       response: Dict[str, Dict],
+                       concept_ids: Set[str],
+                       match_type: str) -> (Dict, Set):
         """Return matched Drug records as a structured response for a given
         collection of concept IDs.
 
@@ -129,7 +129,7 @@ def fetch_records(self,
             to client.
         :param List[str] concept_ids: List of concept IDs to build from.
             Should be all lower-case.
-        :param MatchType match_type: record should be assigned this type of
+        :param str match_type: record should be assigned this type of
             match.
         :return: response Dict with records filled in via provided concept
             IDs, and Set of source names of matched records
@@ -137,20 +137,16 @@ def fetch_records(self,
         matched_sources = set()
         for concept_id in concept_ids:
             try:
-                pk = f'{concept_id.lower()}##identity'
-                filter_exp = Key('label_and_type').eq(pk)
-                result = self.db.therapies.query(
-                    KeyConditionExpression=filter_exp
-                )
-                match = result['Items'][0]
-                (response, src) = self.add_record(response, match, match_type)
+                match = self.db.get_record_by_id(concept_id.lower(),
+                                                 case_sensitive=False)
+                (response, src) = self._add_record(response, match, match_type)
                 matched_sources.add(src)
             except ClientError as e:
-                print(e.response['Error']['Message'])
+                logger.error(e.response['Error']['Message'])
 
         return (response, matched_sources)
 
-    def fill_no_matches(self, resp: Dict[str, Dict]) -> Dict:
+    def _fill_no_matches(self, resp: Dict[str, Dict]) -> Dict:
         """Fill all empty source_matches slots with NO_MATCH results.
 
         :param Dict[str, Dict] resp: incoming response object
@@ -162,16 +158,17 @@ def fill_no_matches(self, resp: Dict[str, Dict]) -> Dict:
                 resp['source_matches'][src_name] = {
                     'match_type': MatchType.NO_MATCH,
                     'records': [],
-                    'meta_': self.fetch_meta(src_name)
+                    'meta_': self._fetch_meta(src_name)
                 }
         return resp
 
-    def check_concept_id(self,
-                         query: str,
-                         resp: Dict,
-                         sources: Set[str]) -> (Dict, Set):
+    def _check_concept_id(self,
+                          query: str,
+                          resp: Dict,
+                          sources: Set[str]) -> (Dict, Set):
         """Check query for concept ID match. Should only find 0 or 1 matches,
         but stores them as a collection to be safe.
+        # TODO is that ^ safe?
 
         :param str query: search string
         :param Dict resp: in-progress response object to return to client
@@ -190,7 +187,7 @@ def check_concept_id(self,
                 if len(result['Items']) > 0:
                     concept_id_items += result['Items']
             except ClientError as e:
-                print(e.response['Error']['Message'])
+                logger.error(e.response['Error']['Message'])
         for prefix in [p for p in NAMESPACE_LOOKUP.keys()
                        if query.startswith(p)]:
             pk = f'{NAMESPACE_LOOKUP[prefix].lower()}:{query}##identity'
@@ -202,43 +199,37 @@ def check_concept_id(self,
                 if len(result['Items']) > 0:  # TODO remove check?
                     concept_id_items += result['Items']
             except ClientError as e:
-                print(e.response['Error']['Message'])
+                logger(e.response['Error']['Message'])
 
         for item in concept_id_items:
-            (resp, src_name) = self.add_record(resp, item,
-                                               MatchType.CONCEPT_ID)
+            (resp, src_name) = self._add_record(resp, item,
+                                                MatchType.CONCEPT_ID.name)
             sources = sources - {src_name}
         return (resp, sources)
 
-    def check_match_type(self,
-                         query: str,
-                         resp: Dict,
-                         sources: Set[str],
-                         match: str) -> (Dict, Set):
+    def _check_match_type(self,
+                          query: str,
+                          resp: Dict,
+                          sources: Set[str],
+                          match_type: str) -> (Dict, Set):
         """Check query for selected match type.
         :param str query: search string
         :param Dict resp: in-progress response object to return to client
         :param Set[str] sources: remaining unmatched sources
-        :param str match: Match type name
+        :param str match_type: Match type to check for. Should be one of
+            ['trade_name', 'label', 'alias']
         :return: Tuple with updated resp object and updated set of unmatched
                  sources
         """
-        filter_exp = Key('label_and_type').eq(f'{query}##{match}')
-        try:
-            db_response = self.db.therapies.query(
-                KeyConditionExpression=filter_exp
-            )
-            if 'Items' in db_response.keys():
-                concept_ids = [i['concept_id'] for i in db_response['Items']]
-                (resp, matched_srcs) = self.fetch_records(
-                    resp, concept_ids, MatchType[match.upper()]
-                )
-                sources = sources - matched_srcs
-        except ClientError as e:
-            print(e.response['Error']['Message'])
+        matches = self.db.get_records_by_type(query, match_type)
+        if matches:
+            concept_ids = {i['concept_id'] for i in matches}
+            (resp, matched_srcs) = self._fetch_records(resp, concept_ids,
+                                                       match_type)
+            sources = sources - matched_srcs
         return (resp, sources)
 
-    def response_keyed(self, query: str, sources: Set[str]) -> Dict:
+    def _response_keyed(self, query: str, sources: Set[str]) -> Dict:
         """Return response as dict where key is source name and value
         is a list of records. Corresponds to `keyed=true` API parameter.
 
@@ -246,36 +237,34 @@ def response_keyed(self, query: str, sources: Set[str]) -> Dict:
         :param Set[str] sources: sources to match from
         :return: completed response object to return to client
         """
-        resp = {
+        response = {
             'query': query,
-            'warnings': self.emit_warnings(query),
+            'warnings': self._emit_warnings(query),
             'source_matches': {
                 source: None for source in sources
             }
         }
         if query == '':
-            resp = self.fill_no_matches(resp)
-            return resp
-        query_l = query.lower()
+            response = self._fill_no_matches(response)
+            return response
+        query = query.lower()
 
         # check if concept ID match
-        (resp, sources) = self.check_concept_id(query_l, resp, sources)
+        (response, sources) = self._check_concept_id(query, response, sources)
         if len(sources) == 0:
-            return resp
+            return response
 
         match_types = ['label', 'trade_name', 'alias']
-        for match in match_types:
-            (resp, sources) = self.check_match_type(
-                query_l, resp, sources, match)
+        for match_type in match_types:
+            (response, sources) = self._check_match_type(query, response,
+                                                         sources, match_type)
             if len(sources) == 0:
-                return resp
+                return response
 
         # remaining sources get no match
-        resp = self.fill_no_matches(resp)
+        return self._fill_no_matches(response)
 
-        return resp
-
-    def response_list(self, query: str, sources: List[str]) -> Dict:
+    def _response_list(self, query: str, sources: List[str]) -> Dict:
         """Return response as list, where the first key-value in each item
         is the source name. Corresponds to `keyed=false` API parameter.
 
@@ -283,7 +272,7 @@ def response_list(self, query: str, sources: List[str]) -> Dict:
         :param List[str] sources: sources to match from
         :return: Completed response object to return to client
         """
-        response_dict = self.response_keyed(query, sources)
+        response_dict = self._response_keyed(query, sources)
         source_list = []
         for src_name in response_dict['source_matches'].keys():
             src = {
@@ -297,21 +286,20 @@ def response_list(self, query: str, sources: List[str]) -> Dict:
 
         return response_dict
 
-    def normalize(self, query_str, keyed=False, incl='', excl='', **params):
+    def search_sources(self, query_str, keyed=False, incl='', excl='') -> Dict:
         """Fetch normalized therapy objects.
 
         :param str query_str: query, a string, to search for
         :param bool keyed: if true, return response as dict keying source names
             to source objects; otherwise, return list of source objects
         :param str incl: str containing comma-separated names of sources to
-            use. Will exclude all other sources. Case-insensitive. Raises
-            InvalidParameterException if both incl and excl args are
-            provided, or if invalid source names are given.
+            use. Will exclude all other sources. Case-insensitive.
         :param str excl: str containing comma-separated names of source to
-            exclude. Will include all other source. Case-insensitive. Raises
-            InvalidParameterException if both incl and excl args are
-            provided, or if invalid source names are given.
+            exclude. Will include all other source. Case-insensitive.
         :return: dict containing all matches found in sources.
+        :rtype: dict
+        :raises InvalidParameterException: if both incl and excl args are
+            provided, or if invalid source names are given.
         """
         sources = {name.value.lower(): name.value for name in
                    SourceName.__members__.values()}
@@ -351,8 +339,126 @@ def normalize(self, query_str, keyed=False, incl='', excl='', **params):
         query_str = query_str.strip()
 
         if keyed:
-            resp = self.response_keyed(query_str, query_sources)
+            response = self._response_keyed(query_str, query_sources)
         else:
-            resp = self.response_list(query_str, query_sources)
+            response = self._response_list(query_str, query_sources)
 
-        return resp
+        return response
+
+    def _add_merged_meta(self, response: Dict) -> Dict:
+        """Add source metadata to response object.
+
+        :param Dict response: in-progress response object
+        :return: completed resopnse object.
+        """
+        sources_meta = {}
+        for concept_id in response['record']['concept_ids']:
+            prefix = concept_id.split(':')[0]
+            src_name = PREFIX_LOOKUP[prefix]
+            if src_name not in sources_meta:
+                sources_meta[src_name] = self._fetch_meta(src_name)
+        response['meta_'] = sources_meta
+        return response
+
+    def _format_merged_record(self, merged_record: Dict) -> Dict:
+        """Transform merged record (as served by DB) into response object
+        format.
+
+        :param Dict merged_record: merged record as returned by DB call
+        :return: formatted record
+        """
+        del merged_record['label_and_type']
+        merged_record['concept_ids'] = merged_record['concept_id'].split('|')
+        del merged_record['concept_id']
+        return merged_record
+
+    def _add_merged_record(self, response: Dict, merge_ref: str) -> Dict:
+        """Add referenced concept ID group to response object.
+
+        :param Dict response: in-progress response object. Should have
+            match_type attribute filled in already.
+        :param str merge_ref: primary key for concept ID group lookup.
+        :return: completed response object.
+        """
+        merged_record = self.db.get_record_by_id(merge_ref, False, True)
+        if not merged_record:
+            logger.warning(f"Could not retrieve merged record for concept "
+                           f"ID group {merge_ref} "
+                           f"by way of query {response['query']}")
+            response['match_type'] = MatchType.NO_MATCH
+            return response
+        response['record'] = self._format_merged_record(merged_record)
+        response = self._add_merged_meta(response)
+        return response
+
+    def _record_order(self, record: Dict) -> (int, str):
+        """Construct priority order for matching. Only called by sort().
+
+        :param Dict record: individual record item in iterable to sort
+        :return: tuple with rank value and concept ID
+        """
+        src = record['src_name']
+        if src == SourceName.RXNORM.value:
+            source_rank = 1
+        elif src == SourceName.NCIT.value:
+            source_rank = 2
+        elif src == SourceName.CHEMIDPLUS.value:
+            source_rank = 3
+        else:
+            source_rank = 4
+        return (source_rank, record['concept_id'])
+
+    def search_groups(self, query_str: str) -> Dict:
+        """Return merged, normalized concept for given search term.
+
+        :param str query_str: string to search against
+        """
+        # prepare basic response
+        response = {
+            'query': query_str,
+            'warnings': self._emit_warnings(query_str),
+        }
+        if query_str == '':
+            response['match_type'] = MatchType.NO_MATCH
+            return response
+        query_str = query_str.lower()
+
+        # check merged concept ID match
+        record = self.db.get_record_by_id(query_str, case_sensitive=False,
+                                          merge=True)
+        if record:
+            response['match_type'] = MatchType.CONCEPT_ID
+            response['record'] = self._format_merged_record(record)
+            response = self._add_merged_meta(response)
+            return response
+
+        # check concept ID match
+        record = self.db.get_record_by_id(query_str, case_sensitive=False)
+        if record:
+            response['match_type'] = MatchType.CONCEPT_ID
+            response = self._add_merged_record(response, record['merge_ref'])
+            return response
+
+        # check other match types
+        for match_type in ['label', 'trade_name', 'alias']:
+            # get matches list for match tier
+            prohibited_sources = (SourceName.CHEMBL.value,
+                                  SourceName.DRUGBANK.value)
+            query_matches = self.db.get_records_by_type(query_str, match_type)
+            query_matches = [self.db.get_record_by_id(m['concept_id'], False)
+                             for m in query_matches
+                             if m['src_name'] not in prohibited_sources]
+            query_matches.sort(key=self._record_order)
+
+            # attempt merge ref resolution until successful
+            for match in query_matches:
+                record = self.db.get_record_by_id(match['concept_id'], False)
+                if record:
+                    merge_ref = record['merge_ref']
+                    if merge_ref:
+                        response['match_type'] = MatchType[match_type.upper()]
+                        return self._add_merged_record(response, merge_ref)
+
+        if not query_matches:
+            response['match_type'] = MatchType.NO_MATCH
+        return response
diff --git a/therapy/schemas.py b/therapy/schemas.py
index 5ded3dcf..84449fa5 100644
--- a/therapy/schemas.py
+++ b/therapy/schemas.py
@@ -294,6 +294,20 @@ def schema_extra(schema: Dict[str, Any],
             }
 
 
+class MergedMatch(BaseModel):
+    """Represent merged concept in response to client.
+
+    Currently, does not include ApprovalStatus data because no included
+    sources provide it.
+    """
+
+    concept_ids: List[str]
+    aliases: List[str]
+    trade_names: Optional[List[str]]
+    xrefs: Optional[List[str]]
+    label: Optional[str]
+
+
 class Service(BaseModel):
     """Core response schema containing matches for each source"""
 
@@ -330,3 +344,15 @@ def schema_extra(schema: Dict[str, Any],
                     }
                 }
             }
+
+
+class NormalizationService(BaseModel):
+    """Response containing one or more merged records and relevant source
+    data.
+    """
+
+    query: str
+    warnings: Optional[Dict]
+    match_type: MatchType
+    record: Optional[MergedMatch]
+    meta_: Optional[Dict[SourceName, Meta]]


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 00000000..cf1e8c3e
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,194 @@
+"""Pytest test config tools."""
+from therapy.database import Database
+from therapy import PROJECT_ROOT
+from typing import Dict, Any, Optional, List
+import json
+import pytest
+
+
+@pytest.fixture(scope='module')
+def mock_database():
+    """Return MockDatabase object."""
+
+    class MockDatabase(Database):
+        """Mock database object to use in test cases."""
+
+        def __init__(self):
+            """Initialize mock database object. This class's method's shadow the
+            actual Database class methods.
+
+            `self.records` loads preexisting DB items.
+            `self.added_records` stores add record requests, with the
+            concept_id as the key and the complete record as the value.
+            `self.updates` stores update requests, with the concept_id as the
+            key and the updated attribute and new value as the value.
+            """
+            infile = PROJECT_ROOT / 'tests' / 'unit' / 'data' / 'therapies.json'  # noqa: E501
+            self.records = {}
+            with open(infile, 'r') as f:
+                records_json = json.load(f)
+            for record in records_json:
+                self.records[record['label_and_type']] = {
+                    record['concept_id']: record
+                }
+            self.added_records: Dict[str, Dict[Any, Any]] = {}
+            self.updates: Dict[str, Dict[Any, Any]] = {}
+            self.cached_sources = {
+                'Wikidata': {
+                    "data_license": "CC0 1.0",
+                    "data_license_url": "https://creativecommons.org/publicdomain/zero/1.0/",  # noqa: E501
+                    "version": "20200812",
+                    "data_url": None,
+                    "rdp_url": None,
+                    "data_license_attributes": {
+                        "non_commercial": False,
+                        "attribution": False,
+                        "share_alike": False
+                    }
+                },
+                'ChemIDplus': {
+                    "data_license": "custom",
+                    "data_license_url": "https://www.nlm.nih.gov/databases/download/terms_and_conditions.html",  # noqa: E501
+                    "version": "20200327",
+                    "data_url": "ftp://ftp.nlm.nih.gov/nlmdata/.chemidlease/",
+                    "rdp_url": None,
+                    "data_license_attributes": {
+                        "non_commercial": False,
+                        "attribution": False,
+                        "share_alike": False
+                    }
+                },
+                'RxNorm': {
+                    "data_license": "UMLS Metathesaurus",
+                    "data_license_url": "https://www.nlm.nih.gov/research/umls/rxnorm/docs/termsofservice.html",  # noqa: E501
+                    "version": "20210104",
+                    "data_url": "https://www.nlm.nih.gov/research/umls/rxnorm/docs/rxnormfiles.html",  # noqa: E501
+                    "rdp_url": None,
+                    "data_license_attributes": {
+                        "non_commercial": False,
+                        "attribution": False,
+                        "share_alike": False
+                    }
+                },
+                'DrugBank': {
+                    "data_license": "CC BY-NC 4.0",
+                    "data_license_url": "https://creativecommons.org/licenses/by-nc/4.0/legalcode",  # noqa: E501
+                    "version": "5.1.7",
+                    "data_url": "https://go.drugbank.com/releases/5-1-7/downloads/all-full-database",  # noqa: E501
+                    "rdp_url": "http://reusabledata.org/drugbank.html",
+                    "data_license_attributes": {
+                        "non_commercial": True,
+                        "attribution": True,
+                        "share_alike": False,
+                    }
+                },
+                'NCIt': {
+                    "data_license": "CC BY 4.0",
+                    "data_license_url": "https://creativecommons.org/licenses/by/4.0/legalcode",  # noqa: E501
+                    "version": "20.09d",
+                    "data_url": "https://evs.nci.nih.gov/ftp1/NCI_Thesaurus/archive/20.09d_Release/",  # noqa: E501
+                    "rdp_url": "http://reusabledata.org/ncit.html",
+                    "data_license_attributes": {
+                        "non_commercial": False,
+                        "attribution": True,
+                        "share_alike": False
+                    }
+                },
+                'ChEMBL': {
+                    "data_license": "CC BY-SA 3.0",
+                    "data_license_url": "https://creativecommons.org/licenses/by-sa/3.0/",  # noqa: E501
+                    "version": "27",
+                    "data_url": "http://ftp.ebi.ac.uk/pub/databases/chembl/ChEMBLdb/releases/chembl_27/",  # noqa: E501
+                    "rdp_url": "http://reusabledata.org/chembl.html",
+                    "data_license_attributes": {
+                        "non_commercial": False,
+                        "attribution": True,
+                        "share_alike": True
+                    }
+                }
+            }
+
+        def get_record_by_id(self, record_id: str,
+                             case_sensitive: bool = True,
+                             merge: bool = False) -> Optional[Dict]:
+            """Fetch record corresponding to provided concept ID.
+
+            :param str concept_id: concept ID for therapy record
+            :param bool case_sensitive: if true, performs exact lookup, which
+                is more efficient. Otherwise, performs filter operation, which
+                doesn't require correct casing.
+            :param bool merge: if true, retrieve merged record
+            :return: complete therapy record, if match is found; None otherwise
+            """
+            if merge:
+                label_and_type = f'{record_id.lower()}##merger'
+            else:
+                label_and_type = f'{record_id.lower()}##identity'
+            record_lookup = self.records.get(label_and_type, None)
+            if record_lookup:
+                if case_sensitive:
+                    record = record_lookup.get(record_id, None)
+                    if record:
+                        return record.copy()
+                    else:
+                        return None
+                elif record_lookup.values():
+                    return list(record_lookup.values())[0].copy()
+            return None
+
+        def get_records_by_type(self, query: str,
+                                match_type: str) -> List[Dict]:
+            """Retrieve records for given query and match type.
+
+            :param query: string to match against
+            :param str match_type: type of match to look for. Should be one
+                of "alias", "trade_name", or "label" (use get_record_by_id for
+                concept ID lookup)
+            :return: list of matching records. Empty if lookup fails.
+            """
+            assert match_type in ('alias', 'trade_name', 'label', 'rx_brand')
+            label_and_type = f'{query}##{match_type.lower()}'
+            records_lookup = self.records.get(label_and_type, None)
+            if records_lookup:
+                return [v.copy() for v in records_lookup.values()]
+            else:
+                return []
+
+        def get_merged_record(self, merge_ref) -> Optional[Dict]:
+            """Fetch merged record from given reference.
+
+            :param str merge_ref: key for merged record, formated as a string
+                of grouped concept IDs separated by vertical bars, ending with
+                `##merger`. Must be correctly-cased.
+            :return: complete merged record if lookup successful, None
+                otherwise
+            """
+            record_lookup = self.records.get(merge_ref, None)
+            if record_lookup:
+                vals = list(record_lookup.values())
+                if vals:
+                    return vals[0].copy()
+            return None
+
+        def add_record(self, record: Dict, record_type: str):
+            """Store add record request sent to database.
+
+            :param Dict record: record (of any type) to upload. Must include
+                `concept_id` key. If record is of the `identity` type, the
+                concept_id must be correctly-cased.
+            :param str record_type: ignored by this function
+            """
+            self.added_records[record['concept_id']] = record
+
+        def update_record(self, concept_id: str, attribute: str,
+                          new_value: Any):
+            """Store update request sent to database.
+
+            :param str concept_id: record to update
+            :param str field: name of field to update
+            :parm str new_value: new value
+            """
+            assert f'{concept_id.lower()}##identity' in self.records
+            self.updates[concept_id] = {attribute: new_value}
+
+    return MockDatabase
diff --git a/tests/unit/data/therapies.json b/tests/unit/data/therapies.json
index b0676b47..31b3d13d 100644
--- a/tests/unit/data/therapies.json
+++ b/tests/unit/data/therapies.json
@@ -250,7 +250,8 @@
       "rxcui:2555"
     ],
     "xrefs": ["pubchem.compound:5702198"],
-    "src_name": "Wikidata"
+    "src_name": "Wikidata",
+    "merge_ref": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001"
   },
   {
     "label_and_type": "cisplatin##label",
@@ -314,23 +315,21 @@
     "label": "atropine",
     "concept_id": "wikidata:Q26272",
     "aliases": [
-        "dl-Hyoscyamine",
-        "dl-tropyltropate",
-        "Mydriasine",
-        "Tropine tropate",
-        "(±)-atropine",
-        "(±)-hyoscyamine",
-        "(+-)-Atropine",
-        "(+-)-Hyoscyamine",
-        "(+,-)-Tropyl tropate",
-        "(3-Endo)-8-methyl-8-azabicyclo[3.2.1]oct-3-yl tropate",
-        "[(1S,5R)-8-Methyl-8-azabicyclo[3.2.1]oct-3-yl] 3-hydroxy-2-phenyl-propanoate",
-        "8-Methyl-8-azabicyclo[3.2.1]oct-3-yl 3-hydroxy-2-phenylpropanoate",
-        "8-Methyl-8-azabicyclo[3.2.1]oct-3-yl tropate"
-    ],
-    "xrefs": [
-      "pubchem.compound:174174"
+      "dl-Hyoscyamine",
+      "dl-tropyltropate",
+      "Mydriasine",
+      "Tropine tropate",
+      "(±)-atropine",
+      "(±)-hyoscyamine",
+      "(+-)-Atropine",
+      "(+-)-Hyoscyamine",
+      "(+,-)-Tropyl tropate",
+      "(3-Endo)-8-methyl-8-azabicyclo[3.2.1]oct-3-yl tropate",
+      "[(1S,5R)-8-Methyl-8-azabicyclo[3.2.1]oct-3-yl] 3-hydroxy-2-phenyl-propanoate",
+      "8-Methyl-8-azabicyclo[3.2.1]oct-3-yl 3-hydroxy-2-phenylpropanoate",
+      "8-Methyl-8-azabicyclo[3.2.1]oct-3-yl tropate"
     ],
+    "xrefs": ["pubchem.compound:174174"],
     "other_identifiers": [
       "drugbank:DB00572",
       "rxcui:1223",
@@ -423,8 +422,9 @@
     "label": "Cisplatin",
     "concept_id": "ncit:C376",
     "other_identifiers": ["chemidplus:15663-27-1"],
-    "xrefs": ["umls:C0008838", "fda:Q20Q21Q62"],
-    "src_name": "NCIt"
+    "xrefs": ["umls:C0008838", "fda:Q20Q21Q62J"],
+    "src_name": "NCIt",
+    "merge_ref": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001"
   },
   {
     "label_and_type": "cisplatin##label",
@@ -477,7 +477,8 @@
     "trade_names": [],
     "other_identifiers": ["drugbank:DB00515"],
     "xrefs": ["fda:Q20Q21Q62J"],
-    "src_name": "ChemIDplus"
+    "src_name": "ChemIDplus",
+    "merge_ref": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001"
   },
   {
     "label_and_type": "cisplatin##label",
@@ -499,18 +500,10 @@
     "concept_id": "rxcui:100213",
     "label": "Bifidobacterium Infantis",
     "approval_status": "approved",
-    "aliases": [
-      "bifidobacterium infantis",
-      "Bifidobacterium infantis"
-    ],
-    "trade_names": [
-      "Align",
-      "Evivo"
-    ],
+    "aliases": ["bifidobacterium infantis", "Bifidobacterium infantis"],
+    "trade_names": ["Align", "Evivo"],
     "other_identifiers": ["drugbank:DB14222"],
-    "xrefs": [
-      "mmsl:d07347"
-    ],
+    "xrefs": ["mmsl:d07347"],
     "src_name": "RxNorm"
   },
   {
@@ -549,14 +542,8 @@
       "CDDP",
       "Cis-DDP"
     ],
-    "trade_names": [
-      "Cisplatin",
-      "Platinol"
-    ],
-    "other_identifiers": [
-      "drugbank:DB00515",
-      "drugbank:DB12117"
-    ],
+    "trade_names": ["Cisplatin", "Platinol"],
+    "other_identifiers": ["drugbank:DB00515", "drugbank:DB12117"],
     "xrefs": [
       "usp:m17910",
       "vandf:4018139",
@@ -567,7 +554,8 @@
       "mmsl:31747",
       "mmsl:4456"
     ],
-    "src_name": "RxNorm"
+    "src_name": "RxNorm",
+    "merge_ref": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001"
   },
   {
     "label_and_type": "cisplatin##label",
@@ -616,9 +604,7 @@
       "Amilamont",
       "Moduretic"
     ],
-    "other_identifiers": [
-      "drugbank:DB00594"
-    ],
+    "other_identifiers": ["drugbank:DB00594"],
     "xrefs": [
       "mesh:D000584",
       "vandf:4019603",
@@ -672,9 +658,7 @@
       "Glaucol",
       "Timoptol"
     ],
-    "other_identifiers": [
-      "drugbank:DB00373"
-    ],
+    "other_identifiers": ["drugbank:DB00373"],
     "xrefs": [
       "vandf:4019949",
       "mthspl:817W3C6175",
@@ -683,7 +667,8 @@
       "atc:C07AA06",
       "atc:S01ED01"
     ],
-    "src_name": "RxNorm"
+    "src_name": "RxNorm",
+    "merge_ref": "rxcui:10600"
   },
   {
     "label_and_type": "timolol##label",
@@ -721,10 +706,7 @@
       "Globulin, Anti-Thymocyte",
       "lymphocyte immune globulin, anti-thy (obs)"
     ],
-    "trade_names": [
-      "ATGAM",
-      "Thymoglobulin"
-    ],
+    "trade_names": ["ATGAM", "Thymoglobulin"],
     "xrefs": [
       "mesh:D000961",
       "vandf:4022194",
@@ -777,9 +759,7 @@
       "o-carboxyphenyl acetate",
       "Acetylsalicylic Acid"
     ],
-    "other_identifiers": [
-      "drugbank:DB00945"
-    ],
+    "other_identifiers": ["drugbank:DB00945"],
     "xrefs": [
       "usp:m6240",
       "vandf:4017536",
@@ -822,11 +802,7 @@
       "2-Mercaptoethanesulphonate, Sodium",
       "Sodium 2-Mercaptoethanesulphonate"
     ],
-    "trade_names": [
-      "Mesna",
-      "Mesnex",
-      "Mesna Nova Plus"
-    ],
+    "trade_names": ["Mesna", "Mesnex", "Mesna Nova Plus"],
     "xrefs": [
       "usp:m49500",
       "mesh:D015080",
@@ -870,11 +846,7 @@
       "3 Aminopropionic Acid",
       "3-Aminopropionic Acid"
     ],
-    "xrefs": [
-      "mesh:D015091",
-      "vandf:4028377",
-      "mthspl:11P2JDE17B"
-    ],
+    "xrefs": ["mesh:D015091", "vandf:4028377", "mthspl:11P2JDE17B"],
     "src_name": "RxNorm"
   },
   {
@@ -898,9 +870,7 @@
       "Alphasone",
       "Dihydroxyprogesterone"
     ],
-    "xrefs": [
-      "mesh:D000523"
-    ],
+    "xrefs": ["mesh:D000523"],
     "src_name": "RxNorm"
   },
   {
@@ -959,9 +929,7 @@
       "Tirosint",
       "Unithroid"
     ],
-    "other_identifiers": [
-      "drugbank:DB00451"
-    ],
+    "other_identifiers": ["drugbank:DB00451"],
     "xrefs": [
       "vandf:4022126",
       "mesh:D013974",
@@ -985,6 +953,324 @@
     "concept_id": "rxcui:10582",
     "src_name": "RxNorm"
   },
+  {
+    "label_and_type": "rxcui:8134##identity",
+    "label": "Phenobarbital",
+    "concept_id": "rxcui:8134",
+    "aliases": [
+      "Phenylethylmalonylurea",
+      "Phenobarbital-containing product",
+      "Phenobarbital",
+      "Phenobarbitone",
+      "Acid, Phenylethylbarbituric",
+      "Phenobarbital (substance)",
+      "PHENobarbital",
+      "Phenylethylbarbituric Acid",
+      "Phenylbarbital",
+      "Phenemal",
+      "Phenylethylbarbiturate",
+      "5-ethyl-5-phenylpyrimidine-2,4,6(1H,3H,5H)-trione",
+      "5-Phenyl-5-ethylbarbituric acid",
+      "5-Ethyl-5-phenylbarbituric acid",
+      "Phenobarbituric Acid",
+      "5-ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione",
+      "5-Ethyl-5-phenyl-pyrimidine-2,4,6-trione",
+      "Phenobarbitol",
+      "Product containing phenobarbital (medicinal product)"
+    ],
+    "other_identifiers": ["drugbank:DB01174"],
+    "xrefs": [
+      "usp:m63400",
+      "gsddb:2179",
+      "snomedct:51073002",
+      "vandf:4017422",
+      "mmsl:2390",
+      "msh:D010634",
+      "snomedct:373505007",
+      "mmsl:5272",
+      "mthspl:YQE403BP4D",
+      "fdbmk:001406",
+      "mmsl:d00340",
+      "atc:N03AA02"
+    ],
+    "approval_status": null,
+    "trade_names": ["Phenobarbital"],
+    "src_name": "RxNorm",
+    "merge_ref": "rxcui:8134|ncit:c739|chemidplus:50-06-6|wikidata:q407241"
+  },
+  {
+    "label_and_type": "phenobarbital##label",
+    "concept_id": "rxcui:8134",
+    "src_name": "RxNorm"
+  },
+  {
+    "label_and_type": "phenobarbital##trade_name",
+    "concept_id": "rxcui:8134",
+    "src_name": "RxNorm"
+  },
+  {
+    "label_and_type": "phenemal##alias",
+    "concept_id": "rxcui:8134",
+    "src_name": "RxNorm"
+  },
+  {
+    "label_and_type": "phenobarbital##alias",
+    "concept_id": "rxcui:8134",
+    "src_name": "RxNorm"
+  },
+  {
+    "label_and_type": "cis diamminedichloroplatinum##alias",
+    "concept_id": "rxcui:2555",
+    "src_name": "RxNorm"
+  },
+  {
+    "label_and_type": "chemidplus:50-06-6##identity",
+    "label": "Phenobarbital",
+    "concept_id": "chemidplus:50-06-6",
+    "aliases": [],
+    "other_identifiers": ["drugbank:DB01174"],
+    "xrefs": ["fda:YQE403BP4D"],
+    "approval_status": null,
+    "trade_names": [],
+    "src_name": "ChemIDplus",
+    "merge_ref": "rxcui:8134|ncit:c739|chemidplus:50-06-6|wikidata:q407241"
+  },
+  {
+    "label": "Spiramycin",
+    "label_and_type": "chemidplus:8025-81-8##identity",
+    "concept_id": "chemidplus:8025-81-8",
+    "aliases": [],
+    "other_identifiers": [],
+    "xrefs": ["fda:71ODY0V87H"],
+    "approval_status": null,
+    "trade_names": [],
+    "src_name": "ChemIDplus",
+    "merge_ref": "ncit:c839|chemidplus:8025-81-8"
+  },
+  {
+    "label": "phenobarbital",
+    "label_and_type": "wikidata:q407241##identity",
+    "concept_id": "wikidata:Q407241",
+    "aliases": [
+      "phenylethylbarbiturate",
+      "fenobarbital",
+      "Phenobarbitol",
+      "Phenylethylbarbituric Acid",
+      "5-Ethyl-5-phenylbarbituric acid",
+      "Phenylethylbarbiturate",
+      "Phenylethylbarbitursaeure",
+      "Phenylethylmalonylurea",
+      "Phenobarbital",
+      "Phenobarbitone",
+      "Phenylaethylbarbitursaeure",
+      "5-Phenyl-5-ethylbarbituric acid",
+      "Phenobarbituric Acid",
+      "5-ethyl-5-phenylpyrimidine-2,4,6(1H,3H,5H)-trione",
+      "5-ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione",
+      "Luminal®",
+      "phenobarbital sodium",
+      "5-Ethyl-5-phenyl-pyrimidine-2,4,6-trione",
+      "PHENYLETHYLMALONYLUREA"
+    ],
+    "other_identifiers": [
+      "chemidplus:50-06-6",
+      "chembl:CHEMBL40",
+      "rxcui:8134",
+      "drugbank:DB01174"
+    ],
+    "xrefs": ["pubchem.compound:4763"],
+    "approval_status": null,
+    "trade_names": [],
+    "src_name": "Wikidata",
+    "merge_ref": "rxcui:8134|ncit:c739|chemidplus:50-06-6|wikidata:q407241"
+  },
+  {
+    "label": "Phenobarbital",
+    "label_and_type": "ncit:c739##identity",
+    "concept_id": "ncit:C739",
+    "aliases": [
+      "5-Ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione",
+      "phenobarbital",
+      "PHENOBARBITAL",
+      "PHENO"
+    ],
+    "other_identifiers": ["chemidplus:50-06-6"],
+    "xrefs": ["umls:C0031412", "fda:YQE403BP4D", "chebi:CHEBI:8069"],
+    "approval_status": null,
+    "trade_names": [],
+    "src_name": "NCIt",
+    "merge_ref": "rxcui:8134|ncit:c739|chemidplus:50-06-6|wikidata:q407241"
+  },
+  {
+    "label_and_type": "pheno##alias",
+    "concept_id": "ncit:c739",
+    "src_name": "NCIt"
+  },
+  {
+    "label": "Spiramycin",
+    "label_and_type": "ncit:c839##identity",
+    "concept_id": "ncit:C839",
+    "aliases": [
+      "SPIRAMYCIN",
+      "Antibiotic 799",
+      "Rovamycin",
+      "Provamycin",
+      "Rovamycine",
+      "RP 5337",
+      "(4R,5S,6R,7R,9R,10R,11E,13E,16R)-10-{[(2R,5S,6R)-5-(dimethylamino)-6-methyltetrahydro-2H-pyran-2-yl]oxy}-9,16-dimethyl-5-methoxy-2-oxo-7-(2-oxoethyl)oxacyclohexadeca-11,13-dien-6-yl 3,6-dideoxy-4-O-(2,6-dideoxy-3-C-methyl-alpha-L-ribo-hexopyranosyl)-3-(dimethylamino)-alpha-D-glucopyranoside"
+    ],
+    "other_identifiers": ["chemidplus:8025-81-8"],
+    "xrefs": ["umls:C0037962", "fda:71ODY0V87H"],
+    "approval_status": null,
+    "trade_names": [],
+    "src_name": "NCIt",
+    "merge_ref": "ncit:c839|chemidplus:8025-81-8"
+  },
+  {
+    "label_and_type": "rp 5337##alias",
+    "concept_id": "ncit:C839",
+    "src_name": "NCIt"
+  },
+  {
+    "label_and_type": "rovamycine##alias",
+    "concept_id": "ncit:C839",
+    "src_name": "NCIt"
+  },
+  {
+    "label_and_type": "rxcui:8134|ncit:c739|chemidplus:50-06-6|wikidata:q407241##merger",
+    "concept_id": "rxcui:8134|ncit:C739|chemidplus:50-06-6|wikidata:Q407241",
+    "aliases": [
+      "5-Ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione",
+      "5-Ethyl-5-phenyl-pyrimidine-2,4,6-trione",
+      "5-Ethyl-5-phenylbarbituric acid",
+      "5-Phenyl-5-ethylbarbituric acid",
+      "5-ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione",
+      "5-ethyl-5-phenylpyrimidine-2,4,6(1H,3H,5H)-trione",
+      "Acid, Phenylethylbarbituric",
+      "Luminal®",
+      "PHENO",
+      "PHENOBARBITAL",
+      "PHENYLETHYLMALONYLUREA",
+      "PHENobarbital",
+      "Phenemal",
+      "Phenobarbital",
+      "Phenobarbital (substance)",
+      "Phenobarbital-containing product",
+      "Phenobarbitol",
+      "Phenobarbitone",
+      "Phenobarbituric Acid",
+      "Phenylaethylbarbitursaeure",
+      "Phenylbarbital",
+      "Phenylethylbarbiturate",
+      "Phenylethylbarbituric Acid",
+      "Phenylethylbarbitursaeure",
+      "Phenylethylmalonylurea",
+      "Product containing phenobarbital (medicinal product)",
+      "fenobarbital",
+      "phenobarbital",
+      "phenobarbital sodium",
+      "phenylethylbarbiturate"
+    ],
+    "trade_names": ["Phenobarbital"],
+    "xrefs": [
+      "pubchem.compound:4763",
+      "usp:m63400",
+      "gsddb:2179",
+      "snomedct:51073002",
+      "vandf:4017422",
+      "mmsl:2390",
+      "msh:D010634",
+      "snomedct:373505007",
+      "mmsl:5272",
+      "mthspl:YQE403BP4D",
+      "fdbmk:001406",
+      "mmsl:d00340",
+      "atc:N03AA02",
+      "fda:YQE403BP4D",
+      "umls:C0031412",
+      "chebi:CHEBI:8069"
+    ],
+    "label": "Phenobarbital"
+  },
+  {
+    "label_and_type": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001##merger",
+    "concept_id": "rxcui:2555|ncit:C376|chemidplus:15663-27-1|wikidata:Q412415|wikidata:Q47522001",
+    "trade_names": ["Cisplatin", "Platinol"],
+    "aliases": [
+      "1,2-Diaminocyclohexaneplatinum II citrate",
+      "CDDP",
+      "CISplatin",
+      "Cis-DDP",
+      "CIS-DDP",
+      "DDP",
+      "Diamminodichloride, Platinum",
+      "Dichlorodiammineplatinum",
+      "Platinum Diamminodichloride",
+      "cis Diamminedichloroplatinum",
+      "cis Platinum",
+      "cis-Diaminedichloroplatinum",
+      "cis-Diamminedichloroplatinum",
+      "cis-Diamminedichloroplatinum(II)",
+      "cis-Dichlorodiammineplatinum(II)",
+      "cis-Platinum",
+      "cis-diamminedichloroplatinum(II)",
+      "Platinol-AQ",
+      "Platinol"
+    ],
+    "label": "cisplatin",
+    "xrefs": [
+      "umls:C0008838",
+      "fda:Q20Q21Q62J",
+      "usp:m17910",
+      "vandf:4018139",
+      "mesh:D002945",
+      "mthspl:Q20Q21Q62J",
+      "mmsl:d00195",
+      "atc:L01XA01",
+      "mmsl:31747",
+      "mmsl:4456",
+      "pubchem.compound:5702198"
+    ]
+  },
+  {
+    "label_and_type": "ncit:c839|chemidplus:8025-81-8##merger",
+    "concept_id": "ncit:C839|chemidplus:8025-81-8",
+    "label": "Spiramycin",
+    "aliases": [
+      "SPIRAMYCIN",
+      "Antibiotic 799",
+      "Rovamycin",
+      "Provamycin",
+      "Rovamycine",
+      "RP 5337",
+      "(4R,5S,6R,7R,9R,10R,11E,13E,16R)-10-{[(2R,5S,6R)-5-(dimethylamino)-6-methyltetrahydro-2H-pyran-2-yl]oxy}-9,16-dimethyl-5-methoxy-2-oxo-7-(2-oxoethyl)oxacyclohexadeca-11,13-dien-6-yl 3,6-dideoxy-4-O-(2,6-dideoxy-3-C-methyl-alpha-L-ribo-hexopyranosyl)-3-(dimethylamino)-alpha-D-glucopyranoside"
+    ],
+    "xrefs": ["umls:C0037962", "fda:71ODY0V87H"]
+  },
+  {
+    "label_and_type": "ncit:c107245##identity",
+    "concept_id": "ncit:C107245",
+    "label": "Agerafenib",
+    "aliases": [
+      "AGERAFENIB",
+      "CEP-32496",
+      "AC013773",
+      "RXDX-105",
+      "1-(3-(6,7-Dimethoxyquinazolin-4-yloxy)phenyl)-3-(5-(1,1,1-trifluoro-2-methylpropan-2-yl)isoxazol-3-yl)urea Hydrochloride"
+    ],
+    "other_identifiers": ["chemidplus:1188910-76-0"],
+    "xrefs": ["fda:78I4VEX88N"]
+  },
+  {
+    "label_and_type": "agerafenib##label",
+    "concept_id": "ncit:C107245",
+    "src_name": "NCIt"
+  },
+  {
+    "label_and_type": "agerafenib##alias",
+    "concept_id": "ncit:C107245",
+    "src_name": "NCIt"
+  },
   {
     "label_and_type": "thyroxin##alias",
     "concept_id": "rxcui:10582",
@@ -1086,10 +1372,7 @@
     "concept_id": "rxcui:227224",
     "label": "fluoxetine hydrochloride",
     "approval_status": "approved",
-    "aliases": [
-      "FLUoxetine hydrochloride",
-      "Fluoxetine Hydrochloride"
-    ],
+    "aliases": ["FLUoxetine hydrochloride", "Fluoxetine Hydrochloride"],
     "xrefs": [
       "usp:m33780",
       "mthspl:I9W7N6B1KJ",
@@ -1099,16 +1382,98 @@
       "mmsl:37675",
       "mesh:D005473"
     ],
-    "trade_names": [
-      "FLUoxetine HCl",
-      "FLUoxetine Hydrochloride",
-      "RECONCILE"
-    ],
+    "trade_names": ["FLUoxetine HCl", "FLUoxetine Hydrochloride", "RECONCILE"],
     "src_name": "RxNorm"
   },
   {
     "label_and_type": "fluoxetine hydrochloride##label",
     "concept_id": "rxcui:227224",
     "src_name": "RxNorm"
+  },
+  {
+    "merge_ref": "wikidata:q47521576",
+    "concept_id": "wikidata:Q47521576",
+    "label_and_type": "wikidata:q47521576##identity",
+    "xrefs": [],
+    "label": "Ethyol",
+    "src_name": "Wikidata",
+    "other_identifiers": ["rxcui:151722"]
+  },
+  {
+    "concept_id": "rxcui:4126",
+    "label_and_type": "rxcui:151722##rx_brand",
+    "src_name": "RxNorm"
+  },
+  {
+    "aliases": ["APAETP"],
+    "merge_ref": "rxcui:4126",
+    "concept_id": "rxcui:4126",
+    "label_and_type": "rxcui:4126##identity",
+    "approval_status": "approved",
+    "xrefs": [
+      "usp:m2600",
+      "mesh:D004999",
+      "mmsl:4164",
+      "mmsl:d03868",
+      "vandf:4020970",
+      "mmsl:133029",
+      "mthspl:M487QF2F4V",
+      "atc:V03AF05"
+    ],
+    "label": "amifostine",
+    "trade_names": ["Amifostine", "Ethyol"],
+    "src_name": "RxNorm",
+    "other_identifiers": ["drugbank:DB01143"]
+  },
+  {
+    "label_and_type": "rxcui:10600##merger",
+    "concept_id": "rxcui:10600",
+    "aliases": [
+      "(S)-1-((1,1-Dimethylethyl)amino)-3-((4-(4-morpholinyl)-1,2,5-thiadazol-3-yl)oxy)-2-propanol",
+      "2-Propanol, 1-((1,1-dimethylethyl)amino)-3-((4-(4-morpholinyl)-1,2,5-thiadiazol-3-yl)oxy)-, (S)-",
+      "(S)-1-(tert-butylamino)-3-[(4-morpholin-4-yl-1,2,5-thiadiazol-3-yl)oxy]propan-2-ol"
+    ],
+    "trade_names": [
+      "Timoptol",
+      "Timolide",
+      "Betimol",
+      "Betim",
+      "Timoptic",
+      "Combigan",
+      "Istalol",
+      "Glaucol",
+      "Cosopt",
+      "Blocadren",
+      "Timoptol LA",
+      "Glau-Opt"
+    ],
+    "xrefs": [
+      "atc:C07AA06",
+      "mthspl:817W3C6175",
+      "vandf:4019949",
+      "atc:S01ED01",
+      "mesh:D013999",
+      "mmsl:d00139"
+    ],
+    "label": "timolol"
+  },
+  {
+    "concept_id": "rxcui:2555",
+    "label_and_type": "rxcui:202856##rx_brand",
+    "src_name": "RxNorm"
+  },
+  {
+    "merge_ref": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001",
+    "concept_id": "wikidata:Q47522001",
+    "label_and_type": "wikidata:q47522001##identity",
+    "xrefs": [],
+    "label": "Platinol",
+    "src_name": "Wikidata",
+    "other_identifiers": ["rxcui:202856"]
+  },
+  {
+    "concept_id": "wikidata:q47522001",
+    "label_and_type": "platinol##label",
+    "src_name": "Wikidata"
   }
 ]
diff --git a/tests/unit/test_chembl_source.py b/tests/unit/test_chembl_source.py
index b89bcaf8..4560a0a8 100644
--- a/tests/unit/test_chembl_source.py
+++ b/tests/unit/test_chembl_source.py
@@ -1,7 +1,7 @@
 """Test that the therapy normalizer works as intended for the ChEMBL source."""
 import pytest
 from therapy.schemas import Drug, MatchType
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 
 
 @pytest.fixture(scope='module')
@@ -10,11 +10,11 @@ def chembl():
     class QueryGetter:
 
         def __init__(self):
-            self.normalizer = Normalizer()
+            self.query_handler = QueryHandler()
 
         def normalize(self, query_str):
-            resp = self.normalizer.normalize(query_str, keyed=True,
-                                             incl='chembl')
+            resp = self.query_handler.search_sources(query_str, keyed=True,
+                                                     incl='chembl')
             return resp['source_matches']['ChEMBL']
     return QueryGetter()
 
diff --git a/tests/unit/test_chemidplus_source.py b/tests/unit/test_chemidplus_source.py
index a99baa14..1ca1dd3d 100644
--- a/tests/unit/test_chemidplus_source.py
+++ b/tests/unit/test_chemidplus_source.py
@@ -1,5 +1,5 @@
 """Test ChemIDplus ETL methods."""
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 from therapy.schemas import Drug, MatchType
 import pytest
 from typing import Dict
@@ -11,11 +11,11 @@ def chemidplus():
     class QueryGetter:
 
         def __init__(self):
-            self.normalizer = Normalizer()
+            self.normalizer = QueryHandler()
 
         def normalize(self, query_str):
-            resp = self.normalizer.normalize(query_str, keyed=True,
-                                             incl='chemidplus')
+            resp = self.normalizer.search_sources(query_str, keyed=True,
+                                                  incl='chemidplus')
             return resp['source_matches']['ChemIDplus']
     return QueryGetter()
 
diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py
index 87e47b3b..e8986195 100644
--- a/tests/unit/test_database.py
+++ b/tests/unit/test_database.py
@@ -11,7 +11,7 @@ def db():
     """Create a DynamoDB test fixture."""
     class DB:
         def __init__(self):
-            self.db = Database(db_url=os.environ['THERAPY_NORM_DB_URL'])
+            self.db = Database()
             if os.environ.get('TEST') is not None:
                 self.load_test_data()
 
@@ -32,11 +32,11 @@ def load_test_data(self):
                         batch.put_item(Item=m)
                 f.close()
 
-    return DB().db
+    return DB()
 
 
 def test_tables_created(db):
     """Check that therapy_concepts and therapy_metadata are created."""
-    existing_tables = db.ddb_client.list_tables()['TableNames']
+    existing_tables = db.db.dynamodb_client.list_tables()['TableNames']
     assert 'therapy_concepts' in existing_tables
     assert 'therapy_metadata' in existing_tables
diff --git a/tests/unit/test_drugbank_source.py b/tests/unit/test_drugbank_source.py
index 263a323d..c925f95c 100644
--- a/tests/unit/test_drugbank_source.py
+++ b/tests/unit/test_drugbank_source.py
@@ -3,7 +3,7 @@
 """
 import pytest
 from therapy.schemas import Drug, MatchType
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 
 
 @pytest.fixture(scope='module')
@@ -12,15 +12,15 @@ def drugbank():
     class QueryGetter:
 
         def __init__(self):
-            self.normalizer = Normalizer()
+            self.query_handler = QueryHandler()
 
         def normalize(self, query_str):
-            resp = self.normalizer.normalize(query_str, keyed=True,
-                                             incl='drugbank')
+            resp = self.query_handler.search_sources(query_str, keyed=True,
+                                                     incl='drugbank')
             return resp['source_matches']['DrugBank']
 
         def fetch_meta(self):
-            return self.normalizer.fetch_meta('DrugBank')
+            return self.query_handler._fetch_meta('DrugBank')
 
     return QueryGetter()
 
diff --git a/tests/unit/test_emit_warnings.py b/tests/unit/test_emit_warnings.py
index 59fb0a43..f90196ae 100644
--- a/tests/unit/test_emit_warnings.py
+++ b/tests/unit/test_emit_warnings.py
@@ -1,5 +1,5 @@
 """Test the emit_warnings function."""
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 
 
 def test_emit_warnings():
@@ -7,21 +7,21 @@ def test_emit_warnings():
     expected_warnings = {
         'nbsp': 'Query contains non breaking space characters.'
     }
-    normalizer = Normalizer()
+    query_handler = QueryHandler()
 
     # Test emit no warnings
-    actual_warnings = normalizer.emit_warnings('CISPLATIN')
+    actual_warnings = query_handler._emit_warnings('CISPLATIN')
     assert actual_warnings is None
 
     # Test emit warnings
-    actual_warnings = normalizer.emit_warnings('CIS PLATIN')
+    actual_warnings = query_handler._emit_warnings('CIS PLATIN')
     assert actual_warnings == actual_warnings
 
-    actual_warnings = normalizer.emit_warnings('CIS\u00A0platin')
+    actual_warnings = query_handler._emit_warnings('CIS\u00A0platin')
     assert expected_warnings == actual_warnings
 
-    actual_warnings = normalizer.emit_warnings('CIS&nbsp;platin')
+    actual_warnings = query_handler._emit_warnings('CIS&nbsp;platin')
     assert expected_warnings == actual_warnings
 
-    actual_warnings = normalizer.emit_warnings('CIS\xa0platin')
+    actual_warnings = query_handler._emit_warnings('CIS\xa0platin')
     assert expected_warnings == actual_warnings
diff --git a/tests/unit/test_merge.py b/tests/unit/test_merge.py
new file mode 100644
index 00000000..388b7f09
--- /dev/null
+++ b/tests/unit/test_merge.py
@@ -0,0 +1,342 @@
+"""Test merged record generation."""
+import pytest
+from therapy.etl.merge import Merge
+from typing import Dict
+
+
+@pytest.fixture(scope='module')
+def merge_handler(mock_database):
+    """Provide Merge instance to test cases."""
+    class MergeHandler():
+        def __init__(self):
+            self.merge = Merge(mock_database())
+
+        def get_merge(self):
+            return self.merge
+
+        def create_merged_concepts(self, record_ids):
+            return self.merge.create_merged_concepts(record_ids)
+
+        def get_added_records(self):
+            return self.merge._database.added_records
+
+        def get_updates(self):
+            return self.merge._database.updates
+
+        def create_record_id_set(self, record_id):
+            return self.merge._create_record_id_set(record_id)
+
+        def generate_merged_record(self, record_id_set):
+            return self.merge._generate_merged_record(record_id_set)
+
+    return MergeHandler()
+
+
+def compare_merged_records(actual_record: Dict, fixture_record: Dict):
+    """Check that records are identical."""
+    assert actual_record['concept_id'] == fixture_record['concept_id']
+    assert actual_record['label_and_type'] == fixture_record['label_and_type']
+    assert ('label' in actual_record) == ('label' in fixture_record)
+    if 'label' in actual_record or 'label' in fixture_record:
+        assert actual_record['label'] == fixture_record['label']
+    assert ('trade_names' in actual_record) == ('trade_names' in fixture_record)  # noqa: E501
+    if 'trade_names' in actual_record or 'trade_names' in fixture_record:
+        assert set(actual_record['trade_names']) == set(fixture_record['trade_names'])  # noqa: E501
+    assert ('aliases' in actual_record) == ('aliases' in fixture_record)
+    if 'aliases' in actual_record or 'aliases' in fixture_record:
+        assert set(actual_record['aliases']) == set(fixture_record['aliases'])
+    assert ('xrefs' in actual_record) == ('xrefs' in fixture_record)
+    if 'xrefs' in actual_record or 'xrefs' in fixture_record:
+        assert set(actual_record['xrefs']) == set(fixture_record['xrefs'])
+
+
+@pytest.fixture(scope='module')
+def phenobarbital_merged():
+    """Create phenobarbital fixture."""
+    return {
+        "label_and_type": "rxcui:8134|ncit:c739|chemidplus:50-06-6|wikidata:q407241##merger",  # noqa: E501
+        "concept_id": "rxcui:8134|ncit:C739|chemidplus:50-06-6|wikidata:Q407241",  # noqa: E501
+        "aliases": [
+            '5-Ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione',
+            '5-Ethyl-5-phenyl-pyrimidine-2,4,6-trione',
+            '5-Ethyl-5-phenylbarbituric acid',
+            '5-Phenyl-5-ethylbarbituric acid',
+            '5-ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione',
+            '5-ethyl-5-phenylpyrimidine-2,4,6(1H,3H,5H)-trione',
+            'Acid, Phenylethylbarbituric',
+            'Luminal®',
+            'PHENO',
+            'PHENOBARBITAL',
+            'PHENYLETHYLMALONYLUREA',
+            'PHENobarbital',
+            'Phenemal',
+            'Phenobarbital',
+            'Phenobarbital (substance)',
+            'Phenobarbital-containing product',
+            'Phenobarbitol',
+            'Phenobarbitone',
+            'Phenobarbituric Acid',
+            'Phenylaethylbarbitursaeure',
+            'Phenylbarbital',
+            'Phenylethylbarbiturate',
+            'Phenylethylbarbituric Acid',
+            'Phenylethylbarbitursaeure',
+            'Phenylethylmalonylurea',
+            'Product containing phenobarbital (medicinal product)',
+            'fenobarbital',
+            'phenobarbital',
+            'phenobarbital sodium',
+            'phenylethylbarbiturate'
+        ],
+        "trade_names": [
+            "Phenobarbital",
+        ],
+        "xrefs": [
+            "pubchem.compound:4763",
+            "usp:m63400",
+            "gsddb:2179",
+            "snomedct:51073002",
+            "vandf:4017422",
+            "mmsl:2390",
+            "msh:D010634",
+            "snomedct:373505007",
+            "mmsl:5272",
+            "mthspl:YQE403BP4D",
+            "fdbmk:001406",
+            "mmsl:d00340",
+            "atc:N03AA02",
+            "fda:YQE403BP4D",
+            "umls:C0031412",
+            "chebi:CHEBI:8069"
+
+        ],
+        "label": "Phenobarbital"
+    }
+
+
+@pytest.fixture(scope='module')
+def cisplatin_merged():
+    """Create cisplatin fixture."""
+    return {
+        "label_and_type": "rxcui:2555|ncit:c376|chemidplus:15663-27-1|wikidata:q412415|wikidata:q47522001##merger",  # noqa: E501
+        "concept_id": "rxcui:2555|ncit:C376|chemidplus:15663-27-1|wikidata:Q412415|wikidata:Q47522001",  # noqa: E501
+        "trade_names": [
+            "Cisplatin",
+            "Platinol"
+        ],
+        "aliases": [
+            '1,2-Diaminocyclohexaneplatinum II citrate',
+            'CDDP',
+            'CISplatin',
+            'Cis-DDP',
+            'CIS-DDP',
+            'DDP',
+            'Diamminodichloride, Platinum',
+            'Dichlorodiammineplatinum',
+            'Platinum Diamminodichloride',
+            'cis Diamminedichloroplatinum',
+            'cis Platinum',
+            'cis-Diaminedichloroplatinum',
+            'cis-Diamminedichloroplatinum',
+            'cis-Diamminedichloroplatinum(II)',
+            'cis-Dichlorodiammineplatinum(II)',
+            'cis-Platinum',
+            'cis-diamminedichloroplatinum(II)',
+            'Platinol-AQ',
+            'Platinol'
+        ],
+        "label": "cisplatin",
+        "xrefs": [
+            "umls:C0008838",
+            "fda:Q20Q21Q62J",
+            "usp:m17910",
+            "vandf:4018139",
+            "mesh:D002945",
+            "mthspl:Q20Q21Q62J",
+            "mmsl:d00195",
+            "atc:L01XA01",
+            "mmsl:31747",
+            "mmsl:4456",
+            "pubchem.compound:5702198"
+        ]
+    }
+
+
+@pytest.fixture(scope='module')
+def spiramycin_merged():
+    """Create fixture for spiramycin. The RxNorm entry should be inaccessible
+    to this group.
+    """
+    return {
+        "label_and_type": "ncit:c839|chemidplus:8025-81-8##merger",
+        "concept_id": "ncit:C839|chemidplus:8025-81-8",
+        "label": "Spiramycin",
+        "aliases": [
+            "SPIRAMYCIN",
+            "Antibiotic 799",
+            "Rovamycin",
+            "Provamycin",
+            "Rovamycine",
+            "RP 5337",
+            "(4R,5S,6R,7R,9R,10R,11E,13E,16R)-10-{[(2R,5S,6R)-5-(dimethylamino)-6-methyltetrahydro-2H-pyran-2-yl]oxy}-9,16-dimethyl-5-methoxy-2-oxo-7-(2-oxoethyl)oxacyclohexadeca-11,13-dien-6-yl 3,6-dideoxy-4-O-(2,6-dideoxy-3-C-methyl-alpha-L-ribo-hexopyranosyl)-3-(dimethylamino)-alpha-D-glucopyranoside"],  # noqa: E501
+        "xrefs": [
+            "umls:C0037962",
+            "fda:71ODY0V87H",
+        ],
+    }
+
+
+@pytest.fixture(scope='module')
+def record_id_groups():
+    """Create fixture for concept group sets."""
+    return {
+        "chemidplus:50-06-6": {
+            "rxcui:8134",
+            "ncit:C739",
+            "chemidplus:50-06-6",
+            "wikidata:Q407241",
+        },
+        "ncit:C739": {
+            "rxcui:8134",
+            "ncit:C739",
+            "chemidplus:50-06-6",
+            "wikidata:Q407241",
+        },
+        "ncit:C839": {
+            "ncit:C839",
+            "chemidplus:8025-81-8",
+        },
+        "rxcui:8134": {
+            "rxcui:8134",
+            "ncit:C739",
+            "chemidplus:50-06-6",
+            "wikidata:Q407241",
+        },
+        "wikidata:Q407241": {
+            "rxcui:8134",
+            "ncit:C739",
+            "chemidplus:50-06-6",
+            "wikidata:Q407241",
+        },
+        "chemidplus:15663-27-1": {
+            "rxcui:2555",
+            "ncit:C376",
+            "chemidplus:15663-27-1",
+            "wikidata:Q412415",
+            "wikidata:Q47522001"
+        },
+        "chemidplus:8025-81-8": {
+            "ncit:C839",
+            "chemidplus:8025-81-8",
+        },
+        "rxcui:2555": {
+            "rxcui:2555",
+            "ncit:C376",
+            "chemidplus:15663-27-1",
+            "wikidata:Q412415",
+            "wikidata:Q47522001"
+        },
+        "ncit:C376": {
+            "rxcui:2555",
+            "ncit:C376",
+            "chemidplus:15663-27-1",
+            "wikidata:Q412415",
+            "wikidata:Q47522001"
+        },
+        "wikidata:Q412415": {
+            "rxcui:2555",
+            "ncit:C376",
+            "chemidplus:15663-27-1",
+            "wikidata:Q412415",
+            "wikidata:Q47522001"
+        },
+        "wikidata:Q47522001": {
+            "rxcui:2555",
+            "ncit:C376",
+            "chemidplus:15663-27-1",
+            "wikidata:Q412415",
+            "wikidata:Q47522001"
+        },
+        "rxcui:4126": {
+            "rxcui:4126",
+            "wikidata:Q47521576"
+        },
+        "wikidata:Q47521576": {
+            "rxcui:4126",
+            "wikidata:Q47521576"
+        },
+    }
+
+
+def test_create_record_id_set(merge_handler, record_id_groups):
+    """Test creation of record ID sets."""
+    for record_id in record_id_groups.keys():
+        new_group = merge_handler.create_record_id_set(record_id)
+        for concept_id in new_group:
+            merge_handler.merge._groups[concept_id] = new_group
+    groups = merge_handler.merge._groups
+
+    for concept_id in groups.keys():
+        assert groups[concept_id] == record_id_groups[concept_id]
+    assert len(groups) == len(record_id_groups)  # check if any are missing
+
+    # test dead reference
+    has_dead_ref = 'ncit:C107245'
+    dead_group = merge_handler.create_record_id_set(has_dead_ref)
+    assert dead_group == {has_dead_ref}
+
+
+def test_generate_merged_record(merge_handler, record_id_groups,
+                                phenobarbital_merged, cisplatin_merged,
+                                spiramycin_merged):
+    """Test generation of merged record method."""
+    phenobarbital_ids = record_id_groups['rxcui:8134']
+    merge_response = merge_handler.generate_merged_record(phenobarbital_ids)
+    compare_merged_records(merge_response, phenobarbital_merged)
+
+    cisplatin_ids = record_id_groups['rxcui:2555']
+    merge_response = merge_handler.generate_merged_record(cisplatin_ids)
+    compare_merged_records(merge_response, cisplatin_merged)
+
+    spiramycin_ids = record_id_groups['ncit:C839']
+    merge_response = merge_handler.generate_merged_record(spiramycin_ids)
+    compare_merged_records(merge_response, spiramycin_merged)
+
+
+def test_create_merged_concepts(merge_handler, record_id_groups,
+                                phenobarbital_merged, cisplatin_merged,
+                                spiramycin_merged):
+    """Test end-to-end creation and upload of merged concepts."""
+    record_ids = record_id_groups.keys()
+    merge_handler.create_merged_concepts(record_ids)
+
+    # check merged record generation and storage
+    added_records = merge_handler.get_added_records()
+    assert len(added_records) == 4
+    phenobarb_merged_id = phenobarbital_merged['concept_id']
+    assert phenobarb_merged_id in added_records.keys()
+    compare_merged_records(added_records[phenobarb_merged_id],
+                           phenobarbital_merged)
+    cispl_merged_id = cisplatin_merged['concept_id']
+    assert cispl_merged_id in added_records.keys()
+    compare_merged_records(added_records[cispl_merged_id], cisplatin_merged)
+    spira_merged_id = spiramycin_merged['concept_id']
+    assert spira_merged_id in added_records.keys()
+    compare_merged_records(added_records[spira_merged_id],
+                           spiramycin_merged)
+
+    # check merged record reference updating
+    updates = merge_handler.get_updates()
+    assert len(updates) == len(record_id_groups)
+    for concept_id in record_id_groups['rxcui:8134']:
+        assert updates[concept_id] == {
+            'merge_ref': phenobarbital_merged['concept_id'].lower()
+        }
+    for concept_id in record_id_groups['rxcui:2555']:
+        assert updates[concept_id] == {
+            'merge_ref': cisplatin_merged['concept_id'].lower()
+        }
+    for concept_id in record_id_groups['ncit:C839']:
+        assert updates[concept_id] == {
+            'merge_ref': spiramycin_merged['concept_id'].lower()
+        }
diff --git a/tests/unit/test_ncit_source.py b/tests/unit/test_ncit_source.py
index 76804365..415f2375 100644
--- a/tests/unit/test_ncit_source.py
+++ b/tests/unit/test_ncit_source.py
@@ -1,7 +1,7 @@
 """Test NCIT source"""
 import pytest
 from therapy.schemas import Drug, MatchType
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 
 
 @pytest.fixture(scope='module')
@@ -10,11 +10,11 @@ def ncit():
     class QueryGetter:
 
         def __init__(self):
-            self.normalizer = Normalizer()
+            self.query_handler = QueryHandler()
 
         def normalize(self, query_str):
-            resp = self.normalizer.normalize(query_str, keyed=True,
-                                             incl='ncit')
+            resp = self.query_handler.search_sources(query_str, keyed=True,
+                                                     incl='ncit')
             return resp['source_matches']['NCIt']
     return QueryGetter()
 
diff --git a/tests/unit/test_query.py b/tests/unit/test_query.py
new file mode 100644
index 00000000..aa09d028
--- /dev/null
+++ b/tests/unit/test_query.py
@@ -0,0 +1,417 @@
+"""Test the therapy querying method."""
+from therapy.query import QueryHandler, InvalidParameterException
+from therapy.schemas import MatchType
+import pytest
+
+
+@pytest.fixture(scope='module')
+def query_handler():
+    """Build query handler test fixture."""
+    class QueryGetter:
+
+        def __init__(self):
+            self.query_handler = QueryHandler()
+
+        def normalize(self, query_str, keyed=False, incl='', excl=''):
+            resp = self.query_handler.search_sources(query_str=query_str,
+                                                     keyed=keyed,
+                                                     incl=incl, excl=excl)
+            return resp
+
+    return QueryGetter()
+
+
+@pytest.fixture(scope='module')
+def merge_query_handler(mock_database):
+    """Provide Merge instance to test cases."""
+    class QueryGetter():
+        def __init__(self):
+            self.query_handler = QueryHandler(db_url='http://localhost:8000')
+            self.query_handler.db = mock_database()  # replace initial DB
+
+        def search_groups(self, query_str):
+            return self.query_handler.search_groups(query_str)
+
+    return QueryGetter()
+
+
+@pytest.fixture(scope='module')
+def phenobarbital():
+    """Create phenobarbital fixture."""
+    return {
+        "concept_ids": [
+            "rxcui:8134",
+            "ncit:C739",
+            "chemidplus:50-06-6",
+            "wikidata:Q407241"
+        ],
+        "aliases": [
+            '5-Ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione',
+            '5-Ethyl-5-phenyl-pyrimidine-2,4,6-trione',
+            '5-Ethyl-5-phenylbarbituric acid',
+            '5-Phenyl-5-ethylbarbituric acid',
+            '5-ethyl-5-phenyl-2,4,6(1H,3H,5H)-pyrimidinetrione',
+            '5-ethyl-5-phenylpyrimidine-2,4,6(1H,3H,5H)-trione',
+            'Acid, Phenylethylbarbituric',
+            'Luminal®',
+            'PHENO',
+            'PHENOBARBITAL',
+            'PHENYLETHYLMALONYLUREA',
+            'PHENobarbital',
+            'Phenemal',
+            'Phenobarbital',
+            'Phenobarbital (substance)',
+            'Phenobarbital-containing product',
+            'Phenobarbitol',
+            'Phenobarbitone',
+            'Phenobarbituric Acid',
+            'Phenylaethylbarbitursaeure',
+            'Phenylbarbital',
+            'Phenylethylbarbiturate',
+            'Phenylethylbarbituric Acid',
+            'Phenylethylbarbitursaeure',
+            'Phenylethylmalonylurea',
+            'Product containing phenobarbital (medicinal product)',
+            'fenobarbital',
+            'phenobarbital',
+            'phenobarbital sodium',
+            'phenylethylbarbiturate'
+        ],
+        "trade_names": [
+            "Phenobarbital",
+        ],
+        "xrefs": [
+            "pubchem.compound:4763",
+            "usp:m63400",
+            "gsddb:2179",
+            "snomedct:51073002",
+            "vandf:4017422",
+            "mmsl:2390",
+            "msh:D010634",
+            "snomedct:373505007",
+            "mmsl:5272",
+            "mthspl:YQE403BP4D",
+            "fdbmk:001406",
+            "mmsl:d00340",
+            "atc:N03AA02",
+            "fda:YQE403BP4D",
+            "umls:C0031412",
+            "chebi:CHEBI:8069"
+
+        ],
+        "label": "Phenobarbital"
+    }
+
+
+@pytest.fixture(scope='module')
+def cisplatin():
+    """Create cisplatin fixture."""
+    return {
+        "concept_ids": [
+            "rxcui:2555",
+            "ncit:C376",
+            "chemidplus:15663-27-1",
+            "wikidata:Q412415",
+            "wikidata:Q47522001"
+        ],
+        "trade_names": [
+            "Cisplatin",
+            "Platinol"
+        ],
+        "aliases": [
+            '1,2-Diaminocyclohexaneplatinum II citrate',
+            'CDDP',
+            'CISplatin',
+            'Cis-DDP',
+            'CIS-DDP',
+            'DDP',
+            'Diamminodichloride, Platinum',
+            'Dichlorodiammineplatinum',
+            'Platinum Diamminodichloride',
+            'cis Diamminedichloroplatinum',
+            'cis Platinum',
+            'cis-Diaminedichloroplatinum',
+            'cis-Diamminedichloroplatinum',
+            'cis-Diamminedichloroplatinum(II)',
+            'cis-Dichlorodiammineplatinum(II)',
+            'cis-Platinum',
+            'cis-diamminedichloroplatinum(II)',
+            'Platinol-AQ',
+            'Platinol'
+        ],
+        "label": "cisplatin",
+        "xrefs": [
+            "umls:C0008838",
+            "fda:Q20Q21Q62J",
+            "usp:m17910",
+            "vandf:4018139",
+            "mesh:D002945",
+            "mthspl:Q20Q21Q62J",
+            "mmsl:d00195",
+            "atc:L01XA01",
+            "mmsl:31747",
+            "mmsl:4456",
+            "pubchem.compound:5702198"
+        ]
+    }
+
+
+@pytest.fixture(scope='module')
+def spiramycin():
+    """Create fixture for spiramycin."""
+    return {
+        "concept_ids": [
+            "ncit:C839",
+            "chemidplus:8025-81-8"
+        ],
+        "label": "Spiramycin",
+        "aliases": [
+            "SPIRAMYCIN",
+            "Antibiotic 799",
+            "Rovamycin",
+            "Provamycin",
+            "Rovamycine",
+            "RP 5337",
+            "(4R,5S,6R,7R,9R,10R,11E,13E,16R)-10-{[(2R,5S,6R)-5-(dimethylamino)-6-methyltetrahydro-2H-pyran-2-yl]oxy}-9,16-dimethyl-5-methoxy-2-oxo-7-(2-oxoethyl)oxacyclohexadeca-11,13-dien-6-yl 3,6-dideoxy-4-O-(2,6-dideoxy-3-C-methyl-alpha-L-ribo-hexopyranosyl)-3-(dimethylamino)-alpha-D-glucopyranoside"],  # noqa: E501
+        "xrefs": [
+            "umls:C0037962",
+            "fda:71ODY0V87H",
+        ],
+    }
+
+
+@pytest.fixture(scope='module')
+def timolol():
+    """Create fixture for timolol."""
+    return {
+        "concept_ids": ["rxcui:10600"],
+        "aliases": [
+            "(S)-1-((1,1-Dimethylethyl)amino)-3-((4-(4-morpholinyl)-1,2,5-thiadazol-3-yl)oxy)-2-propanol",  # noqa: E501
+            "2-Propanol, 1-((1,1-dimethylethyl)amino)-3-((4-(4-morpholinyl)-1,2,5-thiadiazol-3-yl)oxy)-, (S)-",  # noqa: E501
+            "(S)-1-(tert-butylamino)-3-[(4-morpholin-4-yl-1,2,5-thiadiazol-3-yl)oxy]propan-2-ol"  # noqa: E501
+        ],
+        "trade_names": [
+            "Timoptol",
+            "Timolide",
+            "Betimol",
+            "Betim",
+            "Timoptic",
+            "Combigan",
+            "Istalol",
+            "Glaucol",
+            "Cosopt",
+            "Blocadren",
+            "Timoptol LA",
+            "Glau-Opt"
+        ],
+        "xrefs": [
+            "atc:C07AA06",
+            "mthspl:817W3C6175",
+            "vandf:4019949",
+            "atc:S01ED01",
+            "mesh:D013999",
+            "mmsl:d00139"
+        ],
+        "label": "timolol"
+    }
+
+
+def test_query(query_handler):
+    """Test that query returns properly-structured response."""
+    resp = query_handler.normalize('cisplatin', keyed=False)
+    assert resp['query'] == 'cisplatin'
+    matches = resp['source_matches']
+    assert isinstance(matches, list)
+    assert len(matches) == 6
+    wikidata = list(filter(lambda m: m['source'] == 'Wikidata',
+                           matches))[0]
+    assert len(wikidata['records']) == 1
+    wikidata_record = wikidata['records'][0]
+    assert wikidata_record.label == 'cisplatin'
+
+
+def test_query_keyed(query_handler):
+    """Test that query structures matches as dict when requested."""
+    resp = query_handler.normalize('penicillin v', keyed=True)
+    matches = resp['source_matches']
+    assert isinstance(matches, dict)
+    chemidplus = matches['ChemIDplus']
+    chemidplus_record = chemidplus['records'][0]
+    assert chemidplus_record.label == 'Penicillin V'
+
+
+def test_query_specify_query_handlers(query_handler):
+    """Test inclusion and exclusion of sources in query."""
+    # test blank params
+    resp = query_handler.normalize('cisplatin', keyed=True)
+    matches = resp['source_matches']
+    assert len(matches) == 6
+    assert 'Wikidata' in matches
+    assert 'ChEMBL' in matches
+    assert 'NCIt' in matches
+    assert 'DrugBank' in matches
+
+    # test partial inclusion
+    resp = query_handler.normalize('cisplatin', keyed=True, incl='chembl,ncit')
+    matches = resp['source_matches']
+    assert len(matches) == 2
+    assert 'Wikidata' not in matches
+    assert 'ChEMBL' in matches
+    assert 'NCIt' in matches
+    assert 'DrugBank' not in matches
+
+    # test full inclusion
+    sources = 'chembl,ncit,drugbank,wikidata,rxnorm,chemidplus'
+    resp = query_handler.normalize('cisplatin', keyed=True,
+                                   incl=sources, excl='')
+    matches = resp['source_matches']
+    assert len(matches) == 6
+    assert 'Wikidata' in matches
+    assert 'ChEMBL' in matches
+
+    # test partial exclusion
+    resp = query_handler.normalize('cisplatin', keyed=True, excl='chemidplus')
+    matches = resp['source_matches']
+    assert len(matches) == 5
+    assert 'Wikidata' in matches
+    assert 'ChemIDplus' not in matches
+
+    # test full exclusion
+    resp = query_handler.normalize('cisplatin', keyed=True,
+                                   excl='chembl, wikidata, drugbank, ncit, '
+                                   'rxnorm, chemidplus')
+    matches = resp['source_matches']
+    assert len(matches) == 0
+    assert 'Wikidata' not in matches
+    assert 'ChEMBL' not in matches
+    assert 'NCIt' not in matches
+    assert 'DrugBank' not in matches
+
+    # test case insensitive
+    resp = query_handler.normalize('cisplatin', keyed=True, excl='ChEmBl')
+    matches = resp['source_matches']
+    assert 'Wikidata' in matches
+    assert 'ChEMBL' not in matches
+    resp = query_handler.normalize('cisplatin', keyed=True,
+                                   incl='wIkIdAtA,cHeMbL')
+    matches = resp['source_matches']
+    assert 'Wikidata' in matches
+    assert 'ChEMBL' in matches
+
+    # test error on invalid source names
+    with pytest.raises(InvalidParameterException):
+        resp = query_handler.normalize('cisplatin', keyed=True, incl='chambl')
+
+    # test error for supplying both incl and excl args
+    with pytest.raises(InvalidParameterException):
+        resp = query_handler.normalize('cisplatin', keyed=True, incl='chembl',
+                                       excl='wikidata')
+
+
+def compare_records(test_record, record_fixture):
+    """Assert equality of test record against correct record.
+
+    :param Dict test_record: record produced in test case
+    :param Dict record_fixture: expected record output
+    """
+    assert test_record['concept_ids'] == record_fixture['concept_ids']
+    if 'aliases' in test_record or 'aliases' in record_fixture:
+        assert set(test_record['aliases']) == set(record_fixture['aliases'])
+    if 'trade_names' in test_record or 'trade_names' in record_fixture:
+        assert set(test_record['trade_names']) == set(record_fixture['trade_names'])  # noqa: E501
+    if 'xrefs' in test_record or 'xrefs' in record_fixture:
+        assert set(test_record['xrefs']) == set(record_fixture['xrefs'])
+    assert test_record['label'] == record_fixture['label']
+
+
+def test_query_merged(merge_query_handler, phenobarbital, cisplatin,
+                      spiramycin, timolol):
+    """Test that the merged concept endpoint handles queries correctly."""
+    # test merged id match
+    test_query = "rxcui:2555|ncit:C376|chemidplus:15663-27-1|wikidata:Q412415|wikidata:Q47522001"  # noqa: E501
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.CONCEPT_ID
+    compare_records(response['record'], cisplatin)
+
+    # test concept id match
+    test_query = "chemidplus:50-06-6"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.CONCEPT_ID
+    compare_records(response['record'], phenobarbital)
+
+    # # test label match
+    test_query = "phenobarbital"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.LABEL
+    compare_records(response['record'], phenobarbital)
+
+    # test trade name match
+    test_query = "Platinol"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.TRADE_NAME
+    compare_records(response['record'], cisplatin)
+
+    # test alias match
+    test_query = "cis Diamminedichloroplatinum"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.ALIAS
+    compare_records(response['record'], cisplatin)
+
+    test_query = "Rovamycine"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.ALIAS
+    compare_records(response['record'], spiramycin)
+
+    # test no match
+    test_query = "zzzz fake therapy zzzz"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert 'record' not in response
+    assert response['match_type'] == MatchType.NO_MATCH
+
+    # test merge group with single member
+    test_query = "Betimol"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.TRADE_NAME
+    compare_records(response['record'], timolol)
+
+    # test that term with multiple possible resolutions resolves at highest
+    # match
+    test_query = "Cisplatin"
+    response = merge_query_handler.search_groups(test_query)
+    assert response['query'] == test_query
+    assert response['warnings'] is None
+    assert response['match_type'] == MatchType.LABEL
+    compare_records(response['record'], cisplatin)
+
+
+def test_merged_meta(merge_query_handler):
+    """Test population of source metadata in merged querying."""
+    test_query = "pheno"
+    response = merge_query_handler.search_groups(test_query)
+    meta_items = response['meta_']
+    assert 'RxNorm' in meta_items.keys()
+    assert 'Wikidata' in meta_items.keys()
+    assert 'NCIt' in meta_items.keys()
+    assert 'ChemIDplus' in meta_items.keys()
+
+    test_query = "RP 5337"
+    response = merge_query_handler.search_groups(test_query)
+    meta_items = response['meta_']
+    assert 'NCIt' in meta_items.keys()
+    assert 'ChemIDplus' in meta_items.keys()
diff --git a/tests/unit/test_rxnorm_source.py b/tests/unit/test_rxnorm_source.py
index 825f6a9d..e5511486 100644
--- a/tests/unit/test_rxnorm_source.py
+++ b/tests/unit/test_rxnorm_source.py
@@ -1,7 +1,7 @@
 """Test that the therapy normalizer works as intended for the RxNorm source."""
 import pytest
 from therapy.schemas import Drug, MatchType
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 from therapy.database import Database
 import os
 from boto3.dynamodb.conditions import Key
@@ -13,21 +13,20 @@ def rxnorm():
     class QueryGetter:
 
         def __init__(self):
-            self.normalizer = Normalizer()
+            self.normalizer = QueryHandler()
             if 'THERAPY_NORM_DB_URL' in os.environ:
                 db_url = os.environ['THERAPY_NORM_DB_URL']
             else:
                 db_url = 'http://localhost:8000'
-
             self.db = Database(db_url=db_url)
 
         def normalize(self, query_str):
-            resp = self.normalizer.normalize(query_str, keyed=True,
-                                             incl='rxnorm')
+            resp = self.normalizer.search_sources(query_str, keyed=True,
+                                                  incl='rxnorm')
             return resp['source_matches']['RxNorm']
 
         def fetch_meta(self):
-            return self.normalizer.fetch_meta('RxNorm')
+            return self.normalizer._fetch_meta('RxNorm')
 
     return QueryGetter()
 
diff --git a/tests/unit/test_therapy_query.py b/tests/unit/test_therapy_query.py
deleted file mode 100644
index 011f60d9..00000000
--- a/tests/unit/test_therapy_query.py
+++ /dev/null
@@ -1,111 +0,0 @@
-"""Test the therapy querying method."""
-from therapy.query import Normalizer, InvalidParameterException
-import pytest
-
-
-@pytest.fixture(scope='module')
-def normalizer():
-    """Build normalizer test fixture."""
-    class QueryGetter:
-
-        def __init__(self):
-            self.normalizer = Normalizer()
-
-        def normalize(self, query_str, keyed=False, incl='', excl=''):
-            resp = self.normalizer.normalize(query_str=query_str, keyed=keyed,
-                                             incl=incl, excl=excl)
-            return resp
-
-    return QueryGetter()
-
-
-def test_query(normalizer):
-    """Test that query returns properly-structured response."""
-    resp = normalizer.normalize('cisplatin', keyed=False)
-    assert resp['query'] == 'cisplatin'
-    matches = resp['source_matches']
-    assert isinstance(matches, list)
-    assert len(matches) == 6
-    wikidata = list(filter(lambda m: m['source'] == 'Wikidata',
-                           matches))[0]
-    assert len(wikidata['records']) == 1
-    wikidata_record = wikidata['records'][0]
-    assert wikidata_record.label == 'cisplatin'
-
-
-def test_query_keyed(normalizer):
-    """Test that query structures matches as dict when requested."""
-    resp = normalizer.normalize('penicillin v', keyed=True)
-    matches = resp['source_matches']
-    assert isinstance(matches, dict)
-    chemidplus = matches['ChemIDplus']
-    chemidplus_record = chemidplus['records'][0]
-    assert chemidplus_record.label == 'Penicillin V'
-
-
-def test_query_specify_normalizers(normalizer):
-    """Test inclusion and exclusion of normalizers in query."""
-    # test blank params
-    resp = normalizer.normalize('cisplatin', keyed=True)
-    matches = resp['source_matches']
-    assert len(matches) == 6
-    assert 'Wikidata' in matches
-    assert 'ChEMBL' in matches
-    assert 'NCIt' in matches
-    assert 'DrugBank' in matches
-
-    # test partial inclusion
-    resp = normalizer.normalize('cisplatin', keyed=True, incl='chembl,ncit')
-    matches = resp['source_matches']
-    assert len(matches) == 2
-    assert 'Wikidata' not in matches
-    assert 'ChEMBL' in matches
-    assert 'NCIt' in matches
-    assert 'DrugBank' not in matches
-
-    # test full inclusion
-    sources = 'chembl,ncit,drugbank,wikidata,rxnorm,chemidplus'
-    resp = normalizer.normalize('cisplatin', keyed=True,
-                                incl=sources, excl='')
-    matches = resp['source_matches']
-    assert len(matches) == 6
-    assert 'Wikidata' in matches
-    assert 'ChEMBL' in matches
-
-    # test partial exclusion
-    resp = normalizer.normalize('cisplatin', keyed=True, excl='chemidplus')
-    matches = resp['source_matches']
-    assert len(matches) == 5
-    assert 'Wikidata' in matches
-    assert 'ChemIDplus' not in matches
-
-    # test full exclusion
-    resp = normalizer.normalize('cisplatin', keyed=True,
-                                excl='chembl, wikidata, drugbank, ncit, '
-                                     'rxnorm, chemidplus')
-    matches = resp['source_matches']
-    assert len(matches) == 0
-    assert 'Wikidata' not in matches
-    assert 'ChEMBL' not in matches
-    assert 'NCIt' not in matches
-    assert 'DrugBank' not in matches
-
-    # test case insensitive
-    resp = normalizer.normalize('cisplatin', keyed=True, excl='ChEmBl')
-    matches = resp['source_matches']
-    assert 'Wikidata' in matches
-    assert 'ChEMBL' not in matches
-    resp = normalizer.normalize('cisplatin', keyed=True,
-                                incl='wIkIdAtA,cHeMbL')
-    matches = resp['source_matches']
-    assert 'Wikidata' in matches
-    assert 'ChEMBL' in matches
-
-    # test error on invalid normalizer names
-    with pytest.raises(InvalidParameterException):
-        resp = normalizer.normalize('cisplatin', keyed=True, incl='chambl')
-
-    # test error for supplying both incl and excl args
-    with pytest.raises(InvalidParameterException):
-        resp = normalizer.normalize('cisplatin', keyed=True, incl='chembl',
-                                    excl='wikidata')
diff --git a/tests/unit/test_wikidata_source.py b/tests/unit/test_wikidata_source.py
index a19ca5ea..0f39de1b 100644
--- a/tests/unit/test_wikidata_source.py
+++ b/tests/unit/test_wikidata_source.py
@@ -5,7 +5,7 @@
 
 import pytest
 from therapy.schemas import Drug, MatchType
-from therapy.query import Normalizer
+from therapy.query import QueryHandler
 
 
 @pytest.fixture(scope='module')
@@ -14,11 +14,11 @@ def wikidata():
     class QueryGetter:
 
         def __init__(self):
-            self.normalizer = Normalizer()
+            self.query_handler = QueryHandler()
 
         def normalize(self, query_str):
-            resp = self.normalizer.normalize(query_str, keyed=True,
-                                             incl='wikidata')
+            resp = self.query_handler.search_sources(query_str, keyed=True,
+                                                     incl='wikidata')
             return resp['source_matches']['Wikidata']
     return QueryGetter()
 
@@ -52,6 +52,21 @@ def cisplatin():
     return Drug(**params)
 
 
+@pytest.fixture(scope='module')
+def platinol():
+    """Create a platinol drug fixture."""
+    params = {
+        'label': 'Platinol',
+        'concept_id': 'wikidata:Q47522001',
+        'aliases': [],
+        'trade_names': [],
+        'approval_status': None,
+        'other_identifiers': ['rxcui:202856'],
+        'xrefs': []
+    }
+    return Drug(**params)
+
+
 @pytest.fixture(scope='module')
 def interferon_alfacon_1():
     """Create an Interferon alfacon-1 drug fixture."""