fix: Fix LinkedIn get_user() and create_post() method. (#397)

* add a retry to get_user() method to fix bug.

* craft the link to newly created post

Author: tybalex

linkedin/main.py

@@ -1,9 +1,6 @@
 from linkedin_api.clients.restli.client import RestliClient
-from tools import (
-    users,
-    posts,
-)  # pylint: disable=unused-import. These are used in the tool registry.
-from tools.helper import tool_registry
+from tools.users import get_user
+from tools.posts import create_post
 import json
 import sys
 import asyncio
@@ -17,7 +14,14 @@ async def main():
     command = sys.argv[1]
     client = RestliClient()
 
-    response = await tool_registry.get(command)(client)
+    match command:
+        case "GetCurrentUser":
+            response = get_user(client)
+        case "CreatePost":
+            response = await create_post(client)
+        case _:
+            print(f"Unknown command: {command}")
+            sys.exit(1)
     print(json.dumps(response, indent=4))
 
 

linkedin/tools/helper.py

@@ -11,50 +11,6 @@ def str_to_bool(value):
     return str(value).lower() in ("true", "1", "yes")
 
 
-class ToolRegistry:
-    def __init__(self):
-        self._tools = {}
-
-    def _register(self, name, func):
-        """
-        Registers a tool by the given 'name'.
-        Raises a ValueError if a tool with the same name is already registered.
-        """
-        if name in self._tools:
-            raise ValueError(f"Tool '{name}' is already registered.")
-        self._tools[name] = func
-
-    def get(self, name):
-        """
-        Retrieves a registered tool by name.
-        Raises a ValueError if the tool is not found.
-        """
-        if name not in self._tools:
-            raise ValueError(f"Tool '{name}' not found.")
-        return self._tools[name]
-
-    def list_tools(self):
-        """
-        Returns a list of all registered tool names.
-        """
-        return list(self._tools.keys())
-
-    def register_tool(self, name):
-        """
-        A decorator that automatically registers the decorated function
-        under the specified 'name' in the ToolRegistry.
-        """
-
-        def wrapper(func):
-            self._register(name, func)
-            return func
-
-        return wrapper
-
-
-tool_registry = ToolRegistry()
-
-
 def _prepend_base_path(base_path: str, file_path: str):
     """
     Prepend a base path to a file path if it's not already rooted in the base path.

linkedin/tools/posts.py

@@ -1,12 +1,11 @@
-from tools.helper import tool_registry, ACCESS_TOKEN, load_from_gptscript_workspace
+from tools.helper import ACCESS_TOKEN, load_from_gptscript_workspace
 from linkedin_api.clients.restli.client import RestliClient
 from tools.users import get_user
 import os
 import requests
 
 
-@tool_registry.register_tool("CreatePost")
-async def create_post(client: RestliClient):
+async def create_post(client: RestliClient) -> dict:
     r = get_user(client)
     user_id = r["sub"]
     content = os.getenv("CONTENT")
@@ -97,10 +96,14 @@ async def create_post(client: RestliClient):
         raise ValueError(
             f"Error: failed to create post. Status code: {response.status_code}. Response: {response.entity}"
         )
+
+    response.entity["link"] = (
+        f"https://www.linkedin.com/feed/update/{response.entity['id']}"  # craft the link manually
+    )
     return response.entity
 
 
-def _register_upload(client: RestliClient, category: str):
+def _register_upload(client: RestliClient, category: str) -> tuple[str, dict]:
     url = "https://api.linkedin.com/v2/assets?action=registerUpload"
 
     headers = {
@@ -145,7 +148,7 @@ def _register_upload(client: RestliClient, category: str):
         )
 
 
-async def _upload_registered_file(file_path, upload_url):
+async def _upload_registered_file(file_path, upload_url) -> int:
     headers = {
         "Authorization": f"Bearer {ACCESS_TOKEN}",
         "Content-Type": "application/octet-stream",

linkedin/tools/users.py

@@ -1,8 +1,30 @@
-from tools.helper import tool_registry, ACCESS_TOKEN
+from tools.helper import ACCESS_TOKEN
 from linkedin_api.clients.restli.client import RestliClient
+import time
+import sys
 
 
-@tool_registry.register_tool("GetCurrentUser")
-def get_user(client: RestliClient):
-    response = client.get(resource_path="/userinfo", access_token=ACCESS_TOKEN)
+def get_user(client: RestliClient) -> dict:
+
+    MAX_RETRIES = 3
+    DELAY = 1  # seconds between retries
+
+    for attempt in range(1, MAX_RETRIES + 1):
+        try:
+            response = client.get(resource_path="/userinfo", access_token=ACCESS_TOKEN)
+            if response.status_code >= 200 and response.status_code < 300:
+                break
+        except Exception as e:
+            print(f"Attempt {attempt}: Failed with error: {e}", file=sys.stderr)
+            print(
+                f"Response: {response.status_code} - {response.entity}", file=sys.stderr
+            )
+
+        if attempt < MAX_RETRIES:
+            time.sleep(DELAY)  # Wait before retrying
+    else:
+        raise Exception(
+            f"Failed to get user info, Error: {response.status_code} - {response.entity}"
+        )
+
     return response.entity