parameters: (*, projectId, name, contentType, contentLength, expires,
                   readerFactory, maxRetries=5, uploadId=None, objectService)
docstring: 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.
docstring_summary: 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.
docstring_tokens: 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 .
function: 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,
          })
function_tokens: 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 , } )