fix a few type errors

Author: TheCommCraft

scratchattach/cloud/_base.py

@@ -7,15 +7,16 @@
 
 from websocket import WebSocketBadStatusException
 
-from ._base import BaseCloud
+from ._base import BaseCloud, LogCloud
 from scratchattach.utils.requests import requests
 from scratchattach.utils import exceptions, commons
 from scratchattach.site import cloud_activity
 
-class ScratchCloud(BaseCloud):
+
+class ScratchCloud(LogCloud):
     def __init__(self, *, project_id, _session=None):
         super().__init__()
-        
+
         self.project_id = project_id
 
         # Configure this object's attributes specifically for being used with Scratch's cloud:
@@ -28,35 +29,37 @@ def __init__(self, *, project_id, _session=None):
             self.origin = "https://scratch.mit.edu"
 
     def connect(self):
-        self._assert_auth() # Connecting to Scratch's cloud websocket requires a login to the Scratch website
+        self._assert_auth()  # Connecting to Scratch's cloud websocket requires a login to the Scratch website
         try:
             super().connect()
         except WebSocketBadStatusException as e:
             raise exceptions.CloudConnectionError("Error: Scratch's Cloud system may be down. Please try again later.") from e
 
-    def set_var(self, variable, value, *, max_retries : int = 2):
-        self._assert_auth() # Setting a cloud var requires a login to the Scratch website
+    def set_var(self, variable, value, *, max_retries: int = 2):
+        self._assert_auth()  # Setting a cloud var requires a login to the Scratch website
         super().set_var(variable, value, max_retries=max_retries)
 
-    def set_vars(self, var_value_dict, *, intelligent_waits=True, max_retries : int = 2):
-        self._assert_auth() 
+    def set_vars(self, var_value_dict, *, intelligent_waits=True, max_retries: int = 2):
+        self._assert_auth()
         super().set_vars(var_value_dict, intelligent_waits=intelligent_waits, max_retries=max_retries)
 
     def logs(self, *, filter_by_var_named=None, limit=100, offset=0) -> list[cloud_activity.CloudActivity]:
         """
         Gets the data from Scratch's clouddata logs.
-        
+
         Keyword Arguments:
             filter_by_var_named (str or None): If you only want to get data for one cloud variable, set this argument to its name.
             limit (int): Max. amount of returned activity.
             offset (int): Offset of the first activity in the returned list.
             log_url (str): If you want to get the clouddata from a cloud log API different to Scratch's normal cloud log API, set this argument to the URL of the API. Only set this argument if you know what you are doing. If you want to get the clouddata from the normal API, don't put this argument.
         """
         try:
-            data = requests.get(f"https://clouddata.scratch.mit.edu/logs?projectid={self.project_id}&limit={limit}&offset={offset}", timeout=10).json()
+            data = requests.get(
+                f"https://clouddata.scratch.mit.edu/logs?projectid={self.project_id}&limit={limit}&offset={offset}", timeout=10
+            ).json()
             if filter_by_var_named is not None:
                 filter_by_var_named = filter_by_var_named.removeprefix("☁ ")
-                data = list(filter(lambda k: k["name"] == "☁ "+filter_by_var_named, data))
+                data = list(filter(lambda k: k["name"] == "☁ " + filter_by_var_named, data))
             for x in data:
                 x["cloud"] = self
             return commons.parse_object_list(data, cloud_activity.CloudActivity, self._session, "name")
@@ -66,16 +69,16 @@ def logs(self, *, filter_by_var_named=None, limit=100, offset=0) -> list[cloud_a
     def get_var(self, var, *, recorder_initial_values: Optional[dict[str, Any]] = None, use_logs=False):
         var = var.removeprefix("☁ ")
         if self._session is None or use_logs:
-            filtered = self.logs(limit=100, filter_by_var_named="☁ "+var)
+            filtered = self.logs(limit=100, filter_by_var_named="☁ " + var)
             if len(filtered) == 0:
                 return None
             return filtered[0].value
         else:
             if self.recorder is None:
                 initial_values = self.get_all_vars(use_logs=True)
-                return super().get_var("☁ "+var, recorder_initial_values=initial_values)
+                return super().get_var("☁ " + var, recorder_initial_values=initial_values)
             else:
-                return super().get_var("☁ "+var, recorder_initial_values=recorder_initial_values)
+                return super().get_var("☁ " + var, recorder_initial_values=recorder_initial_values)
 
     def get_all_vars(self, *, recorder_initial_values: Optional[dict[str, Any]] = None, use_logs=False):
         if self._session is None or use_logs:
@@ -95,33 +98,33 @@ def get_all_vars(self, *, recorder_initial_values: Optional[dict[str, Any]] = No
     def events(self, *, use_logs=False):
         if self._session is None or use_logs:
             from scratchattach.eventhandlers.cloud_events import CloudLogEvents
+
             return CloudLogEvents(self)
         else:
             return super().events()
 
 
 class TwCloud(BaseCloud):
-    def __init__(self, *, project_id, cloud_host="wss://clouddata.turbowarp.org", purpose="", contact="",
-                 _session=None):
+    def __init__(self, *, project_id, cloud_host="wss://clouddata.turbowarp.org", purpose="", contact="", _session=None):
         super().__init__()
-        
+
         self.project_id = project_id
-        
+
         # Configure this object's attributes specifically for being used with TurboWarp's cloud:
         self.cloud_host = cloud_host
-        self.ws_shortterm_ratelimit = 0 # TurboWarp doesn't enforce a wait time between cloud variable sets
+        self.ws_shortterm_ratelimit = 0  # TurboWarp doesn't enforce a wait time between cloud variable sets
         self.ws_longterm_ratelimit = 0
-        self.length_limit = 100000 # TurboWarp doesn't enforce a cloud variable length
+        self.length_limit = 100000  # TurboWarp doesn't enforce a cloud variable length
         purpose_string = ""
         if purpose != "" or contact != "":
             purpose_string = f" (Purpose:{purpose}; Contact:{contact})"
-        self.header = {"User-Agent":f"scratchattach/2.0.0{purpose_string}"}
+        self.header = {"User-Agent": f"scratchattach/2.0.0{purpose_string}"}
 
-class CustomCloud(BaseCloud):
 
+class CustomCloud(BaseCloud):
     def __init__(self, *, project_id, cloud_host, **kwargs):
         super().__init__()
-        
+
         self.project_id = project_id
         self.cloud_host = cloud_host
 
@@ -145,36 +148,37 @@ def get_cloud(project_id, *, CloudClass: type[BaseCloud] = ScratchCloud) -> Base
 
     Args:
         project_id:
-    
+
     Keyword arguments:
         CloudClass: The class that the returned object should be of. By default this class is scratchattach.cloud.ScratchCloud.
 
     Returns:
         Type[scratchattach.cloud._base.BaseCloud]: An object representing the cloud of a project. Can be of any class inheriting from BaseCloud.
     """
     warnings.warn(
-        "To set Scratch cloud variables, use session.connect_cloud instead of get_cloud",
-        exceptions.CloudAuthenticationWarning
+        "To set Scratch cloud variables, use session.connect_cloud instead of get_cloud", exceptions.CloudAuthenticationWarning
     )
     return CloudClass(project_id=project_id)
 
+
 def get_scratch_cloud(project_id):
     """
     Warning:
         Since this method doesn't connect a login / session to the returned object, setting Scratch cloud variables won't be possible with it.
 
         To set Scratch cloud variables, use `scratchattach.Session.connect_scratch_cloud` instead.
 
-        
+
     Returns:
         scratchattach.cloud.ScratchCloud: An object representing the Scratch cloud of a project.
     """
     warnings.warn(
         "To set Scratch cloud variables, use session.connect_scratch_cloud instead of get_scratch_cloud",
-        exceptions.CloudAuthenticationWarning
+        exceptions.CloudAuthenticationWarning,
     )
     return ScratchCloud(project_id=project_id)
 
+
 def get_tw_cloud(project_id, *, purpose="", contact="", cloud_host="wss://clouddata.turbowarp.org"):
     """
     Returns:

scratchattach/cloud/cloud.py

@@ -1,5 +1,8 @@
 """CloudEvents class"""
+
 from __future__ import annotations
+from scratchattach.site.typed_dicts import CloudActivityDict
+from typing import cast
 
 import traceback
 
@@ -10,10 +13,14 @@
 import time
 from collections.abc import Iterator
 
+
 class CloudEvents(BaseEventHandler):
     """
     Class that calls events when on cloud updates that are received through a websocket connection.
     """
+
+    source_stream: _base.EventStream
+
     def __init__(self, cloud: _base.AnyCloud):
         super().__init__()
         self.cloud = cloud
@@ -29,12 +36,12 @@ def _updater(self):
         """
         A process that listens for cloud activity and executes events on cloud activity
         """
-        
+
         self.call_event("on_ready")
 
         if not self.running:
             return
-        
+
         # TODO: refactor this method. It works, but is hard to read
         while self.running:
             try:
@@ -46,64 +53,77 @@ def _updater(self):
                         # print(f"Got event {data}")
                         self.subsequent_reconnects = 0
                         try:
-                            _a = cloud_activity.CloudActivity(timestamp=time.time()*1000, _session=self._session, cloud=self.cloud)
+                            _a = cloud_activity.CloudActivity(
+                                timestamp=time.time() * 1000,
+                                _session=self._session,
+                                cloud=self.cloud,
+                            )
                             # if _a.timestamp < self.startup_time + 500: # catch the on_connect message sent by TurboWarp's (and sometimes Scratch's) cloud server
                             #     # print(f"Skipped as {_a.timestamp} < {self.startup_time + 500}")
                             #     continue
-                            data["variable_name"] = data["name"]
-                            data["name"] = data["variable_name"].replace("☁ ", "")
-                            _a._update_from_dict(data)
+                            cloud_activity_dict = cast(CloudActivityDict, data)
+                            cloud_activity_dict["variable_name"] = cloud_activity_dict["name"]
+                            cloud_activity_dict["name"] = cloud_activity_dict[
+                                "variable_name"
+                            ].replace("☁ ", "")
+                            _a._update_from_dict(cloud_activity_dict)
                             # print(f"sending event {_a}")
                             self.call_event(f"on_{_a.type}", [_a])
                         except Exception as e:
                             print(f"Cloud events _updated ignored: {e} {traceback.format_exc()}")
                             pass
             except Exception:
                 self.subsequent_reconnects += 1
-                time.sleep(0.1) # cooldown
+                time.sleep(0.1)  # cooldown
 
                 if self.subsequent_reconnects >= 5:
-                    print(f"Warning: {self.subsequent_reconnects} subsequent cloud disconnects. Cloud may be down, causing CloudEvents to not call events.")
+                    print(
+                        f"Warning: {self.subsequent_reconnects} subsequent cloud disconnects. Cloud may be down, causing CloudEvents to not call events."
+                    )
                 self.call_event("on_reconnect", [])
 
+
 class ManualCloudLogEvents:
-    """
-    Class that calls events on cloud updates that are received from a clouddata log.
-    """
+    """Class that calls events on cloud updates that are received from a clouddata log."""
+
     def __init__(self, cloud: _base.LogCloud):
         if not isinstance(cloud, _base.LogCloud):
-            raise ValueError("Cloud log events can't be used with a cloud that has no logs available")
+            raise ValueError(
+                "Cloud log events can't be used with a cloud that has no logs available"
+            )
         self.cloud = cloud
         self.source_cloud = cloud
         self._session = cloud._session
         self.last_timestamp = 0
         self.subsequent_failed_log_fetches = 0
 
     def update(self) -> Iterator[tuple[str, list[cloud_activity.CloudActivity]]]:
-        """
-        Update once and yield all packets
-        """
+        """Update once and yield all packets"""
         try:
             data = self.source_cloud.logs(limit=25)
             self.subsequent_failed_log_fetches = 0
             for _a in data[::-1]:
                 if _a.timestamp <= self.last_timestamp:
                     continue
                 self.last_timestamp = int(_a.timestamp)
-                yield ("on_"+_a.type, [_a])
+                yield ("on_" + _a.type, [_a])
         except Exception:
             self.subsequent_failed_log_fetches += 1
             if self.subsequent_failed_log_fetches == 20:
-                print("Warning: 20 subsequent clouddata log fetches failed. Scrach's cloud logs may be down, causing CloudLogEvents to not call events.")
-        
+                print(
+                    "Warning: 20 subsequent clouddata log fetches failed. Scrach's cloud logs may be down, causing CloudLogEvents to not call events."
+                )
+
+
 class CloudLogEvents(BaseEventHandler):
-    """
-    Class that calls events on cloud updates that are received from a clouddata log.
-    """
+    """Class that calls events on cloud updates that are received from a clouddata log."""
+
     def __init__(self, cloud: _base.LogCloud, *, update_interval=0.1):
         super().__init__()
         if not isinstance(cloud, _base.LogCloud):
-            raise ValueError("Cloud log events can't be used with a cloud that has no logs available")
+            raise ValueError(
+                "Cloud log events can't be used with a cloud that has no logs available"
+            )
         self.cloud = cloud
         self.source_cloud = cloud
         self.update_interval = update_interval

scratchattach/eventhandlers/cloud_events.py

@@ -137,6 +137,7 @@ class CloudRequests(CloudEvents):
     executer_thread: Optional[Thread]
     responder_thread: Optional[Thread]
     extra_executor_threads: list[Thread]
+    cloud: _base.AnyCloud
 
     def __init__(
         self,
@@ -268,19 +269,19 @@ def _set_FROM_HOST_var(self, value):
         self.current_var += 1
         if self.current_var == len(self.used_cloud_vars):
             self.current_var = 0
-        time.sleep(self.cloud.ws_shortterm_ratelimit)
+        time.sleep(getattr(self.cloud, "ws_shortterm_ratelimit", 0.1))
 
     def _respond(self, request_id, response, *, validation=2222):
         """
         Sends back the request response to the Scratch project
         """
-        if (self.cloud.last_var_set + 8 < time.time() # if the cloud connection has been idle for too long, a reconnect is necessary to make sure the first package will not be lost
+        if (getattr(self.cloud, "last_var_set", time.time()) + 8 < time.time() # if the cloud connection has been idle for too long, a reconnect is necessary to make sure the first package will not be lost
             ) or self.no_packet_loss:
             self.cloud.reconnect()
 
         memory = ResponseMemory(rid=request_id, packets={})#{"rid":request_id}
         remaining_response = str(response)
-        length_limit = self.cloud.length_limit - (len(str(request_id))+6) # the subtrahend is the worst-case length of the "."+numbers after the "."
+        length_limit = getattr(self.cloud, "length_limit", 256) - (len(str(request_id))+6) # the subtrahend is the worst-case length of the "."+numbers after the "."
 
         i = 0
         while not remaining_response == "":
@@ -457,11 +458,12 @@ def on_reconnect(self):
         Called when the underlying cloud events reconnect. Makes sure that no requests are missed in this case.
         """
         try:
-            extradata = self.cloud.logs(limit=35)[::-1] # Reverse result so oldest activity is first
-            for activity in extradata:
-                if activity.timestamp < self.startup_time:
-                    continue
-                self.on_set(activity) # Read in the fetched activity
+            if isinstance(self.cloud, _base.LogCloud):
+                extradata = self.cloud.logs(limit=35)[::-1] # Reverse result so oldest activity is first
+                for activity in extradata:
+                    if activity.timestamp < self.startup_time:
+                        continue
+                    self.on_set(activity) # Read in the fetched activity
         except Exception:
             pass
 
@@ -511,7 +513,7 @@ def send(self, data, *, priority=0):
         else:
             time.sleep(0.07)
 
-    def stop(self, wait_extra_threads: bool = True):
+    def stop(self, wait_extra_threads: bool = True):  # ty:ignore[invalid-method-override]
         """
         Stops the request handler and all associated threads forever. Lets running response sending processes finish.
         """
@@ -543,6 +545,10 @@ def hard_stop(self):
 
     def credit_check(self):
         try:
+            if not isinstance(self.cloud, _base.BaseCloud):
+                raise TypeError
+            if self.cloud.project_id is None:
+                raise ValueError
             p = project.Project(id=self.cloud.project_id)
             if not p.update(): # can't get project, probably because it's unshared (no authentication is used for getting it)
                 print("If you use cloud requests or cloud storages, please credit TimMcCool!")