twarc.expansions

Expansions are how the new v2 Twitter API includes optional metadata about Tweets. In contrast to v1.1, where each Tweet JSON object is self-contained, in v2 metadata about a whole "page" of requests is included in the response. This means that to get a self-contained Tweet JSON, additional processing is needed to look up each piece of extra metadata. Different tools and libraries may implement this in different ways. In twarc, the goal was to retain the original JSON format and only append extra fields, so that any code that expects original JSON will still work.

This module contains a list of the known Twitter V2+ API expansions and fields for each expansion, and a function flatten() for "flattening" a result set, including all expansions inline.

ensure_flattened() can be used in tweet processing programs that need to make sure that data is flattened.

ensure_flattened(data)

Will ensure that the supplied data is "flattened". The input data can be a response from the Twitter API, a list of tweet dictionaries, or a single tweet dictionary. It will always return a list of tweet dictionaries. A ValueError will be thrown if the supplied data is not recognizable or it cannot be flattened.

ensure_flattened is designed for use in twarc plugins and other tweet processing applications that want to operate on a stream of tweets, and examine included entities like users and tweets without hunting and pecking in the response data.

Source code in twarc/expansions.py

def ensure_flattened(data):
    """
    Will ensure that the supplied data is "flattened". The input data can be a
    response from the Twitter API, a list of tweet dictionaries, or a single tweet
    dictionary. It will always return a list of tweet dictionaries. A ValueError
    will be thrown if the supplied data is not recognizable or it cannot be
    flattened.

    ensure_flattened is designed for use in twarc plugins and other tweet
    processing applications that want to operate on a stream of tweets, and
    examine included entities like users and tweets without hunting and
    pecking in the response data.
    """

    # If it's a single response from the API, with data and includes, we flatten it:
    if isinstance(data, dict) and "data" in data and "includes" in data:
        return flatten(data)

    # If it's a single response with data, but without includes:
    elif isinstance(data, dict) and "data" in data and "includes" not in data:
        # flatten() will still work, just with {} empty expansions, log a warning.
        log.warning(f"Unable to expand dictionary without includes: {data}")
        return flatten(data)

    # If it's just an object with errors return an empty list
    elif (
        isinstance(data, dict)
        and "data" not in data
        and "includes" not in data
        and "errors" in data
    ):
        return []

    # If it's a single response and both "includes" and "data" are missing, it is already flattened
    elif isinstance(data, dict) and "data" not in data and "includes" not in data:
        return [data]

    # If it's a list of objects (could be list of responses, or tweets, or users):
    elif isinstance(data, list) and len(data) > 0 and isinstance(data[0], dict):
        # Same as above,
        if "data" in data[0] and "includes" in data[0]:
            # but flatten each object individually and return a single list
            return list(chain.from_iterable([flatten(item) for item in data]))
        elif "data" in data[0] and "includes" not in data[0]:
            # same as above, log warnings and return a single list
            log.warning(f"Unable to expand dictionary without includes: {data[0]}")
            return list(chain.from_iterable([flatten(item) for item in data]))
        # Return already flattened data as is
        elif "data" not in data[0] and "includes" not in data[0]:
            return data
    # Unknown format, eg: list of lists, or primitive
    else:
        raise ValueError(f"Cannot flatten unrecognized data: {data}")

flatten(response)

Flatten an API response by moving all "included" entities inline with the tweets they are referenced from. flatten expects an entire page response from the API (data, includes, meta) and will raise a ValueError if what is passed in does not appear to be an API response. It will return a list of dictionaries where each dictionary represents a tweet. Empty objects will be returned for things that are missing in includes, which can happen when protected or delete users or tweets are referenced.

Source code in twarc/expansions.py

def flatten(response):
    """
    Flatten an API response by moving all "included" entities inline with the
    tweets they are referenced from. flatten expects an entire page response
    from the API (data, includes, meta) and will raise a ValueError if what is
    passed in does not appear to be an API response. It will return a list of
    dictionaries where each dictionary represents a tweet. Empty objects will
    be returned for things that are missing in includes, which can happen when
    protected or delete users or tweets are referenced.
    """

    # Users extracted both by id and by username for expanding mentions
    includes_users = defaultdict(
        lambda: {},
        {
            **extract_includes(response, "users", "id"),
            **extract_includes(response, "users", "username"),
        },
    )
    # Media is by media_key, not id
    includes_media = extract_includes(response, "media", "media_key")
    includes_polls = extract_includes(response, "polls")
    includes_places = extract_includes(response, "places")
    # Tweets in includes will themselves be expanded
    includes_tweets = extract_includes(response, "tweets")
    # Errors are returned but unused here for now
    includes_errors = extract_includes(response, "errors")

    def expand_payload(payload):
        """
        Recursively step through an object and sub objects and append extra data.
        Can be applied to any tweet, list of tweets, sub object of tweet etc.
        """

        # Don't try to expand on primitive values, return strings as is:
        if isinstance(payload, (str, bool, int, float)):
            return payload
        # expand list items individually:
        elif isinstance(payload, list):
            payload = [expand_payload(item) for item in payload]
            return payload
        # Try to expand on dicts within dicts:
        elif isinstance(payload, dict):
            for key, value in payload.items():
                payload[key] = expand_payload(value)

        if "author_id" in payload:
            payload["author"] = includes_users[payload["author_id"]]

        if "in_reply_to_user_id" in payload:
            payload["in_reply_to_user"] = includes_users[payload["in_reply_to_user_id"]]

        if "media_keys" in payload:
            payload["media"] = list(
                includes_media[media_key] for media_key in payload["media_keys"]
            )

        if "poll_ids" in payload and len(payload["poll_ids"]) > 0:
            poll_id = payload["poll_ids"][-1]  # only ever 1 poll per tweet.
            payload["poll"] = includes_polls[poll_id]

        if "geo" in payload and "place_id" in payload["geo"]:
            place_id = payload["geo"]["place_id"]
            payload["geo"] = {**payload["geo"], **includes_places[place_id]}

        if "mentions" in payload:
            payload["mentions"] = list(
                {**referenced_user, **includes_users[referenced_user["username"]]}
                for referenced_user in payload["mentions"]
            )

        if "referenced_tweets" in payload:
            payload["referenced_tweets"] = list(
                {**referenced_tweet, **includes_tweets[referenced_tweet["id"]]}
                for referenced_tweet in payload["referenced_tweets"]
            )

        if "pinned_tweet_id" in payload:
            payload["pinned_tweet"] = includes_tweets[payload["pinned_tweet_id"]]

        return payload

    # First expand the tweets in "includes", before processing actual result tweets:
    for included_id, included_tweet in extract_includes(response, "tweets").items():
        includes_tweets[included_id] = expand_payload(included_tweet)

    # Now expand the list of tweets or an individual tweet in "data"
    tweets = []
    if "data" in response:
        data = response["data"]

        if isinstance(data, list):
            tweets = expand_payload(response["data"])
        elif isinstance(data, dict):
            tweets = [expand_payload(response["data"])]

        # Add the __twarc metadata and matching rules to each tweet if it's a result set
        if "__twarc" in response:
            for tweet in tweets:
                tweet["__twarc"] = response["__twarc"]
        if "matching_rules" in response:
            for tweet in tweets:
                tweet["matching_rules"] = response["matching_rules"]
    else:
        raise ValueError(f"missing data stanza in response: {response}")

    return tweets

handler: python