文件 AWS 開發套件範例 GitHub 儲存庫中有更多可用的 [AWS SDK 範例](https://github.com/awsdocs/aws-doc-sdk-examples)。 本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 使用 SDK for Python (Boto3) 的 HealthLake 範例 下列程式碼範例示範如何使用 適用於 Python (Boto3) 的 AWS SDK 搭配 HealthLake 執行動作和實作常見案例。 *Actions* 是大型程式的程式碼摘錄，必須在內容中執行。雖然動作會告訴您如何呼叫個別服務函數，但您可以在其相關情境中查看內容中的動作。 每個範例均包含完整原始碼的連結，您可在連結中找到如何設定和執行內容中程式碼的相關指示。 **Topics** + [動作](#actions) ## 動作 ### `CreateFHIRDatastore` 以下程式碼範例顯示如何使用 `CreateFHIRDatastore`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store. When creating a SMART on FHIR data store, 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 data store %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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [CreateFHIRDatastore](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/CreateFHIRDatastore)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `DeleteFHIRDatastore` 以下程式碼範例顯示如何使用 `DeleteFHIRDatastore`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store. :param datastore_id: The data store ID. """ try: self.health_lake_client.delete_fhir_datastore(DatastoreId=datastore_id) except ClientError as err: logger.exception( "Couldn't delete data store with ID %s. Here's why %s", datastore_id, err.response["Error"]["Message"], ) raise ``` + 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DeleteFHIRDatastore](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/DeleteFHIRDatastore)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `DescribeFHIRDatastore` 以下程式碼範例顯示如何使用 `DescribeFHIRDatastore`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store. :param datastore_id: The data store ID. :return: The data store 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 data store with ID %s. Here's why %s", datastore_id, err.response["Error"]["Message"], ) raise ``` + 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DescribeFHIRDatastore](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/DescribeFHIRDatastore)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `DescribeFHIRExportJob` 以下程式碼範例顯示如何使用 `DescribeFHIRExportJob`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store 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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DescribeFHIRExportJob](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/DescribeFHIRExportJob)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `DescribeFHIRImportJob` 以下程式碼範例顯示如何使用 `DescribeFHIRImportJob`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store 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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DescribeFHIRImportJob](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/DescribeFHIRImportJob)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `ListFHIRDatastores` 以下程式碼範例顯示如何使用 `ListFHIRDatastores`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data stores. :return: A list of data store 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 data stores. Here's why %s", err.response["Error"]["Message"] ) raise ``` + 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListFHIRDatastores](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/ListFHIRDatastores)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `ListFHIRExportJobs` 以下程式碼範例顯示如何使用 `ListFHIRExportJobs`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store 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 ``` + 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListFHIRExportJobs](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/ListFHIRExportJobs)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `ListFHIRImportJobs` 以下程式碼範例顯示如何使用 `ListFHIRImportJobs`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store 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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListFHIRImportJobs](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/ListFHIRImportJobs)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `ListTagsForResource` 以下程式碼範例顯示如何使用 `ListTagsForResource`。 **適用於 Python 的 SDK (Boto3)** ``` @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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListTagsForResource](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/ListTagsForResource)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `StartFHIRExportJob` 以下程式碼範例顯示如何使用 `StartFHIRExportJob`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store 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 ``` + 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [StartFHIRExportJob](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/StartFHIRExportJob)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `StartFHIRImportJob` 以下程式碼範例顯示如何使用 `StartFHIRImportJob`。 **適用於 Python 的 SDK (Boto3)** ``` @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 data store 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 ``` + 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [StartFHIRImportJob](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/StartFHIRImportJob)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `TagResource` 以下程式碼範例顯示如何使用 `TagResource`。 **適用於 Python 的 SDK (Boto3)** ``` @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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [TagResource](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/TagResource)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。 ### `UntagResource` 以下程式碼範例顯示如何使用 `UntagResource`。 **適用於 Python 的 SDK (Boto3)** ``` @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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [UntagResource](https://docs.aws.amazon.com/goto/boto3/healthlake-2017-07-01/UntagResource)。 GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/healthlake#code-examples)中設定和執行。