add return types, more positconnectservers

Author: edavidaja

rsconnect/api.py

@@ -259,15 +259,15 @@ def __init__(
         self.api_key = None
         self.bootstrap_jwt = None
 
-    def token_endpoint(self):
+    def token_endpoint(self) -> str:
         params = get_connection_parameters(self.snowflake_connection_name)
 
         if params is None:
             raise RSConnectException("No Snowflake connection found.")
 
         return "https://{}.snowflakecomputing.com/".format(params["account"])
 
-    def fmt_payload(self):
+    def fmt_payload(self) -> str:
         params = get_connection_parameters(self.snowflake_connection_name)
 
         if params is None:
@@ -282,7 +282,7 @@ def fmt_payload(self):
         payload = urlencode(payload)
         return payload
 
-    def exchange_token(self):
+    def exchange_token(self) -> str | bytes:
         try:
             server = HTTPServer(url=self.token_endpoint())
             payload = self.fmt_payload()
@@ -1880,7 +1880,7 @@ def verify_api_key(connect_server: RSConnectServer) -> str:
         return result["username"]
 
 
-def get_python_info(connect_server: RSConnectServer):
+def get_python_info(connect_server: PositConnectServer):
     """
     Return information about versions of Python that are installed on the indicated
     Connect server.
@@ -1894,7 +1894,7 @@ def get_python_info(connect_server: RSConnectServer):
         return result
 
 
-def get_app_info(connect_server: RSConnectServer, app_id: str):
+def get_app_info(connect_server: PositConnectServer, app_id: str):
     """
     Return information about an application that has been created in Connect.
 
@@ -1915,7 +1915,7 @@ def get_posit_app_info(server: PositServer, app_id: str):
             return response["source"]
 
 
-def get_app_config(connect_server: RSConnectServer, app_id: str):
+def get_app_config(connect_server: PositConnectServer, app_id: str):
     """
     Return the configuration information for an application that has been created
     in Connect.
@@ -1931,7 +1931,7 @@ def get_app_config(connect_server: RSConnectServer, app_id: str):
 
 
 def emit_task_log(
-    connect_server: RSConnectServer,
+    connect_server: PositConnectServer,
     app_id: str,
     task_id: str,
     log_callback: Optional[Callable[[str], None]],
@@ -1967,7 +1967,7 @@ def emit_task_log(
 
 
 def retrieve_matching_apps(
-    connect_server: RSConnectServer,
+    connect_server: PositConnectServer,
     filters: Optional[dict[str, str | int]] = None,
     limit: Optional[int] = None,
     mapping_function: Optional[Callable[[RSConnectClient, ContentItemV0], AbbreviatedAppItem | None]] = None,
@@ -2043,7 +2043,7 @@ class AbbreviatedAppItem(TypedDict):
     config_url: str
 
 
-def override_title_search(connect_server: RSConnectServer, app_id: str, app_title: str):
+def override_title_search(connect_server: PositConnectServer, app_id: str, app_title: str):
     """
     Returns a list of abbreviated app data that contains apps with a title
     that matches the given one and/or the specific app noted by its ID.
@@ -2124,7 +2124,7 @@ def find_unique_name(remote_server: TargetableServer, name: str):
     :param name: the default name for an app.
     :return: the name, potentially with a suffixed number to guarantee uniqueness.
     """
-    if isinstance(remote_server, RSConnectServer):
+    if isinstance(remote_server, PositConnectServer):
         existing_names = retrieve_matching_apps(
             remote_server,
             filters={"search": name},

rsconnect/metadata.py

@@ -28,7 +28,7 @@
 
 
 if TYPE_CHECKING:
-    from .api import RSConnectServer
+    from .api import PositConnectServer
 
 from .exception import RSConnectException
 from .log import logger
@@ -602,7 +602,7 @@ class ContentBuildStore(DataStore[Dict[str, object]]):
 
     def __init__(
         self,
-        server: RSConnectServer,
+        server: PositConnectServer,
         base_dir: str = os.getenv("CONNECT_CONTENT_BUILD_DIR", DEFAULT_BUILD_DIR),
     ):
         # This type declaration is a bit of a hack. It is needed because data model used

rsconnect/snowflake.py

@@ -1,19 +1,20 @@
+# pyright: reportMissingTypeStubs=false, reportUnusedImport=false
 import json
-from subprocess import CalledProcessError, run
-from typing import Optional
+from subprocess import CalledProcessError, CompletedProcess, run
+from typing import Any, Dict, List, Optional
 
 from .exception import RSConnectException
 from .log import logger
 
 
-def snow(*args: str):
+def snow(*args: str) -> CompletedProcess[str]:
     ensure_snow_installed()
     return run(["snow"] + list(args), capture_output=True, text=True, check=True)
 
 
 def ensure_snow_installed() -> None:
     try:
-        import snowflake.cli  # noqa
+        import snowflake.cli  # noqa: F401
 
         logger.debug("snowflake-cli is installed.")
 
@@ -27,7 +28,7 @@ def ensure_snow_installed() -> None:
             raise RSConnectException("snow cannot be found.")
 
 
-def list_connections():
+def list_connections() -> List[Dict[str, Any]]:
 
     try:
         res = snow("connection", "list", "--format", "json")
@@ -37,7 +38,7 @@ def list_connections():
         raise RSConnectException("Could not list snowflake connections.")
 
 
-def get_connection_parameters(name: Optional[str] = None):
+def get_connection_parameters(name: Optional[str] = None) -> Optional[Dict[str, Any]]:
 
     connection_list = list_connections()
     # return parameters for default connection if configured
@@ -55,7 +56,7 @@ def get_connection_parameters(name: Optional[str] = None):
         raise RSConnectException(f"No Snowflake connection found with name '{name}'.")
 
 
-def generate_jwt(name: Optional[str] = None):
+def generate_jwt(name: Optional[str] = None) -> str:
 
     _ = get_connection_parameters(name)
     connection_name = "" if name is None else name