misc: fix typing errors

Makefile

@@ -0,0 +1,6 @@
+.PHONY: test
+
+test:
+	pyre incremental
+test.watch:
+	watchman-make -p "**/*.py" -t test

instauto/api/actions/authentication.py

@@ -37,11 +37,11 @@ def log_in(self) -> None:
         }
         resp = self._request('accounts/login/', Method.POST, body=body, sign_request=True)
         try:
-            # pyre-ignore[6]
             self.state.logged_in_account_data = LoggedInAccountData(**self._json_loads(resp.text)['logged_in_user'])
         except KeyError as e:
-            # The response can be empty if challenge was needed. In that case, the logged_in_account_data
-            # attribute should've been set from within in the challenge handler.
+            # The response can be empty if challenge was needed. In that
+            # case, the logged_in_account_data attribute should've been
+            # set from within in the challenge handler.
             if self.state.logged_in_account_data is None:
                 raise e
 
@@ -104,6 +104,7 @@ def _encode_password(self, password: Optional[str] = None) -> Optional[str]:
         aes = AES.new(key, AES.MODE_GCM, nonce=iv)
         aes.update(str(time).encode('utf-8'))
 
+        # pyre-ignore[6]: we do check if either password or plain_password
         encrypted_password, cipher_tag = aes.encrypt_and_digest(bytes(password or self._plain_password, 'utf-8'))
 
         encrypted = bytes([1,

instauto/api/actions/direct.py

@@ -31,7 +31,6 @@ def direct_update_inbox(self) -> bool:
             bool: True if the inbox has been updated
         """
         resp = self._request('direct_v2/inbox', Method.GET)
-        # pyre-ignore[6]
         stat = self._set_inbox_from_json(self._json_loads(resp.text))
         return resp.ok and stat
 
@@ -45,7 +44,6 @@ def direct_get_thread(self, obj: DirectThread) -> Thread:
             Thread: The retrieved thread.
         """
         resp = self._request(f"direct_v2/threads/{obj.thread_id}", Method.GET)
-        # pyre-ignore[6]
         thread = self._json_loads(resp.text)['thread']
         thread = Thread(thread.pop('thread_id'), thread.pop('thread_v2_id'), thread.pop('users'),
                         thread.pop('left_users'), thread.pop('admin_user_ids'), thread.pop('items'),

instauto/api/actions/feed.py

@@ -18,6 +18,5 @@ def feed_get(self, obj: FeedGet) -> Tuple[Union[FeedGet], requests.Response]:
         if obj.reason == 'cold_start_fetch':
             obj.reason = 'pagination'
 
-        # pyre-ignore[6]
         obj.max_id = data['next_max_id']
         return obj, resp

instauto/api/actions/friendships.py

@@ -50,7 +50,6 @@ def follow_requests_get(self, obj: PendingRequests) -> List[dict]:
         """Retrieve all follow requests"""
         resp = self._request('friendships/pending/', Method.GET)
         parsed = self._json_loads(resp.text)
-        # pyre-ignore[6]
         return parsed['users']
 
     def follow_request_approve(self, obj: ApproveRequest) -> Response:
@@ -85,7 +84,6 @@ def _get_base(self, obj: Union[GetFollowing, GetFollowers]) -> \
         as_json = self._json_loads(resp.text)
         if 'next_max_id' not in as_json:
             return obj, False
-        # pyre-ignore[6]
         obj.max_id = as_json['next_max_id']
         # pyre-ignore[58]
         obj.page = data.get('page', 0) + 1

instauto/api/actions/helpers.py

@@ -33,7 +33,7 @@ def _build_default_rupload_params(self, obj, quality: int, is_sidecar: bool) ->
             "is_sidecar": str(int(is_sidecar))
         }
 
-    def _json_loads(self, text: Union[bytes, bytearray, memoryview, str]) -> Any:
+    def _json_loads(self, text: Union[bytes, str]) -> Any:
         return orjson.loads(text)
 
     def _json_dumps(self, obj: Any) -> str:

instauto/api/actions/post.py

@@ -91,10 +91,8 @@ def post_retrieve_by_user(self, obj: RetrieveByUser) -> Tuple[RetrieveByUser, Un
         resp = self._request(f'feed/user/{obj.user_id}/', Method.GET, query=as_dict)
         resp_as_json = self._json_loads(resp.text)
 
-        # pyre-ignore[16]
         obj.max_id = resp_as_json.get('next_max_id')
         obj.page += 1
-        # pyre-ignore[6]
         return obj, resp_as_json['items']
 
     def post_retrieve_story(self, obj: RetrieveStory) -> requests.Response:
@@ -114,24 +112,20 @@ def post_retrieve_by_tag(self, obj: RetrieveByTag) -> Tuple[RetrieveByTag, Union
         resp = self._request(f'feed/tag/{obj.tag_name}/', Method.GET, query=as_dict)
         resp_as_json = self._json_loads(resp.text)
 
-        # pyre-ignore[16]
         obj.max_id = resp_as_json.get('next_max_id')
         obj.page += 1
-        # pyre-ignore[6]
         return obj, resp_as_json['items']
 
     def post_get_likers(self, obj: RetrieveLikers) -> List[Dict]:
         """Retrieve all likers of specific media_id"""
         endpoint = 'media/{media_id}/likers'.format(media_id=obj.media_id)
         resp = self._request(endpoint=endpoint, method=Method.GET)
-        # pyre-ignore[6]
         users_as_json = self._json_loads(resp.text)['users']
         return users_as_json
 
     def post_get_commenters(self, obj: RetrieveCommenters) -> List[Dict]:
         endpoint = 'media/{media_id}/comments'.format(media_id=obj.media_id)
         resp = self._request(endpoint=endpoint, method=Method.GET)
-        # pyre-ignore[6]
         users_as_json = [c['user'] for c in self._json_loads(resp.text)['comments']]
         return users_as_json
 
@@ -164,11 +158,8 @@ def post_carousel(self, posts: List[PostFeed], caption: str, quality: int) -> Di
                 "timezone_offset": post.timezone_offset,
                 "source_type": str(post.source_type),
                 "scene_type": None,
-                # pyre-ignore[6]
                 "edits": self._json_dumps(post.to_dict()['edits']),
-                # pyre-ignore[6]
                 "extra": self._json_dumps(post.to_dict()['extra']),
-                # pyre-ignore[6]
                 "device": self._json_dumps(post.to_dict()['device'])
             })
             if hasattr(post, 'user_tags'):
@@ -213,11 +204,9 @@ def _request_fb_places_id(self, obj: Location) -> str:
             resp = self._request("location_search", Method.GET, query=query)
 
         as_json = self._json_loads(resp.text)
-        # pyre-ignore[6]
         if as_json['status'] != 'ok':
             raise BadResponse
 
-        # pyre-ignore[6]
         return str(as_json['venues'][0]['external_id'])
 
     def _upload_image(self, obj: Union[PostStory, PostFeed], quality: int, is_sidecar: bool = False) -> Tuple[Response, dict]:

instauto/api/actions/request.py

@@ -61,7 +61,6 @@ def _request(
 
     def _build_body(self, data: Optional[Union[dict, list, bytes]], sign_request):
         if sign_request:
-            # pyre-ignore[6]
             as_json = self._json_dumps(data)
             data = {
                 'ig_sig_key_version': self.ig_profile.signature_key_version,
@@ -74,8 +73,7 @@ def _check_response_for_errors(self, resp: requests.Response) -> None:
             return
 
         try:
-            # pyre-ignore[9]: assume the response is a dictionary
-            parsed: Dict[Any, Any] = self._json_loads(resp.text)
+            parsed = self._json_loads(resp.text)
         except orjson.JSONDecodeError:
             if resp.status_code == 404 and '/friendships/' in resp.url:
                 raise InvalidUserId(f"account id: {resp.url.split('/')[-2]} is not recognized "

instauto/api/actions/structs/direct.py

@@ -54,10 +54,14 @@ def __init__(self, media_id: str, recipients: Optional[List[List[str]]] = None,
 class LinkShare(_Base):
     REQUEST = 'direct/linkshare.json'
 
-    def __init__(self, text: str, links: Union[List[str], str], recipients: Optional[List[List[str]]] = None,
-                 threads: Optional[List[str]] = None, *args, **kwargs):
-        if type(links) != list:
-            # pyre-ignore[9]: we check if it's a list
+    def __init__(
+        self, 
+        text: str, 
+        links: Union[List[str], str], 
+        recipients: Optional[List[List[str]]] = None,
+        threads: Optional[List[str]] = None, *args, **kwargs
+    ):
+        if type(links) == str:
             links = [links]
         self.link_text = text
         self.link_urls = orjson.dumps(links).decode()

instauto/api/actions/structs/post.py

@@ -38,14 +38,12 @@ class UserTag:
     def to_dict(self) -> dict:
         if self.x is None or self.y is None:
             raise Exception("Invalid x, y coordinates.")
+        rounded_x = round(self.x, ndigits=8)
+        # pyre-ignore[6]
+        rounded_y = round(self.y, ndigits=8)
         data = {
             "user_id": self.user_id,
-            "position": [
-                # pyre-ignore[6]
-                round(self.x, ndigits=8),
-                # pyre-ignore[6]
-                round(self.y, ndigits=8)
-            ]
+            "position": [rounded_x, rounded_y]
         }
         return data
 

instauto/api/actions/stub.py

@@ -44,7 +44,7 @@ class StubMixin:
     _request_finished_callbacks: list
     _handle_challenge: Callable
     _2fa_function: Optional[Callable[[str], str]]
-    _handle_2fa: Optional[Callable[[dict], None]]
+    _handle_2fa: Callable[[dict], None]
     _request: _request
     _username: Optional[str]
     _plain_password: Optional[str]

instauto/api/client.py

@@ -8,7 +8,7 @@
 import base64
 import time
 
-from typing import Callable, Optional
+from typing import Callable, Optional, Union
 
 # pyre-ignore[21]
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -139,8 +139,8 @@ def to_json(self) -> str:
             })
 
     @classmethod
-    def from_json(cls, jsondata: str) -> "ApiClient":
-        data = orjson.loads(jsondata)
+    def from_json(cls, j: Union[str, bytes]) -> "ApiClient":
+        data = orjson.loads(j)
 
         state = data['State']
         state['logged_in_account_data'] = data['LoggedInAccountData']