twarc.Client2

Support for the Twitter v2 API.

Twarc2

A client for the Twitter v2 API.

Source code in twarc/client2.py

class Twarc2:
    """
    A client for the Twitter v2 API.
    """

    def __init__(
        self,
        consumer_key=None,
        consumer_secret=None,
        access_token=None,
        access_token_secret=None,
        bearer_token=None,
        connection_errors=0,
        metadata=True,
    ):
        """
        Instantiate a Twarc2 instance to talk to the Twitter V2+ API.

        The client can use either App or User authentication, but only one at a
        time. Whether app auth or user auth is used depends on which credentials
        are provided on initialisation:

        1. If a `bearer_token` is passed, app auth is always used.
        2. If a `consumer_key` and `consumer_secret` are passed without an
        `access_token` and `access_token_secret`, app auth is used.
        3. If `consumer_key`, `consumer_secret`, `access_token` and
        `access_token_secret` are all passed, then user authentication
        is used instead.

        Args:
            consumer_key (str):
                The API key.
            consumer_secret (str):
                The API secret.
            access_token (str):
                The Access Token
            access_token_secret (str):
                The Access Token Secret
            bearer_token (str):
                Bearer Token, can be generated from API keys.
            connection_errors (int):
                Number of retries for GETs
            metadata (bool):
                Append `__twarc` metadata to results.
        """
        self.api_version = "2"
        self.connection_errors = connection_errors
        self.metadata = metadata
        self.bearer_token = None

        if bearer_token:
            self.bearer_token = bearer_token
            self.auth_type = "application"

        elif consumer_key and consumer_secret:
            if access_token and access_token_secret:
                self.consumer_key = consumer_key
                self.consumer_secret = consumer_secret
                self.access_token = access_token
                self.access_token_secret = access_token_secret
                self.auth_type = "user"

            else:
                self.consumer_key = consumer_key
                self.consumer_secret = consumer_secret
                self.auth_type = "application"

        else:
            raise ValueError(
                "Must pass either a bearer_token or consumer/access_token keys and secrets"
            )

        self.client = None
        self.last_response = None

        self.connect()

    def _prepare_params(self, **kwargs):
        """
        Prepare URL parameters and defaults for fields and expansions and others
        """
        params = {}

        # Defaults for fields and expansions
        if "expansions" in kwargs:
            params["expansions"] = (
                kwargs.pop("expansions")
                if kwargs["expansions"]
                else ",".join(EXPANSIONS)
            )

        if "tweet_fields" in kwargs:
            params["tweet.fields"] = (
                kwargs.pop("tweet_fields")
                if kwargs["tweet_fields"]
                else ",".join(TWEET_FIELDS)
            )

        if "user_fields" in kwargs:
            params["user.fields"] = (
                kwargs.pop("user_fields")
                if kwargs["user_fields"]
                else ",".join(USER_FIELDS)
            )

        if "media_fields" in kwargs:
            params["media.fields"] = (
                kwargs.pop("media_fields")
                if kwargs["media_fields"]
                else ",".join(MEDIA_FIELDS)
            )

        if "poll_fields" in kwargs:
            params["poll.fields"] = (
                kwargs.pop("poll_fields")
                if kwargs["poll_fields"]
                else ",".join(POLL_FIELDS)
            )

        if "place_fields" in kwargs:
            params["place.fields"] = (
                kwargs.pop("place_fields")
                if kwargs["place_fields"]
                else ",".join(PLACE_FIELDS)
            )

        if "list_fields" in kwargs:
            params["list.fields"] = (
                kwargs.pop("list_fields")
                if kwargs["list_fields"]
                else ",".join(LIST_FIELDS)
            )

        # Format start_time and end_time
        if "start_time" in kwargs:
            start_time = kwargs["start_time"]
            params["start_time"] = (
                _ts(kwargs.pop("start_time"))
                if start_time and not isinstance(start_time, str)
                else start_time
            )

        if "end_time" in kwargs:
            end_time = kwargs["end_time"]
            params["end_time"] = (
                _ts(kwargs.pop("end_time"))
                if end_time and not isinstance(end_time, str)
                else end_time
            )

        # Any other parameters passed as is,
        # these include backfill_minutes, next_token, pagination_token, sort_order
        params = {**params, **{k: v for k, v in kwargs.items() if v is not None}}

        return params

    def _search(
        self,
        url,
        query,
        since_id,
        until_id,
        start_time,
        end_time,
        max_results,
        expansions,
        tweet_fields,
        user_fields,
        media_fields,
        poll_fields,
        place_fields,
        sort_order,
        next_token=None,
        granularity=None,
        sleep_between=0,
    ):
        """
        Common function for search, counts endpoints.
        """

        params = self._prepare_params(
            query=query,
            max_results=max_results,
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            next_token=next_token,
            sort_order=sort_order,
        )

        if granularity:
            # Do not specify anything else when calling counts endpoint
            params["granularity"] = granularity
            # Mark that we're using counts, to workaround a limitation of the
            # Twitter API with long running counts.
            using_counts = True

            # We need to use these as sentinel values, to differentiate
            # between the count API returning zero prematurely, and queries
            # like "from:<no longer existing user_id>". In the latter case
            # instead of returning counts of 0 per day, it will just return
            # an empty response with a total tweet count of zero. We can
            # disambiguate the two cases by noting that the premature
            # termination will already have counted some tweets correctly,
            # while the latter will return immediately without any data
            # rows.
            time_periods_collected = 0
            last_time_start = None
        else:
            params = self._prepare_params(
                **params,
                expansions=expansions,
                tweet_fields=tweet_fields,
                user_fields=user_fields,
                media_fields=media_fields,
                poll_fields=poll_fields,
                place_fields=place_fields,
            )
            using_counts = False

        # Workaround for observed odd behaviour in the Twitter counts
        # functionality.
        if using_counts:
            while True:
                for response in self.get_paginated(url, params=params):
                    # Note that we're ensuring the appropriate amount of sleep is
                    # taken before yielding every item. This ensures that we won't
                    # exceed the rate limit even in cases where a response generator
                    # is not completely consumed. This might be more conservative
                    # than necessary.
                    time.sleep(sleep_between)

                    # can't return without 'data' if there are no results
                    if "data" in response:
                        last_time_start = response["data"][0]["start"]
                        time_periods_collected += len(response["data"])
                        yield response

                    else:
                        log.info(f"Retrieved an empty page of results.")

                # Check that we've actually reached the end, and restart if necessary.
                # Note we need to exactly match the Twitter format, which is a little
                # fiddly because Python doesn't let you specify milliseconds only for
                # strftime.
                if (
                    # If there's no explicit start time we're getting the last
                    # 30 days by default, so don't need to do the tricky
                    # things.
                    start_time is None
                    # We've actually reached the specified start time
                    or (
                        (start_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z")
                        == last_time_start
                    )
                    # Or, we've hit one of the special cases that returns no rows
                    # of data, and immediately indicates zero tweets returned, like
                    # searching for a tweet that doesn't exist.
                    or (time_periods_collected == 0)
                ):
                    break
                else:
                    # Note that we're passing the Twitter start_time straight
                    # back to it - this avoids parsing and reformatting the date.
                    params["end_time"] = last_time_start

                    # Remove the next_token reference, we're restarting the search.
                    if "next_token" in params:
                        del params["next_token"]

                    log.info(
                        "Detected incomplete counts, restarting with "
                        f"{last_time_start} as the new end_time"
                    )

        else:
            for response in self.get_paginated(url, params=params):
                # Note that we're ensuring the appropriate amount of sleep is
                # taken before yielding every item. This ensures that we won't
                # exceed the rate limit even in cases where a response generator
                # is not completely consumed. This might be more conservative
                # than necessary.
                time.sleep(sleep_between)

                # can't return without 'data' if there are no results
                if "data" in response:
                    yield response

                else:
                    log.info(f"Retrieved an empty page of results.")

        log.info(f"No more results for search {query}.")

    def _lists(
        self,
        url,
        expansions=None,
        list_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Paginates and returns lists
        """
        params = self._prepare_params(
            list_fields=list_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        if expansions:
            params["expansions"] = "owner_id"

        for response in self.get_paginated(url, params=params):
            # can return without 'data' if there are no results
            if "data" in response:
                yield response
            else:
                log.info(f"Retrieved an empty page of results of lists for {url}")

    def list_followers(
        self,
        list_id,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns a list of users who are followers of the specified List.

        Calls [GET /2/lists/:id/followers](https://developer.twitter.com/en/docs/twitter-api/lists/list-follows/api-reference/get-lists-id-followers)

        Args:
            list_id (int): ID of the list.
            expansions enum (pinned_tweet_id): Expansions, include pinned tweets.
            max_results (int): the maximum number of results to retrieve. Between 1 and 100. Default is 100.

        Returns:
            generator[dict]: A generator, dict for each page of results.

        """
        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        if expansions:
            params["expansions"] = "pinned_tweet_id"

        url = f"https://api.twitter.com/2/lists/{list_id}/followers"
        return self.get_paginated(url, params=params)

    def list_members(
        self,
        list_id,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns a list of users who are members of the specified List.

        Calls [GET /2/lists/:id/members](https://developer.twitter.com/en/docs/twitter-api/lists/list-members/api-reference/get-lists-id-members)

        Args:
            list_id (int): ID of the list.
            expansions enum (pinned_tweet_id): Expansions, include pinned tweets.
            max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
            pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

        Returns:
            generator[dict]: A generator, dict for each page of results.

        """

        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        if expansions:
            params["expansions"] = "pinned_tweet_id"

        url = f"https://api.twitter.com/2/lists/{list_id}/members"
        return self.get_paginated(url, params=params)

    def list_memberships(
        self,
        user,
        expansions=None,
        list_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns all Lists a specified user is a member of.

        Calls [GET /2/users/:id/list_memberships](https://developer.twitter.com/en/docs/twitter-api/lists/list-members/api-reference/get-users-id-list_memberships)

        Args:
            user (int): ID of the user.
            expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
            list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
            user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
                This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
            max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
            pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user)
        url = f"https://api.twitter.com/2/users/{user_id}/list_memberships"

        return self._lists(
            url=url,
            expansions=expansions,
            list_fields=list_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

    def owned_lists(
        self,
        user,
        expansions=None,
        list_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns all Lists owned by the specified user.

        Calls [GET /2/users/:id/owned_lists](https://developer.twitter.com/en/docs/twitter-api/lists/list-lookup/api-reference/get-users-id-owned_lists)

        Args:
            user (int): ID of the user.
            expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
            list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
            user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
                This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
            max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
            pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user)
        url = f"https://api.twitter.com/2/users/{user_id}/owned_lists"

        return self._lists(
            url=url,
            expansions=expansions,
            list_fields=list_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

    def followed_lists(
        self,
        user,
        expansions=None,
        list_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns all Lists a specified user follows.

        Calls [GET /2/users/:id/followed_lists](https://developer.twitter.com/en/docs/twitter-api/lists/list-follows/api-reference/get-users-id-followed_lists)

        Args:
            user (int): ID of the user.
            expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
            list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
            user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
                This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
            max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
            pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user)
        url = f"https://api.twitter.com/2/users/{user_id}/followed_lists"

        return self._lists(
            url=url,
            expansions=expansions,
            list_fields=list_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

    def pinned_lists(
        self,
        user,
        expansions=None,
        list_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns the Lists pinned by the authenticating user. Does not work with a Bearer token.

        Calls [GET /2/users/:id/pinned_lists](https://developer.twitter.com/en/docs/twitter-api/lists/pinned-lists/api-reference/get-users-id-pinned_lists)

        Args:
            user (int): ID of the user.
            expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
            list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
            user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
                This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
            max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
            pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user)
        url = f"https://api.twitter.com/2/users/{user_id}/pinned_lists"

        return self._lists(
            url=url,
            expansions=expansions,
            list_fields=list_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

    def list_lookup(self, list_id, expansions=None, list_fields=None, user_fields=None):
        """
        Returns the details of a specified List.

        Calls [GET /2/lists/:id](https://developer.twitter.com/en/docs/twitter-api/lists/list-lookup/api-reference/get-lists-id)

        Args:
            list_id (int): ID of the list.
            expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
            list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
            user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
                This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.

        Returns:
            dict: Result dictionary.
        """

        params = self._prepare_params(
            list_fields=list_fields,
            user_fields=user_fields,
        )

        if expansions:
            params["expansions"] = "owner_id"
        url = f"https://api.twitter.com/2/lists/{list_id}"
        resp = self.get(url, params=params)
        data = resp.json()

        if self.metadata:
            data = _append_metadata(data, resp.url)

        return data

    def list_tweets(
        self,
        list_id,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        max_results=None,
        pagination_token=None,
    ):
        """
        Returns Tweets from the specified List.

        Calls [GET /2/lists/:id/tweets](https://developer.twitter.com/en/docs/twitter-api/lists/list-tweets/api-reference/get-lists-id-tweets)

        Args:
            list_id (int): ID of the list.
            expansions enum (author_id): enable you to request additional data objects that relate to the originally returned List.
            list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
            user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
                This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """

        params = self._prepare_params(
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        url = f"https://api.twitter.com/2/lists/{list_id}/tweets"
        return self.get_paginated(url, params=params)

    def search_recent(
        self,
        query,
        since_id=None,
        until_id=None,
        start_time=None,
        end_time=None,
        max_results=100,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        next_token=None,
        sort_order=None,
    ):
        """
        Search Twitter for the given query in the last seven days,
        using the `/search/recent` endpoint.

        Calls [GET /2/tweets/search/recent](https://developer.twitter.com/en/docs/twitter-api/tweets/search/api-reference/get-tweets-search-recent)

        Args:
            query (str):
                The query string to be passed directly to the Twitter API.
            since_id (int):
                Return all tweets since this tweet_id.
            until_id (int):
                Return all tweets up to this tweet_id.
            start_time (datetime):
                Return all tweets after this time (UTC datetime).
            end_time (datetime):
                Return all tweets before this time (UTC datetime).
            max_results (int):
                The maximum number of results per request. Max is 100.
            sort_order (str):
                Order tweets based on relevancy or recency.

        Returns:
            generator[dict]: a generator, dict for each paginated response.
        """
        return self._search(
            url="https://api.twitter.com/2/tweets/search/recent",
            query=query,
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            max_results=max_results,
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            next_token=next_token,
            sort_order=sort_order,
        )

    @requires_app_auth
    def search_all(
        self,
        query,
        since_id=None,
        until_id=None,
        start_time=None,
        end_time=None,
        max_results=100,  # temp fix for #504
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        next_token=None,
        sort_order=None,
    ):
        """
        Search Twitter for the given query in the full archive,
        using the `/search/all` endpoint (Requires Academic Access).

        Calls [GET /2/tweets/search/all](https://developer.twitter.com/en/docs/twitter-api/tweets/search/api-reference/get-tweets-search-all)

        Args:
            query (str):
                The query string to be passed directly to the Twitter API.
            since_id (int):
                Return all tweets since this tweet_id.
            until_id (int):
                Return all tweets up to this tweet_id.
            start_time (datetime):
                Return all tweets after this time (UTC datetime). If none of start_time, since_id, or until_id
                are specified, this defaults to 2006-3-21 to search the entire history of Twitter.
            end_time (datetime):
                Return all tweets before this time (UTC datetime).
            max_results (int):
                The maximum number of results per request. Max is 500.
            sort_order (str):
                Order tweets based on relevancy or recency.

        Returns:
            generator[dict]: a generator, dict for each paginated response.
        """

        # start time defaults to the beginning of Twitter to override the
        # default of the last month. Only do this if start_time is not already
        # specified and since_id and until_id aren't being used
        if start_time is None and since_id is None and until_id is None:
            start_time = datetime.datetime(2006, 3, 21, tzinfo=datetime.timezone.utc)

        return self._search(
            url="https://api.twitter.com/2/tweets/search/all",
            query=query,
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            max_results=max_results,
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            next_token=next_token,
            sleep_between=1.05,
            sort_order=sort_order,
        )

    @requires_app_auth
    def counts_recent(
        self,
        query,
        since_id=None,
        until_id=None,
        start_time=None,
        end_time=None,
        granularity="hour",
    ):
        """
        Retrieve counts for the given query in the last seven days,
        using the `/counts/recent` endpoint.

        Calls [GET /2/tweets/counts/recent](https://developer.twitter.com/en/docs/twitter-api/tweets/counts/api-reference/get-tweets-counts-recent)

        Args:
            query (str):
                The query string to be passed directly to the Twitter API.
            since_id (int):
                Return all tweets since this tweet_id.
            until_id (int):
                Return all tweets up to this tweet_id.
            start_time (datetime):
                Return all tweets after this time (UTC datetime).
            end_time (datetime):
                Return all tweets before this time (UTC datetime).
            granularity (str):
                Count aggregation level: `day`, `hour`, `minute`.
                Default is `hour`.

        Returns:
            generator[dict]: a generator, dict for each paginated response.
        """
        return self._search(
            url="https://api.twitter.com/2/tweets/counts/recent",
            query=query,
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            max_results=None,
            expansions=None,
            tweet_fields=None,
            user_fields=None,
            media_fields=None,
            poll_fields=None,
            place_fields=None,
            granularity=granularity,
            sort_order=None,
        )

    @requires_app_auth
    def counts_all(
        self,
        query,
        since_id=None,
        until_id=None,
        start_time=None,
        end_time=None,
        granularity="hour",
        next_token=None,
    ):
        """
        Retrieve counts for the given query in the full archive,
        using the `/search/all` endpoint (Requires Academic Access).

        Calls [GET /2/tweets/counts/all](https://developer.twitter.com/en/docs/twitter-api/tweets/counts/api-reference/get-tweets-counts-all)

        Args:
            query (str):
                The query string to be passed directly to the Twitter API.
            since_id (int):
                Return all tweets since this tweet_id.
            until_id (int):
                Return all tweets up to this tweet_id.
            start_time (datetime):
                Return all tweets after this time (UTC datetime).
            end_time (datetime):
                Return all tweets before this time (UTC datetime).
            granularity (str):
                Count aggregation level: `day`, `hour`, `minute`.
                Default is `hour`.

        Returns:
            generator[dict]: a generator, dict for each paginated response.
        """
        return self._search(
            url="https://api.twitter.com/2/tweets/counts/all",
            query=query,
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            max_results=None,
            expansions=None,
            tweet_fields=None,
            user_fields=None,
            media_fields=None,
            poll_fields=None,
            place_fields=None,
            next_token=next_token,
            granularity=granularity,
            sleep_between=1.05,
            sort_order=None,
        )

    def tweet_lookup(
        self,
        tweet_ids,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
    ):
        """
        Lookup tweets, taking an iterator of IDs and returning pages of fully
        expanded tweet objects.

        This can be used to rehydrate a collection shared as only tweet IDs.
        Yields one page of tweets at a time, in blocks of up to 100.

        Calls [GET /2/tweets](https://developer.twitter.com/en/docs/twitter-api/tweets/lookup/api-reference/get-tweets)

        Args:
            tweet_ids (iterable): A list of tweet IDs

        Returns:
            generator[dict]: a generator, dict for each batch of 100 tweets.
        """

        def lookup_batch(tweet_id):
            url = "https://api.twitter.com/2/tweets"

            params = self._prepare_params(
                expansions=expansions,
                tweet_fields=tweet_fields,
                user_fields=user_fields,
                media_fields=media_fields,
                poll_fields=poll_fields,
                place_fields=place_fields,
            )
            params["ids"] = ",".join(tweet_id)

            resp = self.get(url, params=params)
            data = resp.json()

            if self.metadata:
                data = _append_metadata(data, resp.url)

            return data

        tweet_id_batch = []

        for tweet_id in tweet_ids:
            tweet_id_batch.append(str(int(tweet_id)))

            if len(tweet_id_batch) == 100:
                yield lookup_batch(tweet_id_batch)
                tweet_id_batch = []

        if tweet_id_batch:
            yield (lookup_batch(tweet_id_batch))

    def user_lookup(
        self,
        users,
        usernames=False,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
    ):
        """
        Returns fully populated user profiles for the given iterator of
        user_id or usernames. By default user_lookup expects user ids but if
        you want to pass in usernames set usernames = True.

        Yields one page of results at a time (in blocks of at most 100 user
        profiles).

        Calls [GET /2/users](https://developer.twitter.com/en/docs/twitter-api/users/lookup/api-reference/get-users)

        Args:
            users (iterable): User IDs or usernames to lookup.
            usernames (bool): Parse `users` as usernames, not IDs.

        Returns:
            generator[dict]: a generator, dict for each batch of 100 users.
        """

        if isinstance(users, str):
            raise TypeError("users must be an iterable other than a string")

        if usernames:
            url = "https://api.twitter.com/2/users/by"
        else:
            url = "https://api.twitter.com/2/users"

        def lookup_batch(users):
            params = self._prepare_params(
                tweet_fields=tweet_fields,
                user_fields=user_fields,
            )
            if expansions:
                params["expansions"] = "pinned_tweet_id"
            if usernames:
                params["usernames"] = ",".join(users)
            else:
                params["ids"] = ",".join(users)

            resp = self.get(url, params=params)
            data = resp.json()

            if self.metadata:
                data = _append_metadata(data, resp.url)

            return data

        batch = []
        for item in users:
            batch.append(str(item).strip())
            if len(batch) == 100:
                yield lookup_batch(batch)
                batch = []

        if batch:
            yield (lookup_batch(batch))

    @catch_request_exceptions
    @requires_app_auth
    def sample(
        self,
        event=None,
        record_keepalive=False,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        backfill_minutes=None,
    ):
        """
        Returns a sample of all publicly posted tweets.

        The sample is based on slices of each second, not truly randomised. The
        same tweets are returned for all users of this endpoint.

        If a `threading.Event` is provided and the event is set, the
        sample will be interrupted. This can be used for coordination with other
        programs.

        Calls [GET /2/tweets/sample/stream](https://developer.twitter.com/en/docs/twitter-api/tweets/sampled-stream/api-reference/get-tweets-sample-stream)

        Args:
            event (threading.Event): Manages a flag to stop the process.
            record_keepalive (bool): whether to output keep-alive events.

        Returns:
            generator[dict]: a generator, dict for each tweet.
        """
        url = "https://api.twitter.com/2/tweets/sample/stream"
        params = self._prepare_params(
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            backfill_minutes=backfill_minutes,
        )
        yield from self._stream(url, params, event, record_keepalive)

    @requires_app_auth
    def add_stream_rules(self, rules):
        """
        Adds new rules to the filter stream.

        Calls [POST /2/tweets/search/stream/rules](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/post-tweets-search-stream-rules)

        Args:
            rules (list[dict]): A list of rules to add.

        Returns:
            dict: JSON Response from Twitter API.
        """
        url = "https://api.twitter.com/2/tweets/search/stream/rules"
        return self.post(url, {"add": rules}).json()

    @requires_app_auth
    def get_stream_rules(self):
        """
        Returns a list of rules for the filter stream.

        Calls [GET /2/tweets/search/stream/rules](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream-rules)

        Returns:
            dict: JSON Response from Twitter API with a list of defined rules.
        """
        url = "https://api.twitter.com/2/tweets/search/stream/rules"
        return self.get(url).json()

    @requires_app_auth
    def delete_stream_rule_ids(self, rule_ids):
        """
        Deletes rules from the filter stream.

        Calls [POST /2/tweets/search/stream/rules](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/post-tweets-search-stream-rules)

        Args:
            rule_ids (list[int]): A list of rule ids to delete.

        Returns:
            dict: JSON Response from Twitter API.
        """
        url = "https://api.twitter.com/2/tweets/search/stream/rules"
        return self.post(url, {"delete": {"ids": rule_ids}}).json()

    @requires_app_auth
    def stream(
        self,
        event=None,
        record_keepalive=False,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        backfill_minutes=None,
    ):
        """
        Returns a stream of tweets matching the defined rules.

        Rules can be added or removed out-of-band, without disconnecting.
        Tweet results will contain metadata about the rule that matched it.

        If event is set with a threading.Event object, the sample stream
        will be interrupted. This can be used for coordination with other
        programs.

        Calls [GET /2/tweets/search/stream](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream)

        Args:
            event (threading.Event): Manages a flag to stop the process.
            record_keepalive (bool): whether to output keep-alive events.

        Returns:
            generator[dict]: a generator, dict for each tweet.
        """
        url = "https://api.twitter.com/2/tweets/search/stream"
        params = self._prepare_params(
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            backfill_minutes=backfill_minutes,
        )
        yield from self._stream(url, params, event, record_keepalive)

    def _stream(self, url, params, event, record_keepalive, tries=30):
        """
        A generator that handles streaming data from a response and catches and
        logs any request exceptions, sleeps (exponential backoff) and restarts
        the stream.

        Args:
            url (str): the streaming endpoint URL
            params (dict): any query paramters to use with the url
            event (threading.Event): Manages a flag to stop the process.
            record_keepalive (bool): whether to output keep-alive events.
            tries (int): the number of times to retry connecting after an error
        Returns:
            generator[dict]: A generator of tweet dicts.
        """
        errors = 0
        while True:
            log.info(f"connecting to stream {url}")
            resp = self.get(url, params=params, stream=True)

            try:
                for line in resp.iter_lines():
                    errors = 0

                    # quit & close the stream if the event is set
                    if event and event.is_set():
                        log.info("stopping response stream")
                        resp.close()
                        return

                    # return the JSON data w/ optional keep-alive
                    if not line:
                        log.info("keep-alive")
                        if record_keepalive:
                            yield "keep-alive"
                        continue
                    else:
                        data = json.loads(line.decode())
                        if self.metadata:
                            data = _append_metadata(data, resp.url)
                        yield data
                        if self._check_for_disconnect(data):
                            break

            except requests.exceptions.RequestException as e:
                log.warn("caught exception during streaming: %s", e)
                errors += 1
                if errors > tries:
                    log.error(f"too many consecutive errors ({tries}). stopping")
                    return
                else:
                    secs = errors**2
                    log.info("sleeping %s seconds before reconnecting", secs)
                    time.sleep(secs)

    def _timeline(
        self,
        user_id,
        timeline_type,
        since_id,
        until_id,
        start_time,
        end_time,
        exclude_retweets,
        exclude_replies,
        max_results=None,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        pagination_token=None,
    ):
        """
        Helper function for user and mention timelines

        Calls [GET /2/users/:id/tweets](https://developer.twitter.com/en/docs/twitter-api/tweets/timelines/api-reference/get-users-id-tweets)
        or [GET /2/users/:id/mentions](https://developer.twitter.com/en/docs/twitter-api/tweets/timelines/api-reference/get-users-id-mentions)

        Args:
            user_id (int): ID of the user.
            timeline_type (str): timeline type: `tweets` or `mentions`
            since_id (int): results with a Tweet ID greater than (newer) than specified
            until_id (int): results with a Tweet ID less than (older) than specified
            start_time (datetime): oldest UTC timestamp from which the Tweets will be provided
            end_time (datetime): newest UTC timestamp from which the Tweets will be provided
            exclude_retweets (boolean): remove retweets from timeline
            exlucde_replies (boolean): remove replies from timeline
        Returns:
            generator[dict]: A generator, dict for each page of results.
        """

        url = f"https://api.twitter.com/2/users/{user_id}/{timeline_type}"

        params = self._prepare_params(
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            max_results=max_results,
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            pagination_token=pagination_token,
        )

        excludes = []
        if exclude_retweets:
            excludes.append("retweets")
        if exclude_replies:
            excludes.append("replies")
        if len(excludes) > 0:
            params["exclude"] = ",".join(excludes)

        for response in self.get_paginated(url, params=params):
            # can return without 'data' if there are no results
            if "data" in response:
                yield response
            else:
                log.info(f"Retrieved an empty page of results for timeline {user_id}")

        log.info(f"No more results for timeline {user_id}.")

    def timeline(
        self,
        user,
        since_id=None,
        until_id=None,
        start_time=None,
        end_time=None,
        exclude_retweets=False,
        exclude_replies=False,
        max_results=100,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        pagination_token=None,
    ):
        """
        Retrieve up to the 3200 most recent tweets made by the given user.

        Calls [GET /2/users/:id/tweets](https://developer.twitter.com/en/docs/twitter-api/tweets/timelines/api-reference/get-users-id-tweets)

        Args:
            user (int): ID of the user.
            since_id (int): results with a Tweet ID greater than (newer) than specified
            until_id (int): results with a Tweet ID less than (older) than specified
            start_time (datetime): oldest UTC timestamp from which the Tweets will be provided
            end_time (datetime): newest UTC timestamp from which the Tweets will be provided
            exclude_retweets (boolean): remove retweets from timeline results
            exclude_replies (boolean): remove replies from timeline results
            max_results (int): the maximum number of Tweets to retrieve. Between 5 and 100.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user)
        return self._timeline(
            user_id=user_id,
            timeline_type="tweets",
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            exclude_retweets=exclude_retweets,
            exclude_replies=exclude_replies,
            max_results=max_results,
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            pagination_token=pagination_token,
        )

    def mentions(
        self,
        user,
        since_id=None,
        until_id=None,
        start_time=None,
        end_time=None,
        exclude_retweets=False,
        exclude_replies=False,
        max_results=100,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        pagination_token=None,
    ):
        """
        Retrieve up to the 800 most recent tweets mentioning the given user.

        Calls [GET /2/users/:id/mentions](https://developer.twitter.com/en/docs/twitter-api/tweets/timelines/api-reference/get-users-id-mentions)

        Args:
            user (int): ID of the user.
            since_id (int): results with a Tweet ID greater than (newer) than specified
            until_id (int): results with a Tweet ID less than (older) than specified
            start_time (datetime): oldest UTC timestamp from which the Tweets will be provided
            end_time (datetime): newest UTC timestamp from which the Tweets will be provided
            exclude_retweets (boolean): remove retweets from timeline results
            exclude_replies (boolean): remove replies from timeline results
            max_results (int): the maximum number of Tweets to retrieve. Between 5 and 100.


        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user)
        return self._timeline(
            user_id=user_id,
            timeline_type="mentions",
            since_id=since_id,
            until_id=until_id,
            start_time=start_time,
            end_time=end_time,
            exclude_retweets=exclude_retweets,
            exclude_replies=exclude_replies,
            max_results=max_results,
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
            pagination_token=pagination_token,
        )

    def following(
        self,
        user,
        user_id=None,
        max_results=1000,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        pagination_token=None,
    ):
        """
        Retrieve the user profiles of accounts followed by the given user.

        Calls [GET /2/users/:id/following](https://developer.twitter.com/en/docs/twitter-api/users/follows/api-reference/get-users-id-following)

        Args:
            user (int): ID of the user.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user) if not user_id else user_id
        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )
        if expansions:
            params["expansions"] = "pinned_tweet_id"
        url = f"https://api.twitter.com/2/users/{user_id}/following"
        return self.get_paginated(url, params=params)

    def followers(
        self,
        user,
        user_id=None,
        max_results=1000,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        pagination_token=None,
    ):
        """
        Retrieve the user profiles of accounts following the given user.

        Calls [GET /2/users/:id/followers](https://developer.twitter.com/en/docs/twitter-api/users/follows/api-reference/get-users-id-followers)

        Args:
            user (int): ID of the user.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """
        user_id = self._ensure_user_id(user) if not user_id else user_id
        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )
        if expansions:
            params["expansions"] = "pinned_tweet_id"
        url = f"https://api.twitter.com/2/users/{user_id}/followers"
        return self.get_paginated(url, params=params)

    def liking_users(
        self,
        tweet_id,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        max_results=100,
        pagination_token=None,
    ):
        """
        Retrieve the user profiles of accounts that have liked the given tweet.

        """
        url = f"https://api.twitter.com/2/tweets/{tweet_id}/liking_users"

        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        if expansions:
            params["expansions"] = "pinned_tweet_id"

        for page in self.get_paginated(url, params=params):
            if "data" in page:
                yield page
            else:
                log.info(
                    f"Retrieved an empty page of results for liking_users of {tweet_id}"
                )

    def liked_tweets(
        self,
        user_id,
        max_results=100,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        pagination_token=None,
    ):
        """
        Retrieve the tweets liked by the given user_id.

        """
        user_id = self._ensure_user_id(user_id)
        url = f"https://api.twitter.com/2/users/{user_id}/liked_tweets"

        params = self._prepare_params(
            max_results=100,
            expansions=None,
            tweet_fields=None,
            user_fields=None,
            media_fields=None,
            poll_fields=None,
            place_fields=None,
            pagination_token=None,
        )

        for page in self.get_paginated(url, params=params):
            if "data" in page:
                yield page
            else:
                log.info(
                    f"Retrieved an empty page of results for liked_tweets of {user_id}"
                )

    def retweeted_by(
        self,
        tweet_id,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        max_results=100,
        pagination_token=None,
    ):
        """
        Retrieve the user profiles of accounts that have retweeted the given tweet.

        """
        url = f"https://api.twitter.com/2/tweets/{tweet_id}/retweeted_by"

        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        if expansions:
            params["expansions"] = "pinned_tweet_id"

        for page in self.get_paginated(url, params=params):
            if "data" in page:
                yield page
            else:
                log.info(
                    f"Retrieved an empty page of results for retweeted_by of {tweet_id}"
                )

    def quotes(
        self,
        tweet_id,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        max_results=100,
        pagination_token=None,
    ):
        """
        Retrieve the tweets that quote tweet the given tweet.

        """
        url = f"https://api.twitter.com/2/tweets/{tweet_id}/quote_tweets"

        params = self._prepare_params(
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            max_results=max_results,
            pagination_token=pagination_token,
        )

        for page in self.get_paginated(url, params=params):
            if "data" in page:
                yield page
            else:
                log.info(f"Retrieved an empty page of results for quotes of {tweet_id}")

    @catch_request_exceptions
    @rate_limit
    def get(self, *args, **kwargs):
        """
        Make a GET request to a specified URL.

        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.

        Returns:
            requests.Response: Response from Twitter API.
        """
        if not self.client:
            self.connect()
        log.info("getting %s %s", args, kwargs)
        r = self.last_response = self.client.get(*args, timeout=(3.05, 31), **kwargs)
        return r

    def get_paginated(self, *args, **kwargs):
        """
        A wrapper around the `get` method that handles Twitter token based
        pagination.

        Yields one page (one API response) at a time.

        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.

        Returns:
            generator[dict]: A generator, dict for each page of results.
        """

        resp = self.get(*args, **kwargs)
        page = resp.json()

        url = args[0]

        if self.metadata:
            page = _append_metadata(page, resp.url)

        yield page

        # Todo: Maybe this should be backwards.. check for `next_token`
        endings = [
            "mentions",
            "tweets",
            "following",
            "followers",
            "liked_tweets",
            "liking_users",
            "retweeted_by",
            "members",
            "memberships",
            "followed_lists",
            "owned_lists",
            "pinned_lists",
        ]

        # The search endpoints only take a next_token, but the timeline
        # endpoints take a pagination_token instead - this is a bit of a hack,
        # but check the URL ending to see which we should use.
        if any(url.endswith(end) for end in endings):
            token_param = "pagination_token"
        else:
            token_param = "next_token"

        while "meta" in page and "next_token" in page["meta"]:
            if "params" in kwargs:
                kwargs["params"][token_param] = page["meta"]["next_token"]
            else:
                kwargs["params"] = {token_param: page["meta"]["next_token"]}

            resp = self.get(*args, **kwargs)
            page = resp.json()

            if self.metadata:
                page = _append_metadata(page, resp.url)

            yield page

    @catch_request_exceptions
    @rate_limit
    def post(self, url, json_data):
        """
        Make a POST request to the specified URL.

        Args:
            url (str): URL to make a POST request
            json_data (dict): JSON data to send.

        Returns:
            requests.Response: Response from Twitter API.
        """
        if not self.client:
            self.connect()
        return self.client.post(url, json=json_data)

    def connect(self):
        """
        Sets up the HTTP session to talk to Twitter. If one is active it is
        closed and another one is opened.
        """
        if self.last_response:
            self.last_response.close()

        if self.client:
            self.client.close()

        if self.auth_type == "application" and self.bearer_token:
            log.info("creating HTTP session headers for app auth.")
            auth = f"Bearer {self.bearer_token}"
            log.debug("authorization: %s", auth)
            self.client = requests.Session()
            self.client.headers.update({"Authorization": auth})
        elif self.auth_type == "application":
            log.info("creating app auth client via OAuth2")
            log.debug("client_id: %s", self.consumer_key)
            log.debug("client_secret: %s", self.consumer_secret)
            client = BackendApplicationClient(client_id=self.consumer_key)
            self.client = OAuth2Session(client=client)
            self.client.fetch_token(
                token_url="https://api.twitter.com/oauth2/token",
                client_id=self.consumer_key,
                client_secret=self.consumer_secret,
            )
        else:
            log.info("creating user auth client")
            log.debug("client_id: %s", self.consumer_key)
            log.debug("client_secret: %s", self.consumer_secret)
            log.debug("resource_owner_key: %s", self.access_token)
            log.debug("resource_owner_secret: %s", self.access_token_secret)
            self.client = OAuth1Session(
                client_key=self.consumer_key,
                client_secret=self.consumer_secret,
                resource_owner_key=self.access_token,
                resource_owner_secret=self.access_token_secret,
            )

        if self.client:
            self.client.headers.update({"User-Agent": user_agent})

    @requires_app_auth
    def compliance_job_list(self, job_type, status):
        """
        Returns list of compliance jobs.

        Calls [GET /2/compliance/jobs](https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/api-reference/get-compliance-jobs)

        Args:
            job_type (str): Filter by job type - either tweets or users.
            status (str): Filter by job status. Only one of 'created', 'in_progress', 'complete', 'failed' can be specified. If not set, returns all.

        Returns:
            list[dict]: A list of jobs.
        """
        params = {}
        if job_type:
            params["type"] = job_type
        if status:
            params["status"] = status
        result = self.client.get(
            "https://api.twitter.com/2/compliance/jobs", params=params
        ).json()
        if "data" in result or not result:
            return result
        else:
            raise ValueError(f"Unknown response from twitter: {result}")

    @requires_app_auth
    def compliance_job_get(self, job_id):
        """
        Returns a compliance job.

        Calls [GET /2/compliance/jobs/{job_id}](https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/api-reference/get-compliance-jobs-id)

        Args:
            job_id (int): The ID of the compliance job.

        Returns:
            dict: A compliance job.
        """
        result = self.client.get(
            "https://api.twitter.com/2/compliance/jobs/{}".format(job_id)
        )
        if result.status_code == 200:
            result = result.json()
        else:
            raise ValueError(f"Error from API, response: {result.status_code}")
        if "data" in result:
            return result
        else:
            raise ValueError(f"Unknown response from twitter: {result}")

    @requires_app_auth
    def compliance_job_create(self, job_type, job_name, resumable=False):
        """
        Creates a new compliace job.

        Calls [POST /2/compliance/jobs](https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/api-reference/post-compliance-jobs)

        Args:
            job_type (str): The type of job to create. Either 'tweets' or 'users'.
            job_name (str): Optional name for the job.
            resumable (bool): Whether or not the job upload is resumable.
        """
        payload = {}
        payload["type"] = job_type
        payload["resumable"] = resumable
        if job_name:
            payload["name"] = job_name

        result = self.client.post(
            "https://api.twitter.com/2/compliance/jobs", json=payload
        )

        if result.status_code == 200:
            result = result.json()
        else:
            raise ValueError(f"Error from API, response: {result.status_code}")
        if "data" in result:
            return result
        else:
            raise ValueError(f"Unknown response from twitter: {result}")

    def geo(
        self,
        lat=None,
        lon=None,
        query=None,
        ip=None,
        granularity="neighborhood",
        max_results=None,
    ):
        """
        Gets geographic places that can be useful in queries. This is a v1.1
        endpoint but is useful in querying the v2 API.

        Calls [1.1/geo/search.json](https://api.twitter.com/1.1/geo/search.json)

        Args:
            lat (float): latitude to search around
            lon (float): longitude to search around
            query (str): text to match in the place name
            ip (str): use the ip address to locate places
            granularity (str) : neighborhood, city, admin, country
            max_results (int): maximum results to return
        """

        params = {}
        if lat and lon:
            params["lat"] = lat
            params["long"] = lon
        elif query:
            params["query"] = query
        elif ip:
            params["ip"] = ip
        else:
            raise ValueError("geo() needs either lat/lon, query or ip)")

        if granularity not in ["neighborhood", "city", "admin", "country"]:
            raise ValueError(
                "{granularity} is not valid value for granularity, please use neighborhood, city, admin or country"
            )
        params["granularity"] = granularity

        if max_results and type(max_results) != int:
            raise ValueError("max_results must be an int")
        params["max_results"] = max_results

        url = "https://api.twitter.com/1.1/geo/search.json"

        result = self.get(url, params=params)
        if result.status_code == 200:
            result = result.json()
        else:
            raise ValueError(f"Error from API, response: {result.status_code}")

        return result

    def _id_exists(self, user):
        """
        Returns True if the user id exists
        """
        try:
            error_name = next(self.user_lookup([user]))["errors"][0]["title"]
            return error_name != "Not Found Error"
        except KeyError:
            return True

    def _ensure_user_id(self, user):
        """
        Always return a valid user id, look up if not numeric.
        """
        user = str(user)
        is_numeric = re.match(r"^\d+$", user)

        if len(user) > 15 or (is_numeric and self._id_exists(user)):
            return user
        else:
            results = next(self.user_lookup([user], usernames=True))
            if "data" in results and len(results["data"]) > 0:
                return results["data"][0]["id"]
            elif is_numeric:
                return user
            else:
                raise ValueError(f"No such user {user}")

    def _ensure_user(self, user):
        """
        Always return a valid user object.
        """
        user = str(user)
        is_numeric = re.match(r"^\d+$", user)

        lookup = []
        if len(user) > 15 or (is_numeric and self._id_exists(user)):
            lookup = list(self.user_lookup([user]))[0]
        else:
            lookup = list(self.user_lookup([user], usernames=True))[0]

        if "data" in lookup:
            return lookup["data"][0]
        else:
            raise ValueError(f"No such user {user}")

    def _check_for_disconnect(self, data):
        """
        Look for disconnect errors in a response, and reconnect if found. The
        function returns True if a disconnect was found and False otherwise.
        """
        for error in data.get("errors", []):
            if error.get("disconnect_type") == "OperationalDisconnect":
                log.info("Received operational disconnect message, reconnecting")
                self.connect()
                return True
        return False

__init__(consumer_key=None, consumer_secret=None, access_token=None, access_token_secret=None, bearer_token=None, connection_errors=0, metadata=True)

Instantiate a Twarc2 instance to talk to the Twitter V2+ API.

The client can use either App or User authentication, but only one at a time. Whether app auth or user auth is used depends on which credentials are provided on initialisation:

1. If a bearer_token is passed, app auth is always used.
2. If a consumer_key and consumer_secret are passed without an access_token and access_token_secret, app auth is used.
3. If consumer_key, consumer_secret, access_token and access_token_secret are all passed, then user authentication is used instead.

Parameters:

Name	Type	Description	Default
consumer_key	str	The API key.	None
consumer_secret	str	The API secret.	None
access_token	str	The Access Token	None
access_token_secret	str	The Access Token Secret	None
bearer_token	str	Bearer Token, can be generated from API keys.	None
connection_errors	int	Number of retries for GETs	0
metadata	bool	Append __twarc metadata to results.	True

Source code in twarc/client2.py

def __init__(
    self,
    consumer_key=None,
    consumer_secret=None,
    access_token=None,
    access_token_secret=None,
    bearer_token=None,
    connection_errors=0,
    metadata=True,
):
    """
    Instantiate a Twarc2 instance to talk to the Twitter V2+ API.

    The client can use either App or User authentication, but only one at a
    time. Whether app auth or user auth is used depends on which credentials
    are provided on initialisation:

    1. If a `bearer_token` is passed, app auth is always used.
    2. If a `consumer_key` and `consumer_secret` are passed without an
    `access_token` and `access_token_secret`, app auth is used.
    3. If `consumer_key`, `consumer_secret`, `access_token` and
    `access_token_secret` are all passed, then user authentication
    is used instead.

    Args:
        consumer_key (str):
            The API key.
        consumer_secret (str):
            The API secret.
        access_token (str):
            The Access Token
        access_token_secret (str):
            The Access Token Secret
        bearer_token (str):
            Bearer Token, can be generated from API keys.
        connection_errors (int):
            Number of retries for GETs
        metadata (bool):
            Append `__twarc` metadata to results.
    """
    self.api_version = "2"
    self.connection_errors = connection_errors
    self.metadata = metadata
    self.bearer_token = None

    if bearer_token:
        self.bearer_token = bearer_token
        self.auth_type = "application"

    elif consumer_key and consumer_secret:
        if access_token and access_token_secret:
            self.consumer_key = consumer_key
            self.consumer_secret = consumer_secret
            self.access_token = access_token
            self.access_token_secret = access_token_secret
            self.auth_type = "user"

        else:
            self.consumer_key = consumer_key
            self.consumer_secret = consumer_secret
            self.auth_type = "application"

    else:
        raise ValueError(
            "Must pass either a bearer_token or consumer/access_token keys and secrets"
        )

    self.client = None
    self.last_response = None

    self.connect()

add_stream_rules(rules)

Adds new rules to the filter stream.

Calls POST /2/tweets/search/stream/rules

Parameters:

Name	Type	Description	Default
rules	list[dict]	A list of rules to add.	required

Returns:

Name	Type	Description
dict		JSON Response from Twitter API.

Source code in twarc/client2.py

@requires_app_auth
def add_stream_rules(self, rules):
    """
    Adds new rules to the filter stream.

    Calls [POST /2/tweets/search/stream/rules](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/post-tweets-search-stream-rules)

    Args:
        rules (list[dict]): A list of rules to add.

    Returns:
        dict: JSON Response from Twitter API.
    """
    url = "https://api.twitter.com/2/tweets/search/stream/rules"
    return self.post(url, {"add": rules}).json()

compliance_job_create(job_type, job_name, resumable=False)

Creates a new compliace job.

Calls POST /2/compliance/jobs

Parameters:

Name	Type	Description	Default
job_type	str	The type of job to create. Either 'tweets' or 'users'.	required
job_name	str	Optional name for the job.	required
resumable	bool	Whether or not the job upload is resumable.	False

Source code in twarc/client2.py

@requires_app_auth
def compliance_job_create(self, job_type, job_name, resumable=False):
    """
    Creates a new compliace job.

    Calls [POST /2/compliance/jobs](https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/api-reference/post-compliance-jobs)

    Args:
        job_type (str): The type of job to create. Either 'tweets' or 'users'.
        job_name (str): Optional name for the job.
        resumable (bool): Whether or not the job upload is resumable.
    """
    payload = {}
    payload["type"] = job_type
    payload["resumable"] = resumable
    if job_name:
        payload["name"] = job_name

    result = self.client.post(
        "https://api.twitter.com/2/compliance/jobs", json=payload
    )

    if result.status_code == 200:
        result = result.json()
    else:
        raise ValueError(f"Error from API, response: {result.status_code}")
    if "data" in result:
        return result
    else:
        raise ValueError(f"Unknown response from twitter: {result}")

compliance_job_get(job_id)

Returns a compliance job.

Calls GET /2/compliance/jobs/{job_id}

Parameters:

Name	Type	Description	Default
job_id	int	The ID of the compliance job.	required

Returns:

Name	Type	Description
dict		A compliance job.

Source code in twarc/client2.py

@requires_app_auth
def compliance_job_get(self, job_id):
    """
    Returns a compliance job.

    Calls [GET /2/compliance/jobs/{job_id}](https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/api-reference/get-compliance-jobs-id)

    Args:
        job_id (int): The ID of the compliance job.

    Returns:
        dict: A compliance job.
    """
    result = self.client.get(
        "https://api.twitter.com/2/compliance/jobs/{}".format(job_id)
    )
    if result.status_code == 200:
        result = result.json()
    else:
        raise ValueError(f"Error from API, response: {result.status_code}")
    if "data" in result:
        return result
    else:
        raise ValueError(f"Unknown response from twitter: {result}")

compliance_job_list(job_type, status)

Returns list of compliance jobs.

Calls GET /2/compliance/jobs

Parameters:

Name	Type	Description	Default
job_type	str	Filter by job type - either tweets or users.	required
status	str	Filter by job status. Only one of 'created', 'in_progress', 'complete', 'failed' can be specified. If not set, returns all.	required

Returns:

Type	Description
	list[dict]: A list of jobs.

Source code in twarc/client2.py

@requires_app_auth
def compliance_job_list(self, job_type, status):
    """
    Returns list of compliance jobs.

    Calls [GET /2/compliance/jobs](https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/api-reference/get-compliance-jobs)

    Args:
        job_type (str): Filter by job type - either tweets or users.
        status (str): Filter by job status. Only one of 'created', 'in_progress', 'complete', 'failed' can be specified. If not set, returns all.

    Returns:
        list[dict]: A list of jobs.
    """
    params = {}
    if job_type:
        params["type"] = job_type
    if status:
        params["status"] = status
    result = self.client.get(
        "https://api.twitter.com/2/compliance/jobs", params=params
    ).json()
    if "data" in result or not result:
        return result
    else:
        raise ValueError(f"Unknown response from twitter: {result}")

connect()

Sets up the HTTP session to talk to Twitter. If one is active it is closed and another one is opened.

Source code in twarc/client2.py

def connect(self):
    """
    Sets up the HTTP session to talk to Twitter. If one is active it is
    closed and another one is opened.
    """
    if self.last_response:
        self.last_response.close()

    if self.client:
        self.client.close()

    if self.auth_type == "application" and self.bearer_token:
        log.info("creating HTTP session headers for app auth.")
        auth = f"Bearer {self.bearer_token}"
        log.debug("authorization: %s", auth)
        self.client = requests.Session()
        self.client.headers.update({"Authorization": auth})
    elif self.auth_type == "application":
        log.info("creating app auth client via OAuth2")
        log.debug("client_id: %s", self.consumer_key)
        log.debug("client_secret: %s", self.consumer_secret)
        client = BackendApplicationClient(client_id=self.consumer_key)
        self.client = OAuth2Session(client=client)
        self.client.fetch_token(
            token_url="https://api.twitter.com/oauth2/token",
            client_id=self.consumer_key,
            client_secret=self.consumer_secret,
        )
    else:
        log.info("creating user auth client")
        log.debug("client_id: %s", self.consumer_key)
        log.debug("client_secret: %s", self.consumer_secret)
        log.debug("resource_owner_key: %s", self.access_token)
        log.debug("resource_owner_secret: %s", self.access_token_secret)
        self.client = OAuth1Session(
            client_key=self.consumer_key,
            client_secret=self.consumer_secret,
            resource_owner_key=self.access_token,
            resource_owner_secret=self.access_token_secret,
        )

    if self.client:
        self.client.headers.update({"User-Agent": user_agent})

counts_all(query, since_id=None, until_id=None, start_time=None, end_time=None, granularity='hour', next_token=None)

Retrieve counts for the given query in the full archive, using the /search/all endpoint (Requires Academic Access).

Calls GET /2/tweets/counts/all

Parameters:

Name	Type	Description	Default
query	str	The query string to be passed directly to the Twitter API.	required
since_id	int	Return all tweets since this tweet_id.	None
until_id	int	Return all tweets up to this tweet_id.	None
start_time	datetime	Return all tweets after this time (UTC datetime).	None
end_time	datetime	Return all tweets before this time (UTC datetime).	None
granularity	str	Count aggregation level: day, hour, minute. Default is hour.	'hour'

Returns:

Type	Description
	generator[dict]: a generator, dict for each paginated response.

Source code in twarc/client2.py

@requires_app_auth
def counts_all(
    self,
    query,
    since_id=None,
    until_id=None,
    start_time=None,
    end_time=None,
    granularity="hour",
    next_token=None,
):
    """
    Retrieve counts for the given query in the full archive,
    using the `/search/all` endpoint (Requires Academic Access).

    Calls [GET /2/tweets/counts/all](https://developer.twitter.com/en/docs/twitter-api/tweets/counts/api-reference/get-tweets-counts-all)

    Args:
        query (str):
            The query string to be passed directly to the Twitter API.
        since_id (int):
            Return all tweets since this tweet_id.
        until_id (int):
            Return all tweets up to this tweet_id.
        start_time (datetime):
            Return all tweets after this time (UTC datetime).
        end_time (datetime):
            Return all tweets before this time (UTC datetime).
        granularity (str):
            Count aggregation level: `day`, `hour`, `minute`.
            Default is `hour`.

    Returns:
        generator[dict]: a generator, dict for each paginated response.
    """
    return self._search(
        url="https://api.twitter.com/2/tweets/counts/all",
        query=query,
        since_id=since_id,
        until_id=until_id,
        start_time=start_time,
        end_time=end_time,
        max_results=None,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        next_token=next_token,
        granularity=granularity,
        sleep_between=1.05,
        sort_order=None,
    )

counts_recent(query, since_id=None, until_id=None, start_time=None, end_time=None, granularity='hour')

Retrieve counts for the given query in the last seven days, using the /counts/recent endpoint.

Calls GET /2/tweets/counts/recent

Parameters:

Name	Type	Description	Default
query	str	The query string to be passed directly to the Twitter API.	required
since_id	int	Return all tweets since this tweet_id.	None
until_id	int	Return all tweets up to this tweet_id.	None
start_time	datetime	Return all tweets after this time (UTC datetime).	None
end_time	datetime	Return all tweets before this time (UTC datetime).	None
granularity	str	Count aggregation level: day, hour, minute. Default is hour.	'hour'

Returns:

Type	Description
	generator[dict]: a generator, dict for each paginated response.

Source code in twarc/client2.py

@requires_app_auth
def counts_recent(
    self,
    query,
    since_id=None,
    until_id=None,
    start_time=None,
    end_time=None,
    granularity="hour",
):
    """
    Retrieve counts for the given query in the last seven days,
    using the `/counts/recent` endpoint.

    Calls [GET /2/tweets/counts/recent](https://developer.twitter.com/en/docs/twitter-api/tweets/counts/api-reference/get-tweets-counts-recent)

    Args:
        query (str):
            The query string to be passed directly to the Twitter API.
        since_id (int):
            Return all tweets since this tweet_id.
        until_id (int):
            Return all tweets up to this tweet_id.
        start_time (datetime):
            Return all tweets after this time (UTC datetime).
        end_time (datetime):
            Return all tweets before this time (UTC datetime).
        granularity (str):
            Count aggregation level: `day`, `hour`, `minute`.
            Default is `hour`.

    Returns:
        generator[dict]: a generator, dict for each paginated response.
    """
    return self._search(
        url="https://api.twitter.com/2/tweets/counts/recent",
        query=query,
        since_id=since_id,
        until_id=until_id,
        start_time=start_time,
        end_time=end_time,
        max_results=None,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        granularity=granularity,
        sort_order=None,
    )

delete_stream_rule_ids(rule_ids)

Deletes rules from the filter stream.

Calls POST /2/tweets/search/stream/rules

Parameters:

Name	Type	Description	Default
rule_ids	list[int]	A list of rule ids to delete.	required

Returns:

Name	Type	Description
dict		JSON Response from Twitter API.

Source code in twarc/client2.py

@requires_app_auth
def delete_stream_rule_ids(self, rule_ids):
    """
    Deletes rules from the filter stream.

    Calls [POST /2/tweets/search/stream/rules](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/post-tweets-search-stream-rules)

    Args:
        rule_ids (list[int]): A list of rule ids to delete.

    Returns:
        dict: JSON Response from Twitter API.
    """
    url = "https://api.twitter.com/2/tweets/search/stream/rules"
    return self.post(url, {"delete": {"ids": rule_ids}}).json()

followed_lists(user, expansions=None, list_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns all Lists a specified user follows.

Calls GET /2/users/:id/followed_lists

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required
expansions	enum (owner_id	enable you to request additional data objects that relate to the originally returned List.	None
list_fields	enum (created_at, follower_count, member_count, private, description, owner_id	This fields parameter enables you to select which specific List fields will deliver with each returned List objects.	None
user_fields	enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld	This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.	None
max_results	int	The maximum number of results to be returned per page. This can be a number between 1 and 100.	None
pagination_token	string	Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def followed_lists(
    self,
    user,
    expansions=None,
    list_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns all Lists a specified user follows.

    Calls [GET /2/users/:id/followed_lists](https://developer.twitter.com/en/docs/twitter-api/lists/list-follows/api-reference/get-users-id-followed_lists)

    Args:
        user (int): ID of the user.
        expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
        list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
        user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
            This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
        max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
        pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user)
    url = f"https://api.twitter.com/2/users/{user_id}/followed_lists"

    return self._lists(
        url=url,
        expansions=expansions,
        list_fields=list_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

followers(user, user_id=None, max_results=1000, expansions=None, tweet_fields=None, user_fields=None, pagination_token=None)

Retrieve the user profiles of accounts following the given user.

Calls GET /2/users/:id/followers

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def followers(
    self,
    user,
    user_id=None,
    max_results=1000,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    pagination_token=None,
):
    """
    Retrieve the user profiles of accounts following the given user.

    Calls [GET /2/users/:id/followers](https://developer.twitter.com/en/docs/twitter-api/users/follows/api-reference/get-users-id-followers)

    Args:
        user (int): ID of the user.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user) if not user_id else user_id
    params = self._prepare_params(
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )
    if expansions:
        params["expansions"] = "pinned_tweet_id"
    url = f"https://api.twitter.com/2/users/{user_id}/followers"
    return self.get_paginated(url, params=params)

following(user, user_id=None, max_results=1000, expansions=None, tweet_fields=None, user_fields=None, pagination_token=None)

Retrieve the user profiles of accounts followed by the given user.

Calls GET /2/users/:id/following

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def following(
    self,
    user,
    user_id=None,
    max_results=1000,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    pagination_token=None,
):
    """
    Retrieve the user profiles of accounts followed by the given user.

    Calls [GET /2/users/:id/following](https://developer.twitter.com/en/docs/twitter-api/users/follows/api-reference/get-users-id-following)

    Args:
        user (int): ID of the user.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user) if not user_id else user_id
    params = self._prepare_params(
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )
    if expansions:
        params["expansions"] = "pinned_tweet_id"
    url = f"https://api.twitter.com/2/users/{user_id}/following"
    return self.get_paginated(url, params=params)

geo(lat=None, lon=None, query=None, ip=None, granularity='neighborhood', max_results=None)

Gets geographic places that can be useful in queries. This is a v1.1 endpoint but is useful in querying the v2 API.

Calls 1.1/geo/search.json

Parameters:

Name	Type	Description	Default
lat	float	latitude to search around	None
lon	float	longitude to search around	None
query	str	text to match in the place name	None
ip	str	use the ip address to locate places	None
granularity	str)	neighborhood, city, admin, country	'neighborhood'
max_results	int	maximum results to return	None

Source code in twarc/client2.py

def geo(
    self,
    lat=None,
    lon=None,
    query=None,
    ip=None,
    granularity="neighborhood",
    max_results=None,
):
    """
    Gets geographic places that can be useful in queries. This is a v1.1
    endpoint but is useful in querying the v2 API.

    Calls [1.1/geo/search.json](https://api.twitter.com/1.1/geo/search.json)

    Args:
        lat (float): latitude to search around
        lon (float): longitude to search around
        query (str): text to match in the place name
        ip (str): use the ip address to locate places
        granularity (str) : neighborhood, city, admin, country
        max_results (int): maximum results to return
    """

    params = {}
    if lat and lon:
        params["lat"] = lat
        params["long"] = lon
    elif query:
        params["query"] = query
    elif ip:
        params["ip"] = ip
    else:
        raise ValueError("geo() needs either lat/lon, query or ip)")

    if granularity not in ["neighborhood", "city", "admin", "country"]:
        raise ValueError(
            "{granularity} is not valid value for granularity, please use neighborhood, city, admin or country"
        )
    params["granularity"] = granularity

    if max_results and type(max_results) != int:
        raise ValueError("max_results must be an int")
    params["max_results"] = max_results

    url = "https://api.twitter.com/1.1/geo/search.json"

    result = self.get(url, params=params)
    if result.status_code == 200:
        result = result.json()
    else:
        raise ValueError(f"Error from API, response: {result.status_code}")

    return result

get(*args, **kwargs)

Make a GET request to a specified URL.

Parameters:

Name	Type	Description	Default
*args		Variable length argument list.	()
**kwargs		Arbitrary keyword arguments.	{}

Returns:

Type	Description
	requests.Response: Response from Twitter API.

Source code in twarc/client2.py

@catch_request_exceptions
@rate_limit
def get(self, *args, **kwargs):
    """
    Make a GET request to a specified URL.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        requests.Response: Response from Twitter API.
    """
    if not self.client:
        self.connect()
    log.info("getting %s %s", args, kwargs)
    r = self.last_response = self.client.get(*args, timeout=(3.05, 31), **kwargs)
    return r

get_paginated(*args, **kwargs)

A wrapper around the get method that handles Twitter token based pagination.

Yields one page (one API response) at a time.

Parameters:

Name	Type	Description	Default
*args		Variable length argument list.	()
**kwargs		Arbitrary keyword arguments.	{}

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def get_paginated(self, *args, **kwargs):
    """
    A wrapper around the `get` method that handles Twitter token based
    pagination.

    Yields one page (one API response) at a time.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """

    resp = self.get(*args, **kwargs)
    page = resp.json()

    url = args[0]

    if self.metadata:
        page = _append_metadata(page, resp.url)

    yield page

    # Todo: Maybe this should be backwards.. check for `next_token`
    endings = [
        "mentions",
        "tweets",
        "following",
        "followers",
        "liked_tweets",
        "liking_users",
        "retweeted_by",
        "members",
        "memberships",
        "followed_lists",
        "owned_lists",
        "pinned_lists",
    ]

    # The search endpoints only take a next_token, but the timeline
    # endpoints take a pagination_token instead - this is a bit of a hack,
    # but check the URL ending to see which we should use.
    if any(url.endswith(end) for end in endings):
        token_param = "pagination_token"
    else:
        token_param = "next_token"

    while "meta" in page and "next_token" in page["meta"]:
        if "params" in kwargs:
            kwargs["params"][token_param] = page["meta"]["next_token"]
        else:
            kwargs["params"] = {token_param: page["meta"]["next_token"]}

        resp = self.get(*args, **kwargs)
        page = resp.json()

        if self.metadata:
            page = _append_metadata(page, resp.url)

        yield page

get_stream_rules()

Returns a list of rules for the filter stream.

Calls GET /2/tweets/search/stream/rules

Returns:

Name	Type	Description
dict		JSON Response from Twitter API with a list of defined rules.

Source code in twarc/client2.py

@requires_app_auth
def get_stream_rules(self):
    """
    Returns a list of rules for the filter stream.

    Calls [GET /2/tweets/search/stream/rules](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream-rules)

    Returns:
        dict: JSON Response from Twitter API with a list of defined rules.
    """
    url = "https://api.twitter.com/2/tweets/search/stream/rules"
    return self.get(url).json()

liked_tweets(user_id, max_results=100, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, pagination_token=None)

Retrieve the tweets liked by the given user_id.

Source code in twarc/client2.py

def liked_tweets(
    self,
    user_id,
    max_results=100,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    pagination_token=None,
):
    """
    Retrieve the tweets liked by the given user_id.

    """
    user_id = self._ensure_user_id(user_id)
    url = f"https://api.twitter.com/2/users/{user_id}/liked_tweets"

    params = self._prepare_params(
        max_results=100,
        expansions=None,
        tweet_fields=None,
        user_fields=None,
        media_fields=None,
        poll_fields=None,
        place_fields=None,
        pagination_token=None,
    )

    for page in self.get_paginated(url, params=params):
        if "data" in page:
            yield page
        else:
            log.info(
                f"Retrieved an empty page of results for liked_tweets of {user_id}"
            )

liking_users(tweet_id, expansions=None, tweet_fields=None, user_fields=None, max_results=100, pagination_token=None)

Retrieve the user profiles of accounts that have liked the given tweet.

Source code in twarc/client2.py

def liking_users(
    self,
    tweet_id,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    max_results=100,
    pagination_token=None,
):
    """
    Retrieve the user profiles of accounts that have liked the given tweet.

    """
    url = f"https://api.twitter.com/2/tweets/{tweet_id}/liking_users"

    params = self._prepare_params(
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

    if expansions:
        params["expansions"] = "pinned_tweet_id"

    for page in self.get_paginated(url, params=params):
        if "data" in page:
            yield page
        else:
            log.info(
                f"Retrieved an empty page of results for liking_users of {tweet_id}"
            )

list_followers(list_id, expansions=None, tweet_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns a list of users who are followers of the specified List.

Calls GET /2/lists/:id/followers

Parameters:

Name	Type	Description	Default
list_id	int	ID of the list.	required
expansions	enum (pinned_tweet_id	Expansions, include pinned tweets.	None
max_results	int	the maximum number of results to retrieve. Between 1 and 100. Default is 100.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def list_followers(
    self,
    list_id,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns a list of users who are followers of the specified List.

    Calls [GET /2/lists/:id/followers](https://developer.twitter.com/en/docs/twitter-api/lists/list-follows/api-reference/get-lists-id-followers)

    Args:
        list_id (int): ID of the list.
        expansions enum (pinned_tweet_id): Expansions, include pinned tweets.
        max_results (int): the maximum number of results to retrieve. Between 1 and 100. Default is 100.

    Returns:
        generator[dict]: A generator, dict for each page of results.

    """
    params = self._prepare_params(
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

    if expansions:
        params["expansions"] = "pinned_tweet_id"

    url = f"https://api.twitter.com/2/lists/{list_id}/followers"
    return self.get_paginated(url, params=params)

list_lookup(list_id, expansions=None, list_fields=None, user_fields=None)

Returns the details of a specified List.

Calls GET /2/lists/:id

Parameters:

Name	Type	Description	Default
list_id	int	ID of the list.	required
expansions	enum (owner_id	enable you to request additional data objects that relate to the originally returned List.	None
list_fields	enum (created_at, follower_count, member_count, private, description, owner_id	This fields parameter enables you to select which specific List fields will deliver with each returned List objects.	None
user_fields	enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld	This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.	None

Returns:

Name	Type	Description
dict		Result dictionary.

Source code in twarc/client2.py

def list_lookup(self, list_id, expansions=None, list_fields=None, user_fields=None):
    """
    Returns the details of a specified List.

    Calls [GET /2/lists/:id](https://developer.twitter.com/en/docs/twitter-api/lists/list-lookup/api-reference/get-lists-id)

    Args:
        list_id (int): ID of the list.
        expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
        list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
        user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
            This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.

    Returns:
        dict: Result dictionary.
    """

    params = self._prepare_params(
        list_fields=list_fields,
        user_fields=user_fields,
    )

    if expansions:
        params["expansions"] = "owner_id"
    url = f"https://api.twitter.com/2/lists/{list_id}"
    resp = self.get(url, params=params)
    data = resp.json()

    if self.metadata:
        data = _append_metadata(data, resp.url)

    return data

list_members(list_id, expansions=None, tweet_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns a list of users who are members of the specified List.

Calls GET /2/lists/:id/members

Parameters:

Name	Type	Description	Default
list_id	int	ID of the list.	required
expansions	enum (pinned_tweet_id	Expansions, include pinned tweets.	None
max_results	int	The maximum number of results to be returned per page. This can be a number between 1 and 100.	None
pagination_token	string	Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def list_members(
    self,
    list_id,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns a list of users who are members of the specified List.

    Calls [GET /2/lists/:id/members](https://developer.twitter.com/en/docs/twitter-api/lists/list-members/api-reference/get-lists-id-members)

    Args:
        list_id (int): ID of the list.
        expansions enum (pinned_tweet_id): Expansions, include pinned tweets.
        max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
        pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

    Returns:
        generator[dict]: A generator, dict for each page of results.

    """

    params = self._prepare_params(
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

    if expansions:
        params["expansions"] = "pinned_tweet_id"

    url = f"https://api.twitter.com/2/lists/{list_id}/members"
    return self.get_paginated(url, params=params)

list_memberships(user, expansions=None, list_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns all Lists a specified user is a member of.

Calls GET /2/users/:id/list_memberships

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required
expansions	enum (owner_id	enable you to request additional data objects that relate to the originally returned List.	None
list_fields	enum (created_at, follower_count, member_count, private, description, owner_id	This fields parameter enables you to select which specific List fields will deliver with each returned List objects.	None
user_fields	enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld	This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.	None
max_results	int	The maximum number of results to be returned per page. This can be a number between 1 and 100.	None
pagination_token	string	Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def list_memberships(
    self,
    user,
    expansions=None,
    list_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns all Lists a specified user is a member of.

    Calls [GET /2/users/:id/list_memberships](https://developer.twitter.com/en/docs/twitter-api/lists/list-members/api-reference/get-users-id-list_memberships)

    Args:
        user (int): ID of the user.
        expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
        list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
        user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
            This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
        max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
        pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user)
    url = f"https://api.twitter.com/2/users/{user_id}/list_memberships"

    return self._lists(
        url=url,
        expansions=expansions,
        list_fields=list_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

list_tweets(list_id, expansions=None, tweet_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns Tweets from the specified List.

Calls GET /2/lists/:id/tweets

Parameters:

Name	Type	Description	Default
list_id	int	ID of the list.	required
expansions	enum (author_id	enable you to request additional data objects that relate to the originally returned List.	None
list_fields	enum (created_at, follower_count, member_count, private, description, owner_id	This fields parameter enables you to select which specific List fields will deliver with each returned List objects.	required
user_fields	enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld	This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def list_tweets(
    self,
    list_id,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns Tweets from the specified List.

    Calls [GET /2/lists/:id/tweets](https://developer.twitter.com/en/docs/twitter-api/lists/list-tweets/api-reference/get-lists-id-tweets)

    Args:
        list_id (int): ID of the list.
        expansions enum (author_id): enable you to request additional data objects that relate to the originally returned List.
        list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
        user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
            This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """

    params = self._prepare_params(
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

    url = f"https://api.twitter.com/2/lists/{list_id}/tweets"
    return self.get_paginated(url, params=params)

mentions(user, since_id=None, until_id=None, start_time=None, end_time=None, exclude_retweets=False, exclude_replies=False, max_results=100, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, pagination_token=None)

Retrieve up to the 800 most recent tweets mentioning the given user.

Calls GET /2/users/:id/mentions

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required
since_id	int	results with a Tweet ID greater than (newer) than specified	None
until_id	int	results with a Tweet ID less than (older) than specified	None
start_time	datetime	oldest UTC timestamp from which the Tweets will be provided	None
end_time	datetime	newest UTC timestamp from which the Tweets will be provided	None
exclude_retweets	boolean	remove retweets from timeline results	False
exclude_replies	boolean	remove replies from timeline results	False
max_results	int	the maximum number of Tweets to retrieve. Between 5 and 100.	100

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def mentions(
    self,
    user,
    since_id=None,
    until_id=None,
    start_time=None,
    end_time=None,
    exclude_retweets=False,
    exclude_replies=False,
    max_results=100,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    pagination_token=None,
):
    """
    Retrieve up to the 800 most recent tweets mentioning the given user.

    Calls [GET /2/users/:id/mentions](https://developer.twitter.com/en/docs/twitter-api/tweets/timelines/api-reference/get-users-id-mentions)

    Args:
        user (int): ID of the user.
        since_id (int): results with a Tweet ID greater than (newer) than specified
        until_id (int): results with a Tweet ID less than (older) than specified
        start_time (datetime): oldest UTC timestamp from which the Tweets will be provided
        end_time (datetime): newest UTC timestamp from which the Tweets will be provided
        exclude_retweets (boolean): remove retweets from timeline results
        exclude_replies (boolean): remove replies from timeline results
        max_results (int): the maximum number of Tweets to retrieve. Between 5 and 100.


    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user)
    return self._timeline(
        user_id=user_id,
        timeline_type="mentions",
        since_id=since_id,
        until_id=until_id,
        start_time=start_time,
        end_time=end_time,
        exclude_retweets=exclude_retweets,
        exclude_replies=exclude_replies,
        max_results=max_results,
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        media_fields=media_fields,
        poll_fields=poll_fields,
        place_fields=place_fields,
        pagination_token=pagination_token,
    )

owned_lists(user, expansions=None, list_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns all Lists owned by the specified user.

Calls GET /2/users/:id/owned_lists

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required
expansions	enum (owner_id	enable you to request additional data objects that relate to the originally returned List.	None
list_fields	enum (created_at, follower_count, member_count, private, description, owner_id	This fields parameter enables you to select which specific List fields will deliver with each returned List objects.	None
user_fields	enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld	This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.	None
max_results	int	The maximum number of results to be returned per page. This can be a number between 1 and 100.	None
pagination_token	string	Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def owned_lists(
    self,
    user,
    expansions=None,
    list_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns all Lists owned by the specified user.

    Calls [GET /2/users/:id/owned_lists](https://developer.twitter.com/en/docs/twitter-api/lists/list-lookup/api-reference/get-users-id-owned_lists)

    Args:
        user (int): ID of the user.
        expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
        list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
        user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
            This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
        max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
        pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user)
    url = f"https://api.twitter.com/2/users/{user_id}/owned_lists"

    return self._lists(
        url=url,
        expansions=expansions,
        list_fields=list_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

pinned_lists(user, expansions=None, list_fields=None, user_fields=None, max_results=None, pagination_token=None)

Returns the Lists pinned by the authenticating user. Does not work with a Bearer token.

Calls GET /2/users/:id/pinned_lists

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required
expansions	enum (owner_id	enable you to request additional data objects that relate to the originally returned List.	None
list_fields	enum (created_at, follower_count, member_count, private, description, owner_id	This fields parameter enables you to select which specific List fields will deliver with each returned List objects.	None
user_fields	enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld	This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.	None
max_results	int	The maximum number of results to be returned per page. This can be a number between 1 and 100.	None
pagination_token	string	Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.	None

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def pinned_lists(
    self,
    user,
    expansions=None,
    list_fields=None,
    user_fields=None,
    max_results=None,
    pagination_token=None,
):
    """
    Returns the Lists pinned by the authenticating user. Does not work with a Bearer token.

    Calls [GET /2/users/:id/pinned_lists](https://developer.twitter.com/en/docs/twitter-api/lists/pinned-lists/api-reference/get-users-id-pinned_lists)

    Args:
        user (int): ID of the user.
        expansions enum (owner_id): enable you to request additional data objects that relate to the originally returned List.
        list_fields enum (created_at, follower_count, member_count, private, description, owner_id): This fields parameter enables you to select which specific List fields will deliver with each returned List objects.
        user_fields enum (created_at, description, entities, id, location, name, pinned_tweet_id, profile_image_url, protected, public_metrics, url, username, verified, withheld):
            This fields parameter enables you to select which specific user fields will deliver with the users object. Specify the desired fields in a comma-separated list without spaces between commas and fields.
        max_results (int): The maximum number of results to be returned per page. This can be a number between 1 and 100.
        pagination_token (string): Used to request the next page of results if all results weren't returned with the latest request, or to go back to the previous page of results.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user)
    url = f"https://api.twitter.com/2/users/{user_id}/pinned_lists"

    return self._lists(
        url=url,
        expansions=expansions,
        list_fields=list_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

post(url, json_data)

Make a POST request to the specified URL.

Parameters:

Name	Type	Description	Default
url	str	URL to make a POST request	required
json_data	dict	JSON data to send.	required

Returns:

Type	Description
	requests.Response: Response from Twitter API.

Source code in twarc/client2.py

@catch_request_exceptions
@rate_limit
def post(self, url, json_data):
    """
    Make a POST request to the specified URL.

    Args:
        url (str): URL to make a POST request
        json_data (dict): JSON data to send.

    Returns:
        requests.Response: Response from Twitter API.
    """
    if not self.client:
        self.connect()
    return self.client.post(url, json=json_data)

quotes(tweet_id, expansions=None, tweet_fields=None, user_fields=None, max_results=100, pagination_token=None)

Retrieve the tweets that quote tweet the given tweet.

Source code in twarc/client2.py

def quotes(
    self,
    tweet_id,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    max_results=100,
    pagination_token=None,
):
    """
    Retrieve the tweets that quote tweet the given tweet.

    """
    url = f"https://api.twitter.com/2/tweets/{tweet_id}/quote_tweets"

    params = self._prepare_params(
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

    for page in self.get_paginated(url, params=params):
        if "data" in page:
            yield page
        else:
            log.info(f"Retrieved an empty page of results for quotes of {tweet_id}")

retweeted_by(tweet_id, expansions=None, tweet_fields=None, user_fields=None, max_results=100, pagination_token=None)

Retrieve the user profiles of accounts that have retweeted the given tweet.

Source code in twarc/client2.py

def retweeted_by(
    self,
    tweet_id,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    max_results=100,
    pagination_token=None,
):
    """
    Retrieve the user profiles of accounts that have retweeted the given tweet.

    """
    url = f"https://api.twitter.com/2/tweets/{tweet_id}/retweeted_by"

    params = self._prepare_params(
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        max_results=max_results,
        pagination_token=pagination_token,
    )

    if expansions:
        params["expansions"] = "pinned_tweet_id"

    for page in self.get_paginated(url, params=params):
        if "data" in page:
            yield page
        else:
            log.info(
                f"Retrieved an empty page of results for retweeted_by of {tweet_id}"
            )

sample(event=None, record_keepalive=False, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, backfill_minutes=None)

Returns a sample of all publicly posted tweets.

The sample is based on slices of each second, not truly randomised. The same tweets are returned for all users of this endpoint.

If a threading.Event is provided and the event is set, the sample will be interrupted. This can be used for coordination with other programs.

Calls GET /2/tweets/sample/stream

Parameters:

Name	Type	Description	Default
event	threading.Event	Manages a flag to stop the process.	None
record_keepalive	bool	whether to output keep-alive events.	False

Returns:

Type	Description
	generator[dict]: a generator, dict for each tweet.

Source code in twarc/client2.py

@catch_request_exceptions
@requires_app_auth
def sample(
    self,
    event=None,
    record_keepalive=False,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    backfill_minutes=None,
):
    """
    Returns a sample of all publicly posted tweets.

    The sample is based on slices of each second, not truly randomised. The
    same tweets are returned for all users of this endpoint.

    If a `threading.Event` is provided and the event is set, the
    sample will be interrupted. This can be used for coordination with other
    programs.

    Calls [GET /2/tweets/sample/stream](https://developer.twitter.com/en/docs/twitter-api/tweets/sampled-stream/api-reference/get-tweets-sample-stream)

    Args:
        event (threading.Event): Manages a flag to stop the process.
        record_keepalive (bool): whether to output keep-alive events.

    Returns:
        generator[dict]: a generator, dict for each tweet.
    """
    url = "https://api.twitter.com/2/tweets/sample/stream"
    params = self._prepare_params(
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        media_fields=media_fields,
        poll_fields=poll_fields,
        place_fields=place_fields,
        backfill_minutes=backfill_minutes,
    )
    yield from self._stream(url, params, event, record_keepalive)

search_all(query, since_id=None, until_id=None, start_time=None, end_time=None, max_results=100, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, next_token=None, sort_order=None)

Search Twitter for the given query in the full archive, using the /search/all endpoint (Requires Academic Access).

Calls GET /2/tweets/search/all

Parameters:

Name	Type	Description	Default
query	str	The query string to be passed directly to the Twitter API.	required
since_id	int	Return all tweets since this tweet_id.	None
until_id	int	Return all tweets up to this tweet_id.	None
start_time	datetime	Return all tweets after this time (UTC datetime). If none of start_time, since_id, or until_id are specified, this defaults to 2006-3-21 to search the entire history of Twitter.	None
end_time	datetime	Return all tweets before this time (UTC datetime).	None
max_results	int	The maximum number of results per request. Max is 500.	100
sort_order	str	Order tweets based on relevancy or recency.	None

Returns:

Type	Description
	generator[dict]: a generator, dict for each paginated response.

Source code in twarc/client2.py

@requires_app_auth
def search_all(
    self,
    query,
    since_id=None,
    until_id=None,
    start_time=None,
    end_time=None,
    max_results=100,  # temp fix for #504
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    next_token=None,
    sort_order=None,
):
    """
    Search Twitter for the given query in the full archive,
    using the `/search/all` endpoint (Requires Academic Access).

    Calls [GET /2/tweets/search/all](https://developer.twitter.com/en/docs/twitter-api/tweets/search/api-reference/get-tweets-search-all)

    Args:
        query (str):
            The query string to be passed directly to the Twitter API.
        since_id (int):
            Return all tweets since this tweet_id.
        until_id (int):
            Return all tweets up to this tweet_id.
        start_time (datetime):
            Return all tweets after this time (UTC datetime). If none of start_time, since_id, or until_id
            are specified, this defaults to 2006-3-21 to search the entire history of Twitter.
        end_time (datetime):
            Return all tweets before this time (UTC datetime).
        max_results (int):
            The maximum number of results per request. Max is 500.
        sort_order (str):
            Order tweets based on relevancy or recency.

    Returns:
        generator[dict]: a generator, dict for each paginated response.
    """

    # start time defaults to the beginning of Twitter to override the
    # default of the last month. Only do this if start_time is not already
    # specified and since_id and until_id aren't being used
    if start_time is None and since_id is None and until_id is None:
        start_time = datetime.datetime(2006, 3, 21, tzinfo=datetime.timezone.utc)

    return self._search(
        url="https://api.twitter.com/2/tweets/search/all",
        query=query,
        since_id=since_id,
        until_id=until_id,
        start_time=start_time,
        end_time=end_time,
        max_results=max_results,
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        media_fields=media_fields,
        poll_fields=poll_fields,
        place_fields=place_fields,
        next_token=next_token,
        sleep_between=1.05,
        sort_order=sort_order,
    )

search_recent(query, since_id=None, until_id=None, start_time=None, end_time=None, max_results=100, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, next_token=None, sort_order=None)

Search Twitter for the given query in the last seven days, using the /search/recent endpoint.

Calls GET /2/tweets/search/recent

Parameters:

Name	Type	Description	Default
query	str	The query string to be passed directly to the Twitter API.	required
since_id	int	Return all tweets since this tweet_id.	None
until_id	int	Return all tweets up to this tweet_id.	None
start_time	datetime	Return all tweets after this time (UTC datetime).	None
end_time	datetime	Return all tweets before this time (UTC datetime).	None
max_results	int	The maximum number of results per request. Max is 100.	100
sort_order	str	Order tweets based on relevancy or recency.	None

Returns:

Type	Description
	generator[dict]: a generator, dict for each paginated response.

Source code in twarc/client2.py

def search_recent(
    self,
    query,
    since_id=None,
    until_id=None,
    start_time=None,
    end_time=None,
    max_results=100,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    next_token=None,
    sort_order=None,
):
    """
    Search Twitter for the given query in the last seven days,
    using the `/search/recent` endpoint.

    Calls [GET /2/tweets/search/recent](https://developer.twitter.com/en/docs/twitter-api/tweets/search/api-reference/get-tweets-search-recent)

    Args:
        query (str):
            The query string to be passed directly to the Twitter API.
        since_id (int):
            Return all tweets since this tweet_id.
        until_id (int):
            Return all tweets up to this tweet_id.
        start_time (datetime):
            Return all tweets after this time (UTC datetime).
        end_time (datetime):
            Return all tweets before this time (UTC datetime).
        max_results (int):
            The maximum number of results per request. Max is 100.
        sort_order (str):
            Order tweets based on relevancy or recency.

    Returns:
        generator[dict]: a generator, dict for each paginated response.
    """
    return self._search(
        url="https://api.twitter.com/2/tweets/search/recent",
        query=query,
        since_id=since_id,
        until_id=until_id,
        start_time=start_time,
        end_time=end_time,
        max_results=max_results,
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        media_fields=media_fields,
        poll_fields=poll_fields,
        place_fields=place_fields,
        next_token=next_token,
        sort_order=sort_order,
    )

stream(event=None, record_keepalive=False, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, backfill_minutes=None)

Returns a stream of tweets matching the defined rules.

Rules can be added or removed out-of-band, without disconnecting. Tweet results will contain metadata about the rule that matched it.

If event is set with a threading.Event object, the sample stream will be interrupted. This can be used for coordination with other programs.

Calls GET /2/tweets/search/stream

Parameters:

Name	Type	Description	Default
event	threading.Event	Manages a flag to stop the process.	None
record_keepalive	bool	whether to output keep-alive events.	False

Returns:

Type	Description
	generator[dict]: a generator, dict for each tweet.

Source code in twarc/client2.py

@requires_app_auth
def stream(
    self,
    event=None,
    record_keepalive=False,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    backfill_minutes=None,
):
    """
    Returns a stream of tweets matching the defined rules.

    Rules can be added or removed out-of-band, without disconnecting.
    Tweet results will contain metadata about the rule that matched it.

    If event is set with a threading.Event object, the sample stream
    will be interrupted. This can be used for coordination with other
    programs.

    Calls [GET /2/tweets/search/stream](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream)

    Args:
        event (threading.Event): Manages a flag to stop the process.
        record_keepalive (bool): whether to output keep-alive events.

    Returns:
        generator[dict]: a generator, dict for each tweet.
    """
    url = "https://api.twitter.com/2/tweets/search/stream"
    params = self._prepare_params(
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        media_fields=media_fields,
        poll_fields=poll_fields,
        place_fields=place_fields,
        backfill_minutes=backfill_minutes,
    )
    yield from self._stream(url, params, event, record_keepalive)

timeline(user, since_id=None, until_id=None, start_time=None, end_time=None, exclude_retweets=False, exclude_replies=False, max_results=100, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None, pagination_token=None)

Retrieve up to the 3200 most recent tweets made by the given user.

Calls GET /2/users/:id/tweets

Parameters:

Name	Type	Description	Default
user	int	ID of the user.	required
since_id	int	results with a Tweet ID greater than (newer) than specified	None
until_id	int	results with a Tweet ID less than (older) than specified	None
start_time	datetime	oldest UTC timestamp from which the Tweets will be provided	None
end_time	datetime	newest UTC timestamp from which the Tweets will be provided	None
exclude_retweets	boolean	remove retweets from timeline results	False
exclude_replies	boolean	remove replies from timeline results	False
max_results	int	the maximum number of Tweets to retrieve. Between 5 and 100.	100

Returns:

Type	Description
	generator[dict]: A generator, dict for each page of results.

Source code in twarc/client2.py

def timeline(
    self,
    user,
    since_id=None,
    until_id=None,
    start_time=None,
    end_time=None,
    exclude_retweets=False,
    exclude_replies=False,
    max_results=100,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
    pagination_token=None,
):
    """
    Retrieve up to the 3200 most recent tweets made by the given user.

    Calls [GET /2/users/:id/tweets](https://developer.twitter.com/en/docs/twitter-api/tweets/timelines/api-reference/get-users-id-tweets)

    Args:
        user (int): ID of the user.
        since_id (int): results with a Tweet ID greater than (newer) than specified
        until_id (int): results with a Tweet ID less than (older) than specified
        start_time (datetime): oldest UTC timestamp from which the Tweets will be provided
        end_time (datetime): newest UTC timestamp from which the Tweets will be provided
        exclude_retweets (boolean): remove retweets from timeline results
        exclude_replies (boolean): remove replies from timeline results
        max_results (int): the maximum number of Tweets to retrieve. Between 5 and 100.

    Returns:
        generator[dict]: A generator, dict for each page of results.
    """
    user_id = self._ensure_user_id(user)
    return self._timeline(
        user_id=user_id,
        timeline_type="tweets",
        since_id=since_id,
        until_id=until_id,
        start_time=start_time,
        end_time=end_time,
        exclude_retweets=exclude_retweets,
        exclude_replies=exclude_replies,
        max_results=max_results,
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        media_fields=media_fields,
        poll_fields=poll_fields,
        place_fields=place_fields,
        pagination_token=pagination_token,
    )

tweet_lookup(tweet_ids, expansions=None, tweet_fields=None, user_fields=None, media_fields=None, poll_fields=None, place_fields=None)

Lookup tweets, taking an iterator of IDs and returning pages of fully expanded tweet objects.

This can be used to rehydrate a collection shared as only tweet IDs. Yields one page of tweets at a time, in blocks of up to 100.

Calls GET /2/tweets

Parameters:

Name	Type	Description	Default
tweet_ids	iterable	A list of tweet IDs	required

Returns:

Type	Description
	generator[dict]: a generator, dict for each batch of 100 tweets.

Source code in twarc/client2.py

def tweet_lookup(
    self,
    tweet_ids,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
    media_fields=None,
    poll_fields=None,
    place_fields=None,
):
    """
    Lookup tweets, taking an iterator of IDs and returning pages of fully
    expanded tweet objects.

    This can be used to rehydrate a collection shared as only tweet IDs.
    Yields one page of tweets at a time, in blocks of up to 100.

    Calls [GET /2/tweets](https://developer.twitter.com/en/docs/twitter-api/tweets/lookup/api-reference/get-tweets)

    Args:
        tweet_ids (iterable): A list of tweet IDs

    Returns:
        generator[dict]: a generator, dict for each batch of 100 tweets.
    """

    def lookup_batch(tweet_id):
        url = "https://api.twitter.com/2/tweets"

        params = self._prepare_params(
            expansions=expansions,
            tweet_fields=tweet_fields,
            user_fields=user_fields,
            media_fields=media_fields,
            poll_fields=poll_fields,
            place_fields=place_fields,
        )
        params["ids"] = ",".join(tweet_id)

        resp = self.get(url, params=params)
        data = resp.json()

        if self.metadata:
            data = _append_metadata(data, resp.url)

        return data

    tweet_id_batch = []

    for tweet_id in tweet_ids:
        tweet_id_batch.append(str(int(tweet_id)))

        if len(tweet_id_batch) == 100:
            yield lookup_batch(tweet_id_batch)
            tweet_id_batch = []

    if tweet_id_batch:
        yield (lookup_batch(tweet_id_batch))

user_lookup(users, usernames=False, expansions=None, tweet_fields=None, user_fields=None)

Returns fully populated user profiles for the given iterator of user_id or usernames. By default user_lookup expects user ids but if you want to pass in usernames set usernames = True.

Yields one page of results at a time (in blocks of at most 100 user profiles).

Calls GET /2/users

Parameters:

Name	Type	Description	Default
users	iterable	User IDs or usernames to lookup.	required
usernames	bool	Parse users as usernames, not IDs.	False

Returns:

Type	Description
	generator[dict]: a generator, dict for each batch of 100 users.

Source code in twarc/client2.py

def user_lookup(
    self,
    users,
    usernames=False,
    expansions=None,
    tweet_fields=None,
    user_fields=None,
):
    """
    Returns fully populated user profiles for the given iterator of
    user_id or usernames. By default user_lookup expects user ids but if
    you want to pass in usernames set usernames = True.

    Yields one page of results at a time (in blocks of at most 100 user
    profiles).

    Calls [GET /2/users](https://developer.twitter.com/en/docs/twitter-api/users/lookup/api-reference/get-users)

    Args:
        users (iterable): User IDs or usernames to lookup.
        usernames (bool): Parse `users` as usernames, not IDs.

    Returns:
        generator[dict]: a generator, dict for each batch of 100 users.
    """

    if isinstance(users, str):
        raise TypeError("users must be an iterable other than a string")

    if usernames:
        url = "https://api.twitter.com/2/users/by"
    else:
        url = "https://api.twitter.com/2/users"

    def lookup_batch(users):
        params = self._prepare_params(
            tweet_fields=tweet_fields,
            user_fields=user_fields,
        )
        if expansions:
            params["expansions"] = "pinned_tweet_id"
        if usernames:
            params["usernames"] = ",".join(users)
        else:
            params["ids"] = ",".join(users)

        resp = self.get(url, params=params)
        data = resp.json()

        if self.metadata:
            data = _append_metadata(data, resp.url)

        return data

    batch = []
    for item in users:
        batch.append(str(item).strip())
        if len(batch) == 100:
            yield lookup_batch(batch)
            batch = []

    if batch:
        yield (lookup_batch(batch))

handler: python