src/dirbs/importer/base_delta_importer.py
def _count_changes_by_type_single_shard(self, staging_tbl_shard_name, filter_str=None):
        """Calculate staging table counts per change type for a single shard."""
        changes_by_type = {}
        if filter_str is None:
            filter_sql = sql.SQL('')
        else:
            filter_sql = sql.SQL('WHERE {filter_str}'.format(filter_str=filter_str))

        with utils.create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            cursor.execute(sql.SQL("""SELECT change_type, COUNT(*) AS cnt
                                        FROM {staging_tbl_shard}
                                             {change_filter}
                                    GROUP BY change_type
                                   """).format(staging_tbl_shard=sql.Identifier(staging_tbl_shard_name),
                                               change_filter=filter_sql))
            for res in cursor:
                changes_by_type[res.change_type] = res.cnt

        return changes_by_type
Calculate staging table counts per change type for a single shard.
Calculate staging table counts per change type for a single shard.
Calculate staging table counts per change type for a single shard .
def _count_changes_by_type_single_shard(self, staging_tbl_shard_name, filter_str=None):
        changes_by_type = {}
        if filter_str is None:
            filter_sql = sql.SQL('')
        else:
            filter_sql = sql.SQL('WHERE {filter_str}'.format(filter_str=filter_str))
        with utils.create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            cursor.execute(sql.SQL("""SELECT change_type, COUNT(*) AS cnt
                                        FROM {staging_tbl_shard}
                                             {change_filter}
                                    GROUP BY change_type
                                   """).format(staging_tbl_shard=sql.Identifier(staging_tbl_shard_name),
                                               change_filter=filter_sql))
            for res in cursor:
                changes_by_type[res.change_type] = res.cnt
        return changes_by_type
def _count_changes_by_type_single_shard ( self , staging_tbl_shard_name , filter_str = None ) : changes_by_type = { } if filter_str is None : filter_sql = sql . SQL ( '' ) else : filter_sql = sql . SQL ( 'WHERE {filter_str}' . format ( filter_str = filter_str ) ) with utils . create_db_connection ( self . _db_config ) as conn , conn . cursor ( ) as cursor : cursor . execute ( sql . SQL ( """SELECT change_type, COUNT(*) AS cnt
                                        FROM {staging_tbl_shard}
                                             {change_filter}
                                    GROUP BY change_type
                                   """ ) . format ( staging_tbl_shard = sql . Identifier ( staging_tbl_shard_name ) , change_filter = filter_sql ) ) for res in cursor : changes_by_type [ res . change_type ] = res . cnt return changes_by_type
Calculate staging table counts per change type for a single shard.
Calculate staging table counts per change type for a single shard .
"""Calculate staging table counts per change type for a single shard."""
self
staging_tbl_shard_name
filter_str