def get_timestamp_gap_data()

in cosmos-db-migration-utility/src/lambda/gap-watch-request-reader/lambda_function.py [0:0]

• 32 lines of code
• 3 McCabe index (conditional complexity)

def get_timestamp_gap_data(cluster_name):
    client = boto3.resource('dynamodb')
    logger.info("Fetching timestamp gap data for cluster_name: %s.", cluster_name)
    response = client.Table("time_gap").query(
        KeyConditionExpression=Key("cluster_name").eq(cluster_name)
    )
    max_gap = 0
    max_gap_timestamp = datetime(2020,1,1,0,0,0).isoformat()
    timestamp = datetime.utcnow().isoformat()
    items = []
    for item in response["Items"]:
        if item["time_gap_in_seconds"] > max_gap:
            max_gap = item["time_gap_in_seconds"]
            max_gap_timestamp = item["created_timestamp"]
        logger.info("Found timestamp gap data for cluster_name: %s. Item: %s.", cluster_name, json.dumps(item, cls=JSONFriendlyEncoder))
        items.append({
            "cluster_name": item["cluster_name"],
            "namespace": item["namespace"],
            "batch_id": item["batch_id"],
            "created_timestamp": item["created_timestamp"],
            "processed_timestamp": item["processed_timestamp"],
            "time_gap_in_seconds": item["time_gap_in_seconds"]
        })
    items = sorted(items, key = lambda i: i['time_gap_in_seconds'], reverse=True)
    result = {
        "cluster_name": cluster_name,
        "current_time": timestamp,
        "gap_in_seconds": max_gap,
        "details": json.dumps(items, cls=JSONFriendlyEncoder)
    }
    logger.info("Successfully completed computing the time gap for cluster name: %s. Result: %s", cluster_name, json.dumps(result, cls=JSONFriendlyEncoder))
    return result