(aggregate_table_adapter, window)
For a given period (start/end) - populate all data in the aggregate table associated
    with that period.
For a given period (start/end) - populate all data in the aggregate table associated
    with that period.
For a given period ( start / end ) - populate all data in the aggregate table associated with that period .
def populate_aggregate_table_data_for_time_period(aggregate_table_adapter, window):
    """
    For a given period (start/end) - populate all data in the aggregate table associated
    with that period.
    """
    doing_time_aggregation = window is not None
    if doing_time_aggregation:
        aggregation_params = {
            window.start.name: window.start.value,
            window.end.name: window.end.value,
        }
    else:
        aggregation_params = {}

    primary_column_adapters = list(aggregate_table_adapter.config.get_primary_column_adapters())
    primary_table = get_indicator_adapter(aggregate_table_adapter.config.data_source).get_table()
    all_query_columns = [
        pca.to_sqlalchemy_query_column(primary_table, aggregation_params) for pca in primary_column_adapters
    ]
    for secondary_table, sqlalchemy_secondary_table in aggregate_table_adapter.config.get_secondary_tables_and_adapters():
        for column_adapter in secondary_table.get_column_adapters():
            all_query_columns.append(
                column_adapter.to_sqlalchemy_query_column(sqlalchemy_secondary_table, aggregation_params)
            )

    select_statement = sqlalchemy.select(
        all_query_columns
    )

    # now construct join
    select_table = primary_table
    for secondary_table, sqlalchemy_secondary_table in aggregate_table_adapter.config.get_secondary_tables_and_adapters():
        # apply join filters along with period start/end filters (if necessary) for related model
        join_conditions = [
            (primary_table.c[secondary_table.join_column_primary] ==
             sqlalchemy_secondary_table.c[secondary_table.join_column_secondary])
        ]
        if doing_time_aggregation:
            join_conditions.extend([
                sqlalchemy_secondary_table.c[secondary_table.time_window_column] >= window.start.value,
                sqlalchemy_secondary_table.c[secondary_table.time_window_column] < window.end.value
            ])
        select_table = select_table.outerjoin(
            sqlalchemy_secondary_table,
            sqlalchemy.and_(*join_conditions)
        )

    select_statement = select_statement.select_from(select_table)
    # apply period start/end filters for primary model
    # to match, start should be before the end of the period and end should be after the start
    # this makes the first and last periods inclusive.
    if doing_time_aggregation:
        select_statement = select_statement.where(
            primary_table.c[window.start.mapped_column_id] < window.end.value
        )
        select_statement = select_statement.where(
            sqlalchemy.or_(primary_table.c[window.end.mapped_column_id] == None,  # noqa this is sqlalchemy
                           primary_table.c[window.end.mapped_column_id] >= window.start.value))

    for primary_column_adapter in primary_column_adapters:
        if primary_column_adapter.is_groupable():
            select_statement = select_statement.group_by(
                primary_column_adapter.to_sqlalchemy_query_column(
                    primary_table, aggregation_params
                )
            )

    aggregate_table = aggregate_table_adapter.get_table()
    primary_key_columns = [
        aggregate_table.c[spec.column_id] for spec in primary_column_adapters if spec.is_primary_key()
    ]
    secondary_column_ids = [
        spec.column_id for spec in aggregate_table_adapter.config.get_column_adapters()
        if not spec.is_primary_key()
    ]
    insert_statement = insert(aggregate_table).from_select(
        aggregate_table.c, select_statement
    )
    insert_statement = insert_statement.on_conflict_do_update(
        index_elements=primary_key_columns,
        set_={
            k: insert_statement.excluded[k] for k in secondary_column_ids
        }

    )

    with aggregate_table_adapter.session_helper.session_context() as session:
        session.execute(insert_statement)
def populate_aggregate_table_data_for_time_period ( aggregate_table_adapter , window ) : doing_time_aggregation = window is not None if doing_time_aggregation : aggregation_params = { window . start . name : window . start . value , window . end . name : window . end . value , } else : aggregation_params = { } primary_column_adapters = list ( aggregate_table_adapter . config . get_primary_column_adapters ( ) ) primary_table = get_indicator_adapter ( aggregate_table_adapter . config . data_source ) . get_table ( ) all_query_columns = [ pca . to_sqlalchemy_query_column ( primary_table , aggregation_params ) for pca in primary_column_adapters ] for secondary_table , sqlalchemy_secondary_table in aggregate_table_adapter . config . get_secondary_tables_and_adapters ( ) : for column_adapter in secondary_table . get_column_adapters ( ) : all_query_columns . append ( column_adapter . to_sqlalchemy_query_column ( sqlalchemy_secondary_table , aggregation_params ) ) select_statement = sqlalchemy . select ( all_query_columns ) # now construct join select_table = primary_table for secondary_table , sqlalchemy_secondary_table in aggregate_table_adapter . config . get_secondary_tables_and_adapters ( ) : # apply join filters along with period start/end filters (if necessary) for related model join_conditions = [ ( primary_table . c [ secondary_table . join_column_primary ] == sqlalchemy_secondary_table . c [ secondary_table . join_column_secondary ] ) ] if doing_time_aggregation : join_conditions . extend ( [ sqlalchemy_secondary_table . c [ secondary_table . time_window_column ] >= window . start . value , sqlalchemy_secondary_table . c [ secondary_table . time_window_column ] < window . end . value ] ) select_table = select_table . outerjoin ( sqlalchemy_secondary_table , sqlalchemy . and_ ( * join_conditions ) ) select_statement = select_statement . select_from ( select_table ) # apply period start/end filters for primary model # to match, start should be before the end of the period and end should be after the start # this makes the first and last periods inclusive. if doing_time_aggregation : select_statement = select_statement . where ( primary_table . c [ window . start . mapped_column_id ] < window . end . value ) select_statement = select_statement . where ( sqlalchemy . or_ ( primary_table . c [ window . end . mapped_column_id ] == None , # noqa this is sqlalchemy primary_table . c [ window . end . mapped_column_id ] >= window . start . value ) ) for primary_column_adapter in primary_column_adapters : if primary_column_adapter . is_groupable ( ) : select_statement = select_statement . group_by ( primary_column_adapter . to_sqlalchemy_query_column ( primary_table , aggregation_params ) ) aggregate_table = aggregate_table_adapter . get_table ( ) primary_key_columns = [ aggregate_table . c [ spec . column_id ] for spec in primary_column_adapters if spec . is_primary_key ( ) ] secondary_column_ids = [ spec . column_id for spec in aggregate_table_adapter . config . get_column_adapters ( ) if not spec . is_primary_key ( ) ] insert_statement = insert ( aggregate_table ) . from_select ( aggregate_table . c , select_statement ) insert_statement = insert_statement . on_conflict_do_update ( index_elements = primary_key_columns , set_ = { k : insert_statement . excluded [ k ] for k in secondary_column_ids } ) with aggregate_table_adapter . session_helper . session_context ( ) as session : session . execute ( insert_statement )