This was my entry for a recent coding challenge for computing the in-degree of a tweet graph (the competition is over).

The requirement was to compute the in-degree of a graph made from unique hash tags within a given window of time (rolling). Any hash tag (case insensitive) is a node, and two unique hash tags present in the same tweet becomes an edge. There are no weighted edges. Any time a new tweet comes in, the graph is reexamined, and tweets older than the given window time from the latest tweet is discarded.

I would like to get feed back on my coding style, any idioms that I missed, and better ways of doing this. My full repo, including the unit tests is here. The original coding challenge description is here.

#!/usr/bin/env python3
This module computes the rolling average vertex degree of a twitter
tweet hashtag graph.
import itertools
import json
import sys
import time
import argparse
import logging
from typing import Dict, Tuple, List, Any, Optional, cast, Iterable

from heapdict import heapdict

TIME_FMT = "%a %b %d %H:%M:%S +0000 %Y"
logging.basicConfig(level=logging.WARNING, format='%(levelname)s %(message)s', stream=sys.stderr)
LOG = logging.getLogger(__name__)

class TweetGraph:
    Process the tweet, and keeps track of the time. This implementation uses
    a priority queue (heap) containing a tuple (edge,created time) as the
    def __init__(self, curtime: int, window: int) -> None:
        Initialize the TweetGraph
        :param curtime: The starting time
        :param window: The sliding window
        self.latest = curtime
        self.edges = {}  # type: Dict[Tuple[str, str], int]
        self.queue = heapdict()
        self.window = window

    def in_window(self, ctime: int) -> bool:
        Is the passed in time within the window? Note that the formula is
        `(self.latest - ctime) >= window`
        :param ctime: The time which has to be checked.
        :return: boolean indicating whether passed
        return False if (self.latest - ctime) >= self.window else True

    def add_edge(self, ctime: int, edge: Tuple[str, str]) -> None:
        Add or update the given edge with the given time to
        our database of edges.
        :param ctime: The creation time of the tweet
        :param edge: A tuple containing two hash tags
        old_ctime = self.edges.get(edge, None)
        if (not old_ctime) or (ctime > old_ctime):
            self.queue[edge] = ctime
            self.edges[edge] = ctime

    def update_hashtags(self, ctime: int, hashtags: List[str]) -> None:
        Process the given set of hashtags for the given time.
        :param ctime: The creation time of the tweet
        :param hashtags: The unique hashtags associated with this tweet.
        if not self.in_window(ctime):
        if ctime > self.latest:
            self.latest = ctime

        # Ensure that we perform gc _before_ checking if the
        # prerequisite number of hashtags are present.

        # very nicely, we do not need to check for hashtags being
        # atleast two because itertools.combinations() will not
        # produce an item in that case.
        for edge in cast(Iterable, itertools.combinations(hashtags, 2)):
            self.add_edge(ctime, edge)

    def gc_complete(self) -> bool:
        Check if the gc is complete.
        if len(self.queue) == 0:
            return True
        _, ctime = self.queue.peekitem()
        return self.in_window(ctime)

    def collect_garbage(self) -> None:
        Perform garbage collection.
        LOG.info('start gc edges: %d queue: %d', len(self.edges.keys()), len(self.queue))
        while not self.gc_complete():
            min_edge, _ = self.queue.popitem()
            del self.edges[min_edge]
            LOG.info('- %s', min_edge)
        LOG.info('finished gc edges: %d queue: %d', len(self.edges.keys()), len(self.queue))

    def avg_vdegree(self) -> float:
        Compute the average degree of a vertex using the formula 2*edges/nodes.
        if not self.edges:
            return 0
        # Our edge.keys are tuples of hashtags. We flatten them.
        nodes = set(itertools.chain.from_iterable(self.edges.keys()))
        return (2.0 * len(self.edges)) / len(nodes)

    def process_tweet(self, tweet: Dict[str, Any]) -> float:
        Process a tweet and return the current average vertex degree
        :param tweet: the dict containing the stripped tweet.
        :return: The current vertex degree
        ctime, htags = self.trim_tweet(tweet)
        self.update_hashtags(ctime, htags)
        # We have to print average each time a new tweet makes its
        # appearance irrespective of whether it can be ignored or not.
        return self.avg_vdegree

    def trim_tweet(my_hash: Dict[str, Any]) -> Tuple[int, List[str]]:
        Initial processing of the json line. Remove all the fluf
        except created_at, and hashtags.
        :param my_hash: The tweet dict to be de-fluffed
        :return: A tuple containing ctime and hashtags if
        the number of unique hashtags is at least two. None otherwise.

        htags = my_hash.get('entities', {}).get('hashtags', [])
        # We sort to make sure that any two
        # keys always have a well defined edge name.
        hashtags = sorted(set(h['text'] for h in htags))
        return my_hash['ctime'], hashtags

def get_tweet(line: str) -> Optional[Dict[str, Any]]:
    Parse the line into json, and check that it is a valid tweet
    and not a limit message.
    :param line: The json line to be parsed.
    :return: If this is a valid tweet, the dict containing creation
    time and hashtags. None otherwise.
        j = json.loads(line)
        created_at = j.get('created_at', None)
        if not created_at:
            return None

        # We validate the creation time here. If the creation time
        # is in invalid format, it is an invalid tweet.
        ctime = int(time.mktime(time.strptime(created_at, TIME_FMT)))
        j['ctime'] = ctime
        return j
    except ValueError:
        # We do not expect any records to be malformed. However, if there
        # are any, it is important not to abort the whole process, and
        # instead, just discard the record and let the user know through
        # another channel.
        LOG.warning('malformed: %s', line)
        return None

def main():
    The entry point. We require a single parameter: the window length.
    We also accept tweets in stdin, and write to stdout.
    pcmd = argparse.ArgumentParser()
    pcmd.add_argument('window', type=int, help='window for rolling average')
    args = pcmd.parse_args()
    tweetgraph = TweetGraph(0, args.window)
    for line in sys.stdin:
        tweet = get_tweet(line)
        # Do not print rolling average in case this is not a valid tweet
        if tweet:

if __name__ == "__main__":
  • \$\begingroup\$ I've made an edit to your title as it was pretty generic. Nice question, by the way! \$\endgroup\$
    – Phrancis
    Apr 19, 2016 at 21:16

1 Answer 1


Just looking at the function get_tweet:

  1. The docstring says, "Parse the line into json" but actually it parses the line from JSON (and into a Python dictionary).

  2. This function has multiple responsibilities: it parses a tweet, and also logs failures. This makes the code hard to re-use if you have a different logging environment or different requirements about how to handle failures. It would be better to separate these responsibilities and have the caller do the logging (or otherwise handling) of failures.

  3. If you are going to log an exception and continue, it's a good idea to log a full traceback (using the traceback module) so that you have as much information as possible. At the moment, the logging in get_tweet doesn't indicate whether the failure happened in json.loads or in time.strptime.

  4. The code assumes that json.loads returns a dictionary, but in fact this function can also return a number, string, Boolean, list, or None, and in these cases j.get('created_at', None) would fail with AttributeError.

  5. The code parses the time and uses it to construct a Unix timestamp. But this time representation is not very convenient to use in Python. It would be better to use datetime.datetime objects throughout.

  6. The name j is not very clear: presumably j is short for JSON but the value of this variable is not a JSON string, but a Python dictionary representing a tweet. So tweet would be a better name.

  • \$\begingroup\$ Thank you for the review. I agree with (1). However, I have reservations with (2), because parsing is done externally using json. For (3), due to the environment of application, I would argue it makes more sense to spit out a single line for each malformed record than the traceback. Agreew with (4), (5) and perhaps with (6), but I would argue t is as good as tweet because its use is explicit from the context. \$\endgroup\$ May 15, 2016 at 5:58

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.