Pydantic types for API responses (#59)

Author: Danie1

README.md

@@ -19,15 +19,17 @@ Inspired by [NPM Threads-API](https://github.com/junhoyeo/threads-api)
 
 Threads API is an unofficial Python client for Meta's Threads API. It allows you to interact with the API to login, read and publish posts, view who liked a post, retrieve user profile information, follow/unfollow and much more.
 
-It allows you to configure the session object. Choose between:
-* `aiohttp` - Python library to ease asynchronous execution of the API, for ⚡ super-fast ⚡ results. (*default*)
-* `requests` - Python library for standard ease of use (supports `HTTP_PROXY` env var functionality)
-* `instagrapi` - utilize the same connection all the way for private api
-* (Advanced) Implement your own and call ThreadsAPI like this: `ThreadsAPI(http_session_class=YourOwnHTTPSessionImpl)`
+✅ Configurable underlying HTTP Client (`aiohttp` / `requests` / `instagrapi`'s client / implement your own)
+
+✅ Authentication Methods supported via `instagrapi`
+
+✅ Stores token and settings locally, to reduce login-attempts (*uses the same token for all authenticated requests for up to 24 hrs*)
+
+✅ Pydantic structures for API responses (for IDE auto-completion) (at [types.py](https://github.com/Danie1/threads-api/blob/main/threads_api/src/types.py))
+
+✅ Actively Maintained since Threads.net Release (responsive in Github Issues, try it out!)
 
-> **Note** Since v1.1.10 you can use `requests` or `instagrapi` as HTTP clients, not just `aiohttp`.
 
-> **Note** Since v1.1.12 a `.session.json` file will be created by-default to save default settings (to reduce risk of being flagged). You can disable it by passing `ThreadsAPI(settings_path=None)`
 
 > **Important Tip** Use the same `cached_token_path` for connections, to reduce the number of actual login attempts. When needed, threads-api will reconnect and update the file in `cached_token_path`.  
 
@@ -36,12 +38,12 @@ Table of content:
 * [Demo](#demo)
 * [Getting started](#getting-started)
   * [Installation](#installation)
-  * [Set Log Level](#set-desired-log-level)
+  * [Set Log Level & Troubleshooting](#set-desired-log-level)
   * [Set HTTP Client](#choose-a-different-http-client)
-* [Contributions](#contributing-to-danie1threads-api)
 * [Supported Features](#supported-features)
 * [Usage Examples](#usage-examples)
 * [Roadmap](#📌-roadmap)
+* [Contributions](#contributing-to-danie1threads-api)
 * [License](#license)
 
 # Demo
@@ -119,42 +121,45 @@ export LOG_LEVEL=INFO
 export LOG_LEVEL=DEBUG
 ```
 
-# Contributing to Danie1/threads-api
-## Getting Started
+<details>
+  <summary>Example of Request when LOG_LEVEL=DEBUG</summary>
 
-With Poetry (*Recommended*)
 ``` bash
-# Step 1: Clone the project
-git clone git@github.com:Danie1/threads-api.git
+<---- START ---->
+Keyword arguments:
+  [title]: ["PUBLIC REQUEST"]
+  [type]: ["GET"]
+  [url]: ["https://www.instagram.com/instagram"]
+  [headers]: [{
+    "Authority": "www.threads.net",
+    "Accept": "*/*",
+    "Accept-Language": "en-US,en;q=0.9",
+    "Cache-Control": "no-cache",
+    "Content-Type": "application/x-www-form-urlencoded",
+    "Origin": "https://www.threads.net",
+    "Pragma": "no-cache",
+    "Sec-Fetch-Dest": "document",
+    "Sec-Fetch-Mode": "navigate",
+    "Sec-Fetch-Site": "cross-site",
+    "Sec-Fetch-User": "?1",
+    "Upgrade-Insecure-Requests": "1",
+    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/11.1.2 Safari/605.1.15",
+    "X-ASBD-ID": "129477",
+    "X-IG-App-ID": "238260118697367"
+}]
+<---- END ---->
 
-# Step 2: Install dependencies to virtual environment
-poetry install
-
-# Step 3: Activate virtual environment
-poetry shell
 ```
-or
-
-Without Poetry
 
-``` bash
-# Step 1: Clone the project
-git clone git@github.com:Danie1/threads-api.git
-
-# Step 2: Create virtual environment
-python3 -m venv env
+</details>
 
-# Step 3 (Unix/MacOS): Activate virtual environment
-source env/bin/activate # Unix/MacOS
+### Troubleshooting threads-api
+Upon unexpected error, or upon receiving an exception with:  `Oops, this is an error that hasn't yet been properly handled.\nPlease open an issue on Github at https://github.com/Danie1/threads-api.`
 
-# Step 3 (Windows): Activate virtual environment
-.\env\Scripts\activate # Windows
-
-# Step 4: Install dependencies
-pip install -r requirements.txt
-```
+Please open a Github Issue with all of the information you can provide, which includes the last Request and Response (Set `LOG_LEVEL=DEBUG`)
 
 # Supported Features
+- [x] ✅ Pydantic typing API responses at [types.py](https://github.com/Danie1/threads-api/blob/main/threads_api/src/types.py)
 - [x] ✅ Login functionality, including 2FA 🔒
   - [x] ✅ Cache login token securely (reduce login requests / due to restrictive limits)
   - [x] ✅ Saves settings locally, such as device information and timezone to use along your sessions
@@ -200,25 +205,63 @@ pip install -r requirements.txt
   - [x] ✅  Instagrapi
 
 ## Usage Examples
-View [examples/public_api_examples.py](https://github.com/Danie1/threads-api/blob/main/examples/public_api_examples.py) for Public API code examples. For the Private API usage (requires login), head over to [examples/private_api_examples.py](https://github.com/Danie1/threads-api/blob/main/examples/private_api_examples.py)
-
-At the end of the file you will be able to uncomment and run the individual examples with ease.
+View [examples/public_api_examples.py](https://github.com/Danie1/threads-api/blob/main/examples/public_api_examples.py) for Public API code examples. 
 
-Then simply run as:
-```
+Run as:
+``` bash
 python3 examples/public_api_examples.py
+```
 
-# or
+View [examples/private_api_examples.py](https://github.com/Danie1/threads-api/blob/main/examples/private_api_examples.py) for Private API code examples.  (🔒 Requires Authentication 🔒)
 
-# Pass the credentials as environment variables
+Run as:
+```
 USERNAME=<Instagram Username> PASSWORD=<Instagram Password> python3 examples/private_api_examples.py
 ```
 
+> **Note:**
+> At the end of the file you will be able to uncomment and run the individual examples with ease.
+
 ## 📌 Roadmap
-- [ ] 🚧 Post text and share a video
+- [ ] 🚧 Share a video
 - [ ] 🚧 Documentation Improvements
-- [ ] 🚧 CI/CD Improvements
-  - [ ] 🚧 Add coverage Pytest + Coverage Widget to README
+- [ ] 🚧 Add coverage Pytest + Coverage Widget to README
+
+
+# Contributing to Danie1/threads-api
+## Getting Started
+
+With Poetry (*Recommended*)
+``` bash
+# Step 1: Clone the project
+git clone git@github.com:Danie1/threads-api.git
+
+# Step 2: Install dependencies to virtual environment
+poetry install
+
+# Step 3: Activate virtual environment
+poetry shell
+```
+or
+
+Without Poetry
+
+``` bash
+# Step 1: Clone the project
+git clone git@github.com:Danie1/threads-api.git
+
+# Step 2: Create virtual environment
+python3 -m venv env
+
+# Step 3 (Unix/MacOS): Activate virtual environment
+source env/bin/activate # Unix/MacOS
+
+# Step 3 (Windows): Activate virtual environment
+.\env\Scripts\activate # Windows
+
+# Step 4: Install dependencies
+pip install -r requirements.txt
+```
 
 # License
 This project is licensed under the MIT license.

examples/private_api_examples.py

@@ -1,13 +1,10 @@
 from threads_api.src.threads_api import ThreadsAPI
 import asyncio
 import os
-import logging
 from threads_api.src.http_sessions.instagrapi_session import InstagrapiSession
 from threads_api.src.http_sessions.requests_session import RequestsSession
 from threads_api.src.http_sessions.aiohttp_session import AioHTTPSession
 
-from threads_api.src.anotherlogger import format_log
-
 # Code example of logging in while storing token encrypted on the file-system
 async def login_with_cache():
 
@@ -33,9 +30,9 @@ async def login_with_cache():
 # Asynchronously posts a message
 async def post(api):
     result = await api.post("Hello World!")
-
-    if result:
-        print("Post has been successfully posted")
+    
+    if result.media.pk:
+        print(f"Post has been successfully posted with id: [{result.media.pk}]")
     else:
         print("Unable to post.")
 
@@ -46,8 +43,8 @@ async def post_and_quote(api):
 
     result = await api.post("What do you think of this?", quoted_post_id=post_id)
 
-    if result:
-        print("Post has been successfully posted")
+    if result.media.pk:
+        print(f"Post has been successfully posted with id: [{result.media.pk}]")
     else:
         print("Unable to post.")
 
@@ -56,17 +53,17 @@ async def post_and_quote(api):
 async def post_include_image(api):
     result = await api.post("Hello World with an image!", image_path=".github/logo.jpg")
 
-    if result:
-        print("Post has been successfully posted")
+    if result.media.pk:
+        print(f"Post has been successfully posted with id: [{result.media.pk}]")
     else:
         print("Unable to post.")
 
 # Asynchronously posts a message with an image
 async def post_include_image_from_url(api):
     result = await api.post("Hello World with an image!", image_path="https://upload.wikimedia.org/wikipedia/commons/thumb/4/47/PNG_transparency_demonstration_1.png/280px-PNG_transparency_demonstration_1.png")
 
-    if result:
-        print("Post has been successfully posted")
+    if result.media.pk:
+        print(f"Post has been successfully posted with id: [{result.media.pk}]")
     else:
         print("Unable to post.")
 
@@ -76,8 +73,8 @@ async def post_include_multiple_images(api):
                                                                       "https://upload.wikimedia.org/wikipedia/commons/b/b5/Baby.tux.sit-black-800x800.png", 
                                                                       "https://upload.wikimedia.org/wikipedia/commons/thumb/4/47/PNG_transparency_demonstration_1.png/280px-PNG_transparency_demonstration_1.png"])
 
-    if result:
-        print("Post has been successfully posted")
+    if result.media.pk:
+        print(f"Post has been successfully posted with id: [{result.media.pk}]")
     else:
         print("Unable to post.")
 
@@ -87,8 +84,8 @@ async def post_include_url(api):
     result = False
     result = await api.post("Hello World with a link!", url="https://threads.net")
 
-    if result:
-        print("Post has been successfully posted")
+    if result.media.pk:
+        print(f"Post has been successfully posted with id: [{result.media.pk}]")
     else:
         print("Unable to post.")
 
@@ -184,13 +181,13 @@ async def block_and_unblock_user(api):
     return
 
 async def get_timeline(api):
-    def _print_post(post):
-        caption = post['thread_items'][0]['post']['caption']
+    def _print_post(timeline_item):
+        caption = timeline_item.thread_items[0].post.caption
 
         if caption == None:
             caption = "<Unable to print non-textual posts>"
         else:
-            caption = caption['text']
+            caption = caption.text
         print(f"Post -> Caption: [{caption}]\n")
 
     async def _print_posts_in_feed(next_max_id=None, posts_to_go=0):
@@ -200,11 +197,12 @@ async def _print_posts_in_feed(next_max_id=None, posts_to_go=0):
             else:
                 resp = await api.get_timeline()
 
-            for post in resp['items'][:resp['num_results']]:
-                _print_post(post)
+            
+            for timeline_item in resp.items[:resp.num_results]:
+                _print_post(timeline_item)
 
-            posts_to_go -= resp['num_results']
-            await _print_posts_in_feed(resp['next_max_id'], posts_to_go)
+            posts_to_go -= resp.num_results
+            await _print_posts_in_feed(resp.next_max_id, posts_to_go)
 
     await _print_posts_in_feed(posts_to_go=20)
 
@@ -216,25 +214,33 @@ async def get_user_threads(api):
     user_id = await api.get_user_id_from_username(username)
 
     if user_id:
-        resp = await api.get_user_threads(user_id)
+        threads = await api.get_user_threads(user_id)
+
         print(f"The threads for user '{username}' are:")
-        for thread in resp['threads']:
-            print(f"{username}'s Post: {thread['thread_items'][0]['post']['caption']['text']} || Likes: {thread['thread_items'][0]['post']['like_count']}")
+        for thread in threads.threads:
+            for thread_item in thread.thread_items:
+                print(f"{username}'s Post: {thread_item.post.caption.text} || Likes: {thread_item.post.like_count}")
     else:
         print(f"User ID not found for username '{username}'")
 
 # Asynchronously gets the replies for a user
-async def get_user_replies(api):
+async def get_user_replies(api : ThreadsAPI):
     username = "zuck"
     user_id = await api.get_user_id_from_username(username)
 
     if user_id:
         threads = await api.get_user_replies(user_id)
         print(f"The replies for user '{username}' are:")
-        for thread in threads['threads']:
-            print(f"-\n{thread['thread_items'][0]['post']['user']['username']}'s Post: {thread['thread_items'][0]['post']['caption']['text']} || Likes: {thread['thread_items'][0]['post']['like_count']}")
+        for thread in threads.threads:
+            post = thread.thread_items[0].post
+            print(f"-\n{post.user.username}'s Post: {post.caption.text} || Likes: {post.like_count}")
+
+            if len(thread.thread_items) > 1:
+                first_reply = thread.thread_items[1].post
+                print(f"{username}'s Reply: {first_reply.caption.text} || Likes: {first_reply.like_count}\n-")
+            else:
+                print(f"-> You will need to sign up / login to see more.")
 
-            print(f"{username}'s Reply: {thread['thread_items'][1]['post']['caption']['text']} || Likes: {thread['thread_items'][1]['post']['like_count']}\n-")
     else:
         print(f"User ID not found for username '{username}'")
 
@@ -244,24 +250,26 @@ async def get_post(api : ThreadsAPI):
 
     post_id = await api.get_post_id_from_url(post_url)
 
-    thread = await api.get_post(post_id)
-    print(f"{thread['containing_thread']['thread_items'][0]['post']['user']['username']}'s post {thread['containing_thread']['thread_items'][0]['post']['caption']}:")
+    response = await api.get_post(post_id)
+
+    thread = response.containing_thread.thread_items[0].post
+    print(f"{thread.user.username}'s post {thread.caption.text}: || Likes: {thread.like_count}")
 
-    for thread in thread["reply_threads"]:
-        if 'thread_items' in thread and len(thread['thread_items']) > 1:
-            print(f"-\n{thread['thread_items'][0]['post']['user']['username']}'s Reply: {thread['thread_items'][0]['post']['caption']} || Likes: {thread['thread_items'][0]['post']['like_count']}")
+    for reply in response.reply_threads:
+        if reply.thread_items is not None and len(reply.thread_items) >= 1:
+            print(f"-\n{reply.thread_items[0].post.user.username}'s Reply: {reply.thread_items[0].post.caption.text} || Likes: {reply.thread_items[0].post.like_count}")
 
 # Asynchronously gets the likes for a post
 async def get_post_likes(api : ThreadsAPI):
     post_url = "https://www.threads.net/t/CuZsgfWLyiI"
 
     post_id = await api.get_post_id_from_url(post_url)
 
-    likes = await api.get_post_likes(post_id)
+    users_list = await api.get_post_likes(post_id)
     number_of_likes_to_display = 10
 
-    for user_info in likes['users'][:number_of_likes_to_display]:
-        print(f'Username: {user_info["username"]} || Full Name: {user_info["full_name"]}')
+    for user in users_list.users[:number_of_likes_to_display]:
+        print(f'Username: {user.username} || Full Name: {user.full_name} || Follower Count: {user.follower_count} ')
 
 # Asynchronously reposts and deletes the repost
 async def repost_and_delete(api : ThreadsAPI):