run_fetch_dac5_codes.py

from collections import defaultdict
from typing import Dict, List, Set

from sqlalchemy.orm.session import Session
from db.config import AppSession
from db.models import Project
from ingest.assistance.iati.core import (
    DPortalTable,
    get_act_data_rows_from_dportal_api,
)


def __collate_sector_codes(res, collated_res, sector_field_name):
    for r in res:
        if r[sector_field_name] not in (None, ""):
            collated_res[r["aid"]].add(r[sector_field_name])


if __name__ == "__main__":

    session: Session = AppSession()
    aid_codes = (
        session.query(Project.notes)
        .filter(("IATI Registry" == Project.sources.any_()))
        .all()
    )
    aid_codes = [t[0] for t in aid_codes]

    res_trans = get_act_data_rows_from_dportal_api(
        aid_codes,
        fields=["aid", "trans_sector"],
        table=DPortalTable.trans,
    )
    res = get_act_data_rows_from_dportal_api(
        aid_codes,
        fields=["aid", "sector_code"],
        table=DPortalTable.sector,
    )
    # collate
    collated_res: Dict[str, Set[str]] = defaultdict(set)

    __collate_sector_codes(res, collated_res, "sector_code")
    __collate_sector_codes(res_trans, collated_res, "trans_sector")
    sqls: List[str] = list()
    with open("dac_5_sector_codes_by_aid.csv", "w") as file:
        file.write('"aid","sector_codes"\n')
        for r in collated_res:
            code_list: str = ",".join(sorted(collated_res[r]))
            file.write(f""""{r}","{code_list}"\n""")