get_job_details async ¶

async def get_job_details(
    self,
    job_id: "capo_codepipeline.types.job_id.JobId",
    *,
    config_overrides: Optional[AsyncCodePipelineClientConfig] = None,
) -> "capo_codepipeline.types.get_job_details_output.GetJobDetailsOutput":
    """<p>Returns information about a job. Used for custom actions only.</p> <important> <p>When this API is called, CodePipeline returns temporary credentials for the S3 bucket used to store artifacts for the pipeline, if the action requires access to that S3 bucket for input or output artifacts. This API also returns any secret values defined for the action.</p> </important>

    Args:
        job_id: <p>The unique system-generated ID for the job.</p>

    Raises:
        capo_codepipeline.errors.job_not_found_exception.JobNotFoundException: <p>The job was specified in an invalid format or cannot be found.</p>
        capo_codepipeline.errors.validation_exception.ValidationException: <p>The validation was specified in an invalid format.</p>
        capo_codepipeline.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_codepipeline.types.get_job_details_input.GetJobDetailsInput]",
    ) -> AsyncOperationResponse[
        "capo_codepipeline.types.get_job_details_output.GetJobDetailsOutput"
    ]:
        import capo_codepipeline._operations.code_pipeline_20150709.get_job_details

        (
            output,
            http_response,
        ) = await capo_codepipeline._operations.code_pipeline_20150709.get_job_details.async_get_job_details(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    def _redact_output(
        output: "capo_codepipeline.types.get_job_details_output.GetJobDetailsOutput",
    ) -> dict[str, Any]:
        import capo_codepipeline.types.get_job_details_output

        return capo_codepipeline.types.get_job_details_output.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("GetJobDetails", redact_output=_redact_output),
    )
    input_: capo_codepipeline.types.get_job_details_input.GetJobDetailsInput = {
        "job_id": job_id
    }

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output