🔥 Threads API added, 1.0.5

Author: rocketapi-io

examples_threads.py

@@ -0,0 +1,14 @@
+from rocketapi import ThreadsAPI
+from rocketapi.exceptions import NotFoundException, BadResponseException
+
+api = ThreadsAPI(token="put your token here")
+
+# Get user feed by id
+user_id = 35670846775
+try:
+    user = api.get_user_feed(user_id)
+    print(user)
+except NotFoundException:
+    print(f"User {user_id} not found")
+except BadResponseException:
+    print(f"Can't get {user_id} feed from API")

rocketapi/__init__.py

@@ -1 +1,2 @@
 from .instagramapi import InstagramAPI
+from .threadsapi import ThreadsAPI

rocketapi/instagramapi.py

@@ -3,13 +3,12 @@
 
 
 class InstagramAPI(RocketAPI):
-    def __init__(self, token, threads=1, max_timeout=30):
+    def __init__(self, token, max_timeout=30):
         """
         Instagram API client.
 
         Args:
             token (str): Your RocketAPI token (https://rocketapi.io/dashboard/)
-            threads (int): Number of threads to use for requests (it's beta, use with caution, every thread charges 1 request)
             max_timeout (int): Maximum timeout for requests. Please, don't use values lower than 15 seconds, it may cause problems with API.
 
         For debugging purposes you can use the following variables:
@@ -20,7 +19,7 @@ def __init__(self, token, threads=1, max_timeout=30):
         """
         self.last_response = None
         self.counter = 0
-        super().__init__(token, threads=threads, max_timeout=max_timeout)
+        super().__init__(token, max_timeout=max_timeout)
 
     def request(self, method, data):
         response = super().request(method, data)

rocketapi/rocketapi.py

@@ -2,7 +2,7 @@
 
 
 class RocketAPI:
-    def __init__(self, token, threads=1, max_timeout=30):
+    def __init__(self, token, max_timeout=30):
         """
         RocketAPI client.
 
@@ -12,11 +12,9 @@ def __init__(self, token, threads=1, max_timeout=30):
         """
         self.base_url = "https://v1.rocketapi.io/"
         self.token = token
-        self.threads = threads
         self.max_timeout = max_timeout
 
     def request(self, method, data):
-        data["_threads"] = self.threads
         return requests.post(
             url=self.base_url + method,
             json=data,

rocketapi/threadsapi.py

@@ -0,0 +1,185 @@
+from rocketapi.exceptions import NotFoundException, BadResponseException
+from rocketapi.rocketapi import RocketAPI
+
+
+class ThreadsAPI(RocketAPI):
+    def __init__(self, token, max_timeout=30):
+        """
+        Threads API client.
+
+        Args:
+            token (str): Your RocketAPI token (https://rocketapi.io/dashboard/)
+            max_timeout (int): Maximum timeout for requests. Please, don't use values lower than 15 seconds, it may cause problems with API.
+
+        For debugging purposes you can use the following variables:
+            last_response (dict): contains the last response from the API.
+            counter (int): contains the number of requests made in the current session.
+
+        For more information, see documentation: https://docs.rocketapi.io/api/
+        """
+        self.last_response = None
+        self.counter = 0
+        super().__init__(token, max_timeout=max_timeout)
+
+    def request(self, method, data):
+        response = super().request(method, data)
+        self.last_response = response
+        self.counter += 1
+        if response["status"] == "done":
+            if (
+                response["response"]["status_code"] == 200
+                and response["response"]["content_type"] == "application/json"
+            ):
+                return response["response"]["body"]
+            elif response["response"]["status_code"] == 404:
+                raise NotFoundException("Instagram resource not found")
+            else:
+                raise BadResponseException("Bad response from Threads")
+        raise BadResponseException("Bad response from RocketAPI")
+
+    def search_users(self, query):
+        """
+        Search for a specific user in Threads
+
+        Args:
+            query (str): Username to search for
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/search_users
+        """
+        return self.request("threads/search_users", {"query": query})
+
+    def get_user_info(self, user_id):
+        """
+        Retrieve Threads user information by id.
+
+        Args:
+            user_id (int): User id
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_info
+        """
+        return self.request("threads/user/get_info", {"id": user_id})
+
+    def get_user_feed(self, user_id, max_id=None):
+        """
+        Retrieve Threads user feed by id.
+
+        Args:
+            user_id (int): User id
+            max_id (str): Use for pagination
+
+        You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response).
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_feed
+        """
+        payload = {"id": user_id}
+        if max_id is not None:
+            payload["max_id"] = max_id
+        return self.request("threads/user/get_feed", payload)
+
+    def get_user_replies(self, user_id, max_id=None):
+        """
+        Retrieve Threads user replies by id.
+
+        Args:
+            user_id (int): User id
+            max_id (str): Use for pagination
+
+        You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response).
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_replies
+        """
+        payload = {"id": user_id}
+        if max_id is not None:
+            payload["max_id"] = max_id
+        return self.request("threads/user/get_replies", payload)
+
+    def get_user_followers(self, user_id, max_id=None):
+        """
+        Retrieve Threads user followers by id.
+
+        Args:
+            user_id (int): User id
+            max_id (str): Use for pagination
+
+        You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response).
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_followers
+        """
+        payload = {"id": user_id}
+        if max_id is not None:
+            payload["max_id"] = max_id
+        return self.request("threads/user/get_followers", payload)
+
+    def search_user_followers(self, user_id, query):
+        """
+        Search Threads user followers by user id.
+
+        Args:
+            user_id (int): User id
+            query (str): Search query
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_followers
+        """
+        return self.request(
+            "threads/user/get_followers", {"id": user_id, "query": query}
+        )
+
+    def get_user_following(self, user_id, max_id=None):
+        """
+        Retrieve Threads user following by id.
+
+        Args:
+            user_id (int): User id
+            max_id (str): Use for pagination
+
+        You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response).
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_following
+        """
+        payload = {"id": user_id}
+        if max_id is not None:
+            payload["max_id"] = max_id
+        return self.request("threads/user/get_following", payload)
+
+    def search_user_following(self, user_id, query):
+        """
+        Search Threads user following by user id.
+
+        Args:
+            user_id (int): User id
+            query (str): Search query
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_following
+        """
+        return self.request(
+            "threads/user/get_following", {"id": user_id, "query": query}
+        )
+
+    def get_thread_replies(self, thread_id, max_id=None):
+        """
+        Retrieve thread replies by id.
+
+        Args:
+            thread_id (int): Thread id
+            max_id (str): Use for pagination
+
+        You can use the `max_id` parameter to paginate through the media (take from the `paging_tokens["downwards"]` field of the response).
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/thread/get_replies
+        """
+        payload = {"id": thread_id}
+        if max_id is not None:
+            payload["max_id"] = max_id
+        return self.request("threads/thread/get_replies", payload)
+
+    def get_thread_likes(self, thread_id):
+        """
+        Retrieve thread likes by id.
+
+        Args:
+            thread_id (int): Thread id
+
+        For more information, see documentation: https://docs.rocketapi.io/api/threads/thread/get_likes
+        """
+        payload = {"id": thread_id}
+        return self.request("threads/thread/get_likes", payload)

setup.py

@@ -3,12 +3,12 @@
 
 setuptools.setup(
     name="rocketapi",
-    version="1.0.4",
+    version="1.0.5",
     author="RocketAPI",
     author_email="[email protected]",
     description="RocketAPI Python SDK",
     packages=["rocketapi"],
     url="https://github.com/rocketapi-io/rocketapi-python",
-    download_url="https://github.com/rocketapi-io/rocketapi-python/archive/refs/tags/v1.0.4.tar.gz",
+    download_url="https://github.com/rocketapi-io/rocketapi-python/archive/refs/tags/v1.0.5.tar.gz",
     install_requires=["requests"],
 )