def upload_files()

in api/runtime/app.py [0:0]

• 35 lines of code
• 1 McCabe index (conditional complexity)

def upload_files():
    """
    Returns an S3 pre-signed URL and the associated bucket name for the client to upload an object to S3 directly
    Authorization: IAM
    This endpoint requires AWS SigV4 authorization. For more details see:
    https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-websocket-control-access-iam.html
    https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html
    ---
    tags:
        - S3 Uploader Helper API
    parameters:
        - name: Authorization
          description: AWS SigV4-signed authorization header
          in: header
          type: string
          required: true
        - name: x-amz-content-sha256
          in: header
          type: string
          required: true
        - name: x-amz-content-sha256
          description: SHA-256 hash of payload
          in: header
          type: string
          required: true
        - name: x-amz-date
          description: Date/Timestamp
          in: header
          type: string
          required: true

    responses:
        200:
            description: Signed certificate and assigned tenant
            schema:
                properties:
                    presignedUrl:
                        type: string
                        description: AWS S3 Pre-signed URL with "put-object" permissions
                    uploadBucket:
                        type: string
                        description: Name of bucket used with Pre-signed URL
        403:
            description: See IAM auth responses

    """

    request = app.current_request
    context = request.context
    print("Context:")
    print(context)
    caller_identity = context['identity']['caller']
    caller_ex = ":(.+)$"
    cert_id = re.search(caller_ex, caller_identity).group(1)

    iot = boto3.client("iot")

    certificate_arn = iot.describe_certificate(
        certificateId=cert_id
    )['certificateDescription']['certificateArn']

    things = iot.list_principal_things(
        principal=certificate_arn
    )['things']
    thing_name = things[0]
    thing_info = iot.describe_thing(
        thingName=thing_name
    )
    s3 = boto3.client('s3')

    presigned_url = s3.generate_presigned_url(
        "put_object",
        Params={
            'Bucket': upload_bucket,
            'Key': "{0}/{1}/sampledata_{2}".format(
                thing_info['attributes']['tenant'], thing_name, str(uuid.uuid4())[:5]
            )
        },
        ExpiresIn=600
    )
    response = {
        "presignedUrl": presigned_url,
        "uploadBucket": upload_bucket
    }
    return response