我想使用单个资源(一个 csv 文件)创建一个数据集“人道主义数据交换”(HDX)。我想使用 HDX Python API。我已经查看了文档,但需要一个更完整的示例来说明如何做到这一点。如何创建数据集?
其中 xxx 是您的用户名。 您还需要使用以下方式上传您想要上传到的组织的 ID:
https://data.humdata.org/api/3/action/organization_show?id=xxx其中 xxx 是组织名称。为了能够上传到组织,您需要成为该组织的编辑或管理员。 要运行下面的代码,您需要提供命令行参数 hdx_token(您的 API 令牌)和 hdx_site(默认为“stage”)(此处为测试服务器:
https://stage.data-humdata-org.ahconu。组织/)。您可以将其设置为“prod”作为主要生产 HDX 站点。 在下面的代码中,无论您在何处看到“我的组织”,都应将其替换为您的组织名称。
#!/usr/bin/python
"""
Creates a dataset on HDX.
"""
import argparse
import csv
import logging
from os.path import join
from hdx.data.dataset import Dataset
from hdx.data.resource import Resource
from hdx.facades.simple import facade
from hdx.utilities.dateparse import parse_date
from hdx.utilities.path import get_temp_dir
logger = logging.getLogger(__name__)
def main():
"""Generate dataset and create it in HDX"""
dataset = Dataset(
{
"name": "my-test",
"title": "My Test Dataset",
"license_id": "cc-by-igo",
"methodology": "Other",
"private": False,
"dataset_source": "My Org"
}
)
dataset["notes"] = "Long description of dataset goes here!"
dataset["methodology_other"] = "Describe methodology here!"
dataset["caveats"] = "Any caveats or comments about the data go here!"
dataset.set_maintainer("xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")
dataset.set_organization(
"xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
)
dataset.set_expected_update_frequency("Every year")
dataset.set_subnational(False)
dataset.add_tags(["displacement"])
dataset.set_reference_period(parse_date("2020-03-05"), parse_date("2021-02-25"))
dataset.add_country_location("AFG")
# or
dataset.add_country_locations(["AFG"])
# or
dataset.add_other_location("world")
logger.info("Dataset metadata created!")
path = join(get_temp_dir(), "test.csv")
with open(path, "w", encoding="UTF8") as f:
writer = csv.writer(f)
# write the header
writer.writerow(["heading1", "heading2", "heading3", "heading4"])
# write the data
writer.writerow([1, 2, 3, 4])
writer.writerow([5, 6, 7, 8])
logger.info(f"Test file {path} created!")
resource = Resource(
{"name": "test file", "description": "description of test file"}
)
resource.set_file_type("csv")
resource.set_file_to_upload(path)
logger.info("Resource metadata created!")
dataset.add_update_resource(resource)
dataset.create_in_hdx(
remove_additional_resources=True,
updated_by_script="My Org Script",
)
logger.info("Completed!")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="My Script")
parser.add_argument("-ht", "--hdx_token", default=None, help="HDX api token")
parser.add_argument("-hs", "--hdx_site", default=None, help="HDX site to use")
args = parser.parse_args()
hdx_site = args.hdx_site
if hdx_site is None:
hdx_site = "stage"
facade(
main,
hdx_key=args.hdx_token,
hdx_site=hdx_site,
user_agent="My Org",
)