def increment_dynamodb_counter()

in lambdas/scale_out_runner/app.py [0:0]

• 14 lines of code
• 2 McCabe index (conditional complexity)

def increment_dynamodb_counter(delta: int = 1) -> int:
    dynamodb = boto3.client('dynamodb')
    args = dict(
        TableName=TABLE_NAME,
        Key={'id': {'S': 'queued_jobs'}},
        ExpressionAttributeValues={':delta': {'N': str(delta)}},
        UpdateExpression='ADD queued :delta',
        ReturnValues='UPDATED_NEW',
    )

    if delta < 0:
        # Make sure it never goes below zero!
        args['ExpressionAttributeValues'][':limit'] = {'N': str(-delta)}
        args['ConditionExpression'] = 'queued >= :limit'

    resp = dynamodb.update_item(**args)
    return int(resp['Attributes']['queued']['N'])