def _update_item()

in backend/lambdas/jobs/status_updater.py [0:0]

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

def _update_item(job_id, attr_updates):
    try:
        update_expression = "set " + ", ".join(
            ["#{k} = :{k}".format(k=k) for k, v in attr_updates.items()]
        )
        attr_names = {}
        attr_values = {}

        for k, v in attr_updates.items():
            attr_names["#{}".format(k)] = k
            attr_values[":{}".format(k)] = v

        unlocked_states_condition = " OR ".join(
            ["#JobStatus = :{}".format(s) for s in unlocked_states]
        )

        return table.update_item(
            Key={"Id": job_id, "Sk": job_id,},
            UpdateExpression=update_expression,
            ConditionExpression="#Id = :Id AND #Sk = :Sk AND ({})".format(
                unlocked_states_condition
            ),
            ExpressionAttributeNames={
                "#Id": "Id",
                "#Sk": "Sk",
                "#JobStatus": "JobStatus",
                **attr_names,
            },
            ExpressionAttributeValues={
                ":Id": job_id,
                ":Sk": job_id,
                **{":{}".format(s): s for s in unlocked_states},
                **attr_values,
            },
            ReturnValues="ALL_NEW",
        )["Attributes"]
    except ddb.meta.client.exceptions.ConditionalCheckFailedException:
        logger.warning("Job %s is already in a status which cannot be updated", job_id)