typecheck yourself before you wreck yourself

Author: edavidaja

rsconnect/actions_content.py

@@ -9,11 +9,11 @@
 import traceback
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta
-from typing import Iterator, Literal, Optional, Sequence, cast
+from typing import Iterator, Literal, Optional, Sequence, cast, Union
 
 import semver
 
-from .api import PositConnectServer, RSConnectClient, emit_task_log
+from .api import RSConnectServer, SPCSConnectServer, RSConnectClient, emit_task_log
 from .exception import RSConnectException
 from .log import logger
 from .metadata import ContentBuildStore, ContentItemWithBuildState
@@ -33,7 +33,7 @@ def content_build_store() -> ContentBuildStore:
     return _content_build_store
 
 
-def ensure_content_build_store(connect_server: PositConnectServer) -> ContentBuildStore:
+def ensure_content_build_store(connect_server: Union[RSConnectServer, SPCSConnectServer]) -> ContentBuildStore:
     global _content_build_store
     if not _content_build_store:
         logger.info("Initializing ContentBuildStore for %s" % connect_server.url)
@@ -42,7 +42,7 @@ def ensure_content_build_store(connect_server: PositConnectServer) -> ContentBui
 
 
 def build_add_content(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     content_guids_with_bundle: Sequence[ContentGuidWithBundle],
 ):
     """
@@ -85,7 +85,7 @@ def _validate_build_rm_args(guid: Optional[str], all: bool, purge: bool):
 
 
 def build_remove_content(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     guid: Optional[str],
     all: bool,
     purge: bool,
@@ -109,20 +109,20 @@ def build_remove_content(
     return guids
 
 
-def build_list_content(connect_server: PositConnectServer, guid: str, status: Optional[str]):
+def build_list_content(connect_server: Union[RSConnectServer, SPCSConnectServer], guid: str, status: Optional[str]):
     build_store = ensure_content_build_store(connect_server)
     if guid:
         return [build_store.get_content_item(g) for g in guid]
     else:
         return build_store.get_content_items(status=status)
 
 
-def build_history(connect_server: PositConnectServer, guid: str):
+def build_history(connect_server: Union[RSConnectServer, SPCSConnectServer], guid: str):
     return ensure_content_build_store(connect_server).get_build_history(guid)
 
 
 def build_start(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     parallelism: int,
     aborted: bool = False,
     error: bool = False,
@@ -251,7 +251,9 @@ def build_start(
             build_monitor.shutdown()
 
 
-def _monitor_build(connect_server: PositConnectServer, content_items: list[ContentItemWithBuildState]):
+def _monitor_build(
+    connect_server: Union[RSConnectServer, SPCSConnectServer], content_items: list[ContentItemWithBuildState]
+):
     """
     :return bool: True if the build completed without errors, False otherwise
     """
@@ -296,7 +298,9 @@ def _monitor_build(connect_server: PositConnectServer, content_items: list[Conte
     return True
 
 
-def _build_content_item(connect_server: PositConnectServer, content: ContentItemWithBuildState, poll_wait: int):
+def _build_content_item(
+    connect_server: Union[RSConnectServer, SPCSConnectServer], content: ContentItemWithBuildState, poll_wait: int
+):
     build_store = ensure_content_build_store(connect_server)
     with RSConnectClient(connect_server) as client:
         # Pending futures will still try to execute when ThreadPoolExecutor.shutdown() is called
@@ -351,7 +355,7 @@ def write_log(line: str):
 
 
 def emit_build_log(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     guid: str,
     format: str,
     task_id: Optional[str] = None,
@@ -369,7 +373,7 @@ def emit_build_log(
         raise RSConnectException("Log file not found for content: %s" % guid)
 
 
-def download_bundle(connect_server: PositConnectServer, guid_with_bundle: ContentGuidWithBundle):
+def download_bundle(connect_server: Union[RSConnectServer, SPCSConnectServer], guid_with_bundle: ContentGuidWithBundle):
     """
     :param guid_with_bundle: models.ContentGuidWithBundle
     """
@@ -387,7 +391,7 @@ def download_bundle(connect_server: PositConnectServer, guid_with_bundle: Conten
         return client.download_bundle(guid_with_bundle.guid, guid_with_bundle.bundle_id)
 
 
-def get_content(connect_server: PositConnectServer, guid: str | list[str]):
+def get_content(connect_server: Union[RSConnectServer, SPCSConnectServer], guid: str | list[str]):
     """
     :param guid: a single guid as a string or list of guids.
     :return: a list of content items.
@@ -401,7 +405,7 @@ def get_content(connect_server: PositConnectServer, guid: str | list[str]):
 
 
 def search_content(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     published: bool,
     unpublished: bool,
     content_type: Sequence[str],

rsconnect/api.py

@@ -323,7 +323,6 @@ def exchange_token(self) -> str:
 
 
 TargetableServer = typing.Union[ShinyappsServer, RSConnectServer, CloudServer, SPCSConnectServer]
-PositConnectServer = typing.Union[RSConnectServer, SPCSConnectServer]
 
 
 class S3Server(AbstractRemoteServer):
@@ -340,7 +339,7 @@ class RSConnectClientDeployResult(TypedDict):
 
 
 class RSConnectClient(HTTPServer):
-    def __init__(self, server: PositConnectServer, cookies: Optional[CookieJar] = None):
+    def __init__(self, server: Union[RSConnectServer, SPCSConnectServer], cookies: Optional[CookieJar] = None):
         if cookies is None:
             cookies = server.cookie_jar
         super().__init__(
@@ -1007,7 +1006,7 @@ def deploy_bundle(self):
         if self.bundle is None:
             raise RSConnectException("A bundle must be created before deploying it.")
 
-        if isinstance(self.remote_server, PositConnectServer):
+        if isinstance(self.remote_server, (RSConnectServer, SPCSConnectServer)):
             if not isinstance(self.client, RSConnectClient):
                 raise RSConnectException("client must be an RSConnectClient.")
             result = self.client.deploy(
@@ -1086,7 +1085,7 @@ def emit_task_log(
         :param raise_on_error: whether to raise an exception when a task is failed, otherwise we
         return the task_result so we can record the exit code.
         """
-        if isinstance(self.remote_server, PositConnectServer):
+        if isinstance(self.remote_server, (RSConnectServer, SPCSConnectServer)):
             if not isinstance(self.client, RSConnectClient):
                 raise RSConnectException("To emit task log, client must be a RSConnectClient.")
 
@@ -1128,7 +1127,7 @@ def save_deployed_info(self):
 
     @cls_logged("Verifying deployed content...")
     def verify_deployment(self):
-        if isinstance(self.remote_server, PositConnectServer):
+        if isinstance(self.remote_server, (RSConnectServer, SPCSConnectServer)):
             if not isinstance(self.client, RSConnectClient):
                 raise RSConnectException("To verify deployment, client must be a RSConnectClient.")
             deployed_info = self.deployed_info
@@ -1883,7 +1882,7 @@ def verify_api_key(connect_server: RSConnectServer) -> str:
         return result["username"]
 
 
-def get_python_info(connect_server: PositConnectServer):
+def get_python_info(connect_server: Union[RSConnectServer, SPCSConnectServer]):
     """
     Return information about versions of Python that are installed on the indicated
     Connect server.
@@ -1897,7 +1896,7 @@ def get_python_info(connect_server: PositConnectServer):
         return result
 
 
-def get_app_info(connect_server: PositConnectServer, app_id: str):
+def get_app_info(connect_server: Union[RSConnectServer, SPCSConnectServer], app_id: str):
     """
     Return information about an application that has been created in Connect.
 
@@ -1918,7 +1917,7 @@ def get_posit_app_info(server: PositServer, app_id: str):
             return response["source"]
 
 
-def get_app_config(connect_server: PositConnectServer, app_id: str):
+def get_app_config(connect_server: Union[RSConnectServer, SPCSConnectServer], app_id: str):
     """
     Return the configuration information for an application that has been created
     in Connect.
@@ -1934,7 +1933,7 @@ def get_app_config(connect_server: PositConnectServer, app_id: str):
 
 
 def emit_task_log(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     app_id: str,
     task_id: str,
     log_callback: Optional[Callable[[str], None]],
@@ -1970,7 +1969,7 @@ def emit_task_log(
 
 
 def retrieve_matching_apps(
-    connect_server: PositConnectServer,
+    connect_server: Union[RSConnectServer, SPCSConnectServer],
     filters: Optional[dict[str, str | int]] = None,
     limit: Optional[int] = None,
     mapping_function: Optional[Callable[[RSConnectClient, ContentItemV0], AbbreviatedAppItem | None]] = None,
@@ -2046,7 +2045,7 @@ class AbbreviatedAppItem(TypedDict):
     config_url: str
 
 
-def override_title_search(connect_server: PositConnectServer, app_id: str, app_title: str):
+def override_title_search(connect_server: Union[RSConnectServer, SPCSConnectServer], 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.
@@ -2127,7 +2126,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, PositConnectServer):
+    if isinstance(remote_server, (RSConnectServer, SPCSConnectServer)):
         existing_names = retrieve_matching_apps(
             remote_server,
             filters={"search": name},

rsconnect/main.py

@@ -50,7 +50,6 @@
     search_content,
 )
 from .api import (
-    PositConnectServer,
     RSConnectClient,
     RSConnectExecutor,
     RSConnectServer,
@@ -670,7 +669,7 @@ def details(
     set_verbosity(verbose)
 
     ce = RSConnectExecutor(ctx, name, server, api_key, snowflake_connection_name, insecure, cacert).validate_server()
-    if not isinstance(ce.remote_server, PositConnectServer):
+    if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
         raise RSConnectException("`rsconnect details` requires a Posit Connect server.")
 
     click.echo("    Posit Connect URL: %s" % ce.remote_server.url)
@@ -2422,7 +2421,7 @@ def content_search(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content search` requires a Posit Connect server.")
         result = search_content(
             ce.remote_server, published, unpublished, content_type, r_version, py_version, title_contains, order_by
@@ -2472,7 +2471,7 @@ def content_describe(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content describe` requires a Posit Connect server.")
         result = get_content(ce.remote_server, guid)
         json.dump(result, sys.stdout, indent=2)
@@ -2532,7 +2531,7 @@ def content_bundle_download(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content download-bundle` requires a Posit Connect server.")
         if exists(output) and not overwrite:
             raise RSConnectException("The output file already exists: %s" % output)
@@ -2589,7 +2588,7 @@ def add_content_build(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content build add` requires a Posit Connect server.")
         build_add_content(ce.remote_server, guid)
         if len(guid) == 1:
@@ -2708,7 +2707,7 @@ def list_content_build(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content build ls` requires a Posit Connect server.")
         result = build_list_content(ce.remote_server, guid, status)
         json.dump(result, sys.stdout, indent=2)
@@ -2753,7 +2752,7 @@ def get_build_history(
             logger=None,
         )
         ce.validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content build history` requires a Posit Connect server.")
         result = build_history(ce.remote_server, guid)
         json.dump(result, sys.stdout, indent=2)
@@ -2815,7 +2814,7 @@ def get_build_logs(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("`rsconnect content build logs` requires a Posit Connect server.")
         for line in emit_build_log(ce.remote_server, guid, format, task_id):
             sys.stdout.write(line)
@@ -2901,7 +2900,7 @@ def start_content_build(
             cacert=cacert,
             logger=None,
         ).validate_server()
-        if not isinstance(ce.remote_server, PositConnectServer):
+        if not isinstance(ce.remote_server, (RSConnectServer, SPCSConnectServer)):
             raise RSConnectException("rsconnect content build run` requires a Posit Connect server.")
         build_start(ce.remote_server, parallelism, aborted, error, running, retry, all, poll_wait, debug, force)
 

rsconnect/metadata.py

@@ -15,7 +15,7 @@
 from io import BufferedWriter
 from os.path import abspath, basename, dirname, exists, join
 from threading import Lock
-from typing import TYPE_CHECKING, Callable, Dict, Generic, Mapping, Optional, TypeVar
+from typing import TYPE_CHECKING, Callable, Dict, Generic, Mapping, Optional, TypeVar, Union
 from urllib.parse import urlparse
 
 # Even though TypedDict is available in Python 3.8, because it's used with NotRequired,
@@ -28,7 +28,7 @@
 
 
 if TYPE_CHECKING:
-    from .api import PositConnectServer
+    from .api import RSConnectServer, SPCSConnectServer
 
 from .exception import RSConnectException
 from .log import logger
@@ -602,7 +602,7 @@ class ContentBuildStore(DataStore[Dict[str, object]]):
 
     def __init__(
         self,
-        server: PositConnectServer,
+        server: Union[RSConnectServer, SPCSConnectServer],
         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