我想使用单个资源(一个 csv 文件)创建一个数据集“人道主义数据交换”(HDX)。我想使用 HDX Python API。我已经查看了文档,但需要一个更完整的示例来说明如何做到这一点。如何创建数据集?
要运行下面的代码,您需要提供命令行参数
hdx_token
,这是您的 API 令牌。您需要通过访问
https://data.humdata.org/user/xxx/api-tokens将 xxx 替换为您的用户名来为您的用户生成 API 令牌。您还可以提供
hdx_site
,默认为“stage”。 hdx_site
可以“prod”上传到主要生产 HDX 站点 (https://data.humdata.org/)。您可以使用“stage”或“feature”来测试上传到我们的测试服务器之一(https://stage.data-humdata-org.ahconu.org/ 或 https://feature.data-humdata- org.ahconu.org/). 如果使用测试服务器,需要记住的一件事是它们会定期从生产环境中刷新,因此新创建的用户、组织、API 令牌或生产数据集可能尚不存在于测试服务器上。
在下面的代码中,无论您在何处看到“我的组织”,都应将其替换为您的组织名称。
您需要使用以下方式获取与您的用户关联的 ID:
https://data.humdata.org/api/3/action/user_show?id=xxx 其中 xxx 是您的用户名,并将其用作传递给的参数dataset.set_maintainer
电话。
其中 xxx 是组织名称。为了能够上传到组织,您需要是该组织的编辑或管理员。该 id 应该用作传递给 dataset.set_organization
调用的参数。
#!/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"
) # user id
dataset.set_organization(
"xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
) # organisation id
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",
)