Source code for cms.aggregates

#  Copyright (c) 2021. Harvard University
#
#  Developed by Research Software Engineering,
#  Faculty of Arts and Sciences, Research Computing (FAS RC)
#  Author: Michael A Bouzinier
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

from typing import Dict

import logging

from nsaph import init_logging
from nsaph.db import Connection
from nsaph.loader.common import DBConnectionConfig


[docs]class Aggregator: QUERY = """ SELECT 1 as ORD, 'cms.ps' as table, COUNT(*) FROM "cms"."ps" UNION SELECT 2 as ORD, 'medicaid.beneficiaries' as table, COUNT(*) FROM "medicaid"."beneficiaries" UNION SELECT 3 as ORD, 'medicaid.enrollments' as table, COUNT(*) FROM "medicaid"."enrollments" UNION SELECT 4 as ORD, 'medicaid.eligibility' as table, COUNT(*) FROM "medicaid"."eligibility" ORDER BY 1 """ def __init__(self, context: DBConnectionConfig = None): init_logging() if not context: context = DBConnectionConfig(None, __doc__).instantiate() self.context = context
[docs] def count(self) -> Dict[str,int]: counts = dict() with Connection(self.context.db, self.context.connection, silent=True).connect() as connection: with connection.cursor() as cursor: cursor.execute(self.QUERY) for row in cursor: table = row[1] n = row[2] logging.info("{}: {:,d}".format(table, n)) counts[table] = n return counts
[docs] def verify(self, expected): actual = self.count() for key in expected: if expected[key] != actual[key]: msg = "Verification failed for {}. " \ "Expected: {:,d}; Actual: {:,d}".format( key, expected[key], actual[key] ) raise ValueError(msg) logging.debug("{} count OK".format(key)) logging.info("All counts OK.")
[docs]class ExpectedData: def __init__(self): self.micro_random_counts = { 'cms.ps': 341233, 'medicaid.beneficiaries': 313777, 'medicaid.enrollments': 314272, 'medicaid.eligibility': 3771264 }
if __name__ == '__main__': Aggregator().verify(ExpectedData().micro_random_counts)