(*, projectId, name, contentType, contentLength, expires,
                 readerFactory, maxRetries=5, uploadId=None, objectService)
Upload the given data to the object service with the given metadata.
    The `maxRetries` parameter has the same meaning as for service clients.
    The `objectService` parameter is an instance of the Object class,
    configured with credentials for the upload.
Upload the given data to the object service with the given metadata.
    The `maxRetries` parameter has the same meaning as for service clients.
    The `objectService` parameter is an instance of the Object class,
    configured with credentials for the upload.
Upload the given data to the object service with the given metadata . The maxRetries parameter has the same meaning as for service clients . The objectService parameter is an instance of the Object class configured with credentials for the upload .
async def upload(*, projectId, name, contentType, contentLength, expires,
                 readerFactory, maxRetries=5, uploadId=None, objectService):
    """
    Upload the given data to the object service with the given metadata.
    The `maxRetries` parameter has the same meaning as for service clients.
    The `objectService` parameter is an instance of the Object class,
    configured with credentials for the upload.
    """
    # wrap the readerFactory with one that will also hash the data
    hashingReader = None

    async def hashingReaderFactory():
        nonlocal hashingReader
        hashingReader = HashingReader(await readerFactory())
        return hashingReader

    async with aiohttp.ClientSession() as session:
        if not uploadId:
            uploadId = taskcluster.slugid.nice()
        proposedUploadMethods = {}

        if contentLength < DATA_INLINE_MAX_SIZE:
            reader = await hashingReaderFactory()
            writer = BufferWriter()
            await streamingCopy(reader, writer)
            encoded = base64.b64encode(writer.getbuffer())
            proposedUploadMethods['dataInline'] = {
                "contentType": contentType,
                "objectData": encoded,
            }

        proposedUploadMethods['putUrl'] = {
            "contentType": contentType,
            "contentLength": contentLength,
        }

        uploadResp = await ensureCoro(objectService.createUpload)(name, {
            "expires": expires,
            "projectId": projectId,
            "uploadId": uploadId,
            "proposedUploadMethods": proposedUploadMethods,
        })

        async def tryUpload(retryFor):
            try:
                uploadMethod = uploadResp["uploadMethod"]
                if 'dataInline' in uploadMethod:
                    # data is already uploaded -- nothing to do
                    pass
                elif 'putUrl' in uploadMethod:
                    reader = await hashingReaderFactory()
                    await _putUrlUpload(uploadMethod['putUrl'], reader, session)
                else:
                    raise RuntimeError("Could not negotiate an upload method")
            except aiohttp.ClientResponseError as exc:
                # treat 4xx's as fatal, and retry others
                if 400 <= exc.status < 500:
                    raise exc
                return retryFor(exc)
            except aiohttp.ClientError as exc:
                # retry for all other aiohttp errors
                return retryFor(exc)
            # .. anything else is considered fatal

        await retry(maxRetries, tryUpload)

        hashes = hashingReader.hashes(contentLength)

        await ensureCoro(objectService.finishUpload)(name, {
            "projectId": projectId,
            "uploadId": uploadId,
            "hashes": hashes,
        })
async def upload ( * , projectId , name , contentType , contentLength , expires , readerFactory , maxRetries = 5 , uploadId = None , objectService ) : # wrap the readerFactory with one that will also hash the data hashingReader = None async def hashingReaderFactory ( ) : nonlocal hashingReader hashingReader = HashingReader ( await readerFactory ( ) ) return hashingReader async with aiohttp . ClientSession ( ) as session : if not uploadId : uploadId = taskcluster . slugid . nice ( ) proposedUploadMethods = { } if contentLength < DATA_INLINE_MAX_SIZE : reader = await hashingReaderFactory ( ) writer = BufferWriter ( ) await streamingCopy ( reader , writer ) encoded = base64 . b64encode ( writer . getbuffer ( ) ) proposedUploadMethods [ 'dataInline' ] = { "contentType" : contentType , "objectData" : encoded , } proposedUploadMethods [ 'putUrl' ] = { "contentType" : contentType , "contentLength" : contentLength , } uploadResp = await ensureCoro ( objectService . createUpload ) ( name , { "expires" : expires , "projectId" : projectId , "uploadId" : uploadId , "proposedUploadMethods" : proposedUploadMethods , } ) async def tryUpload ( retryFor ) : try : uploadMethod = uploadResp [ "uploadMethod" ] if 'dataInline' in uploadMethod : # data is already uploaded -- nothing to do pass elif 'putUrl' in uploadMethod : reader = await hashingReaderFactory ( ) await _putUrlUpload ( uploadMethod [ 'putUrl' ] , reader , session ) else : raise RuntimeError ( "Could not negotiate an upload method" ) except aiohttp . ClientResponseError as exc : # treat 4xx's as fatal, and retry others if 400 <= exc . status < 500 : raise exc return retryFor ( exc ) except aiohttp . ClientError as exc : # retry for all other aiohttp errors return retryFor ( exc ) # .. anything else is considered fatal await retry ( maxRetries , tryUpload ) hashes = hashingReader . hashes ( contentLength ) await ensureCoro ( objectService . finishUpload ) ( name , { "projectId" : projectId , "uploadId" : uploadId , "hashes" : hashes , } )