Skip to content

start_content_upload async

start_content_upload(
    knowledge_base_id: UuidOrArn,
    content_type: ContentType,
    *,
    config_overrides: Optional[AsyncQConnectClientConfig] = None,
    presigned_url_time_to_live: Optional[TimeToLive] = None,
) -> StartContentUploadResponse

Get a URL to upload content to a knowledge base. To upload content, first make a PUT request to the returned URL with your file, making sure to include the required headers. Then use CreateContent to finalize the content creation process or UpdateContent to modify an existing resource. You can only upload content to a knowledge base of type CUSTOM.

Parameters:

  • knowledge_base_id (UuidOrArn) –

    The identifier of the knowledge base. Can be either the ID or the ARN. URLs cannot contain the ARN.

  • content_type (ContentType) –

    The type of content to upload.

  • presigned_url_time_to_live (Optional[TimeToLive], default: None ) –

    The expected expiration time of the generated presigned URL, specified in minutes.

Raises:

Source code in docs_src/capo_qconnect/_services/async_q_connect.py
async def start_content_upload(
    self,
    knowledge_base_id: "capo_qconnect.types.uuid_or_arn.UuidOrArn",
    content_type: "capo_qconnect.types.content_type.ContentType",
    *,
    config_overrides: Optional[AsyncQConnectClientConfig] = None,
    presigned_url_time_to_live: Optional[
        "capo_qconnect.types.time_to_live.TimeToLive"
    ] = None,
) -> "capo_qconnect.types.start_content_upload_response.StartContentUploadResponse":
    """<p>Get a URL to upload content to a knowledge base. To upload content, first make a PUT request to the returned URL with your file, making sure to include the required headers. Then use <a href="https://docs.aws.amazon.com/amazon-q-connect/latest/APIReference/API_CreateContent.html">CreateContent</a> to finalize the content creation process or <a href="https://docs.aws.amazon.com/amazon-q-connect/latest/APIReference/API_UpdateContent.html">UpdateContent</a> to modify an existing resource. You can only upload content to a knowledge base of type CUSTOM.</p>

    Args:
        knowledge_base_id: <p>The identifier of the knowledge base. Can be either the ID or the ARN. URLs cannot contain the ARN.</p>
        content_type: <p>The type of content to upload.</p>
        presigned_url_time_to_live: <p>The expected expiration time of the generated presigned URL, specified in minutes.</p>

    Raises:
        capo_qconnect.errors.access_denied_exception.AccessDeniedException: <p>You do not have sufficient access to perform this action.</p>
        capo_qconnect.errors.resource_not_found_exception.ResourceNotFoundException: <p>The specified resource does not exist.</p>
        capo_qconnect.errors.unauthorized_exception.UnauthorizedException: <p>You do not have permission to perform this action.</p>
        capo_qconnect.errors.validation_exception.ValidationException: <p>The input fails to satisfy the constraints specified by a service.</p>
        capo_qconnect.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_qconnect.types.start_content_upload_request.StartContentUploadRequest]",
    ) -> AsyncOperationResponse[
        "capo_qconnect.types.start_content_upload_response.StartContentUploadResponse"
    ]:
        import capo_qconnect._operations.wisdom_service.start_content_upload

        (
            output,
            http_response,
        ) = await capo_qconnect._operations.wisdom_service.start_content_upload.async_start_content_upload(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    def _redact_output(
        output: "capo_qconnect.types.start_content_upload_response.StartContentUploadResponse",
    ) -> dict[str, Any]:
        import capo_qconnect.types.start_content_upload_response

        return capo_qconnect.types.start_content_upload_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("StartContentUpload", redact_output=_redact_output),
    )
    input_: capo_qconnect.types.start_content_upload_request.StartContentUploadRequest = {
        "knowledge_base_id": knowledge_base_id,
        "content_type": content_type,
    }
    if presigned_url_time_to_live is not None:
        input_["presigned_url_time_to_live"] = presigned_url_time_to_live

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