HealthLake 用SDK于 Python 的示例 (Boto3) - AWS SDK代码示例

AWS 文档 AWS SDK示例 GitHub 存储库中还有更多SDK示例

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

HealthLake 用SDK于 Python 的示例 (Boto3)

以下代码示例向您展示了如何使用with来执行操作和实现常见场景 HealthLake。 AWS SDK for Python (Boto3)

操作是大型程序的代码摘录,必须在上下文中运行。您可以通过操作了解如何调用单个服务函数,还可以通过函数相关场景的上下文查看操作。

每个示例都包含一个指向完整源代码的链接,您可以在其中找到有关如何在上下文中设置和运行代码的说明。

主题

操作

以下代码示例演示如何使用 CreateFHIRDatastore

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def create_fhir_datastore( self, datastore_name: str, sse_configuration: dict[str, any] = None, identity_provider_configuration: dict[str, any] = None, ) -> dict[str, str]: """ Creates a new HealthLake datastore. When creating a SMART on FHIR datastore, the following parameters are required: - sse_configuration: The server-side encryption configuration for a SMART on FHIR-enabled data store. - identity_provider_configuration: The identity provider configuration for a SMART on FHIR-enabled data store. :param datastore_name: The name of the data store. :param sse_configuration: The server-side encryption configuration for a SMART on FHIR-enabled data store. :param identity_provider_configuration: The identity provider configuration for a SMART on FHIR-enabled data store. :return: A dictionary containing the data store information. """ try: parameters = {"DatastoreName": datastore_name, "DatastoreTypeVersion": "R4"} if ( sse_configuration is not None and identity_provider_configuration is not None ): # Creating a SMART on FHIR-enabled data store parameters["SseConfiguration"] = sse_configuration parameters[ "IdentityProviderConfiguration" ] = identity_provider_configuration response = self.health_lake_client.create_fhir_datastore(**parameters) return response except ClientError as err: logger.exception( "Couldn't create datastore %s. Here's why %s", datastore_name, err.response["Error"]["Message"], ) raise

以下代码显示了启用了FHIR开启 HealthLake 的数据存储SMART的参数示例。

sse_configuration = { "KmsEncryptionConfig": {"CmkType": "AWS_OWNED_KMS_KEY"} } # TODO: Update the metadata to match your environment. metadata = { "issuer": "https://ehr.example.com", "jwks_uri": "https://ehr.example.com/.well-known/jwks.json", "authorization_endpoint": "https://ehr.example.com/auth/authorize", "token_endpoint": "https://ehr.token.com/auth/token", "token_endpoint_auth_methods_supported": [ "client_secret_basic", "foo", ], "grant_types_supported": ["client_credential", "foo"], "registration_endpoint": "https://ehr.example.com/auth/register", "scopes_supported": ["openId", "profile", "launch"], "response_types_supported": ["code"], "management_endpoint": "https://ehr.example.com/user/manage", "introspection_endpoint": "https://ehr.example.com/user/introspect", "revocation_endpoint": "https://ehr.example.com/user/revoke", "code_challenge_methods_supported": ["S256"], "capabilities": [ "launch-ehr", "sso-openid-connect", "client-public", ], } # TODO: Update the IdpLambdaArn. identity_provider_configuration = { "AuthorizationStrategy": "SMART_ON_FHIR_V1", "FineGrainedAuthorizationEnabled": True, "IdpLambdaArn": "arn:aws:lambda:your-region:your-account-id:function:your-lambda-name", "Metadata": json.dumps(metadata), } data_store = self.create_fhir_datastore( datastore_name, sse_configuration, identity_provider_configuration )
  • 有关API详细信息,请参阅 Python (Boto3) API 参考reateFHIRDatastore中的 AWS SDKC 部分。

以下代码示例演示如何使用 DeleteFHIRDatastore

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def delete_fhir_datastore(self, datastore_id: str) -> None: """ Deletes a HealthLake datastore. :param datastore_id: The datastore ID. """ try: self.health_lake_client.delete_fhir_datastore(DatastoreId=datastore_id) except ClientError as err: logger.exception( "Couldn't delete datastore with ID %s. Here's why %s", datastore_id, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅 Python (Boto3) API 参考eleteFHIRDatastore中的 AWS SDKD

以下代码示例演示如何使用 DescribeFHIRDatastore

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def describe_fhir_datastore(self, datastore_id: str) -> dict[str, any]: """ Describes a HealthLake datastore. :param datastore_id: The datastore ID. :return: The datastore description. """ try: response = self.health_lake_client.describe_fhir_datastore( DatastoreId=datastore_id ) return response["DatastoreProperties"] except ClientError as err: logger.exception( "Couldn't describe datastore with ID %s. Here's why %s", datastore_id, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅 Python (Boto3) API 参考escribeFHIRDatastore中的 AWS SDKD

以下代码示例演示如何使用 DescribeFHIRExportJob

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def describe_fhir_export_job( self, datastore_id: str, job_id: str ) -> dict[str, any]: """ Describes a HealthLake export job. :param datastore_id: The datastore ID. :param job_id: The export job ID. :return: The export job description. """ try: response = self.health_lake_client.describe_fhir_export_job( DatastoreId=datastore_id, JobId=job_id ) return response["ExportJobProperties"] except ClientError as err: logger.exception( "Couldn't describe export job with ID %s. Here's why %s", job_id, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅中的 D escribeFHIRExport J ob AWS SDK以获取 Python (Boto3) API 参考。

以下代码示例演示如何使用 DescribeFHIRImportJob

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def describe_fhir_import_job( self, datastore_id: str, job_id: str ) -> dict[str, any]: """ Describes a HealthLake import job. :param datastore_id: The datastore ID. :param job_id: The import job ID. :return: The import job description. """ try: response = self.health_lake_client.describe_fhir_import_job( DatastoreId=datastore_id, JobId=job_id ) return response["ImportJobProperties"] except ClientError as err: logger.exception( "Couldn't describe import job with ID %s. Here's why %s", job_id, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅中的 D escribeFHIRImport J ob AWS SDK以获取 Python (Boto3) API 参考。

以下代码示例演示如何使用 ListFHIRDatastoreImportJobs

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def list_fhir_import_jobs( self, datastore_id: str, job_name: str = None, job_status: str = None, submitted_before: datetime = None, submitted_after: datetime = None, ) -> list[dict[str, any]]: """ Lists HealthLake import jobs satisfying the conditions. :param datastore_id: The datastore ID. :param job_name: The import job name. :param job_status: The import job status. :param submitted_before: The import job submitted before the specified date. :param submitted_after: The import job submitted after the specified date. :return: A list of import jobs. """ try: parameters = {"DatastoreId": datastore_id} if job_name is not None: parameters["JobName"] = job_name if job_status is not None: parameters["JobStatus"] = job_status if submitted_before is not None: parameters["SubmittedBefore"] = submitted_before if submitted_after is not None: parameters["SubmittedAfter"] = submitted_after next_token = None jobs = [] # Loop through paginated results. while True: if next_token is not None: parameters["NextToken"] = next_token response = self.health_lake_client.list_fhir_import_jobs(**parameters) jobs.extend(response["ImportJobPropertiesList"]) if "NextToken" in response: next_token = response["NextToken"] else: break return jobs except ClientError as err: logger.exception( "Couldn't list import jobs. Here's why %s", err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅 Python (Boto3) API 参考istFHIRDatastoreImportJobs中的 AWS SDKL

以下代码示例演示如何使用 ListFHIRDatastores

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def list_fhir_datastores(self) -> list[dict[str, any]]: """ Lists all HealthLake datastores. :return: A list of datastore descriptions. """ try: next_token = None datastores = [] # Loop through paginated results. while True: parameters = {} if next_token is not None: parameters["NextToken"] = next_token response = self.health_lake_client.list_fhir_datastores(**parameters) datastores.extend(response["DatastorePropertiesList"]) if "NextToken" in response: next_token = response["NextToken"] else: break return datastores except ClientError as err: logger.exception( "Couldn't list datastores. Here's why %s", err.response["Error"]["Message"] ) raise
  • 有关API详细信息,请参阅 Python (Boto3) API 参考istFHIRDatastores中的 AWS SDKL

以下代码示例演示如何使用 ListFHIRExportJobs

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def list_fhir_export_jobs( self, datastore_id: str, job_name: str = None, job_status: str = None, submitted_before: datetime = None, submitted_after: datetime = None, ) -> list[dict[str, any]]: """ Lists HealthLake export jobs satisfying the conditions. :param datastore_id: The datastore ID. :param job_name: The export job name. :param job_status: The export job status. :param submitted_before: The export job submitted before the specified date. :param submitted_after: The export job submitted after the specified date. :return: A list of export jobs. """ try: parameters = {"DatastoreId": datastore_id} if job_name is not None: parameters["JobName"] = job_name if job_status is not None: parameters["JobStatus"] = job_status if submitted_before is not None: parameters["SubmittedBefore"] = submitted_before if submitted_after is not None: parameters["SubmittedAfter"] = submitted_after next_token = None jobs = [] # Loop through paginated results. while True: if next_token is not None: parameters["NextToken"] = next_token response = self.health_lake_client.list_fhir_export_jobs(**parameters) jobs.extend(response["ExportJobPropertiesList"]) if "NextToken" in response: next_token = response["NextToken"] else: break return jobs except ClientError as err: logger.exception( "Couldn't list export jobs. Here's why %s", err.response["Error"]["Message"], ) raise

以下代码示例演示如何使用 ListTagsForResource

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def list_tags_for_resource(self, resource_arn: str) -> dict[str, str]: """ Lists the tags for a HealthLake resource. :param resource_arn: The resource ARN. :return: The tags for the resource. """ try: response = self.health_lake_client.list_tags_for_resource( ResourceARN=resource_arn ) return response["Tags"] except ClientError as err: logger.exception( "Couldn't list tags for resource %s. Here's why %s", resource_arn, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅ListTagsForResource中的 AWS SDKPython (Boto3) API 参考。

以下代码示例演示如何使用 StartFHIRExportJob

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def start_fhir_export_job( self, job_name: str, datastore_id: str, output_s3_uri: str, kms_key_id: str, data_access_role_arn: str, ) -> dict[str, str]: """ Starts a HealthLake export job. :param job_name: The export job name. :param datastore_id: The datastore ID. :param output_s3_uri: The output S3 URI. :param kms_key_id: The KMS key ID associated with the output S3 bucket. :param data_access_role_arn: The data access role ARN. :return: The export job. """ try: response = self.health_lake_client.start_fhir_export_job( OutputDataConfig={ "S3Configuration": {"S3Uri": output_s3_uri, "KmsKeyId": kms_key_id} }, DataAccessRoleArn=data_access_role_arn, DatastoreId=datastore_id, JobName=job_name, ) return response except ClientError as err: logger.exception( "Couldn't start export job. Here's why %s", err.response["Error"]["Message"], ) raise

以下代码示例演示如何使用 StartFHIRImportJob

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def start_fhir_import_job( self, job_name: str, datastore_id: str, input_s3_uri: str, job_output_s3_uri: str, kms_key_id: str, data_access_role_arn: str, ) -> dict[str, str]: """ Starts a HealthLake import job. :param job_name: The import job name. :param datastore_id: The datastore ID. :param input_s3_uri: The input S3 URI. :param job_output_s3_uri: The job output S3 URI. :param kms_key_id: The KMS key ID associated with the output S3 bucket. :param data_access_role_arn: The data access role ARN. :return: The import job. """ try: response = self.health_lake_client.start_fhir_import_job( JobName=job_name, InputDataConfig={"S3Uri": input_s3_uri}, JobOutputDataConfig={ "S3Configuration": { "S3Uri": job_output_s3_uri, "KmsKeyId": kms_key_id, } }, DataAccessRoleArn=data_access_role_arn, DatastoreId=datastore_id, ) return response except ClientError as err: logger.exception( "Couldn't start import job. Here's why %s", err.response["Error"]["Message"], ) raise

以下代码示例演示如何使用 TagResource

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def tag_resource(self, resource_arn: str, tags: list[dict[str, str]]) -> None: """ Tags a HealthLake resource. :param resource_arn: The resource ARN. :param tags: The tags to add to the resource. """ try: self.health_lake_client.tag_resource(ResourceARN=resource_arn, Tags=tags) except ClientError as err: logger.exception( "Couldn't tag resource %s. Here's why %s", resource_arn, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅TagResource中的 AWS SDKPython (Boto3) API 参考。

以下代码示例演示如何使用 UntagResource

SDK适用于 Python (Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

@classmethod def from_client(cls) -> "HealthLakeWrapper": """ Creates a HealthLakeWrapper instance with a default AWS HealthLake client. :return: An instance of HealthLakeWrapper initialized with the default HealthLake client. """ health_lake_client = boto3.client("healthlake") return cls(health_lake_client) def untag_resource(self, resource_arn: str, tag_keys: list[str]) -> None: """ Untags a HealthLake resource. :param resource_arn: The resource ARN. :param tag_keys: The tag keys to remove from the resource. """ try: self.health_lake_client.untag_resource( ResourceARN=resource_arn, TagKeys=tag_keys ) except ClientError as err: logger.exception( "Couldn't untag resource %s. Here's why %s", resource_arn, err.response["Error"]["Message"], ) raise
  • 有关API详细信息,请参阅UntagResource中的 AWS SDKPython (Boto3) API 参考。