https://dagster.io/ logo
#ask-community
Title
# ask-community
u

吳冠頡

08/01/2023, 6:59 AM
@asset() def Weather_Data(): url = "https://opendata.cwb.gov.tw/api/v1/rest/datastore/F-C0032-001?Authorization=CWB-F4D9DF87-A7E6-4F8E-9EC9-133CAAA4A73D&format=XML" response = requests.get(url) response.raise_for_status() xml_data = response.text root = ET.fromstring(xml_data) weather_data = [] columns = ['locationName', 'elementName', 'startTime', 'endTime', 'parameterName', 'parameterUnit'] for location in root.findall(".//location"): location_name = location.findtext("locationName") for weather_element in location.findall(".//weatherElement"): element_name = weather_element.findtext("elementName") for time_element in weather_element.findall(".//time"): start_time = time_element.findtext("startTime") end_time = time_element.findtext("endTime") for parameter in time_element.findall("parameter"): parameter_name = parameter.findtext("parameterName") #parameter_value = parameter.findtext("parameterValue") parameter_unit = parameter.findtext("parameterUnit") weather_data.append([location_name, element_name, start_time, end_time, parameter_name, parameter_unit]) df = pd.DataFrame(weather_data, columns=columns) return Output( value=df, metadata={ "num_records": len(df), "preview": MetadataValue.md(df.head().to_markdown()), }, )
3 Views