Physics (async)¤
Example script showing how to use the AsyncPhysics class to asynchronously extract information from the KCDB.
Script¤
"""Asynchronously search the "General Physics" metrology domain of the KCDB."""
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING
from msl.kcdb import AsyncPhysics
if TYPE_CHECKING:
from msl.kcdb.types import MetrologyArea, ResultsPhysics
async def all_cmc_results(country: str) -> tuple[list[MetrologyArea], list[ResultsPhysics]]:
"""Asynchronously get all CMCs for a country."""
physics = AsyncPhysics()
areas = await physics.metrology_areas()
searches = [physics.search(area, countries=country, page_size=1000) for area in areas]
results = await asyncio.gather(*searches)
return areas, results
areas, results = asyncio.run(all_cmc_results("DE"))
for area, result in zip(areas, results):
print(area.label, result)
Output¤
Running this script outputs the following, although, some values may change from what you observe when you run this script since information in the KCDB is continually changing.
AUV ResultsPhysics(number_of_elements=52, page_number=0, page_size=1000, total_elements=52, total_pages=1, version_api_kcdb='1.0.13')
EM ResultsPhysics(number_of_elements=185, page_number=0, page_size=1000, total_elements=185, total_pages=1, version_api_kcdb='1.0.13')
L ResultsPhysics(number_of_elements=102, page_number=0, page_size=1000, total_elements=102, total_pages=1, version_api_kcdb='1.0.13')
M ResultsPhysics(number_of_elements=187, page_number=0, page_size=1000, total_elements=187, total_pages=1, version_api_kcdb='1.0.13')
PR ResultsPhysics(number_of_elements=95, page_number=0, page_size=1000, total_elements=95, total_pages=1, version_api_kcdb='1.0.13')
T ResultsPhysics(number_of_elements=119, page_number=0, page_size=1000, total_elements=119, total_pages=1, version_api_kcdb='1.0.13')
TF ResultsPhysics(number_of_elements=25, page_number=0, page_size=1000, total_elements=25, total_pages=1, version_api_kcdb='1.0.13')