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

Issue:
feat!: support PostgreSQL as optional storage backend
Finalizes a dev release implementing an optional Postgres DB backend to use in TheraPy CI. Once tests are passing for everything, there will be another full release.

Otherwise, everything is basically the same as the gene PR. I've noticed that all three normalizers have slightly different query method structure, which would be good to fix up at some point, but not pressing rn.


Discussion / hints:
@jsstevenson It also looks like there's some conflicts that need resolved before merging

Fix patch:
diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
index 84ca7810..85a412b1 100644
--- a/.github/workflows/checks.yml
+++ b/.github/workflows/checks.yml
@@ -8,32 +8,51 @@ jobs:
       AWS_SECRET_ACCESS_KEY: ${{ secrets.DUMMY_AWS_SECRET_ACCESS_KEY }}
       AWS_DEFAULT_REGION: us-east-2
       AWS_DEFAULT_OUTPUT: text
-      DISEASE_NORM_DB_URL: http://localhost:8002
+      DISEASE_NORM_DB_URL: ${{ matrix.db_url }}
       DISEASE_TEST: true
     strategy:
       matrix:
+        db_url: ["http://localhost:8002", "postgres://postgres:postgres@localhost:5432/disease_normalizer_test"]
         python-version: ['3.8', '3.9', '3.10', '3.11']
+    services:
+      postgres:
+        image: postgres:14
+        env:
+          POSTGRES_USER: 'postgres'
+          POSTGRES_DB: 'disease_normalizer_test'
+          POSTGRES_PASSWORD: 'postgres'
+        ports:
+          - 5432:5432
     steps:
     - uses: actions/checkout@v3
 
-    - name: Setup Python
+    - name: Set up Python
       uses: actions/setup-python@v4
       with:
-          python-version: ${{ matrix.python-version }}
-
-    - name: Install dependencies
-      run: python3 -m pip install ".[dev,test]"
+        python-version: ${{ matrix.python-version }}
 
-    - name: Build local DynamoDB server
+    - name: Build local DynamoDB
+      if: ${{ env.DISEASE_NORM_DB_URL == 'http://localhost:8002' }}
       run: |
-          chmod +x ./tests/scripts/dynamodb_run.sh
-          ./tests/scripts/dynamodb_run.sh
+        chmod +x ./tests/scripts/dynamodb_run.sh
+        ./tests/scripts/dynamodb_run.sh
+
+    - name: Install DynamoDB dependencies
+      if: ${{ env.DISEASE_NORM_DB_URL == 'http://localhost:8002' }}
+      run: python3 -m pip install ".[etl,test]"
+
+    - name: Install PG dependencies
+      if: ${{ env.DISEASE_NORM_DB_URL != 'http://localhost:8002' }}
+      run: python3 -m pip install ".[pg,etl,test]"
 
     - name: Run tests
-      run: python3 -m pytest
+      run: python3 -m pytest tests/
 
   lint:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.8', '3.9', '3.10', '3.11']
     env:
       AWS_ACCESS_KEY_ID: ${{ secrets.DUMMY_AWS_ACCESS_KEY_ID }}
       AWS_SECRET_ACCESS_KEY: ${{ secrets.DUMMY_AWS_SECRET_ACCESS_KEY }}
@@ -45,7 +64,7 @@ jobs:
     - name: Setup Python
       uses: actions/setup-python@v4
       with:
-        python-version: '3.10'
+        python-version: ${{ matrix.python-version }}
 
     - name: Install dependencies
       run: python3 -m pip install ".[dev]"
diff --git a/.gitignore b/.gitignore
index fce6636c..36c18880 100644
--- a/.gitignore
+++ b/.gitignore
@@ -165,3 +165,5 @@ Pipfile.lock
 # Misc
 tests/data/mondo/*.owl
 tests/scripts/robot*
+analysis/civic-data/*.json
+analysis/pmkb-data/*.json
diff --git a/Pipfile b/Pipfile
index 8b4701c0..8bfca4d4 100644
--- a/Pipfile
+++ b/Pipfile
@@ -22,6 +22,7 @@ ipykernel = "*"
 matplotlib = "*"
 lxml = "*"
 xmlformatter = "*"
+psycopg = {version = "*", extras=["binary"]}
 
 [packages]
 pydantic = "*"
diff --git a/README.md b/README.md
index 0eca4399..556ad660 100644
--- a/README.md
+++ b/README.md
@@ -1,137 +1,163 @@
-# Disease Normalization
-Services and guidelines for normalizing disease terms
+# Disease Normalizer
 
-## Developer instructions
-Following are sections include instructions specifically for developers.
+Services and guidelines for normalizing disease terms
 
-### Installation
-For a development install, we recommend using Pipenv. See the
-[pipenv docs](https://pipenv-fork.readthedocs.io/en/latest/#install-pipenv-today)
-for direction on installing pipenv in your compute environment.
+## Installation
 
-Once installed, from the project root dir, just run:
+The Disease Normalizer is available via PyPI:
 
 ```commandline
-pipenv sync
+
+pip install disease-normalizer[etl,pg]
 ```
 
-### Deploying DynamoDB Locally
+The [etl,pg] argument tells pip to install packages to fulfill the dependencies of the gene.etl package and the PostgreSQL data storage implementation alongside the default DynamoDB data storage implementation.
 
-We use Amazon DynamoDB for our database. To deploy locally, follow [these instructions](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBLocal.DownloadingAndRunning.html).
+### External requirements
 
-### Init coding style tests
+The Disease Normalizer can retrieve most required data itself. The exception is disease terms from OMIM, for which a source file must be manually acquired and placed in the `disease/data/omim` folder within the library root. In order to access OMIM data, users must submit a request [here](https://www.omim.org/downloads). Once approved, the relevant OMIM file (`mimTitles.txt`) should be renamed according to the convention `omim_YYYYMMDD.tsv`, where `YYYYMMDD` indicates the date that the file was generated, and placed in the appropriate location.
 
-Code style is managed by [flake8](https://github.com/PyCQA/flake8) and checked prior to commit.
+### Database Initialization
 
-We use [pre-commit](https://pre-commit.com/#usage) to run conformance tests.
+The Disease Normalizer supports two data storage options:
 
-This ensures:
+* [DynamoDB](https://aws.amazon.com/dynamodb), a NoSQL service provided by AWS. This is our preferred storage solution. In addition to cloud deployment, Amazon also provides a tool for local service, which can be installed [here](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBLocal.DownloadingAndRunning.html). Once downloaded, you can start service by running `java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -sharedDb` in a terminal (add a `-port <VALUE>` option to use a different port)
+* [PostgreSQL](https://www.postgresql.org/), a well-known relational database technology. Once starting the Postgres server process, [ensure that a database is created](https://www.postgresql.org/docs/current/sql-createdatabase.html) (we typically name ours `disease_normalizer`).
 
-* Check code style
-* Check for added large files
-* Detect AWS Credentials
-* Detect Private Key
+By default, the Disease Normalizer expects to find a DynamoDB instance listening at `http://localhost:8000`. Alternative locations can be specified in two ways:
 
-Before first commit run:
+The first way is to set the `--db_url` command-line option to the URL endpoint.
 
 ```commandline
-pre-commit install
+disease_norm_update --update_all --db_url="http://localhost:8001"
 ```
 
+The second way is to set the `DISEASE_NORM_DB_URL` environment variable to the URL endpoint.
+```commandline
+export DISEASE_NORM_DB_URL="http://localhost:8001"
+```
 
-### Running unit tests
-
-Tests are provided via pytest.
+To use a PostgreSQL instance instead of DynamoDB, provide a PostgreSQL connection URL instead, e.g.
 
 ```commandline
-pytest
+export DISEASE_NORM_DB_URL="postgresql://postgres@localhost:5432/disease_normalizer"
 ```
 
-By default, tests will employ an existing DynamoDB database. For test environments where this is unavailable (e.g. in CI), the `DISEASE_TEST` environment variable can be set to initialize a local DynamoDB instance with miniature versions of input data files before tests are executed.
+### Adding and refreshing data
 
+Use the `disease_norm_update` command in a shell to update the database.
 
-```comandline
-export DISEASE_TEST=true
-pytest
-```
+#### Update source(s)
 
-Sometimes, sources will update their data, and our test fixtures and data will become incorrect. The `tests/scripts/` subdirectory includes scripts to rebuild data files, although most fixtures will need to be updated manually.
+The Disease Normalizer currently uses data from the following sources:
 
-### Updating the disease normalization database
+ * The [National Cancer Institute Thesaurus (NCIt)](https://ncithesaurus.nci.nih.gov/ncitbrowser/)
+ * The [Mondo Disease Ontology](https://mondo.monarchinitiative.org/)
+ * The [Online Mendelian Inheritance in Man (OMIM)](https://www.omim.org/)
+ * [OncoTree](http://oncotree.mskcc.org/)
+ * The [Disease Ontology](https://disease-ontology.org/)
 
-Before you use the CLI to update the database, run the following in a separate terminal to start DynamoDB on `port 8000`:
+As described above, all source data other than OMIM can be acquired automatically.
 
-```
-java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -sharedDb
+To update one source, simply set `--normalizer` to the source you wish to update. The normalizer will check to see if local source data is up-to-date, acquire the most recent data if not, and use it to populate the database.
+
+For example, run the following to acquire the latest NCIt data if necessary, and update the NCIt disease records in the normalizer database:
+
+```commandline
+disease_norm_update --normalizer="ncit"
 ```
 
-To change the port, simply add `-port value`.
+To update multiple sources, you can use the `--normalizer` option with the source names separated by spaces.
 
-#### Update source(s)
+#### Update all sources
 
-The sources we currently use are: OncoTree, OMIM, Disease Ontology, and Mondo.
+To update all sources, use the `--update_all` flag:
 
-The application will automatically retrieve input data for all sources but OMIM, for which a source file must be manually acquired and placed in the `disease/data/omim` folder within the library root. In order to access OMIM data, users must submit a request [here](https://www.omim.org/downloads). Once approved, the relevant OMIM file (`mimTitles.txt`) should be renamed according to the convention `omim_YYYYMMDD.tsv`, where `YYYYMMDD` indicates the date that the file was generated, and placed in the appropriate location.
+```commandline
+disease_norm_update --update_all
+```
 
-To update one source, simply set `--normalizer` to the source you wish to update. Accepted source names are `DO` (for Disease Ontology), `Mondo`, `OncoTree`, and `OMIM`.
+### Create Merged Concept Groups
+The `normalize` endpoint relies on merged concept groups.
 
-From the project root, run the following to update the Mondo source:
+To create merged concept groups, use the `--update_merged` flag with the `--update_all` flag.
 
 ```commandline
-python3 -m disease.cli --normalizer="Mondo"
+python3 -m disease.cli --update_all --update_merged
 ```
 
-To update multiple sources, you can use the `normalizer` flag with the source names separated by spaces.
+### Starting the disease normalization service
+
+Once the Disease Normalizer database has been loaded, from the project root, run the following:
 
 ```commandline
-python3 -m disease.cli --normalizer="Mondo OMIM DO"
+uvicorn disease.main:app --reload
 ```
 
-#### Update all sources
+Next, view the OpenAPI docs on your local machine:
+
+http://127.0.0.1:8000/disease
 
-To update all sources, use the `--update_all` flag.
+## Developer instructions
+Following are sections include instructions specifically for developers.
 
-From the project root, run the following to update all sources:
+### Installation
+For a development install, we recommend using Pipenv. See the
+[pipenv docs](https://pipenv-fork.readthedocs.io/en/latest/#install-pipenv-today)
+for direction on installing pipenv in your compute environment.
+
+To get started, clone the repo and initialize the environment:
 
 ```commandline
-python3 -m disease.cli --update_all
+git clone https://github.com/cancervariants/disease-normalization
+cd disease-normalization
+pipenv shell
+pipenv update
+pipenv install --dev
 ```
 
-### Create Merged Concept Groups
-The `normalize` endpoint relies on merged concept groups.
-
-To create merged concept groups, use the `--update_merged` flag with the `--update_all` flag.
+Alternatively, install the `pg`, `etl`, `dev`, and test dependency groups in a virtual environment:
 
 ```commandline
-python3 -m disease.cli --update_all --update_merged
+git clone https://github.com/cancervariants/gene-normalization
+cd gene-normalization
+python3 -m virtualenv venv
+source venv/bin/activate
+pip install -e ".[pg,etl,dev,test]"
 ```
 
-#### Specifying the database URL endpoint
+### Init coding style tests
 
-The default URL endpoint is `http://localhost:8000`.
+Code style is managed by [flake8](https://github.com/PyCQA/flake8) and checked prior to commit.
 
-There are two different ways to specify the database URL endpoint.
+We use [pre-commit](https://pre-commit.com/#usage) to run conformance tests.
 
-The first way is to set the `--db_url` flag to the URL endpoint.
+This ensures:
 
-```commandline
-python3 -m disease.cli --update_all --db_url="http://localhost:8001"
-```
+* Check code style
+* Check for added large files
+* Detect AWS Credentials
+* Detect Private Key
+
+Before first commit run:
 
-The second way is to set the `DISEASE_NORM_DB_URL` to the URL endpoint.
 ```commandline
-export DISEASE_NORM_DB_URL="http://localhost:8001"
-python3 -m disease.cli --update_all
+pre-commit install
 ```
 
-### Starting the disease normalization service
+### Running unit tests
 
-From the project root, run the following:
+Tests are provided via pytest.
 
 ```commandline
-uvicorn disease.main:app --reload
+pytest
 ```
 
-Next, view the OpenAPI docs on your local machine:
+By default, tests will employ an existing DynamoDB database. For test environments where this is unavailable (e.g. in CI), the `DISEASE_TEST` environment variable can be set to initialize a local DynamoDB instance with miniature versions of input data files before tests are executed.
 
-http://127.0.0.1:8000/disease
+```comandline
+export DISEASE_TEST=true
+pytest
+```
+
+Sometimes, sources will update their data, and our test fixtures and data will become incorrect. The `tests/scripts/` subdirectory includes scripts to rebuild data files, although most fixtures will need to be updated manually.
diff --git a/disease/__init__.py b/disease/__init__.py
index 986d204e..7c4d2942 100644
--- a/disease/__init__.py
+++ b/disease/__init__.py
@@ -16,14 +16,14 @@
 logger.handlers = []
 
 
-from disease.schemas import (  # noqa: E402
-    ItemTypes,
+from disease.schemas import (  # noqa: E402 E501
     NamespacePrefix,
+    RefType,
     SourceIDAfterNamespace,
     SourceName,
 )
 
-ITEM_TYPES = {k.lower(): v.value for k, v in ItemTypes.__members__.items()}
+ITEM_TYPES = {k.lower(): v.value for k, v in RefType.__members__.items()}
 
 # use to lookup source name from lower-case string
 # technically the same as PREFIX_LOOKUP, but source namespace prefixes
@@ -50,4 +50,4 @@
 }
 
 # Use for checking whether to pull IDs for merge group generation
-SOURCES_FOR_MERGE = {SourceName.MONDO.value}
+SOURCES_FOR_MERGE = {SourceName.MONDO}
diff --git a/disease/cli.py b/disease/cli.py
index f1cb5ce8..5d6856da 100644
--- a/disease/cli.py
+++ b/disease/cli.py
@@ -1,19 +1,18 @@
 """This module provides a CLI util to make updates to normalizer database."""
-from os import environ
+import os
+from pathlib import Path
 from timeit import default_timer as timer
-from typing import List, Optional
+from typing import Collection, List, Optional, Set
 
 import click
-from boto3.dynamodb.conditions import Key
-from botocore.exceptions import ClientError
-
-from disease import logger
-from disease.database import (
-    AWS_ENV_VAR_NAME,
-    SKIP_AWS_DB_ENV_NAME,
-    VALID_AWS_ENV_NAMES,
-    Database,
-    confirm_aws_db_use,
+
+from disease import SOURCES_FOR_MERGE, SOURCES_LOWER_LOOKUP, logger
+from disease.database.database import (
+    AbstractDatabase,
+    DatabaseException,
+    DatabaseReadException,
+    DatabaseWriteException,
+    create_db,
 )
 from disease.etl import DO, OMIM, Mondo, NCIt, OncoTree  # noqa: F401
 from disease.etl.merge import Merge
@@ -26,219 +25,273 @@
 }
 
 
-class CLI:
-    """Class for updating the normalizer database via Click"""
+@click.command()
+@click.option("--db_url", help="URL endpoint for the application database.")
+@click.option("--verbose", "-v", is_flag=True, help="Print result to console if set.")
+def check_db(db_url: str, verbose: bool = False) -> None:
+    """Perform basic checks on DB health and population. Exits with status code 1
+    if DB schema is uninitialized or if critical tables appear to be empty.
 
-    @staticmethod
-    @click.command()
-    @click.option(
-        "--normalizer", help="The source(s) you wish to update separated by spaces."
-    )
-    @click.option(
-        "--aws_instance",
-        help=" Must be `Dev`, `Staging`, or `Prod`. This determines the AWS instance to"
-        " use. `Dev` uses nonprod. `Staging` and `Prod` uses prod.",
-    )
-    @click.option("--db_url", help="URL endpoint for the application database.")
-    @click.option("--update_all", is_flag=True, help="Update all normalizer sources.")
-    @click.option(
-        "--update_merged", is_flag=True, help="Update concepts for /normalize endpoint."
-    )
-    @click.option(
-        "--from_local",
-        is_flag=True,
-        default=False,
-        help="Use most recent local source data instead of fetching latest versions.",
-    )
-    def update_normalizer_db(
-        normalizer, aws_instance, db_url, update_all, update_merged, from_local
-    ):
-        """Update selected source(s) in the Disease Normalizer database."""
-        # If SKIP_AWS_CONFIRMATION is accidentally set, we should verify that the
-        # aws instance should actually be used
-        invalid_aws_msg = f"{AWS_ENV_VAR_NAME} must be set to one of {VALID_AWS_ENV_NAMES}"  # noqa: E501
-        aws_env_name = environ.get(AWS_ENV_VAR_NAME) or aws_instance
-        if aws_env_name:
-            assert aws_env_name in VALID_AWS_ENV_NAMES, invalid_aws_msg
-            environ[AWS_ENV_VAR_NAME] = aws_env_name
-            confirm_aws_db_use(aws_env_name.upper())
-            environ[SKIP_AWS_DB_ENV_NAME] = "true"  # this is already checked above
-            db: Database = Database()
-        else:
-            if db_url:
-                endpoint_url = db_url
-            elif "DISEASE_NORM_DB_URL" in environ:
-                endpoint_url = environ["DISEASE_NORM_DB_URL"]
-            else:
-                endpoint_url = "http://localhost:8000"
-            db = Database(db_url=endpoint_url)
-
-        if update_all:
-            sources_to_update = list(src for src in SOURCES_CLASS_LOOKUP)
-            CLI()._update_sources(sources_to_update, db, update_merged, from_local)
-        elif not normalizer:
-            if update_merged:
-                CLI()._update_merged(db, [])
-            else:
-                CLI()._help_msg()
-        else:
-            sources_to_update = str(normalizer).lower().split()
-            if len(sources_to_update) == 0:
-                CLI()._help_msg()
-
-            invalid_sources = set(sources_to_update) - {
-                src for src in SOURCES_CLASS_LOOKUP
-            }
-            if len(invalid_sources) != 0:
-                raise Exception(f"Not valid sources: {invalid_sources}")
-
-            CLI()._update_sources(sources_to_update, db, update_merged, from_local)
-
-    @staticmethod
-    def _help_msg(message: Optional[str] = None):
-        """Display help message."""
-        ctx = click.get_current_context()
-        if message:
-            click.echo(message)
-        click.echo(ctx.get_help())
-        ctx.exit()
-
-    @staticmethod
-    def _update_sources(
-        sources: List[str], db: Database, update_merged: bool, from_local: bool = False
-    ):
-        """Update selected normalizer sources."""
-        added_ids = []
-        for source_name in sources:
-            msg = f"Deleting {source_name}..."
-            click.echo(f"\n{msg}")
-            logger.info(msg)
-            start_delete = timer()
-            CLI()._delete_data(source_name, db)
-            end_delete = timer()
-            delete_time = end_delete - start_delete
-            msg = f"Deleted {source_name} in {delete_time:.5f} seconds."
-            click.echo(f"{msg}\n")
-            logger.info(msg)
-
-            msg = f"Loading {source_name}..."
-            click.echo(msg)
-            logger.info(msg)
-            start_load = timer()
-            source = SOURCES_CLASS_LOOKUP[source_name](database=db)
-            if isinstance(source, Mondo):
-                added_ids = source.perform_etl(from_local)
-            else:
-                source.perform_etl()
-            end_load = timer()
-            load_time = end_load - start_load
-            msg = f"Loaded {source_name} in {load_time:.5f} seconds."
-            click.echo(msg)
-            logger.info(msg)
-            msg = f"Total time for {source_name}: {(delete_time + load_time):.5f} seconds."  # noqa: E501
-            click.echo(msg)
-            logger.info(msg)
-        if update_merged:
-            CLI()._update_merged(db, added_ids)
-
-    def _update_merged(self, db: Database, added_ids: List[str]) -> None:
-        """Update merged concepts and references.
-        :param db: Database client
-        :param added_ids: list of concept IDs to use for normalized groups.
-            Should consist solely of MONDO IDs. If empty, will be fetched.
-        """
-        start_merge = timer()
-        if not added_ids:
-            CLI()._delete_merged_data(db)
-            added_ids = db.get_ids_for_merge()
-        merge = Merge(database=db)
-        click.echo("Constructing normalized records...")
-        merge.create_merged_concepts(added_ids)
-        end_merge = timer()
+    \f
+    :param db_url: URL to normalizer database
+    :param verbose: if true, print result to console
+    """  # noqa: D301
+    db = create_db(db_url, False)
+    if not db.check_schema_initialized():
+        if verbose:
+            click.echo("Health check failed: DB schema uninitialized.")
+        click.get_current_context().exit(1)
+
+    if not db.check_tables_populated():
+        if verbose:
+            click.echo("Health check failed: DB is incompletely populated.")
+        click.get_current_context().exit(1)
+
+    if verbose:
+        click.echo("DB health check successful: tables appear complete.")
+
+
+@click.command()
+@click.option("--data_url", help="URL to data dump")
+@click.option("--db_url", help="URL endpoint for the application database.")
+def update_from_remote(data_url: Optional[str], db_url: str) -> None:
+    """Update data from remotely-hosted DB dump. By default, fetches from latest
+    available dump on VICC S3 bucket; specific URLs can be provided instead by
+    command line option or DISEASE_NORM_REMOTE_DB_URL environment variable.
+
+    \f
+    :param data_url: user-specified location to pull DB dump from
+    :param db_url: URL to normalizer database
+    """  # noqa: D301
+    if not click.confirm("Are you sure you want to overwrite existing data?"):
+        click.get_current_context().exit()
+    if not data_url:
+        data_url = os.environ.get("DISEASE_NORM_REMOTE_DB_URL")
+    db = create_db(db_url, False)
+    try:
+        db.load_from_remote(data_url)
+    except NotImplementedError:
         click.echo(
-            f"Merged concept generation completed in"
-            f" {(end_merge - start_merge):.5f} seconds."
+            "Error: Fetching remote data dump not supported for "
+            f"{db.__class__.__name__}"
         )
+        click.get_current_context().exit(1)
+    except DatabaseException as e:
+        click.echo(f"Encountered exception during update: {str(e)}")
+        click.get_current_context().exit(1)
+
+
+@click.command()
+@click.option(
+    "--output_directory",
+    "-o",
+    help="Output location to write to",
+    type=click.Path(exists=True, path_type=Path),
+)
+@click.option("--db_url", help="URL endpoint for the application database.")
+def dump_database(output_directory: Path, db_url: str):
+    """Dump data from database into file.
+
+    \f
+    :param output_directory: path to existing directory
+    :param db_url: URL to normalizer database
+    """  # noqa: D301
+    if not output_directory:
+        output_directory = Path(".")
+
+    db = create_db(db_url, False)
+    try:
+        db.export_db(output_directory)
+    except NotImplementedError:
+        click.echo(
+            f"Error: Dumping data to file not supported for {db.__class__.__name__}"
+        )
+        click.get_current_context().exit(1)
+    except DatabaseException as e:
+        click.echo(f"Encountered exception during update: {str(e)}")
+        click.get_current_context().exit(1)
+
+
+def _update_normalizers(
+    normalizers: Collection[SourceName],
+    db: AbstractDatabase,
+    update_merged: bool,
+    from_local: bool,
+) -> None:
+    """Update selected normalizer sources.
+
+    :param normalizers: names of sources to update
+    :param db: database instance
+    :param update_merged: if true, retain processed records to use in updating merged
+        records
+    :param from_local: if true, use locally available data only
+    """
+    processed_ids = list()
+    for n in normalizers:
+        delete_time = _delete_source(n, db)
+        _load_source(n, db, delete_time, processed_ids, from_local)
+
+    if update_merged:
+        _load_merge(db, set(processed_ids))
+
+
+def _delete_source(n: SourceName, db: AbstractDatabase) -> float:
+    """Delete individual source data.
+
+    :param n: name of source to delete
+    :param db: database instance
+    :return: time taken (in seconds) to delete
+    """
+    msg = f"Deleting {n.value}..."
+    click.echo(f"\n{msg}")
+    logger.info(msg)
+    start_delete = timer()
+    db.delete_source(n)
+    end_delete = timer()
+    delete_time = end_delete - start_delete
+    msg = f"Deleted {n.value} in {delete_time:.5f} seconds."
+    click.echo(f"{msg}\n")
+    logger.info(msg)
+    return delete_time
+
+
+def _load_source(
+    n: SourceName,
+    db: AbstractDatabase,
+    delete_time: float,
+    processed_ids: List[str],
+    from_local: bool,
+) -> None:
+    """Load individual source data.
+
+    :param n: name of source
+    :param db: database instance
+    :param delete_time: time taken (in seconds) to run deletion
+    :param processed_ids: in-progress list of processed disease IDs
+    :param from_local: if true, use locally available data
+    """
+    msg = f"Loading {n.value}..."
+    click.echo(msg)
+    logger.info(msg)
+    start_load = timer()
+
+    # used to get source class name from string
+    SourceClass = eval(n.value)
+
+    source = SourceClass(database=db)
+    processed_ids += source.perform_etl(use_existing=from_local)
+    end_load = timer()
+    load_time = end_load - start_load
+    msg = f"Loaded {n.value} in {load_time:.5f} seconds."
+    click.echo(msg)
+    logger.info(msg)
+    msg = f"Total time for {n.value}: {(delete_time + load_time):.5f} seconds."
+    click.echo(msg)
+    logger.info(msg)
+
+
+def _delete_normalized_data(database: AbstractDatabase) -> None:
+    """Delete normalized concepts
+
+    :param database: DB instance
+    """
+    click.echo("\nDeleting normalized records...")
+    start_delete = timer()
+    try:
+        database.delete_normalized_concepts()
+    except (DatabaseReadException, DatabaseWriteException) as e:
+        click.echo(f"Encountered exception during normalized data deletion: {e}")
+        click.get_current_context().exit(1)
+    end_delete = timer()
+    delete_time = end_delete - start_delete
+    click.echo(f"Deleted normalized records in {delete_time:.5f} seconds.")
+
+
+def _load_merge(db: AbstractDatabase, processed_ids: Set[str]) -> None:
+    """Load merged concepts
+
+    :param db: database instance
+    :param processed_ids: in-progress list of processed disease IDs
+    """
+    start = timer()
+    _delete_normalized_data(db)
+    if not processed_ids:
+        processed_ids = set()
+        for source in SOURCES_FOR_MERGE:
+            processed_ids |= db.get_all_concept_ids(source)
+    merge = Merge(database=db)
+    click.echo("Constructing normalized records...")
+    merge.create_merged_concepts(processed_ids)
+    end = timer()
+    click.echo(
+        f"Merged concept generation completed in " f"{(end - start):.5f} seconds"
+    )
+
+
+@click.command()
+@click.option(
+    "--normalizer", help="The normalizer(s) you wish to update separated by spaces."
+)
+@click.option("--aws_instance", is_flag=True, help="Using AWS DynamodDB instance.")
+@click.option("--db_url", help="URL endpoint for the application database.")
+@click.option("--update_all", is_flag=True, help="Update all normalizer sources.")
+@click.option(
+    "--update_merged",
+    is_flag=True,
+    help="Update concepts for normalize endpoint from accepted sources.",
+)
+@click.option(
+    "--from_local",
+    is_flag=True,
+    default=False,
+    help="Use most recent local source data instead of fetching latest versions.",
+)
+def update_normalizer_db(
+    normalizer: str,
+    aws_instance: bool,
+    db_url: str,
+    update_all: bool,
+    update_merged: bool,
+    from_local: bool,
+) -> None:
+    """Update selected normalizer source(s) in the disease database.
+
+    \f
+    :param normalizer: names of sources to update, comma-separated
+    :param aws_instance: if true, use cloud instance
+    :param db_url: URI pointing to database
+    :param update_all: if true, update all sources (ignore `normalizer` parameter)
+    :param update_merged: if true, update normalized records
+    :param from_local: if true, use locally available data only
+    """  # noqa: D301
+    db = create_db(db_url, aws_instance)
+
+    if update_all:
+        _update_normalizers(list(SourceName), db, update_merged, from_local)
+    elif not normalizer:
+        if update_merged:
+            _load_merge(db, set())
+        else:
+            ctx = click.get_current_context()
+            click.echo(
+                "Must either enter 1 or more sources, or use `--update_all` parameter"
+            )  # noqa: E501
+            click.echo(ctx.get_help())
+            ctx.exit()
+    else:
+        normalizers = normalizer.lower().split()
+
+        if len(normalizers) == 0:
+            raise Exception("Must enter a normalizer")
+
+        non_sources = set(normalizers) - set(SOURCES_LOWER_LOOKUP)
+
+        if len(non_sources) != 0:
+            raise Exception(f"Not valid source(s): {non_sources}")
 
-    @staticmethod
-    def _delete_merged_data(database: Database):
-        """Delete data pertaining to merged records.
-        :param database: DynamoDB client
-        """
-        click.echo("\nDeleting normalized records...")
-        start_delete = timer()
-        try:
-            while True:
-                with database.diseases.batch_writer(
-                    overwrite_by_pkeys=["label_and_type", "concept_id"]
-                ) as batch:
-                    response = database.diseases.query(
-                        IndexName="type_index",
-                        KeyConditionExpression=Key("item_type").eq("merger"),
-                    )
-                    records = response["Items"]
-                    if not records:
-                        break
-                    for record in records:
-                        batch.delete_item(
-                            Key={
-                                "label_and_type": record["label_and_type"],
-                                "concept_id": record["concept_id"],
-                            }
-                        )
-        except ClientError as e:
-            click.echo(e.response["Error"]["Message"])
-        end_delete = timer()
-        delete_time = end_delete - start_delete
-        click.echo(f"Deleted normalized records in {delete_time:.5f} seconds.")
-
-    @staticmethod
-    def _delete_data(source: str, database: Database):
-        # Delete source's metadata
-        try:
-            metadata = database.metadata.query(
-                KeyConditionExpression=Key("src_name").eq(
-                    SourceName[f"{source.upper()}"].value
-                )
-            )
-            if metadata["Items"]:
-                database.metadata.delete_item(
-                    Key={"src_name": metadata["Items"][0]["src_name"]},
-                    ConditionExpression="src_name = :src",
-                    ExpressionAttributeValues={
-                        ":src": SourceName[f"{source.upper()}"].value
-                    },
-                )
-        except ClientError as e:
-            click.echo(e.response["Error"]["Message"])
-
-        # Delete source's data from diseases table
-        try:
-            while True:
-                response = database.diseases.query(
-                    IndexName="src_index",
-                    KeyConditionExpression=Key("src_name").eq(
-                        SourceName[f"{source.upper()}"].value
-                    ),
-                )
-
-                records = response["Items"]
-                if not records:
-                    break
-
-                with database.diseases.batch_writer(
-                    overwrite_by_pkeys=["label_and_type", "concept_id"]
-                ) as batch:
-                    for record in records:
-                        batch.delete_item(
-                            Key={
-                                "label_and_type": record["label_and_type"],
-                                "concept_id": record["concept_id"],
-                            }
-                        )
-        except ClientError as e:
-            click.echo(e.response["Error"]["Message"])
+        sources_to_update = {SourceName(SOURCES_LOWER_LOOKUP[s]) for s in normalizers}
+        _update_normalizers(sources_to_update, db, update_merged, from_local)
 
 
 if __name__ == "__main__":
-    CLI().update_normalizer_db()
+    update_normalizer_db()
diff --git a/disease/database/__init__.py b/disease/database/__init__.py
new file mode 100644
index 00000000..5d40355d
--- /dev/null
+++ b/disease/database/__init__.py
@@ -0,0 +1,10 @@
+"""Provide database clients."""
+from .database import AWS_ENV_VAR_NAME  # noqa: F401
+from .database import (
+    AbstractDatabase,
+    DatabaseException,
+    DatabaseInitializationException,
+    DatabaseReadException,
+    DatabaseWriteException,
+    create_db,
+)
diff --git a/disease/database/database.py b/disease/database/database.py
new file mode 100644
index 00000000..bcbc4423
--- /dev/null
+++ b/disease/database/database.py
@@ -0,0 +1,290 @@
+"""Provide core database classes and parameters."""
+import abc
+import sys
+from enum import Enum
+from os import environ
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Union
+
+import click
+
+from disease.schemas import RefType, SourceMeta, SourceName
+
+
+class DatabaseException(Exception):
+    """Create custom class for handling database exceptions"""
+
+
+class DatabaseInitializationException(DatabaseException):
+    """Create custom exception for errors during DB connection initialization."""
+
+
+class DatabaseReadException(DatabaseException):
+    """Create custom exception for lookup/read errors"""
+
+
+class DatabaseWriteException(DatabaseException):
+    """Create custom exception for write errors"""
+
+
+class AbstractDatabase(abc.ABC):
+    """Define a database interface."""
+
+    @abc.abstractmethod
+    def __init__(self, db_url: Optional[str] = None, **db_args):
+        """Initialize database instance.
+
+        Generally, implementing classes should be able to construct a connection by
+        something like a libpq URL. Any additional arguments or DB-specific parameters
+        can be passed as keywords.
+
+        :param db_url: address/connection description for database
+        :param db_args: any DB implementation-specific parameters
+        :raise DatabaseInitializationException: if initial setup fails
+        """
+
+    @abc.abstractmethod
+    def list_tables(self) -> List[str]:
+        """Return names of tables in database.
+
+        :return: Table names in database
+        """
+
+    @staticmethod
+    def _check_delete_okay() -> bool:
+        """Check that environmental conditions permit DB deletion, and require
+        confirmation.
+
+        :raise DatabaseWriteException: if skip confirmation variable is set -- manual
+        approval is required.
+        """
+        if environ.get(AWS_ENV_VAR_NAME, "") == AwsEnvName.PRODUCTION:
+            if environ.get(SKIP_AWS_DB_ENV_NAME, "") == "true":
+                raise DatabaseWriteException(
+                    f"Must unset {SKIP_AWS_DB_ENV_NAME} env variable to enable drop_db()"  # noqa: E501
+                )
+            return click.confirm("Are you sure you want to delete existing data?")
+        else:
+            return True
+
+    @abc.abstractmethod
+    def drop_db(self) -> None:
+        """Initiate total teardown of DB. Useful for quickly resetting the entirety of
+        the data. Requires manual confirmation.
+
+        :raise DatabaseWriteException: if called in a protected setting with
+            confirmation silenced.
+        """
+
+    @abc.abstractmethod
+    def check_schema_initialized(self) -> bool:
+        """Check if database schema is properly initialized.
+
+        :return: True if DB appears to be fully initialized, False otherwise
+        """
+
+    @abc.abstractmethod
+    def check_tables_populated(self) -> bool:
+        """Perform rudimentary checks to see if tables are populated.
+        Emphasis is on rudimentary -- if some fiendish element has deleted half of the
+        disease aliases, this method won't pick it up. It just wants to see if a few
+        critical tables have at least a small number of records.
+
+        :return: True if queries successful, false if DB appears empty
+        """
+
+    @abc.abstractmethod
+    def initialize_db(self) -> None:
+        """Perform all necessary parts of database setup. Should be tolerant of
+        existing content -- ie, this method is also responsible for checking whether
+        the DB is already set up.
+
+        :raise DatabaseInitializationException: if initialization fails
+        """
+
+    @abc.abstractmethod
+    def get_source_metadata(self, src_name: Union[str, SourceName]) -> Optional[Dict]:
+        """Get license, versioning, data lookup, etc information for a source.
+
+        :param src_name: name of the source to get data for
+        :return: Dict containing metadata if lookup is successful
+        """
+
+    @abc.abstractmethod
+    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 concept_id: concept ID for record
+        :param case_sensitive: if true, performs exact lookup, which may be quicker.
+            Otherwise, performs filter operation, which doesn't require correct casing.
+        :param merge: if true, look for merged record; look for identity
+            record otherwise.
+        :return: complete record, if match is found; None otherwise
+        """
+
+    @abc.abstractmethod
+    def get_refs_by_type(self, search_term: str, ref_type: RefType) -> List[str]:
+        """Retrieve concept IDs for records matching the user's query. Other methods
+        are responsible for actually retrieving full records.
+
+        :param search_term: string to match against
+        :param ref_type: type of match to look for.
+        :return: list of associated concept IDs. Empty if lookup fails.
+        """
+
+    @abc.abstractmethod
+    def get_all_concept_ids(self, source: Optional[SourceName] = None) -> Set[str]:
+        """Retrieve all available concept IDs for use in generating normalized records.
+
+        :param source: optionally, just get all IDs for a specific source
+        :return: Set of concept IDs as strings.
+        """
+
+    @abc.abstractmethod
+    def add_source_metadata(self, src_name: SourceName, data: SourceMeta) -> None:
+        """Add new source metadata entry.
+
+        :param src_name: name of source
+        :param data: known source attributes
+        :raise DatabaseWriteException: if write fails
+        """
+
+    @abc.abstractmethod
+    def add_record(self, record: Dict, src_name: SourceName) -> None:
+        """Add new record to database.
+
+        :param record: record to upload
+        :param src_name: name of source for record.
+        """
+
+    @abc.abstractmethod
+    def add_merged_record(self, record: Dict) -> None:
+        """Add merged record to database.
+
+        :param record: merged record to add
+        """
+
+    @abc.abstractmethod
+    def update_merge_ref(self, concept_id: str, merge_ref: Any) -> None:
+        """Update the merged record reference of an individual record to a new value.
+
+        :param concept_id: record to update
+        :param merge_ref: new ref value
+        :raise DatabaseWriteException: if attempting to update non-existent record
+        """
+
+    @abc.abstractmethod
+    def delete_normalized_concepts(self) -> None:
+        """Remove merged records from the database. Use when performing a new update
+        of normalized data.
+
+        :raise DatabaseReadException: if DB client requires separate read calls and
+            encounters a failure in the process
+        :raise DatabaseWriteException: if deletion call fails
+        """
+
+    @abc.abstractmethod
+    def delete_source(self, src_name: SourceName) -> None:
+        """Delete all data for a source. Use when updating source data.
+
+        :param src_name: name of source to delete
+        :raise DatabaseReadException: if DB client requires separate read calls and
+            encounters a failure in the process
+        :raise DatabaseWriteException: if deletion call fails
+        """
+
+    @abc.abstractmethod
+    def complete_write_transaction(self) -> None:
+        """Conclude transaction or batch writing if relevant."""
+
+    @abc.abstractmethod
+    def close_connection(self) -> None:
+        """Perform any manual connection closure procedures if necessary."""
+
+    @abc.abstractmethod
+    def load_from_remote(self, url: Optional[str] = None) -> None:
+        """Load DB from remote dump. Warning: Deletes all existing data.
+
+        :param url: remote location to retrieve gzipped dump file from
+        :raise: NotImplementedError if not supported by DB
+        """
+
+    @abc.abstractmethod
+    def export_db(self, export_location: Path) -> None:
+        """Dump DB to specified location.
+
+        :param export_location: path to save DB dump at
+        :raise: NotImplementedError if not supported by DB
+        """
+
+
+# can be set to either `Dev`, `Staging`, or `Prod`
+# ONLY set when wanting to access aws instance
+AWS_ENV_VAR_NAME = "DISEASE_NORM_ENV"
+
+# Set to "true" if want to skip db confirmation check. Should ONLY be used for
+# deployment needs
+SKIP_AWS_DB_ENV_NAME = "SKIP_AWS_CONFIRMATION"
+
+
+class AwsEnvName(str, Enum):
+    """AWS environment name that is being used"""
+
+    DEVELOPMENT = "Dev"
+    STAGING = "Staging"
+    PRODUCTION = "Prod"
+
+
+VALID_AWS_ENV_NAMES = {v.value for v in AwsEnvName.__members__.values()}
+
+
+def confirm_aws_db_use(env_name: str) -> None:
+    """Check to ensure that AWS instance should actually be used."""
+    if click.confirm(
+        f"Are you sure you want to use the AWS {env_name} database?", default=False
+    ):
+        click.echo(f"***DISEASE AWS {env_name.upper()} DATABASE IN USE***")
+    else:
+        click.echo("Exiting.")
+        sys.exit()
+
+
+def create_db(
+    db_url: Optional[str] = None, aws_instance: bool = False
+) -> AbstractDatabase:
+    """Database factory method. Checks environment variables and provided parameters
+    and creates a DB instance.
+
+    Generally prefers to return a DynamoDB instance, unless all DDB-relevant
+    environment variables are unset and a libpq-compliant URI is passed to
+    `db_url`.
+
+    :param db_url: address to database instance
+    :param aws_instance: use hosted DynamoDB instance, not local DB
+    :return: constructed Database instance
+    """
+    aws_env_var_set = AWS_ENV_VAR_NAME in environ
+    if aws_env_var_set or aws_instance:
+        from disease.database.dynamodb import DynamoDbDatabase
+
+        db = DynamoDbDatabase()
+    else:
+        if db_url:
+            endpoint_url = db_url
+        elif "DISEASE_NORM_DB_URL" in environ.keys():
+            endpoint_url = environ["DISEASE_NORM_DB_URL"]
+        else:
+            endpoint_url = "http://localhost:8000"
+
+        # prefer DynamoDB unless connection explicitly reads like a libpq URI
+        if endpoint_url.startswith("postgres"):
+            from disease.database.postgresql import PostgresDatabase
+
+            db = PostgresDatabase(endpoint_url)
+        else:
+            from disease.database.dynamodb import DynamoDbDatabase
+
+            db = DynamoDbDatabase(endpoint_url)
+    return db
diff --git a/disease/database/dynamodb.py b/disease/database/dynamodb.py
new file mode 100644
index 00000000..a54c7cfc
--- /dev/null
+++ b/disease/database/dynamodb.py
@@ -0,0 +1,536 @@
+"""Provide DynamoDB client."""
+import atexit
+import logging
+import sys
+from os import environ
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Union
+
+import boto3
+import click
+from boto3.dynamodb.conditions import Attr, Equals, Key
+from botocore.exceptions import ClientError
+
+from disease import ITEM_TYPES, PREFIX_LOOKUP
+from disease.database.database import (
+    AWS_ENV_VAR_NAME,
+    SKIP_AWS_DB_ENV_NAME,
+    VALID_AWS_ENV_NAMES,
+    AbstractDatabase,
+    AwsEnvName,
+    DatabaseException,
+    DatabaseReadException,
+    DatabaseWriteException,
+    confirm_aws_db_use,
+)
+from disease.schemas import RefType, SourceMeta, SourceName
+
+_logger = logging.getLogger()
+_logger.setLevel(logging.DEBUG)
+
+
+class DynamoDbDatabase(AbstractDatabase):
+    """Database class employing DynamoDB."""
+
+    def __init__(self, db_url: Optional[str] = None, **db_args):
+        """Initialize Database class.
+
+        :param str db_url: URL endpoint for DynamoDB source
+        :Keyword Arguments:
+            * region_name: AWS region (defaults to "us-east-2")
+        """
+        disease_concepts_table = "disease_concepts"  # default
+        disease_metadata_table = "disease_metadata"  # default
+
+        region_name = db_args.get("region_name", "us-east-2")
+
+        if AWS_ENV_VAR_NAME in environ:
+            if "DISEASE_TEST" in environ:
+                raise DatabaseException(
+                    f"Cannot have both DISEASE_TEST and {AWS_ENV_VAR_NAME} set."
+                )  # noqa: E501
+
+            aws_env = environ[AWS_ENV_VAR_NAME]
+            if aws_env not in VALID_AWS_ENV_NAMES:
+                raise DatabaseException(
+                    f"{AWS_ENV_VAR_NAME} must be one of {VALID_AWS_ENV_NAMES}"
+                )  # noqa: E501
+
+            skip_confirmation = environ.get(SKIP_AWS_DB_ENV_NAME)
+            if (not skip_confirmation) or (
+                skip_confirmation and skip_confirmation != "true"
+            ):  # noqa: E501
+                confirm_aws_db_use(environ[AWS_ENV_VAR_NAME])
+
+            boto_params = {"region_name": region_name}
+
+            if aws_env == AwsEnvName.DEVELOPMENT:
+                disease_concepts_table = "disease_concepts_nonprod"
+                disease_metadata_table = "disease_metadata_nonprod"
+        else:
+            if db_url:
+                endpoint_url = db_url
+            elif "DISEASE_NORM_DB_URL" in environ:
+                endpoint_url = environ["DISEASE_NORM_DB_URL"]
+            else:
+                endpoint_url = "http://localhost:8000"
+            click.echo(f"***Using Disease Database Endpoint: {endpoint_url}***")
+            boto_params = {"region_name": region_name, "endpoint_url": endpoint_url}
+
+        self.dynamodb = boto3.resource("dynamodb", **boto_params)
+        self.dynamodb_client = boto3.client("dynamodb", **boto_params)
+
+        # Only create tables for local instance
+        envs_do_not_create_tables = {AWS_ENV_VAR_NAME, "DISEASE_TEST"}
+        if not set(envs_do_not_create_tables) & set(environ):
+            self.initialize_db()
+
+        self.diseases = self.dynamodb.Table(disease_concepts_table)
+        self.metadata = self.dynamodb.Table(disease_metadata_table)
+        self.batch = self.diseases.batch_writer()
+        self._cached_sources = {}
+        atexit.register(self.close_connection)
+
+    def list_tables(self) -> List[str]:
+        """Return names of tables in database.
+
+        :return: Table names in DynamoDB
+        """
+        return self.dynamodb_client.list_tables()["TableNames"]
+
+    def drop_db(self) -> None:
+        """Delete all tables from database. Requires manual confirmation.
+
+        :raise DatabaseWriteException: if called in a protected setting with
+            confirmation silenced.
+        """
+        try:
+            if not self._check_delete_okay():
+                return
+        except DatabaseWriteException as e:
+            raise e
+
+        existing_tables = self.list_tables()
+        for table_name in existing_tables:
+            self.dynamodb.Table(table_name).delete()
+
+    def _create_diseases_table(self):
+        """Create Diseases table."""
+        self.dynamodb.create_table(
+            TableName="disease_concepts",
+            KeySchema=[
+                {"AttributeName": "label_and_type", "KeyType": "HASH"},  # Partition key
+                {"AttributeName": "concept_id", "KeyType": "RANGE"},  # Sort key
+            ],
+            AttributeDefinitions=[
+                {"AttributeName": "label_and_type", "AttributeType": "S"},
+                {"AttributeName": "concept_id", "AttributeType": "S"},
+                {"AttributeName": "src_name", "AttributeType": "S"},
+                {"AttributeName": "item_type", "AttributeType": "S"},
+            ],
+            GlobalSecondaryIndexes=[
+                {
+                    "IndexName": "src_index",
+                    "KeySchema": [{"AttributeName": "src_name", "KeyType": "HASH"}],
+                    "Projection": {"ProjectionType": "KEYS_ONLY"},
+                    "ProvisionedThroughput": {
+                        "ReadCapacityUnits": 10,
+                        "WriteCapacityUnits": 10,
+                    },
+                },
+                {
+                    "IndexName": "item_type_index",
+                    "KeySchema": [{"AttributeName": "item_type", "KeyType": "HASH"}],
+                    "Projection": {"ProjectionType": "KEYS_ONLY"},
+                    "ProvisionedThroughput": {
+                        "ReadCapacityUnits": 10,
+                        "WriteCapacityUnits": 10,
+                    },
+                },
+            ],
+            ProvisionedThroughput={"ReadCapacityUnits": 10, "WriteCapacityUnits": 10},
+        )
+
+    def _create_meta_data_table(self) -> None:
+        """Create MetaData table."""
+        self.dynamodb.create_table(
+            TableName="disease_metadata",
+            KeySchema=[
+                {"AttributeName": "src_name", "KeyType": "HASH"}  # Partition key
+            ],
+            AttributeDefinitions=[
+                {"AttributeName": "src_name", "AttributeType": "S"},
+            ],
+            ProvisionedThroughput={"ReadCapacityUnits": 10, "WriteCapacityUnits": 10},
+        )
+
+    def check_schema_initialized(self) -> bool:
+        """Check if database schema is properly initialized.
+
+        :return: True if DB appears to be fully initialized, False otherwise
+        """
+        existing_tables = self.list_tables()
+        exists = (
+            "disease_concepts" in existing_tables
+            and "disease_metadata" in existing_tables
+        )
+        if not exists:
+            _logger.info("Disease tables are missing or unavailable.")
+        return exists
+
+    def check_tables_populated(self) -> bool:
+        """Perform rudimentary checks to see if tables are populated.
+        Emphasis is on rudimentary -- if some fiendish element has deleted half of the
+        disease aliases, this method won't pick it up. It just wants to see if a few
+        critical tables have at least a small number of records.
+
+        :return: True if queries successful, false if DB appears empty
+        """
+        sources = self.metadata.scan().get("Items", [])
+        if len(sources) < len(SourceName):
+            _logger.info("Disease sources table is missing expected sources.")
+            return False
+
+        records = self.diseases.query(
+            IndexName="item_type_index",
+            KeyConditionExpression=Key("item_type").eq("identity"),
+            Limit=1,
+        )
+        if len(records.get("Items", [])) < 1:
+            _logger.info("Disease records index is empty.")
+            return False
+
+        normalized_records = self.diseases.query(
+            IndexName="item_type_index",
+            KeyConditionExpression=Key("item_type").eq("merger"),
+            Limit=1,
+        )
+        if len(normalized_records.get("Items", [])) < 1:
+            _logger.info("Normalized disease records index is empty.")
+            return False
+
+        return True
+
+    def initialize_db(self) -> None:
+        """Create disease_concepts and disease_metadata tables if needed."""
+        if not self.check_schema_initialized():
+            self._create_diseases_table()
+            self._create_meta_data_table()
+
+    def get_source_metadata(self, src_name: Union[str, SourceName]) -> Optional[Dict]:
+        """Get license, versioning, data lookup, etc information for a source.
+
+        :param src_name: name of the source to get data for
+        :return: Dict containing metadata if lookup is successful
+        """
+        if isinstance(src_name, SourceName):
+            src_name = src_name.value
+        if src_name in self._cached_sources:
+            return self._cached_sources[src_name]
+        else:
+            metadata = self.metadata.get_item(Key={"src_name": src_name}).get("Item")
+            self._cached_sources[src_name] = metadata
+            return metadata
+
+    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 disease 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 record; look for identity record
+        otherwise.
+        :return: complete 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:
+                response = self.diseases.get_item(
+                    Key={"label_and_type": pk, "concept_id": concept_id}
+                )
+                record = response.get("Item")
+                if not record:
+                    return None
+            else:
+                exp = Key("label_and_type").eq(pk)
+                response = self.diseases.query(KeyConditionExpression=exp)
+                record = response["Items"][0]
+            del record["label_and_type"]
+            return record
+        except ClientError as e:
+            _logger.error(
+                f"boto3 client error on get_records_by_id for "
+                f"search term {concept_id}: "
+                f"{e.response['Error']['Message']}"
+            )
+            return None
+        except (KeyError, IndexError):  # record doesn't exist
+            return None
+
+    def get_refs_by_type(self, search_term: str, ref_type: RefType) -> List[str]:
+        """Retrieve concept IDs for records matching the user's query. Other methods
+        are responsible for actually retrieving full records.
+
+        :param search_term: string to match against
+        :param ref_type: type of match to look for.
+        :return: list of associated concept IDs. Empty if lookup fails.
+        """
+        pk = f"{search_term}##{ref_type.value.lower()}"
+        filter_exp = Key("label_and_type").eq(pk)
+        try:
+            matches = self.diseases.query(KeyConditionExpression=filter_exp)
+            return [m["concept_id"] for m in matches.get("Items", None)]
+        except ClientError as e:
+            _logger.error(
+                f"boto3 client error on get_refs_by_type for "
+                f"search term {search_term}: "
+                f"{e.response['Error']['Message']}"
+            )
+            return []
+
+    def get_all_concept_ids(self, source: Optional[SourceName] = None) -> Set[str]:
+        """Retrieve concept IDs for use in generating normalized records.
+
+        :param source: optionally, just get all IDs for a specific source
+        :return: Set of concept IDs as strings.
+        """
+        last_evaluated_key = None
+        concept_ids = []
+        params: Dict[str, Union[str, Equals]] = {
+            "ProjectionExpression": "concept_id",
+        }
+        if source is not None:
+            params["FilterExpression"] = Attr("src_name").eq(source.value)
+        while True:
+            if last_evaluated_key:
+                response = self.diseases.scan(
+                    ExclusiveStartKey=last_evaluated_key, **params
+                )
+            else:
+                response = self.diseases.scan(**params)
+            records = response["Items"]
+            for record in records:
+                concept_ids.append(record["concept_id"])
+            last_evaluated_key = response.get("LastEvaluatedKey")
+            if not last_evaluated_key:
+                break
+        return set(concept_ids)
+
+    def add_source_metadata(self, src_name: SourceName, metadata: SourceMeta) -> None:
+        """Add new source metadata entry.
+
+        :param src_name: name of source
+        :param data: known source attributes
+        :raise DatabaseWriteException: if write fails
+        """
+        metadata_item = metadata.dict()
+        metadata_item["src_name"] = src_name.value
+        try:
+            self.metadata.put_item(Item=metadata_item)
+        except ClientError as e:
+            raise DatabaseWriteException(e)
+
+    def add_record(self, record: Dict, src_name: SourceName) -> None:
+        """Add new record to database.
+
+        :param Dict record: record to upload
+        :param SourceName src_name: name of source for record
+        """
+        concept_id = record["concept_id"]
+        record["src_name"] = src_name.value
+        label_and_type = f"{concept_id.lower()}##identity"
+        record["label_and_type"] = label_and_type
+        record["item_type"] = "identity"
+        try:
+            self.batch.put_item(Item=record)
+        except ClientError as e:
+            _logger.error(
+                "boto3 client error on add_record for "
+                f"{concept_id}: {e.response['Error']['Message']}"
+            )
+        for attr_type, item_type in ITEM_TYPES.items():
+            if attr_type in record:
+                value = record.get(attr_type)
+                if not value:
+                    continue
+                if isinstance(value, str):
+                    items = [value.lower()]
+                else:
+                    items = {item.lower() for item in value}
+                for item in items:
+                    self._add_ref_record(
+                        item, record["concept_id"], item_type, src_name
+                    )
+
+    def add_merged_record(self, record: Dict) -> None:
+        """Add merged record to database.
+
+        :param record: merged record to add
+        """
+        concept_id = record["concept_id"]
+        id_prefix = concept_id.split(":")[0].lower()
+        record["src_name"] = PREFIX_LOOKUP[id_prefix]
+        label_and_type = f"{concept_id.lower()}##merger"
+        record["label_and_type"] = label_and_type
+        record["item_type"] = "merger"
+        try:
+            self.batch.put_item(Item=record)
+        except ClientError as e:
+            _logger.error(
+                "boto3 client error on add_record for "
+                f"{concept_id}: {e.response['Error']['Message']}"
+            )
+
+    def _add_ref_record(
+        self, term: str, concept_id: str, ref_type: str, src_name: SourceName
+    ) -> None:
+        """Add auxiliary/reference record to database.
+
+        :param str term: referent term
+        :param str concept_id: concept ID to refer to
+        :param str ref_type: one of {'alias', 'label', 'xref',
+            'associated_with'}
+        :param src_name: name of source for record
+        """
+        label_and_type = f"{term.lower()}##{ref_type}"
+        record = {
+            "label_and_type": label_and_type,
+            "concept_id": concept_id.lower(),
+            "src_name": src_name.value,
+            "item_type": ref_type,
+        }
+        try:
+            self.batch.put_item(Item=record)
+        except ClientError as e:
+            _logger.error(
+                f"boto3 client error adding reference {term} for "
+                f"{concept_id} with match type {ref_type}: "
+                f"{e.response['Error']['Message']}"
+            )
+
+    def update_merge_ref(self, concept_id: str, merge_ref: Any) -> None:
+        """Update the merged record reference of an individual record to a new value.
+
+        :param concept_id: record to update
+        :param merge_ref: new ref value
+        :raise DatabaseWriteException: if attempting to update non-existent record
+        """
+        label_and_type = f"{concept_id.lower()}##identity"
+        key = {"label_and_type": label_and_type, "concept_id": concept_id}
+        update_expression = "set merge_ref=:r"
+        update_values = {":r": merge_ref.lower()}
+        condition_expression = "attribute_exists(label_and_type)"
+        try:
+            self.diseases.update_item(
+                Key=key,
+                UpdateExpression=update_expression,
+                ExpressionAttributeValues=update_values,
+                ConditionExpression=condition_expression,
+            )
+        except ClientError as e:
+            code = e.response.get("Error", {}).get("Code")
+            if code == "ConditionalCheckFailedException":
+                raise DatabaseWriteException(
+                    f"No such record exists for keys {label_and_type}, {concept_id}"
+                )
+            else:
+                _logger.error(
+                    f"boto3 client error in `database.update_record()`: "
+                    f"{e.response['Error']['Message']}"
+                )
+
+    def delete_normalized_concepts(self) -> None:
+        """Remove merged records from the database. Use when performing a new update
+        of normalized data.
+
+        :raise DatabaseReadException: if DB client requires separate read calls and
+            encounters a failure in the process
+        :raise DatabaseWriteException: if deletion call fails
+        """
+        while True:
+            with self.diseases.batch_writer(
+                overwrite_by_pkeys=["label_and_type", "concept_id"]
+            ) as batch:
+                try:
+                    response = self.diseases.query(
+                        IndexName="item_type_index",
+                        KeyConditionExpression=Key("item_type").eq("merger"),
+                    )
+                except ClientError as e:
+                    raise DatabaseReadException(e)
+                records = response["Items"]
+                if not records:
+                    break
+                for record in records:
+                    batch.delete_item(
+                        Key={
+                            "label_and_type": record["label_and_type"],
+                            "concept_id": record["concept_id"],
+                        }
+                    )
+
+    def delete_source(self, src_name: SourceName) -> None:
+        """Delete all data for a source. Use when updating source data.
+
+        :param src_name: name of source to delete
+        :raise DatabaseReadException: if DB client requires separate read calls and
+            encounters a failure in the process
+        :raise DatabaseWriteException: if deletion call fails
+        """
+        while True:
+            try:
+                response = self.diseases.query(
+                    IndexName="src_index",
+                    KeyConditionExpression=Key("src_name").eq(src_name.value),
+                )
+            except ClientError as e:
+                raise DatabaseReadException(e)
+            records = response["Items"]
+            if not records:
+                break
+            with self.diseases.batch_writer(
+                overwrite_by_pkeys=["label_and_type", "concept_id"]
+            ) as batch:
+                for record in records:
+                    try:
+                        batch.delete_item(
+                            Key={
+                                "label_and_type": record["label_and_type"],
+                                "concept_id": record["concept_id"],
+                            }
+                        )
+                    except ClientError as e:
+                        raise DatabaseWriteException(e)
+
+        try:
+            self.metadata.delete_item(Key={"src_name": src_name.value})
+        except ClientError as e:
+            raise DatabaseWriteException(e)
+
+    def complete_write_transaction(self) -> None:
+        """Conclude transaction or batch writing if relevant."""
+        self.batch.__exit__(*sys.exc_info())
+        self.batch = self.diseases.batch_writer()
+
+    def close_connection(self) -> None:
+        """Perform any manual connection closure procedures if necessary."""
+        self.batch.__exit__(*sys.exc_info())
+
+    def load_from_remote(self, url: Optional[str] = None) -> None:
+        """Load DB from remote dump. Not available for DynamoDB database backend.
+
+        :param url: remote location to retrieve gzipped dump file from
+        """
+        raise NotImplementedError
+
+    def export_db(self, export_location: Path) -> None:
+        """Dump DB to specified location. Not available for DynamoDB database backend.
+
+        :param export_location: path to save DB dump at
+        """
+        raise NotImplementedError
diff --git a/disease/database/postgresql.py b/disease/database/postgresql.py
new file mode 100644
index 00000000..e15d11d8
--- /dev/null
+++ b/disease/database/postgresql.py
@@ -0,0 +1,711 @@
+"""Provide PostgreSQL client."""
+import atexit
+import logging
+import os
+import tarfile
+import tempfile
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Union
+
+import psycopg
+import requests
+from psycopg.errors import (
+    DuplicateObject,
+    DuplicateTable,
+    UndefinedTable,
+    UniqueViolation,
+)
+
+from disease.database import AbstractDatabase, DatabaseException, DatabaseWriteException
+from disease.schemas import RefType, SourceMeta, SourceName
+
+logger = logging.getLogger()
+
+
+SCRIPTS_DIR = Path(__file__).parent / "postgresql"
+
+
+class PostgresDatabase(AbstractDatabase):
+    """Database class employing PostgreSQL."""
+
+    def __init__(self, db_url: Optional[str] = None, **db_args) -> None:
+        """Initialize Postgres connection.
+
+        :param db_url: libpq compliant database connection URI
+        :param **db_args: see below
+
+        :Keyword Arguments:
+            * user: Postgres username
+            * password: Postgres password (optional or blank if unneeded)
+            * db_name: name of database to connect to
+
+        :raise DatabaseInitializationException: if initial setup fails
+        """
+        if db_url:
+            conninfo = db_url
+        elif "DISEASE_NORM_DB_URL" in os.environ:
+            conninfo = os.environ["DISEASE_NORM_DB_URL"]
+        else:
+            user = db_args.get("user", "postgres")
+            password = db_args.get("password", "")
+            db_name = db_args.get("db_name", "disease_normalizer")
+            if password:
+                conninfo = f"dbname={db_name} user={user} password={password}"
+            else:
+                conninfo = f"dbname={db_name} user={user}"
+
+        self.conn = psycopg.connect(conninfo)
+        self.initialize_db()
+        self._cached_sources = {}
+
+        atexit.register(self.close_connection)
+
+    _list_tables_query = b"""
+    SELECT table_name FROM information_schema.tables
+    WHERE table_schema = 'public'
+    AND table_type = 'BASE TABLE';
+    """
+
+    def list_tables(self) -> List[str]:
+        """Return names of tables in database.
+
+        :return: Table names in database
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(self._list_tables_query)
+            tables = cur.fetchall()
+        return [t[0] for t in tables]
+
+    _drop_db_query = b"""
+    DROP MATERIALIZED VIEW IF EXISTS record_lookup_view;
+    DROP TABLE IF EXISTS
+        disease_aliases,
+        disease_associations,
+        disease_concepts,
+        disease_labels,
+        disease_merged,
+        disease_sources,
+        disease_xrefs;
+    """
+
+    def drop_db(self) -> None:
+        """Perform complete teardown of DB. Useful for quickly resetting all data or
+        reconstructing after apparent schema error. If in a protected environment,
+        require confirmation.
+
+        :raise DatabaseWriteException: if called in a protected setting with
+            confirmation silenced.
+        """
+        try:
+            if not self._check_delete_okay():
+                return
+        except DatabaseWriteException as e:
+            raise e
+
+        with self.conn.cursor() as cur:
+            cur.execute(self._drop_db_query)
+            self.conn.commit()
+        logger.info("Dropped all existing disease normalizer tables.")
+
+    def check_schema_initialized(self) -> bool:
+        """Check if database schema is properly initialized.
+
+        :return: True if DB appears to be fully initialized, False otherwise
+        """
+        try:
+            with self.conn.cursor() as cur:
+                cur.execute((SCRIPTS_DIR / "create_tables.sql").read_bytes())
+        except DuplicateTable:
+            self.conn.rollback()
+        else:
+            logger.info("Disease table existence check failed.")
+            self.conn.rollback()
+            return False
+        try:
+            with self.conn.cursor() as cur:
+                cur.execute((SCRIPTS_DIR / "add_fkeys.sql").read_bytes())
+        except DuplicateObject:
+            self.conn.rollback()
+        else:
+            logger.info("Disease foreign key existence check failed.")
+            self.conn.rollback()
+            return False
+        try:
+            with self.conn.cursor() as cur:
+                cur.execute(
+                    (SCRIPTS_DIR / "create_record_lookup_view.sql").read_bytes()
+                )
+        except DuplicateTable:
+            self.conn.rollback()
+        else:
+            logger.info("Disease normalized view lookup failed.")
+            self.conn.rollback()
+            return False
+        try:
+            with self.conn.cursor() as cur:
+                cur.execute((SCRIPTS_DIR / "add_indexes.sql").read_bytes())
+        except DuplicateTable:
+            self.conn.rollback()
+        else:
+            logger.info("Disease indexes check failed.")
+            self.conn.rollback()
+            return False
+
+        return True
+
+    _check_sources_query = b"SELECT name FROM disease_sources;"
+    _check_concepts_query = b"SELECT COUNT(1) FROM disease_concepts LIMIT 1;"
+    _check_merged_query = b"SELECT COUNT(1) FROM disease_merged LIMIT 1;"
+
+    def check_tables_populated(self) -> bool:
+        """Perform rudimentary checks to see if tables are populated.
+        Emphasis is on rudimentary -- if some fiendish element has deleted half of the
+        disease aliases, this method won't pick it up. It just wants to see if a few
+        critical tables have at least a small number of records.
+
+        :return: True if queries successful, false if DB appears empty
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(self._check_sources_query)
+            results = cur.fetchall()
+        if len(results) < len(SourceName):
+            logger.info("Disease sources table is missing expected sources.")
+            return False
+
+        with self.conn.cursor() as cur:
+            cur.execute(self._check_concepts_query)
+            result = cur.fetchone()
+        if not result or result[0] < 1:
+            logger.info("Disease records table is empty.")
+            return False
+
+        with self.conn.cursor() as cur:
+            cur.execute(self._check_merged_query)
+            result = cur.fetchone()
+        if not result or result[0] < 1:
+            logger.info("Normalized disease records table is empty.")
+            return False
+
+        return True
+
+    def initialize_db(self) -> None:
+        """Check if DB is set up. If not, create tables/indexes/views."""
+        if not self.check_schema_initialized():
+            self.drop_db()
+            self._create_tables()
+            self._create_views()
+            self._add_indexes()
+
+    def _create_views(self) -> None:
+        """Create materialized views."""
+        create_view_query = (SCRIPTS_DIR / "create_record_lookup_view.sql").read_bytes()
+        with self.conn.cursor() as cur:
+            cur.execute(create_view_query)
+            self.conn.commit()
+
+    _refresh_views_query = b"REFRESH MATERIALIZED VIEW record_lookup_view;"
+
+    def _refresh_views(self) -> None:
+        """Update materialized views.
+
+        Not responsible for ensuring existence of views. Calling functions should
+        either check beforehand or catch psycopg.UndefinedTable.
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(self._refresh_views_query)
+            self.conn.commit()
+
+    def _add_fkeys(self) -> None:
+        """Add fkey relationships."""
+        add_fkey_query = (SCRIPTS_DIR / "add_fkeys.sql").read_bytes()
+        with self.conn.cursor() as cur:
+            cur.execute(add_fkey_query)
+            self.conn.commit()
+
+    def _drop_fkeys(self) -> None:
+        """Drop fkey relationships."""
+        drop_fkey_query = (SCRIPTS_DIR / "drop_fkeys.sql").read_bytes()
+        with self.conn.cursor() as cur:
+            cur.execute(drop_fkey_query)
+            self.conn.commit()
+
+    def _add_indexes(self) -> None:
+        """Create core search indexes."""
+        add_indexes_query = (SCRIPTS_DIR / "add_indexes.sql").read_bytes()
+        with self.conn.cursor() as cur:
+            cur.execute(add_indexes_query)
+            self.conn.commit()
+
+    def _drop_indexes(self) -> None:
+        """Drop all custom indexes."""
+        drop_indexes_query = (SCRIPTS_DIR / "drop_indexes.sql").read_bytes()
+        with self.conn.cursor() as cur:
+            cur.execute(drop_indexes_query)
+            self.conn.commit()
+
+    def _create_tables(self) -> None:
+        """Create all tables, indexes, and views."""
+        logger.debug("Creating new disease normalizer tables.")
+        tables_query = (SCRIPTS_DIR / "create_tables.sql").read_bytes()
+
+        with self.conn.cursor() as cur:
+            cur.execute(tables_query)
+            self.conn.commit()
+
+    _source_metadata_query = b"SELECT * FROM disease_sources WHERE name = %s;"
+
+    def get_source_metadata(self, src_name: Union[str, SourceName]) -> Optional[Dict]:
+        """Get license, versioning, data lookup, etc information for a source.
+
+        :param src_name: name of the source to get data for
+        :return: Dict containing metadata if lookup is successful
+        """
+        if isinstance(src_name, SourceName):
+            src_name = src_name.value
+
+        if src_name in self._cached_sources:
+            return self._cached_sources[src_name]
+
+        with self.conn.cursor() as cur:
+            cur.execute(self._source_metadata_query, [src_name])
+            metadata_result = cur.fetchone()
+            if not metadata_result:
+                return None
+            metadata = {
+                "data_license": metadata_result[1],
+                "data_license_url": metadata_result[2],
+                "version": metadata_result[3],
+                "data_url": metadata_result[4],
+                "rdp_url": metadata_result[5],
+                "data_license_attributes": {
+                    "non_commercial": metadata_result[6],
+                    "attribution": metadata_result[7],
+                    "share_alike": metadata_result[8],
+                },
+            }
+            self._cached_sources[src_name] = metadata
+            return metadata
+
+    _record_query = b"SELECT * FROM record_lookup_view WHERE lower(concept_id) = %s;"
+
+    def _get_record(self, concept_id: str, case_sensitive: bool) -> Optional[Dict]:
+        """Retrieve non-merged record. The query is pretty different, so this method
+        is broken out for PostgreSQL.
+
+        :param concept_id: ID of concept to get
+        :param case_sensitive: record lookups are performed using a case-insensitive
+            index, so this parameter isn't used by Postgres
+        :return: complete record object if successful
+        """
+        concept_id_param = concept_id.lower()
+
+        with self.conn.cursor() as cur:
+            cur.execute(self._record_query, [concept_id_param])
+            result = cur.fetchone()
+        if not result:
+            return None
+
+        disease_record = {
+            "concept_id": result[0],
+            "label": result[1],
+            "aliases": result[2],
+            "associated_with": result[3],
+            "xrefs": result[4],
+            "src_name": result[5],
+            "merge_ref": result[6],
+            "pediatric_disease": result[7],
+            "item_type": "identity",
+        }
+        return {k: v for k, v in disease_record.items() if v}
+
+    _merged_record_query = b"SELECT * FROM disease_merged WHERE lower(concept_id) = %s;"
+
+    def _get_merged_record(
+        self, concept_id: str, case_sensitive: bool
+    ) -> Optional[Dict]:
+        """Retrieve normalized record from DB.
+
+        :param concept_id: normalized ID for the merged record
+        :param case_sensitive: record lookups are performed using a case-insensitive
+            index, so this parameter isn't used by Postgres
+        :return: normalized record if successful
+        """
+        concept_id = concept_id.lower()
+        with self.conn.cursor() as cur:
+            cur.execute(self._merged_record_query, [concept_id])
+            result = cur.fetchone()
+        if not result:
+            return None
+
+        merged_record = {
+            "concept_id": result[0],
+            "label": result[1],
+            "aliases": result[2],
+            "associated_with": result[3],
+            "xrefs": result[4],
+            "pediatric_disease": result[5],
+            "item_type": "merger",
+        }
+        return {k: v for k, v in merged_record.items() if v}
+
+    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 disease record
+        :param bool case_sensitive: not used by this implementation -- lookups use
+        case-insensitive index
+        :param bool merge: if true, look for merged record; look for identity record
+        otherwise.
+        :return: complete disease record, if match is found; None otherwise
+        """
+        if merge:
+            return self._get_merged_record(concept_id, case_sensitive)
+        else:
+            return self._get_record(concept_id, case_sensitive)
+
+    _ref_types_query = {
+        RefType.LABEL: b"SELECT concept_id FROM disease_labels WHERE lower(label) = %s;",  # noqa: E501
+        RefType.ALIASES: b"SELECT concept_id FROM disease_aliases WHERE lower(alias) = %s;",  # noqa: E501
+        RefType.XREFS: "SELECT concept_id FROM disease_xrefs WHERE lower(xref) = %s;",
+        RefType.ASSOCIATED_WITH: "SELECT concept_id FROM disease_associations WHERE lower(associated_with) = %s;",  # noqa: E501
+    }
+
+    def get_refs_by_type(self, search_term: str, ref_type: RefType) -> List[str]:
+        """Retrieve concept IDs for records matching the user's query. Other methods
+        are responsible for actually retrieving full records.
+
+        :param search_term: string to match against
+        :param ref_type: type of match to look for.
+        :return: list of associated concept IDs. Empty if lookup fails.
+        """
+        query = self._ref_types_query.get(ref_type)
+        if not query:
+            raise ValueError("invalid reference type")
+
+        with self.conn.cursor() as cur:
+            cur.execute(query, (search_term.lower(),))
+            concept_ids = cur.fetchall()
+        if concept_ids:
+            return [i[0] for i in concept_ids]
+        else:
+            return []
+
+    _source_ids_query = b"""
+    SELECT concept_id FROM disease_concepts dc
+    LEFT JOIN disease_sources ds ON ds.name = dc.source
+    WHERE ds.name = %s;
+    """
+    _all_ids_query = b"SELECT concept_id FROM disease_concepts;"
+
+    def get_all_concept_ids(self, source: Optional[SourceName] = None) -> Set[str]:
+        """Retrieve concept IDs for use in generating normalized records.
+
+        :param source: optionally, just get all IDs for a specific source
+        :return: Set of concept IDs as strings.
+        """
+        if source is not None:
+            ids_query = self._source_ids_query
+        else:
+            ids_query = self._all_ids_query
+        with self.conn.cursor() as cur:
+            if source is None:
+                cur.execute(ids_query)
+            else:
+                cur.execute(ids_query, (source,))
+            ids_tuple = cur.fetchall()
+        return {i[0] for i in ids_tuple}
+
+    _add_source_metadata_query = b"""
+    INSERT INTO disease_sources(
+        name, data_license, data_license_url, version, data_url, rdp_url,
+        data_license_nc, data_license_attr, data_license_sa
+    )
+    VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s );
+    """
+
+    def add_source_metadata(self, src_name: SourceName, meta: SourceMeta) -> None:
+        """Add new source metadata entry.
+
+        :param src_name: name of source
+        :param meta: known source attributes
+        :raise DatabaseWriteException: if write fails
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(
+                self._add_source_metadata_query,
+                [
+                    src_name.value,
+                    meta.data_license,
+                    meta.data_license_url,
+                    meta.version,
+                    meta.data_url,
+                    meta.rdp_url,
+                    meta.data_license_attributes["non_commercial"],
+                    meta.data_license_attributes["attribution"],
+                    meta.data_license_attributes["share_alike"],
+                ],
+            )
+        self.conn.commit()
+
+    _add_record_query = b"""
+        INSERT INTO disease_concepts (concept_id, source, pediatric_disease)
+        VALUES (%s, %s, %s);
+    """
+    _insert_label_query = (
+        b"INSERT INTO disease_labels (label, concept_id) VALUES (%s, %s)"  # noqa: E501
+    )
+    _insert_alias_query = (
+        b"INSERT INTO disease_aliases (alias, concept_id) VALUES (%s, %s)"  # noqa: E501
+    )
+    _insert_xref_query = b"INSERT INTO disease_xrefs (xref, concept_id) VALUES (%s, %s)"
+    _insert_assoc_query = b"INSERT INTO disease_associations (associated_with, concept_id) VALUES (%s, %s)"  # noqa: E501
+
+    def add_record(self, record: Dict, src_name: SourceName) -> None:
+        """Add new record to database.
+
+        :param record: record to upload
+        :param src_name: name of source for record. Not used by PostgreSQL instance.
+        """
+        concept_id = record["concept_id"]
+        with self.conn.cursor() as cur:
+            try:
+                cur.execute(
+                    self._add_record_query,
+                    [concept_id, src_name.value, record.get("pediatric_disease")],
+                )
+                cur.execute(self._insert_label_query, [record["label"], concept_id])
+                for a in record.get("aliases", []):
+                    cur.execute(self._insert_alias_query, [a, concept_id])
+                for x in record.get("xrefs", []):
+                    cur.execute(self._insert_xref_query, [x, concept_id])
+                for a in record.get("associated_with", []):
+                    cur.execute(self._insert_assoc_query, [a, concept_id])
+                self.conn.commit()
+            except UniqueViolation:
+                logger.error(f"Record with ID {concept_id} already exists")
+                self.conn.rollback()
+
+    _add_merged_record_query = b"""
+    INSERT INTO disease_merged (
+        concept_id, label, aliases, associated_with, xrefs, pediatric_disease
+    )
+    VALUES (%s, %s, %s, %s, %s, %s);
+    """
+
+    def add_merged_record(self, record: Dict) -> None:
+        """Add merged record to database.
+
+        :param record: merged record to add
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(
+                self._add_merged_record_query,
+                [
+                    record["concept_id"],
+                    record["label"],
+                    record.get("aliases"),
+                    record.get("associated_with"),
+                    record.get("xrefs"),
+                    record.get("pediatric_disease"),
+                ],
+            )
+            self.conn.commit()
+
+    _update_merge_ref_query = b"""
+    UPDATE disease_concepts
+    SET merge_ref = %(merge_ref)s
+    WHERE concept_id = %(concept_id)s;
+    """
+
+    def update_merge_ref(self, concept_id: str, merge_ref: Any) -> None:
+        """Update the merged record reference of an individual record to a new value.
+
+        :param concept_id: record to update
+        :param merge_ref: new ref value
+        :raise DatabaseWriteException: if attempting to update non-existent record
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(
+                self._update_merge_ref_query,
+                {"merge_ref": merge_ref, "concept_id": concept_id},
+            )
+            row_count = cur.rowcount
+            self.conn.commit()
+
+        # UPDATE will fail silently unless we check the # of affected rows
+        if row_count < 1:
+            raise DatabaseWriteException(
+                f"No such record exists for primary key {concept_id}"
+            )
+
+    def delete_normalized_concepts(self) -> None:
+        """Remove merged records from the database. Use when performing a new update
+        of normalized data.
+
+        It would be faster to drop the entire table and do a cascading delete onto the
+        merge_ref column in disease_concepts, but that requires an exclusive access lock
+        on the DB, which can be annoying (ie you couldn't have multiple processes
+        accessing it, or PgAdmin, etc...). Instead, we'll take down each merge table
+        dependency and rebuild afterwards.
+
+        :raise DatabaseReadException: if DB client requires separate read calls and
+            encounters a failure in the process
+        :raise DatabaseWriteException: if deletion call fails
+        """
+        with self.conn.cursor() as cur:
+            cur.execute((SCRIPTS_DIR / "delete_normalized_concepts.sql").read_bytes())
+        self.conn.commit()
+
+    _drop_aliases_query = b"""
+    DELETE FROM disease_aliases WHERE id IN (
+        SELECT da.id FROM disease_aliases da LEFT JOIN disease_concepts dc
+            ON dc.concept_id = da.concept_id
+        WHERE dc.source = %s
+    );
+    """
+    _drop_associations_query = b"""
+    DELETE FROM disease_associations WHERE id IN (
+        SELECT da.id FROM disease_associations da LEFT JOIN disease_concepts dc
+            ON dc.concept_id = da.concept_id
+        WHERE dc.source = %s
+    );
+    """
+    _drop_labels_query = b"""
+    DELETE FROM disease_labels WHERE id IN (
+        SELECT ds.id FROM disease_labels ds LEFT JOIN disease_concepts dc
+            ON dc.concept_id = ds.concept_id
+        WHERE dc.source = %s
+    );
+    """
+    _drop_xrefs_query = b"""
+    DELETE FROM disease_xrefs WHERE id IN (
+        SELECT dx.id FROM disease_xrefs dx LEFT JOIN disease_concepts dc
+            ON dc.concept_id = dx.concept_id
+        WHERE dc.source = %s
+    );
+    """
+    _drop_concepts_query = b"DELETE FROM disease_concepts WHERE source = %s;"
+    _drop_source_query = b"DELETE FROM disease_sources ds WHERE ds.name = %s;"
+
+    def delete_source(self, src_name: SourceName) -> None:
+        """Delete all data for a source. Use when updating source data.
+
+        All of the foreign key relations make deletes *extremely* slow, so this method
+        drops and then re-adds them once deletes are finished. This makes it a little
+        brittle, and it'd be nice to revisit in the future to perform as a single
+        atomic transaction.
+
+        Refreshing the materialized view at the end might be redundant, because
+        this method will almost always be called right before more data is written,
+        but it's probably necessary just in case that doesn't happen.
+
+        :param src_name: name of source to delete
+        :raise DatabaseWriteException: if deletion call fails
+        """
+        with self.conn.cursor() as cur:
+            cur.execute(self._drop_aliases_query, [src_name.value])
+            cur.execute(self._drop_associations_query, [src_name.value])
+            cur.execute(self._drop_labels_query, [src_name.value])
+            cur.execute(self._drop_xrefs_query, [src_name.value])
+        self._drop_fkeys()
+        self._drop_indexes()
+
+        with self.conn.cursor() as cur:
+            cur.execute(self._drop_concepts_query, [src_name.value])
+            cur.execute(self._drop_source_query, [src_name.value])
+            self.conn.commit()
+
+        self._add_fkeys()
+        self._add_indexes()
+        self._refresh_views()
+
+    def complete_write_transaction(self) -> None:
+        """Conclude transaction or batch writing if relevant."""
+        if not self.conn.closed:
+            try:
+                self._refresh_views()
+            except UndefinedTable:
+                self.conn.rollback()
+
+    def close_connection(self) -> None:
+        """Perform any manual connection closure procedures if necessary."""
+        if not self.conn.closed:
+            self.conn.commit()
+            self.conn.close()
+
+    def load_from_remote(self, url: Optional[str]) -> None:
+        """Load DB from remote dump. Warning: Deletes all existing data. If not
+        passed as an argument, will try to grab latest release from VICC S3 bucket.
+
+        :param url: location of .tar.gz file created from output of pg_dump
+        :raise DatabaseException: if unable to retrieve file from URL or if psql
+            command fails
+        """
+        if not url:
+            url = "https://vicc-normalizers.s3.us-east-2.amazonaws.com/disease_normalization/postgresql/disease_norm_latest.sql.tar.gz"  # noqa: E501
+        with tempfile.TemporaryDirectory() as tempdir:
+            tempdir_path = Path(tempdir)
+            temp_tarfile = tempdir_path / "disease_norm_latest.tar.gz"
+            with requests.get(url, stream=True) as r:
+                try:
+                    r.raise_for_status()
+                except requests.HTTPError:
+                    raise DatabaseException(
+                        f"Unable to retrieve PostgreSQL dump file from {url}"
+                    )
+                with open(temp_tarfile, "wb") as h:
+                    for chunk in r.iter_content(chunk_size=8192):
+                        if chunk:
+                            h.write(chunk)
+            tar = tarfile.open(temp_tarfile, "r:gz")
+            tar_dump_file = [
+                f for f in tar.getmembers() if f.name.startswith("disease_norm_")
+            ][0]
+            tar.extractall(path=tempdir_path, members=[tar_dump_file])
+            dump_file = tempdir_path / tar_dump_file.name
+
+            if self.conn.info.password:
+                pw_param = f"-W {self.conn.info.password}"
+            else:
+                pw_param = "-w"
+
+            self.drop_db()
+            system_call = f"psql -d {self.conn.info.dbname} -U {self.conn.info.user} {pw_param} -f {dump_file.absolute()}"  # noqa: E501
+            result = os.system(system_call)
+        if result != 0:
+            raise DatabaseException(
+                f"System call '{result}' returned failing exit code."
+            )
+
+    def export_db(self, output_directory: Path) -> None:
+        """Dump DB to specified location.
+
+        :param export_location: path to directory to save DB dump in
+        :return: Nothing, but saves results of pg_dump to file named
+            `disease_norm_<date and time>.sql`
+        :raise ValueError: if output directory isn't a directory or doesn't exist
+        :raise DatabaseException: if psql call fails
+        """
+        if not output_directory.is_dir() or not output_directory.exists():
+            raise ValueError(
+                f"Output location {output_directory} isn't a directory or doesn't exist"
+            )  # noqa: E501
+        now = datetime.now().strftime("%Y%m%d%H%M%S")
+        output_location = output_directory / f"disease_norm_{now}.sql"
+        user = self.conn.info.user
+        host = self.conn.info.host
+        port = self.conn.info.port
+        database_name = self.conn.info.dbname
+        if self.conn.info.password:
+            pw_param = f"-W {self.conn.info.password}"
+        else:
+            pw_param = "-w"
+
+        system_call = f"pg_dump -E UTF8 -f {output_location} -U {user} {pw_param} -h {host} -p {port} {database_name}"  # noqa: E501
+        result = os.system(system_call)
+        if result != 0:
+            raise DatabaseException(
+                f"System call '{system_call}' returned failing exit code."
+            )
diff --git a/disease/database/postgresql/add_fkeys.sql b/disease/database/postgresql/add_fkeys.sql
new file mode 100644
index 00000000..08ca9b6a
--- /dev/null
+++ b/disease/database/postgresql/add_fkeys.sql
@@ -0,0 +1,8 @@
+ALTER TABLE disease_aliases ADD CONSTRAINT disease_aliases_concept_id_fkey
+    FOREIGN KEY (concept_id) REFERENCES disease_concepts (concept_id);
+ALTER TABLE disease_associations ADD CONSTRAINT disease_associations_concept_id_fkey
+    FOREIGN KEY (concept_id) REFERENCES disease_concepts (concept_id);
+ALTER TABLE disease_labels ADD CONSTRAINT disease_labels_concept_id_fkey
+    FOREIGN KEY (concept_id) REFERENCES disease_concepts (concept_id);
+ALTER TABLE disease_xrefs ADD CONSTRAINT disease_xrefs_concept_id_fkey
+    FOREIGN KEY (concept_id) REFERENCES disease_concepts (concept_id);
diff --git a/disease/database/postgresql/add_indexes.sql b/disease/database/postgresql/add_indexes.sql
new file mode 100644
index 00000000..7e72900e
--- /dev/null
+++ b/disease/database/postgresql/add_indexes.sql
@@ -0,0 +1,8 @@
+CREATE INDEX idx_d_concept_id_low ON disease_concepts (lower(concept_id));
+-- see also: delete_normalized_concepts.sql
+CREATE INDEX idx_dm_concept_id_low ON disease_merged (lower(concept_id));
+CREATE INDEX idx_dl_label_low ON disease_labels (lower(label));
+CREATE INDEX idx_da_alias_low ON disease_aliases (lower(alias));
+CREATE INDEX idx_dx_xref_low ON disease_xrefs (lower(xref));
+CREATE INDEX idx_d_as_association_low ON disease_associations (lower(associated_with));
+CREATE INDEX idx_rlv_concept_id_low ON record_lookup_view (lower(concept_id));
diff --git a/disease/database/postgresql/create_record_lookup_view.sql b/disease/database/postgresql/create_record_lookup_view.sql
new file mode 100644
index 00000000..f8142102
--- /dev/null
+++ b/disease/database/postgresql/create_record_lookup_view.sql
@@ -0,0 +1,27 @@
+CREATE MATERIALIZED VIEW record_lookup_view AS
+SELECT dc.concept_id,
+       dl.label,
+       da.aliases,
+       das.associated_with,
+       dx.xrefs,
+       dc.source,
+       dc.merge_ref,
+       dc.pediatric_disease,
+       lower(dc.concept_id) AS concept_id_lowercase
+FROM disease_concepts dc
+FULL JOIN (
+    SELECT da_1.concept_id, array_agg(da_1.alias) AS aliases
+    FROM disease_aliases da_1
+    GROUP BY da_1.concept_id
+) da ON dc.concept_id::text = da.concept_id::text
+FULL JOIN (
+    SELECT das_1.concept_id, array_agg(das_1.associated_with) AS associated_with
+    FROM disease_associations das_1
+    GROUP BY das_1.concept_id
+) das ON dc.concept_id::text = das.concept_id::text
+FULL JOIN disease_labels dl ON dc.concept_id::text = dl.concept_id::text
+FULL JOIN (
+    SELECT dx_1.concept_id, array_agg(dx_1.xref) AS xrefs
+    FROM disease_xrefs dx_1
+    GROUP BY dx_1.concept_id
+) dx ON dc.concept_id::text = dx.concept_id::text;
diff --git a/disease/database/postgresql/create_tables.sql b/disease/database/postgresql/create_tables.sql
new file mode 100644
index 00000000..45ae3ad1
--- /dev/null
+++ b/disease/database/postgresql/create_tables.sql
@@ -0,0 +1,46 @@
+CREATE TABLE disease_sources (
+    name VARCHAR(127) PRIMARY KEY,
+    data_license TEXT NOT NULL,
+    data_license_url TEXT NOT NULL,
+    version TEXT NOT NULL,
+    data_url TEXT NOT NULL,
+    rdp_url TEXT,
+    data_license_nc BOOLEAN NOT NULL,
+    data_license_attr BOOLEAN NOT NULL,
+    data_license_sa BOOLEAN NOT NULL
+);
+-- see also: delete_normalized_concepts.sql
+CREATE TABLE disease_merged (
+    concept_id VARCHAR(127) PRIMARY KEY,
+    label TEXT,
+    aliases TEXT [],
+    associated_with TEXT [],
+    xrefs TEXT [],
+    pediatric_disease BOOLEAN
+);
+CREATE TABLE disease_concepts (
+    concept_id VARCHAR(127) PRIMARY KEY,
+    source VARCHAR(127) NOT NULL REFERENCES disease_sources (name),
+    pediatric_disease BOOLEAN,
+    merge_ref VARCHAR(127) REFERENCES disease_merged (concept_id)
+);
+CREATE TABLE disease_labels (
+    id SERIAL PRIMARY KEY,
+    label TEXT NOT NULL,
+    concept_id VARCHAR(127) REFERENCES disease_concepts (concept_id)
+);
+CREATE TABLE disease_aliases (
+    id SERIAL PRIMARY KEY,
+    alias TEXT NOT NULL,
+    concept_id VARCHAR(127) NOT NULL REFERENCES disease_concepts (concept_id)
+);
+CREATE TABLE disease_xrefs (
+    id SERIAL PRIMARY KEY,
+    xref TEXT NOT NULL,
+    concept_id VARCHAR(127) NOT NULL REFERENCES disease_concepts (concept_id)
+);
+CREATE TABLE disease_associations (
+    id SERIAL PRIMARY KEY,
+    associated_with TEXT NOT NULL,
+    concept_ID VARCHAR(127) NOT NULL REFERENCES disease_concepts (concept_id)
+);
diff --git a/disease/database/postgresql/delete_normalized_concepts.sql b/disease/database/postgresql/delete_normalized_concepts.sql
new file mode 100644
index 00000000..7c4fd148
--- /dev/null
+++ b/disease/database/postgresql/delete_normalized_concepts.sql
@@ -0,0 +1,17 @@
+-- some redundancy between here and create_tables.sql, drop_indexes.sql,
+-- add_indexes.sql.
+DROP INDEX IF EXISTS idx_dm_concept_id_low;
+ALTER TABLE disease_concepts DROP CONSTRAINT IF EXISTS disease_concepts_merge_ref_fkey;
+UPDATE disease_concepts SET merge_ref = NULL;
+DROP TABLE disease_merged;
+CREATE TABLE disease_merged (
+    concept_id VARCHAR(127) PRIMARY KEY,
+    label TEXT,
+    aliases TEXT [],
+    associated_with TEXT [],
+    xrefs TEXT [],
+    pediatric_disease BOOLEAN
+);
+ALTER TABLE disease_concepts ADD CONSTRAINT disease_concepts_merge_ref_fkey
+    FOREIGN KEY (merge_ref) REFERENCES disease_merged (concept_id);
+CREATE INDEX idx_dm_concept_id_low ON disease_merged (lower(concept_id));
diff --git a/disease/database/postgresql/drop_fkeys.sql b/disease/database/postgresql/drop_fkeys.sql
new file mode 100644
index 00000000..65749002
--- /dev/null
+++ b/disease/database/postgresql/drop_fkeys.sql
@@ -0,0 +1,4 @@
+ALTER TABLE disease_aliases DROP CONSTRAINT disease_aliases_concept_id_fkey;
+ALTER TABLE disease_associations DROP CONSTRAINT disease_associations_concept_id_fkey;
+ALTER TABLE disease_labels DROP CONSTRAINT disease_labels_concept_id_fkey;
+ALTER TABLE disease_xrefs DROP CONSTRAINT disease_xrefs_concept_id_fkey;
diff --git a/disease/database/postgresql/drop_indexes.sql b/disease/database/postgresql/drop_indexes.sql
new file mode 100644
index 00000000..ad31963d
--- /dev/null
+++ b/disease/database/postgresql/drop_indexes.sql
@@ -0,0 +1,7 @@
+DROP INDEX IF EXISTS idx_d_concept_id_low;
+DROP INDEX IF EXISTS idx_dm_concept_id_low; -- see also: delete_normalized_concepts.sql
+DROP INDEX IF EXISTS idx_dl_label_low;
+DROP INDEX IF EXISTS idx_dx_xref_low;
+DROP INDEX IF EXISTS idx_da_alias_low;
+DROP INDEX IF EXISTS idx_d_as_association_low;
+DROP INDEX IF EXISTS idx_rlv_concept_id_low;
diff --git a/disease/etl/base.py b/disease/etl/base.py
index 341d4d2d..075c2889 100644
--- a/disease/etl/base.py
+++ b/disease/etl/base.py
@@ -13,8 +13,8 @@
 from owlready2.rdflib_store import TripleLiteRDFlibGraph as RDFGraph
 
 from disease import APP_ROOT, ITEM_TYPES, SOURCES_FOR_MERGE, logger
-from disease.database import Database
-from disease.schemas import Disease
+from disease.database import AbstractDatabase
+from disease.schemas import Disease, SourceName
 
 DEFAULT_DATA_PATH = APP_ROOT / "data"
 
@@ -22,14 +22,14 @@
 class Base(ABC):
     """The ETL base class."""
 
-    def __init__(self, database: Database, data_path: Path = DEFAULT_DATA_PATH):
+    def __init__(self, database: AbstractDatabase, data_path: Path = DEFAULT_DATA_PATH):
         """Extract from sources.
-        :param database: DDB client
+        :param database: database client
         :param data_path: location of data directory
         """
-        self.name = self.__class__.__name__.lower()
-        self.database = database
-        self._src_dir: Path = Path(data_path / self.name)
+        self._src_name = SourceName(self.__class__.__name__)
+        self._database = database
+        self._src_dir: Path = Path(data_path / self._src_name.value.lower())
         self._store_ids = self.__class__.__name__ in SOURCES_FOR_MERGE
         if self._store_ids:
             self._added_ids = []
@@ -43,6 +43,7 @@ def perform_etl(self, use_existing: bool = False) -> List:
         self._extract_data(use_existing)
         self._load_meta()
         self._transform_data()
+        self._database.complete_write_transaction()
         if self._store_ids:
             return self._added_ids
         else:
@@ -53,7 +54,7 @@ def get_latest_version(self) -> str:
         sources not added to Bioversions yet, or other special-case sources.
         :return: most recent version, as a str
         """
-        return bioversions.get_version(self.name)
+        return bioversions.get_version(self._src_name)
 
     @abstractmethod
     def _download_data(self):
@@ -149,7 +150,7 @@ def _parse_version(
     def _get_latest_data_file(self) -> Path:
         """Acquire most recent source data file."""
         self._version = self.get_latest_version()
-        fglob = f"{self.name}_{self._version}.*"
+        fglob = f"{self._src_name.lower()}_{self._version}.*"
         latest = list(self._src_dir.glob(fglob))
         if not latest:
             self._download_data()
@@ -194,7 +195,7 @@ def _load_disease(self, disease: Dict):
         assert Disease(**disease)
         concept_id = disease["concept_id"]
 
-        for attr_type, item_type in ITEM_TYPES.items():
+        for attr_type in ITEM_TYPES:
             if attr_type in disease:
                 value = disease[attr_type]
                 if value is not None and value != []:
@@ -209,15 +210,13 @@ def _load_disease(self, disease: Dict):
                                 del disease[attr_type]
                                 continue
 
-                    for item in items:
-                        self.database.add_ref_record(item, concept_id, item_type)
                 else:
                     del disease[attr_type]
 
         if "pediatric_disease" in disease and disease["pediatric_disease"] is None:
             del disease["pediatric_disease"]
 
-        self.database.add_record(disease)
+        self._database.add_record(disease, self._src_name)
         if self._store_ids:
             self._added_ids.append(concept_id)
 
diff --git a/disease/etl/do.py b/disease/etl/do.py
index 9efc985e..d0ad08d5 100644
--- a/disease/etl/do.py
+++ b/disease/etl/do.py
@@ -4,7 +4,7 @@
 
 from disease import PREFIX_LOOKUP, logger
 from disease.etl.base import OWLBase
-from disease.schemas import NamespacePrefix, SourceMeta, SourceName
+from disease.schemas import NamespacePrefix, SourceMeta
 
 DO_PREFIX_LOOKUP = {
     "EFO": NamespacePrefix.EFO.value,
@@ -43,21 +43,19 @@ def _download_data(self):
 
     def _load_meta(self):
         """Load metadata"""
-        metadata_params = {
-            "data_license": "CC0 1.0",
-            "data_license_url": "https://creativecommons.org/publicdomain/zero/1.0/legalcode",  # noqa: E501
-            "version": self._version,
-            "data_url": "http://www.obofoundry.org/ontology/doid.html",
-            "rdp_url": None,
-            "data_license_attributes": {
+        metadata = SourceMeta(
+            data_license="CC0 1.0",
+            data_license_url="https://creativecommons.org/publicdomain/zero/1.0/legalcode",  # noqa: E501
+            version=self._version,
+            data_url="http://www.obofoundry.org/ontology/doid.html",
+            rdp_url=None,
+            data_license_attributes={
                 "non_commercial": False,
                 "share_alike": False,
                 "attribution": False,
             },
-        }
-        assert SourceMeta(**metadata_params)
-        metadata_params["src_name"] = SourceName.DO.value
-        self.database.metadata.put_item(Item=metadata_params)
+        )
+        self._database.add_source_metadata(self._src_name, metadata)
 
     def _transform_data(self):
         """Transform source data and send to loading method."""
diff --git a/disease/etl/merge.py b/disease/etl/merge.py
index 829fe33e..6e24cf0a 100644
--- a/disease/etl/merge.py
+++ b/disease/etl/merge.py
@@ -1,35 +1,32 @@
 """Create concept groups and merged records."""
 from timeit import default_timer as timer
-from typing import Dict, List, Set, Tuple
+from typing import Collection, Dict, List, Set, Tuple
 
 from disease import logger
-from disease.database import Database
+from disease.database.database import AbstractDatabase
 from disease.schemas import SourcePriority
 
 
 class Merge:
     """Handles record merging."""
 
-    def __init__(self, database: Database):
+    def __init__(self, database: AbstractDatabase):
         """Initialize Merge instance.
 
-        :param Database database: db instance to use for record retrieval
-            and creation.
+        :param Database database: db instance to use for record retrieval and creation.
         """
         self._database = database
         self._groups = []  # list of tuples: (mondo_concept_id, set_of_ids)
 
-    def create_merged_concepts(self, record_ids: List[str]):
-        """Create concept groups, generate merged concept records, and
-        update database.
+    def create_merged_concepts(self, record_ids: Collection[str]):
+        """Create concept groups, generate merged concept records, and update database.
+        Our normalization protocols only generate record ID sets that include Mondo
+        terms, meaning only Mondo IDs should be submitted to this method.
 
-        :param List[str] record_ids: concept identifiers from which groups
-            should be generated.
+        :param record_ids: concept identifiers from which groups should be generated.
         """
         # build groups
-        logger.info(
-            f"Generating record ID sets from {len(record_ids)} records"
-        )  # noqa E501
+        logger.info(f"Generating record ID sets from {len(record_ids)} records")
         start = timer()
         for concept_id in record_ids:
             try:
@@ -50,6 +47,7 @@ def create_merged_concepts(self, record_ids: List[str]):
                 group = {concept_id}
             self._groups.append((concept_id, group))
         end = timer()
+        self._database.complete_write_transaction()
         logger.debug(f"Built record ID sets in {end - start} seconds")
 
         # build merged concepts
@@ -64,11 +62,12 @@ def create_merged_concepts(self, record_ids: List[str]):
                     f"{group} for concept {record_id}"
                 )
                 continue
-            self._database.add_record(merged_record, "merger")
-            merge_ref = merged_record["concept_id"].lower()
+            self._database.add_merged_record(merged_record)
+            merge_ref = merged_record["concept_id"]
 
             for concept_id in merged_ids:
-                self._database.update_record(concept_id, "merge_ref", merge_ref)
+                self._database.update_merge_ref(concept_id, merge_ref)
+        self._database.complete_write_transaction()
         end = timer()
         logger.info("merged concept generation successful.")
         logger.debug(f"Generated and added concepts in {end - start} seconds)")
@@ -112,7 +111,7 @@ def record_order(record):
             "associated_with": set(),
         }
         if len(records) > 1:
-            merged_properties["xrefs"] = [r["concept_id"] for r in records[1:]]
+            merged_properties["xrefs"] = list({r["concept_id"] for r in records[1:]})
 
         set_fields = ["aliases", "associated_with"]
         scalar_fields = ["label", "pediatric_disease"]
@@ -131,7 +130,4 @@ def record_order(record):
             else:
                 del merged_properties[field]
 
-        merged_properties[
-            "label_and_type"
-        ] = f'{merged_properties["concept_id"].lower()}##merger'
         return merged_properties, final_ids
diff --git a/disease/etl/mondo.py b/disease/etl/mondo.py
index 36fa4d95..35ae661b 100644
--- a/disease/etl/mondo.py
+++ b/disease/etl/mondo.py
@@ -7,7 +7,7 @@
 from owlready2.rdflib_store import TripleLiteRDFlibGraph as RDFGraph
 
 from disease import PREFIX_LOOKUP, logger
-from disease.schemas import NamespacePrefix, SourceMeta, SourceName
+from disease.schemas import NamespacePrefix, SourceMeta
 
 from .base import OWLBase
 
@@ -84,9 +84,9 @@ def _load_meta(self):
         """Load metadata"""
         metadata = SourceMeta(
             data_license="CC BY 4.0",
-            data_license_url="https://creativecommons.org/licenses/by/4.0/legalcode",
+            data_license_url="https://creativecommons.org/licenses/by/4.0/legalcode",  # noqa: E501
             version=self._version,
-            data_url="https://mondo.monarchinitiative.org/pages/download/",
+            data_url="https://mondo.monarchinitiative.org/pages/download/",  # noqa: E501
             rdp_url="http://reusabledata.org/monarch.html",
             data_license_attributes={
                 "non_commercial": False,
@@ -94,9 +94,7 @@ def _load_meta(self):
                 "attribution": True,
             },
         )
-        params = dict(metadata)
-        params["src_name"] = SourceName.MONDO.value
-        self.database.metadata.put_item(Item=params)
+        self._database.add_source_metadata(self._src_name, metadata)
 
     def _get_concept_id(self, ref: str) -> Optional[str]:
         """Format concept ID for given reference.
diff --git a/disease/etl/ncit.py b/disease/etl/ncit.py
index 186bba1d..40faafc6 100644
--- a/disease/etl/ncit.py
+++ b/disease/etl/ncit.py
@@ -68,9 +68,7 @@ def _load_meta(self):
                 "attribution": True,
             },
         )
-        params = dict(metadata)
-        params["src_name"] = SourceName.NCIT.value
-        self.database.metadata.put_item(Item=params)
+        self._database.add_source_metadata(self._src_name, metadata)
 
     def _get_disease_classes(self) -> Set[str]:
         """Get all nodes with semantic_type 'Neoplastic Process' or 'Disease
diff --git a/disease/etl/omim.py b/disease/etl/omim.py
index 249babc4..e21d6d50 100644
--- a/disease/etl/omim.py
+++ b/disease/etl/omim.py
@@ -1,8 +1,7 @@
 """Module to load disease data from OMIM."""
 from disease import logger
-from disease.schemas import Disease, NamespacePrefix, SourceMeta, SourceName
-
-from .base import Base, DownloadException
+from disease.etl.base import Base, DownloadException
+from disease.schemas import NamespacePrefix, SourceMeta
 
 
 class OMIM(Base):
@@ -47,9 +46,7 @@ def _load_meta(self):
                 "attribution": True,
             },
         )
-        params = dict(metadata)
-        params["src_name"] = SourceName.OMIM.value
-        self.database.metadata.put_item(Item=params)
+        self._database.add_source_metadata(self._src_name, metadata)
 
     def _transform_data(self):
         """Modulate data and prepare for loading."""
@@ -82,5 +79,4 @@ def _transform_data(self):
             }
             disease["aliases"] = [a.lstrip() for a in aliases]
 
-            assert Disease(**disease)
             self._load_disease(disease)
diff --git a/disease/etl/oncotree.py b/disease/etl/oncotree.py
index a4b9296b..912ab27b 100644
--- a/disease/etl/oncotree.py
+++ b/disease/etl/oncotree.py
@@ -2,9 +2,8 @@
 import json
 
 from disease import logger
-from disease.schemas import Disease, NamespacePrefix, SourceMeta, SourceName
-
-from .base import Base
+from disease.etl.base import Base
+from disease.schemas import Disease, NamespacePrefix, SourceMeta
 
 
 class OncoTree(Base):
@@ -23,9 +22,9 @@ def _load_meta(self):
         """Load metadata"""
         metadata = SourceMeta(
             data_license="CC BY 4.0",
-            data_license_url="https://creativecommons.org/licenses/by/4.0/legalcode",
+            data_license_url="https://creativecommons.org/licenses/by/4.0/legalcode",  # noqa F401
             version=self._version,
-            data_url="http://oncotree.mskcc.org/#/home?tab=api",
+            data_url="http://oncotree.mskcc.org/#/home?tab=api",  # noqa: E501
             rdp_url=None,
             data_license_attributes={
                 "non_commercial": False,
@@ -33,9 +32,7 @@ def _load_meta(self):
                 "attribution": True,
             },
         )
-        params = dict(metadata)
-        params["src_name"] = SourceName.ONCOTREE.value
-        self.database.metadata.put_item(Item=params)
+        self._database.add_source_metadata(self._src_name, metadata)
 
     def _traverse_tree(self, disease_node):
         """Traverse JSON tree and load diseases where possible.
diff --git a/disease/main.py b/disease/main.py
index 399bde4d..5e36c9a3 100644
--- a/disease/main.py
+++ b/disease/main.py
@@ -6,10 +6,13 @@
 from fastapi.openapi.utils import get_openapi
 
 from disease import __version__
+from disease.database.database import create_db
 from disease.query import InvalidParameterException, QueryHandler
 from disease.schemas import NormalizationService, SearchService
 
-query_handler = QueryHandler()
+db = create_db()
+query_handler = QueryHandler(db)
+
 app = FastAPI(
     docs_url="/disease",
     openapi_url="/disease/openapi.json",
diff --git a/disease/query.py b/disease/query.py
index ce70a861..efc79b50 100644
--- a/disease/query.py
+++ b/disease/query.py
@@ -6,21 +6,15 @@
 
 from botocore.exceptions import ClientError
 
-from disease import (
-    ITEM_TYPES,
-    NAMESPACE_LOOKUP,
-    PREFIX_LOOKUP,
-    SOURCES_LOWER_LOOKUP,
-    logger,
-)
-from disease.database import Database
+from disease import NAMESPACE_LOOKUP, PREFIX_LOOKUP, SOURCES_LOWER_LOOKUP, logger
+from disease.database.database import AbstractDatabase
 from disease.schemas import (
     Disease,
     MatchType,
     NormalizationService,
+    RefType,
     SearchService,
     ServiceMeta,
-    SourceMeta,
     SourceName,
 )
 
@@ -42,12 +36,12 @@ class QueryHandler:
     and normalizes query input.
     """
 
-    def __init__(self, db_url: str = "", db_region: str = "us-east-2"):
+    def __init__(self, database: AbstractDatabase):
         """Initialize Normalizer instance.
-        :param str db_url: URL to database source.
-        :param str db_region: AWS default region.
+
+        :param database: storage backend to query against
         """
-        self.db = Database(db_url=db_url, region_name=db_region)
+        self.db = database
 
     def _emit_warnings(self, query_str) -> Optional[Dict]:
         """Emit warnings if query contains non breaking space characters.
@@ -63,26 +57,11 @@ def _emit_warnings(self, query_str) -> Optional[Dict]:
             )
         return warnings
 
-    def _fetch_meta(self, src_name: str) -> SourceMeta:
-        """Fetch metadata for src_name.
-        :param str src_name: name of source to get metadata for
-        :return: Meta object containing source metadata
-        """
-        if src_name in self.db.cached_sources.keys():
-            return self.db.cached_sources[src_name]
-        else:
-            try:
-                db_response = self.db.metadata.get_item(Key={"src_name": src_name})
-                response = SourceMeta(**db_response["Item"])
-                self.db.cached_sources[src_name] = response
-                return response
-            except ClientError as e:
-                logger.error(e.response["Error"]["Message"])
-
     def _add_record(
         self, response: Dict[str, Dict], item: Dict, match_type: str
     ) -> Tuple[Dict, str]:
         """Add individual record to response object
+
         :param Dict[str, Dict] response: in-progress response object to return
             to client
         :param Dict item: Item retrieved from DynamoDB
@@ -90,7 +69,6 @@ def _add_record(
         :return: Tuple containing updated response object, and string
             containing name of the source of the match
         """
-        del item["label_and_type"]
         disease = Disease(**item)
         src_name = item["src_name"]
 
@@ -101,7 +79,7 @@ def _add_record(
             matches[src_name] = {
                 "match_type": MatchType[match_type.upper()],
                 "records": [disease],
-                "source_meta_": self._fetch_meta(src_name),
+                "source_meta_": self.db.get_source_metadata(src_name),
             }
         elif matches[src_name]["match_type"] == MatchType[match_type.upper()]:
             matches[src_name]["records"].append(disease)
@@ -113,12 +91,12 @@ def _fetch_records(
     ) -> Tuple[Dict, Set]:
         """Return matched Disease records as a structured response for a given
         collection of concept IDs.
+
         :param Dict[str, Dict] response: in-progress response object to return
             to client.
         :param List[str] concept_ids: List of concept IDs to build from.
             Should be all lower-case.
-        :param str match_type: record should be assigned this type of
-            match.
+        :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
         """
@@ -139,6 +117,7 @@ def _fetch_records(
 
     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
         :return: response object with empty source slots filled with
                 NO_MATCH results and corresponding source metadata
@@ -148,14 +127,15 @@ def _fill_no_matches(self, resp: Dict[str, Dict]) -> Dict:
                 resp["source_matches"][src_name] = {
                     "match_type": MatchType.NO_MATCH,
                     "records": [],
-                    "source_meta_": self._fetch_meta(src_name),
+                    "source_meta_": self.db.get_source_metadata(src_name),
                 }
         return resp
 
     def _check_concept_id(
         self, query: str, resp: Dict, sources: Set[str]
-    ) -> (Dict, Set):
+    ) -> Tuple[Dict, Set]:
         """Check query for concept ID match. Should only find 0 or 1 matches.
+
         :param str query: search string
         :param Dict resp: in-progress response object to return to client
         :param Set[str] sources: remaining unmatched sources
@@ -178,21 +158,23 @@ def _check_concept_id(
         return resp, sources
 
     def _check_match_type(
-        self, query: str, resp: Dict, sources: Set[str], match_type: str
+        self, query: str, resp: Dict, sources: Set[str], match_type: RefType
     ) -> Tuple[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_type: Match type to check for. Should be one of
+        :param RefType match_type: Match type to check for. Should be one of
             {'label', 'alias', 'xref', 'associated_with'}
         :return: Tuple with updated resp object and updated set of unmatched
                  sources
         """
-        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)
+        matching_ids = self.db.get_refs_by_type(query, match_type)
+        if matching_ids:
+            (resp, matched_srcs) = self._fetch_records(
+                resp, set(matching_ids), match_type
+            )
             sources = sources - matched_srcs
         return resp, sources
 
@@ -218,7 +200,7 @@ def _response_keyed(self, query: str, sources: Set[str]) -> Dict:
         if len(sources) == 0:
             return response
 
-        for match_type in ITEM_TYPES.values():
+        for match_type in RefType:
             (response, sources) = self._check_match_type(
                 query, response, sources, match_type
             )
@@ -265,7 +247,7 @@ def search(self, query_str, keyed=False, incl="", excl="") -> SearchService:
         """
         sources = dict()
         for k, v in SOURCES_LOWER_LOOKUP.items():
-            if self.db.metadata.get_item(Key={"src_name": v}).get("Item"):
+            if self.db.get_source_metadata(v):
                 sources[k] = v
         if not incl and not excl:
             query_sources = set(sources.values())
@@ -325,7 +307,7 @@ def _add_merged_meta(self, response: Dict) -> Dict:
             prefix = concept_id.split(":")[0]
             src_name = PREFIX_LOOKUP[prefix.lower()]
             if src_name not in sources_meta:
-                sources_meta[src_name] = self._fetch_meta(src_name)
+                sources_meta[src_name] = self.db.get_source_metadata(src_name)
         response["source_meta_"] = sources_meta
         return response
 
@@ -374,7 +356,7 @@ def _add_vod(
         response = self._add_merged_meta(response)
         return NormalizationService(**response)
 
-    def _record_order(self, record: Dict) -> (int, str):
+    def _record_order(self, record: Dict) -> Tuple[int, str]:
         """Construct priority order for matching. Only called by sort().
 
         :param Dict record: individual record item in iterable to sort
@@ -418,6 +400,7 @@ def normalize(self, query: str) -> NormalizationService:
         """
         # prepare basic response
         response = {
+            "match_type": MatchType.NO_MATCH,
             "query": query,
             "warnings": self._emit_warnings(query),
             "service_meta_": ServiceMeta(
@@ -426,7 +409,6 @@ def normalize(self, query: str) -> NormalizationService:
             ),
         }
         if query == "":
-            response["match_type"] = MatchType.NO_MATCH
             return NormalizationService(**response)
         query_str = query.lower().strip()
 
@@ -452,18 +434,19 @@ def normalize(self, query: str) -> NormalizationService:
                 non_merged_match = (record, "concept_id")
 
         # check other match types
-        for match_type in ["label", "alias", "xref", "associated_with"]:
+        for match_type in RefType:
             # get matches list for match tier
-            query_matches = self.db.get_records_by_type(query_str, match_type)
-            query_matches = [
-                self.db.get_record_by_id(m["concept_id"], case_sensitive=False)
-                for m in query_matches
+            matching_refs = self.db.get_refs_by_type(query_str, match_type)
+            matching_records = [
+                self.db.get_record_by_id(ref, False) for ref in matching_refs
             ]
-            query_matches.sort(key=self._record_order)
+            matching_records.sort(key=self._record_order)  # type: ignore
 
             # attempt merge ref resolution until successful
-            for match in query_matches:
-                record = self.db.get_record_by_id(match["concept_id"], False)
+            for match in matching_records:
+                record = self.db.get_record_by_id(
+                    match["concept_id"], False  # type: ignore
+                )
                 if record:
                     if "merge_ref" in record:
                         merge = self.db.get_record_by_id(
@@ -486,6 +469,4 @@ def normalize(self, query: str) -> NormalizationService:
             match_type = MatchType[non_merged_match[1].upper()]
             return self._add_vod(response, non_merged_match[0], query, match_type)
 
-        if not query_matches:
-            response["match_type"] = MatchType.NO_MATCH
         return NormalizationService(**response)
diff --git a/disease/schemas.py b/disease/schemas.py
index 1db53a7f..13da6bce 100644
--- a/disease/schemas.py
+++ b/disease/schemas.py
@@ -139,8 +139,8 @@ class DataLicenseAttributes(BaseModel):
     attribution: StrictBool
 
 
-class ItemTypes(str, Enum):
-    """Item types used in DynamoDB."""
+class RefType(str, Enum):
+    """Reference item types."""
 
     # Must be in descending MatchType order.
     LABEL = "label"
diff --git a/disease/version.py b/disease/version.py
index e4fbd9d1..760d3c07 100644
--- a/disease/version.py
+++ b/disease/version.py
@@ -1,2 +1,2 @@
 """Disease normalizer version"""
-__version__ = "0.3.2"
+__version__ = "0.3.3.dev0"
diff --git a/requirements-dev.txt b/requirements-dev.txt
index aa55371a..c8265b02 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -4,13 +4,13 @@ appnope==0.1.3 ; platform_system == 'Darwin'
 asttokens==2.2.1
 backcall==0.2.0
 beautifulsoup4==4.12.2 ; python_full_version >= '3.6.0'
-bioregistry[align]==0.8.14 ; python_version >= '3.7'
-bioversions==0.5.110
+bioregistry[align]==0.9.11 ; python_version >= '3.7'
+bioversions==0.5.131
 black==23.3.0
-boto3==1.26.116
-botocore==1.29.116 ; python_version >= '3.7'
+boto3==1.26.133
+botocore==1.29.133 ; python_version >= '3.7'
 cachier==2.1.1
-certifi==2022.12.7 ; python_version >= '3.6'
+certifi==2023.5.7 ; python_version >= '3.6'
 cfgv==3.3.1 ; python_full_version >= '3.6.1'
 charset-normalizer==3.1.0 ; python_full_version >= '3.7.0'
 class-resolver==0.4.2
@@ -20,7 +20,7 @@ comm==0.1.3 ; python_version >= '3.6'
 contourpy==1.0.7 ; python_version >= '3.8'
 coverage==6.5.0
 coveralls==3.3.1
-curies==0.5.3 ; python_version >= '3.7'
+curies==0.5.4 ; python_version >= '3.7'
 cycler==0.11.0 ; python_version >= '3.6'
 dataclasses-json==0.5.7 ; python_version >= '3.6'
 debugpy==1.6.7 ; python_version >= '3.7'
@@ -36,16 +36,16 @@ fastapi==0.95.1
 filelock==3.12.0 ; python_version >= '3.7'
 flake8==6.0.0
 flake8-docstrings==1.7.0
-fonttools==4.39.3 ; python_version >= '3.8'
+fonttools==4.39.4 ; python_version >= '3.8'
 ga4gh.vrsatile.pydantic==0.1.0.dev8
 h11==0.14.0 ; python_version >= '3.7'
-identify==2.5.22 ; python_version >= '3.7'
+identify==2.5.24 ; python_version >= '3.7'
 idna==3.4 ; python_version >= '3.5'
-importlib-metadata==6.5.0 ; python_version < '3.10'
+importlib-metadata==6.6.0 ; python_version < '3.10'
 importlib-resources==5.12.0 ; python_version < '3.10'
 iniconfig==2.0.0 ; python_version >= '3.7'
-ipykernel==6.22.0
-ipython==8.12.0 ; python_version >= '3.8'
+ipykernel==6.23.0
+ipython==8.13.2 ; python_version >= '3.9'
 isodate==0.6.1
 isort==5.12.0
 jedi==0.18.2 ; python_version >= '3.6'
@@ -63,8 +63,8 @@ more-click==0.1.2 ; python_version >= '3.7'
 more-itertools==9.1.0 ; python_version >= '3.7'
 mypy-extensions==1.0.0 ; python_version >= '3.5'
 nest-asyncio==1.5.6 ; python_version >= '3.5'
-nodeenv==1.7.0 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5, 3.6'
-numpy==1.24.2 ; python_version >= '3.8'
+nodeenv==1.8.0 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5, 3.6'
+numpy==1.24.3 ; python_version >= '3.8'
 owlready2==0.41
 packaging==23.1 ; python_version >= '3.7'
 parso==0.8.3 ; python_version >= '3.6'
@@ -73,12 +73,14 @@ pathtools==0.1.2
 pexpect==4.8.0 ; sys_platform != 'win32'
 pickleshare==0.7.5
 pillow==9.5.0 ; python_version >= '3.7'
-platformdirs==3.2.0 ; python_version >= '3.7'
+platformdirs==3.5.1 ; python_version >= '3.7'
 pluggy==1.0.0 ; python_version >= '3.6'
 portalocker==2.7.0 ; python_version >= '3.5'
-pre-commit==3.2.2
+pre-commit==3.3.1
 prompt-toolkit==3.0.38 ; python_full_version >= '3.7.0'
 psutil==5.9.5 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
+psycopg[binary]==3.1.9
+psycopg-binary==3.1.9
 psycopg2-binary==2.9.6 ; python_version >= '3.6'
 ptyprocess==0.7.0 ; os_name != 'nt'
 pure-eval==0.2.2
@@ -96,10 +98,10 @@ pytrie==0.4.0
 pyyaml==6.0 ; python_version >= '3.6'
 pyzmq==25.0.2 ; python_version >= '3.6'
 rdflib==6.3.2
-requests==2.28.2
+requests==2.30.0
 requests-ftp==0.3.1
-s3transfer==0.6.0 ; python_version >= '3.7'
-setuptools==67.6.1 ; python_version >= '3.7'
+s3transfer==0.6.1 ; python_version >= '3.7'
+setuptools==67.7.2 ; python_version >= '3.7'
 six==1.16.0 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 sniffio==1.3.0 ; python_version >= '3.7'
 snowballstemmer==2.2.0
@@ -109,14 +111,14 @@ stack-data==0.6.2
 starlette==0.26.1 ; python_version >= '3.7'
 tabulate==0.9.0 ; python_version >= '3.7'
 tomli==2.0.1 ; python_version < '3.11'
-tornado==6.3 ; python_version >= '3.8'
+tornado==6.3.1 ; python_version >= '3.8'
 tqdm==4.65.0 ; python_version >= '3.7'
 traitlets==5.9.0 ; python_version >= '3.7'
 typing-extensions==4.5.0 ; python_version >= '3.7'
 typing-inspect==0.8.0
 urllib3==1.26.15 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'
-uvicorn==0.21.1
-virtualenv==20.22.0 ; python_version >= '3.7'
+uvicorn==0.22.0
+virtualenv==20.23.0 ; python_version >= '3.7'
 watchdog==3.0.0 ; python_version >= '3.7'
 wcwidth==0.2.6
 xmlformatter==0.2.4
diff --git a/requirements.txt b/requirements.txt
index 4158c051..9911cfa2 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,7 +1,7 @@
 -i https://pypi.org/simple
 anyio==3.6.2 ; python_full_version >= '3.6.2'
-boto3==1.26.116
-botocore==1.29.116 ; python_version >= '3.7'
+boto3==1.26.133
+botocore==1.29.133 ; python_version >= '3.7'
 click==8.1.3
 fastapi==0.95.1
 ga4gh.vrsatile.pydantic==0.1.0.dev8
@@ -10,10 +10,10 @@ idna==3.4 ; python_version >= '3.5'
 jmespath==1.0.1 ; python_version >= '3.7'
 pydantic==1.10.7
 python-dateutil==2.8.2 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
-s3transfer==0.6.0 ; python_version >= '3.7'
+s3transfer==0.6.1 ; python_version >= '3.7'
 six==1.16.0 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'
 sniffio==1.3.0 ; python_version >= '3.7'
 starlette==0.26.1 ; python_version >= '3.7'
 typing-extensions==4.5.0 ; python_version >= '3.7'
 urllib3==1.26.15 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'
-uvicorn==0.21.1
+uvicorn==0.22.0
diff --git a/setup.cfg b/setup.cfg
index 5e0e3c92..e49b2253 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -31,13 +31,26 @@ install_requires =
     ga4gh.vrsatile.pydantic >= 0.1.dev3
     click
 
+[options.package_data]
+disease =
+    database/postgresql/*
+
+[options.entry_points]
+console_scripts =
+    disease_norm_update = disease.cli:update_normalizer_db
+    disease_norm_update_remote = disease.cli:update_from_remote
+    disease_norm_dump = disease.cli:dump_database
+    disease_norm_check_db = disease.cli:check_db
+
 [options.extras_require]
-dev =
+etl =
     owlready2==0.40
     rdflib
     requests
     typing-extensions
     bioversions
+
+dev =
     pre-commit
     flake8
     flake8-docstrings
@@ -46,6 +59,9 @@ dev =
     lxml
     xmlformatter
 
+pg =
+    psycopg[binary]
+
 test =
     pytest
     pytest-cov
@@ -54,4 +70,4 @@ test =
 
 [tool:pytest]
 testpaths = tests/unit
-addopts = --ignore=setup.py --ignore=codebuild/ --ignore=analysis/ --doctest-modules --cov-report term-missing --cov .
+addopts = --ignore=setup.py --ignore=analysis/ --ignore=tests/scripts/ --doctest-modules --cov-report term-missing --cov=.


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 4a0efea0..bafbb2a5 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,4 +1,5 @@
 """Pytest test config tools."""
+import logging
 import os
 import tarfile
 from pathlib import Path
@@ -6,11 +7,14 @@
 
 import pytest
 
-from disease.database import AWS_ENV_VAR_NAME, Database
+from disease.database import AWS_ENV_VAR_NAME, create_db
+from disease.database.database import AbstractDatabase
 from disease.etl.base import Base
 from disease.query import QueryHandler
 from disease.schemas import Disease, MatchesKeyed, MatchType, SourceName
 
+_logger = logging.getLogger(__name__)
+
 
 def pytest_collection_modifyitems(items):
     """Modify test items in place to ensure test modules run in a given order.
@@ -34,30 +38,35 @@ def pytest_collection_modifyitems(items):
 
 TEST_ROOT = Path(__file__).resolve().parents[1]
 TEST_DATA_DIRECTORY = TEST_ROOT / "tests" / "data"
+IS_TEST_ENV = os.environ.get("DISEASE_TEST", "").lower() == "true"
+
+
+def pytest_sessionstart():
+    """Wipe DB before testing if in test environment."""
+    if IS_TEST_ENV:
+        if os.environ.get(AWS_ENV_VAR_NAME):
+            assert False, f"Cannot have both DISEASE_TEST and {AWS_ENV_VAR_NAME} set."
+        db = create_db()
+        db.drop_db()
+        db.initialize_db()
 
 
 @pytest.fixture(scope="session")
-def test_data():
-    """Provide test data location to test modules"""
-    return TEST_DATA_DIRECTORY
+def is_test_env():
+    """If true, currently in test environment (i.e. okay to overwrite DB). Downstream
+    users should also make sure to check if in a production environment.
+
+    Provided here to be accessible directly within test modules.
+    """
+    return IS_TEST_ENV
 
 
-@pytest.fixture(scope="session", autouse=True)
-def db():
+@pytest.fixture(scope="module")
+def database():
     """Provide a database instance to be used by tests."""
-    database = Database()
-    if os.environ.get("DISEASE_TEST", "").lower() == "true":
-        if os.environ.get(AWS_ENV_VAR_NAME):
-            assert False, f"Cannot have both DISEASE_TEST and {AWS_ENV_VAR_NAME} set."
-        existing_tables = database.dynamodb_client.list_tables()["TableNames"]
-        if "disease_concepts" in existing_tables:
-            database.dynamodb_client.delete_table(TableName="disease_concepts")
-        if "disease_metadata" in existing_tables:
-            database.dynamodb_client.delete_table(TableName="disease_metadata")
-        existing_tables = database.dynamodb_client.list_tables()["TableNames"]
-        database.create_diseases_table(existing_tables)
-        database.create_meta_data_table(existing_tables)
-    return database
+    db = create_db()
+    yield db
+    db.close_connection()
 
 
 def decompress_mondo_tar():
@@ -83,28 +92,28 @@ def decompress_mondo_tar():
             raise FileNotFoundError("Unable to find compressed Mondo OWL file")
 
 
-@pytest.fixture(scope="session")
-def test_source(db: Database, test_data: Path):
+@pytest.fixture(scope="module")
+def test_source(database: AbstractDatabase, is_test_env: bool):
     """Provide query endpoint for testing sources. If DISEASE_TEST is set, will try to
     load DB from test data.
 
-    :param db: database fixture
-    :param test_data: test data directory location
+    :param database: test database instance
+    :param is_test_env: if true, load from test data
     :return: factory function that takes an ETL class instance and returns a query
     endpoint.
     """
 
     def test_source_factory(EtlClass: Base):
-        if os.environ.get("DISEASE_TEST", "").lower() == "true":
-            test_class = EtlClass(db, test_data)  # type: ignore
+        if IS_TEST_ENV:
+            _logger.debug(f"Reloading DB with data from {TEST_DATA_DIRECTORY}")
+            test_class = EtlClass(database, TEST_DATA_DIRECTORY)  # type: ignore
             if EtlClass.__name__ == SourceName.MONDO:  # type: ignore
                 decompress_mondo_tar()
             test_class.perform_etl(use_existing=True)
-            test_class.database.flush_batch()
 
         class QueryGetter:
             def __init__(self):
-                self.query_handler = QueryHandler()
+                self.query_handler = QueryHandler(database)
                 self._src_name = EtlClass.__name__  # type: ignore
 
             def search(self, query_str: str):
diff --git a/tests/scripts/build_do_data.py b/tests/scripts/build_do_data.py
index f81f24b0..a937603c 100644
--- a/tests/scripts/build_do_data.py
+++ b/tests/scripts/build_do_data.py
@@ -6,10 +6,10 @@
 
 import requests
 
-from disease.database import Database
+from disease.database import create_db
 from disease.etl import DO
 
-do = DO(Database())
+do = DO(create_db())
 do._extract_data()
 
 infile = do._data_file.absolute()
diff --git a/tests/scripts/build_mondo_data.py b/tests/scripts/build_mondo_data.py
index a44050ae..a3cc5bb3 100644
--- a/tests/scripts/build_mondo_data.py
+++ b/tests/scripts/build_mondo_data.py
@@ -13,10 +13,10 @@
 
 import requests
 
-from disease.database import Database
+from disease.database import create_db
 from disease.etl import Mondo
 
-mondo = Mondo(Database())
+mondo = Mondo(create_db())
 mondo._extract_data()
 
 infile = mondo._data_file.absolute()
diff --git a/tests/scripts/build_ncit_data.py b/tests/scripts/build_ncit_data.py
index f3203a63..72b14476 100644
--- a/tests/scripts/build_ncit_data.py
+++ b/tests/scripts/build_ncit_data.py
@@ -7,7 +7,7 @@
 import owlready2 as owl
 import xmlformatter
 
-from disease.database import Database
+from disease.database import create_db
 from disease.etl import NCIt
 
 # define captured ids in `test_classes` variable
@@ -26,7 +26,7 @@
 CLASS_TAG = f"{OWL_PREFIX}Class"
 
 
-ncit = NCIt(Database())
+ncit = NCIt(create_db())
 ncit._extract_data()
 TEST_DATA_DIR = Path(__file__).resolve().parents[1] / "data" / "ncit"
 outfile_path = TEST_DATA_DIR / ncit._data_file.name
diff --git a/tests/scripts/build_omim_data.py b/tests/scripts/build_omim_data.py
index 4455211b..5ef5b1a7 100644
--- a/tests/scripts/build_omim_data.py
+++ b/tests/scripts/build_omim_data.py
@@ -2,7 +2,7 @@
 import csv
 from pathlib import Path
 
-from disease.database import Database
+from disease.database import create_db
 from disease.etl import OMIM
 
 TEST_IDS = [
@@ -11,7 +11,7 @@
     "247640",
 ]
 
-omim = OMIM(Database())
+omim = OMIM(create_db())
 omim._extract_data()
 TEST_DATA_DIR = Path(__file__).resolve().parents[1] / "data" / "omim"
 outfile_path = TEST_DATA_DIR / omim._data_file.name
diff --git a/tests/scripts/build_oncotree_data.py b/tests/scripts/build_oncotree_data.py
index 4a44348d..77f4ab04 100644
--- a/tests/scripts/build_oncotree_data.py
+++ b/tests/scripts/build_oncotree_data.py
@@ -5,10 +5,10 @@
 import shutil
 from pathlib import Path
 
-from disease.database import Database
+from disease.database import create_db
 from disease.etl import OncoTree
 
-oncotree = OncoTree(Database())
+oncotree = OncoTree(create_db())
 oncotree._extract_data()
 TEST_DATA_DIR = Path(__file__).resolve().parents[1] / "data" / "oncotree"
 outfile_path = TEST_DATA_DIR / oncotree._data_file.name
diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py
index 14aef0ae..46ee136d 100644
--- a/tests/unit/test_database.py
+++ b/tests/unit/test_database.py
@@ -1,42 +1,61 @@
-"""Test DynamoDB"""
-from boto3.dynamodb.conditions import Key
+"""Test database classes."""
+import os
 
+import pytest
 
-def test_tables_created(db):
-    """Check that disease_concepts and disease_metadata are created."""
-    existing_tables = db.dynamodb_client.list_tables()["TableNames"]
-    assert "disease_concepts" in existing_tables
-    assert "disease_metadata" in existing_tables
 
+def test_tables_created(database):
+    """Check that required tables are created."""
+    existing_tables = database.list_tables()
+    if database.__class__.__name__ == "PostgresDatabase":
+        assert set(existing_tables) == {
+            "disease_associations",
+            "disease_labels",
+            "disease_aliases",
+            "disease_xrefs",
+            "disease_concepts",
+            "disease_merged",
+            "disease_sources",
+        }
+    else:
+        assert "disease_concepts" in existing_tables
+        assert "disease_metadata" in existing_tables
 
-def test_item_type(db):
+
+IS_DDB = not os.environ.get("DISEASE_NORM_DB_URL", "").lower().startswith("postgres")
+
+
+@pytest.mark.skipif(not IS_DDB, reason="only applies to DynamoDB in test env")
+def test_item_type(database):
     """Check that objects are tagged with item_type attribute."""
+    from boto3.dynamodb.conditions import Key
+
     filter_exp = Key("label_and_type").eq("ncit:c2926##identity")
-    item = db.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
+    item = database.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
     assert "item_type" in item
     assert item["item_type"] == "identity"
 
     filter_exp = Key("label_and_type").eq("neuroblastoma##label")
-    item = db.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
+    item = database.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
     assert "item_type" in item
     assert item["item_type"] == "label"
 
     filter_exp = Key("label_and_type").eq("umls:c0027819##associated_with")
-    item = db.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
+    item = database.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
     assert "item_type" in item
     assert item["item_type"] == "associated_with"
 
     filter_exp = Key("label_and_type").eq("oncotree:nbl##xref")
-    item = db.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
+    item = database.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
     assert "item_type" in item
     assert item["item_type"] == "xref"
 
     filter_exp = Key("label_and_type").eq("childhood liposarcoma##alias")
-    item = db.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
+    item = database.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
     assert "item_type" in item
     assert item["item_type"] == "alias"
 
     filter_exp = Key("label_and_type").eq("ncit:c9012##merger")
-    item = db.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
+    item = database.diseases.query(KeyConditionExpression=filter_exp)["Items"][0]
     assert "item_type" in item
     assert item["item_type"] == "merger"
diff --git a/tests/unit/test_emit_warnings.py b/tests/unit/test_emit_warnings.py
index 0489db52..37ca6f18 100644
--- a/tests/unit/test_emit_warnings.py
+++ b/tests/unit/test_emit_warnings.py
@@ -2,10 +2,10 @@
 from disease.query import QueryHandler
 
 
-def test_emit_warnings():
+def test_emit_warnings(database):
     """Test that emit_warnings works correctly."""
     expected_warnings = {"nbsp": "Query contains non breaking space characters."}
-    query_handler = QueryHandler()
+    query_handler = QueryHandler(database)
 
     # Test emit no warnings
     actual_warnings = query_handler._emit_warnings("CISPLATIN")
diff --git a/tests/unit/test_merge.py b/tests/unit/test_merge.py
index a6aa8630..ad24296a 100644
--- a/tests/unit/test_merge.py
+++ b/tests/unit/test_merge.py
@@ -1,52 +1,41 @@
 """Test merged record generation."""
 import os
-from typing import Any, Dict
+from typing import Callable
 
 import pytest
 
-from disease.database import AWS_ENV_VAR_NAME, Database
+from disease import SOURCES_FOR_MERGE
+from disease.database import AWS_ENV_VAR_NAME
+from disease.database.database import create_db
 from disease.etl import DO, OMIM, Merge, Mondo, NCIt, OncoTree
+from disease.schemas import SourceName
 
 
 @pytest.fixture(scope="module")
-def merge_instance(test_source):
-    """Provide fixture for ETL merge class"""
-    update_db = os.environ.get("DISEASE_TEST", "").lower() == "true"
-    if update_db and os.environ.get(AWS_ENV_VAR_NAME):
-        assert False, (
-            f"Running the full disease ETL pipeline test on an AWS environment is "
-            f"forbidden -- either unset {AWS_ENV_VAR_NAME} or unset DISEASE_TEST"
-        )
-
-    class TrackingDatabase(Database):
-        """Provide injection for DB instance to track added/updated records"""
-
-        def __init__(self, **kwargs):
-            self.additions = {}
-            self.updates = {}
-            super().__init__(**kwargs)
-
-        def add_record(self, record: Dict, record_type: str):
-            if update_db:
-                super().add_record(record, record_type)
-            self.additions[record["concept_id"]] = record
-
-        def update_record(
-            self,
-            concept_id: str,
-            field: str,
-            new_value: Any,  # noqa: ANN401
-            item_type: str = "identity",
-        ):
-            if update_db:
-                super().update_record(concept_id, field, new_value, item_type)
-            self.updates[concept_id] = {field: new_value}
-
-    if update_db:
-        for SourceClass in (DO, Mondo, NCIt, OncoTree, OMIM):
-            test_source(SourceClass)
-
-    m = Merge(TrackingDatabase())
+def merge_instance(test_source: Callable, is_test_env: bool):
+    """Provide fixture for ETL merge class.
+
+    If in a test environment (e.g. CI) this method will attempt to load any missing
+    source data, and then perform merged record generation.
+    """
+    database = create_db()
+    if is_test_env:
+        if os.environ.get(AWS_ENV_VAR_NAME):
+            assert False, (
+                f"Running the full disease ETL pipeline test on an AWS environment is "
+                f"forbidden -- either unset {AWS_ENV_VAR_NAME} or unset DISEASE_TEST"
+            )
+        else:
+            for SourceClass in (Mondo, DO, NCIt, OncoTree, OMIM):
+                if not database.get_source_metadata(SourceName(SourceClass.__name__)):
+                    test_source(SourceClass)
+
+    m = Merge(database)
+    if is_test_env:
+        concept_ids = set()
+        for source in SOURCES_FOR_MERGE:
+            concept_ids |= database.get_all_concept_ids(source)
+        m.create_merged_concepts(concept_ids)
     return m
 
 
@@ -54,7 +43,6 @@ def update_record(
 def neuroblastoma():
     """Create neuroblastoma fixture."""
     return {
-        "label_and_type": "ncit:c3270##merger",
         "concept_id": "ncit:C3270",
         "item_type": "merger",
         "xrefs": ["mondo:0005072", "oncotree:NBL", "DOID:769"],
@@ -90,7 +78,6 @@ def neuroblastoma():
 def lnscc():
     """Create lung non small cell carcinoma fixture"""
     return {
-        "label_and_type": "ncit:c2926##merger",
         "concept_id": "ncit:C2926",
         "xrefs": ["mondo:0005233", "oncotree:NSCLC", "DOID:3908"],
         "label": "Lung Non-Small Cell Carcinoma",
@@ -130,7 +117,6 @@ def lnscc():
 def richter():
     """Create Richter Syndrome fixture"""
     return {
-        "label_and_type": "ncit:c35424##merger",
         "concept_id": "ncit:C35424",
         "xrefs": ["mondo:0002083", "DOID:1703"],
         "label": "Richter Syndrome",
@@ -156,7 +142,6 @@ def richter():
 def ped_liposarcoma():
     """Create pediatric liposarcoma fixture."""
     return {
-        "label_and_type": "ncit:c8091##merger",
         "concept_id": "ncit:C8091",
         "xrefs": ["mondo:0003587", "DOID:5695"],
         "label": "Childhood Liposarcoma",
@@ -176,7 +161,6 @@ def ped_liposarcoma():
 def teratoma():
     """Create fixture for adult cystic teratoma."""
     return {
-        "label_and_type": "ncit:c9012##merger",
         "concept_id": "ncit:C9012",
         "xrefs": ["mondo:0004099", "DOID:7079"],
         "label": "Adult Cystic Teratoma",
@@ -192,7 +176,6 @@ def mafd2():
     deprecated DO reference is filtered out.
     """
     return {
-        "label_and_type": "mondo:0010648##merger",
         "item_type": "merger",
         "concept_id": "mondo:0010648",
         "label": "major affective disorder 2",
@@ -233,8 +216,6 @@ def compare_merged_records(actual, fixture):
     if "xrefs" in actual:
         assert set(actual["xrefs"]) == set(fixture["xrefs"])
 
-    assert actual["label_and_type"] == fixture["label_and_type"]
-
     assert ("label" in actual) == ("label" in fixture)
     if "label" in actual or "label" in fixture:
         assert actual["label"] == fixture["label"]
@@ -292,48 +273,3 @@ def test_generate_merged_record(
     response, r_ids = merge_instance._generate_merged_record(mafd2_ids)
     assert set(r_ids) == {"mondo:0010648", "omim:309200"}
     compare_merged_records(response, mafd2)
-
-
-def test_create_merged_concepts(
-    merge_instance, neuroblastoma, lnscc, richter, ped_liposarcoma, teratoma, mafd2
-):
-    """Test end-to-end creation and upload of merged concepts."""
-    mondo_ids = [
-        "mondo:0005072",
-        "mondo:0005233",
-        "mondo:0002083",
-        "mondo:0004099",
-        "mondo:0003587",
-        "mondo:0010648",
-    ]
-    merge_instance.create_merged_concepts(mondo_ids)
-    merge_instance._database.flush_batch()
-
-    # check merged record generation and storage
-    added_records = merge_instance._database.additions
-
-    assert len(added_records) == 6
-
-    neuroblastoma_id = neuroblastoma["concept_id"]
-    assert neuroblastoma_id in added_records
-    compare_merged_records(added_records[neuroblastoma_id], neuroblastoma)
-
-    lnscc_id = lnscc["concept_id"]
-    assert lnscc_id in added_records
-    compare_merged_records(added_records[lnscc_id], lnscc)
-
-    richter_id = richter["concept_id"]
-    assert richter_id in added_records
-    compare_merged_records(added_records[richter_id], richter)
-
-    ped_liposarcoma_id = ped_liposarcoma["concept_id"]
-    assert ped_liposarcoma_id in added_records
-    compare_merged_records(added_records[ped_liposarcoma_id], ped_liposarcoma)
-
-    teratoma_id = teratoma["concept_id"]
-    assert teratoma_id in added_records
-    compare_merged_records(added_records[teratoma_id], teratoma)
-
-    mafd2_id = mafd2["concept_id"]
-    assert mafd2_id in added_records
-    compare_merged_records(added_records[mafd2_id], mafd2)
diff --git a/tests/unit/test_omim.py b/tests/unit/test_omim.py
index 72f78555..a704c4eb 100644
--- a/tests/unit/test_omim.py
+++ b/tests/unit/test_omim.py
@@ -100,10 +100,7 @@ def test_label_match(omim, mafd2, acute_ll, lall, compare_response):
 
 def test_alias_match(omim, mafd2, acute_ll, lall, compare_response):
     """Test alias search resolution."""
-    response = omim.search("bipolar affective disorder")
-    compare_response(response, MatchType.ALIAS, mafd2)
-
-    response = omim.search("bpad")
+    response = omim.search("manic-depressive psychosis, x-linked")
     compare_response(response, MatchType.ALIAS, mafd2)
 
     response = omim.search("mafd2")
diff --git a/tests/unit/test_query.py b/tests/unit/test_query.py
index c90c772c..f57c9894 100644
--- a/tests/unit/test_query.py
+++ b/tests/unit/test_query.py
@@ -9,9 +9,9 @@
 
 
 @pytest.fixture(scope="module")
-def query_handler():
+def query_handler(database):
     """Build query handler test fixture"""
-    return QueryHandler()
+    return QueryHandler(database)
 
 
 @pytest.fixture(scope="module")