ref(logs): Updating logs to use console util

devservices/commands/list_dependencies.py

@@ -4,6 +4,7 @@
 from argparse import ArgumentParser
 from argparse import Namespace
 
+from devservices.utils.console import Console
 from devservices.utils.services import find_matching_service
 
 
@@ -22,20 +23,21 @@ def add_parser(subparsers: _SubParsersAction[ArgumentParser]) -> None:
 
 def list_dependencies(args: Namespace) -> None:
     """List the dependencies of a service."""
+    console = Console()
     service_name = args.service_name
 
     try:
         service = find_matching_service(service_name)
     except Exception as e:
-        print(e)
+        console.failure(str(e))
         exit(1)
 
     dependencies = service.config.dependencies
 
     if not dependencies:
-        print(f"No dependencies found for {service.name}")
+        console.info(f"No dependencies found for {service.name}")
         return
 
-    print(f"Dependencies of {service.name}:")
+    console.info(f"Dependencies of {service.name}:")
     for dependency_key, dependency_info in dependencies.items():
-        print("-", dependency_key, ":", dependency_info.description)
+        console.info("-" + dependency_key + ":" + dependency_info.description)

devservices/commands/list_services.py

@@ -4,6 +4,7 @@
 from argparse import ArgumentParser
 from argparse import Namespace
 
+from devservices.utils.console import Console
 from devservices.utils.devenv import get_coderoot
 from devservices.utils.services import get_local_services
 from devservices.utils.state import State
@@ -24,35 +25,35 @@
 
 def list_services(args: Namespace) -> None:
     """List the services installed locally."""
-
+    console = Console()
     # Get all of the services installed locally
     coderoot = get_coderoot()
     services = get_local_services(coderoot)
     state = State()
     running_services = state.get_started_services()
 
     if not services:
-        print("No services found")
+        console.warning("No services found")
         return
 
     services_to_show = (
         services if args.all else [s for s in services if s.name in running_services]
     )
 
     if args.all:
-        print("Services installed locally:")
+        console.info("Services installed locally:")
     else:
-        print("Running services:")
+        console.info("Running services:")
 
     for service in services_to_show:
         status = "running" if service.name in running_services else "stopped"
-        print(f"- {service.name}")
-        print(f"  status: {status}")
-        print(f"  location: {service.repo_path}")
+        console.info(f"- {service.name}")
+        console.info(f"  status: {status}")
+        console.info(f"  location: {service.repo_path}")
 
     if not args.all:
         stopped_count = len(services) - len(services_to_show)
         if stopped_count > 0:
-            print(
+            console.info(
                 f"\n{stopped_count} stopped service(s) not shown. Use --all/-a to see them."
             )

devservices/commands/logs.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
-import sys
 from argparse import _SubParsersAction
 from argparse import ArgumentParser
 from argparse import Namespace
 
 from devservices.constants import MAX_LOG_LINES
 from devservices.exceptions import DependencyError
 from devservices.exceptions import DockerComposeError
+from devservices.utils.console import Console
 from devservices.utils.docker_compose import run_docker_compose_command
 from devservices.utils.services import find_matching_service
 from devservices.utils.state import State
@@ -26,11 +26,12 @@ def add_parser(subparsers: _SubParsersAction[ArgumentParser]) -> None:
 
 def logs(args: Namespace) -> None:
     """View the logs for a specified service."""
+    console = Console()
     service_name = args.service_name
     try:
         service = find_matching_service(service_name)
     except Exception as e:
-        print(e)
+        console.failure(str(e))
         exit(1)
 
     modes = service.config.modes
@@ -41,19 +42,18 @@ def logs(args: Namespace) -> None:
     state = State()
     running_services = state.get_started_services()
     if service_name not in running_services:
-        print(f"Service {service_name} is not running")
+        console.warning(f"Service {service_name} is not running")
         return
 
     try:
         logs_output = run_docker_compose_command(
             service, "logs", mode_dependencies, options=["-n", MAX_LOG_LINES]
         )
     except DependencyError as de:
-        print(str(de))
+        console.failure(str(de))
         exit(1)
     except DockerComposeError as dce:
-        print(f"Failed to get logs for {service.name}: {dce.stderr}")
+        console.failure(f"Failed to get logs for {service.name}: {dce.stderr}")
         exit(1)
     for log in logs_output:
-        sys.stdout.write(log.stdout)
-        sys.stdout.flush()
+        console.info(log.stdout)

devservices/commands/purge.py

@@ -7,6 +7,7 @@
 from argparse import Namespace
 
 from devservices.constants import DEVSERVICES_CACHE_DIR
+from devservices.utils.console import Console
 
 
 def add_parser(subparsers: _SubParsersAction[ArgumentParser]) -> None:
@@ -16,10 +17,11 @@
 
 def purge(args: Namespace) -> None:
     """Purge the local devservices cache."""
+    console = Console()
     if os.path.exists(DEVSERVICES_CACHE_DIR):
         try:
             shutil.rmtree(DEVSERVICES_CACHE_DIR)
         except PermissionError as e:
-            print(f"Failed to purge cache: {e}")
+            console.failure(f"Failed to purge cache: {e}")
             exit(1)
-    print("The local devservices cache has been purged")
+    console.success("The local devservices cache has been purged")

devservices/commands/start.py

@@ -6,6 +6,7 @@
 
 from devservices.exceptions import DependencyError
 from devservices.exceptions import DockerComposeError
+from devservices.utils.console import Console
 from devservices.utils.console import Status
 from devservices.utils.docker_compose import run_docker_compose_command
 from devservices.utils.services import find_matching_service
@@ -22,19 +23,23 @@
 
 def start(args: Namespace) -> None:
     """Start a service and its dependencies."""
+    console = Console()
     service_name = args.service_name
     try:
         service = find_matching_service(service_name)
     except Exception as e:
-        print(e)
+        console.failure(str(e))
         exit(1)
 
     modes = service.config.modes
     # TODO: allow custom modes to be used
     mode_to_start = "default"
     mode_dependencies = modes[mode_to_start]
 
-    with Status(f"Starting {service.name}", f"{service.name} started") as status:
+    with Status(
+        lambda: console.warning(f"Starting {service.name}"),
+        lambda: console.success(f"{service.name} started"),
+    ) as status:
         try:
             run_docker_compose_command(
                 service,
@@ -44,10 +49,10 @@
                 force_update_dependencies=True,
             )
         except DependencyError as de:
-            status.print(str(de))
+            status.failure(str(de))
             exit(1)
         except DockerComposeError as dce:
-            status.print(f"Failed to start {service.name}: {dce.stderr}")
+            status.failure(f"Failed to start {service.name}: {dce.stderr}")
             exit(1)
     # TODO: We should factor in healthchecks here before marking service as running
     state = State()

devservices/commands/status.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
 import json
-import sys
 from argparse import _SubParsersAction
 from argparse import ArgumentParser
 from argparse import Namespace
 
 from devservices.exceptions import DependencyError
 from devservices.exceptions import DockerComposeError
+from devservices.utils.console import Console
 from devservices.utils.docker_compose import run_docker_compose_command
 from devservices.utils.services import find_matching_service
 
@@ -61,11 +61,12 @@ def format_status_output(status_json: str) -> str:
 
 def status(args: Namespace) -> None:
     """Start a service and its dependencies."""
+    console = Console()
     service_name = args.service_name
     try:
         service = find_matching_service(service_name)
     except Exception as e:
-        print(e)
+        console.failure(str(e))
         exit(1)
 
     modes = service.config.modes
@@ -78,22 +79,21 @@ def status(args: Namespace) -> None:
             service, "ps", mode_dependencies, options=["--format", "json"]
         )
     except DependencyError as de:
-        print(str(de))
+        console.failure(str(de))
         exit(1)
     except DockerComposeError as dce:
-        print(f"Failed to get status for {service.name}: {dce.stderr}")
+        console.failure(f"Failed to get status for {service.name}: {dce.stderr}")
         exit(1)
 
     # Filter out empty stdout to help us determine if the service is running
     status_json_results = [
         status_json for status_json in status_json_results if status_json.stdout
     ]
     if len(status_json_results) == 0:
-        print(f"{service.name} is not running")
+        console.warning(f"{service.name} is not running")
         return
     output = f"Service: {service.name}\n\n"
     for status_json in status_json_results:
         output += format_status_output(status_json.stdout)
     output += "=" * LINE_LENGTH
-    sys.stdout.write(output + "\n")
-    sys.stdout.flush()
+    console.info(output + "\n")

devservices/commands/stop.py

@@ -6,6 +6,7 @@
 
 from devservices.exceptions import DependencyError
 from devservices.exceptions import DockerComposeError
+from devservices.utils.console import Console
 from devservices.utils.console import Status
 from devservices.utils.docker_compose import run_docker_compose_command
 from devservices.utils.services import find_matching_service
@@ -22,11 +23,12 @@
 
 def stop(args: Namespace) -> None:
     """Stop a service and its dependencies."""
+    console = Console()
     service_name = args.service_name
     try:
         service = find_matching_service(service_name)
     except Exception as e:
-        print(e)
+        console.failure(str(e))
         exit(1)
 
     modes = service.config.modes
@@ -36,17 +38,20 @@
     state = State()
     started_services = state.get_started_services()
     if service.name not in started_services:
-        print(f"{service.name} is not running")
+        console.warning(f"{service.name} is not running")
         exit(0)
 
-    with Status(f"Stopping {service.name}", f"{service.name} stopped") as status:
+    with Status(
+        lambda: console.warning(f"Stopping {service.name}"),
+        lambda: console.success(f"{service.name} stopped"),
+    ) as status:
         try:
             run_docker_compose_command(service, "down", mode_dependencies)
         except DependencyError as de:
-            status.print(str(de))
+            status.failure(str(de))
             exit(1)
         except DockerComposeError as dce:
-            status.print(f"Failed to stop {service.name}: {dce.stderr}")
+            status.failure(f"Failed to stop {service.name}: {dce.stderr}")
             exit(1)
 
     # TODO: We should factor in healthchecks here before marking service as stopped

devservices/commands/update.py

@@ -11,6 +11,7 @@
 from devservices.constants import DEVSERVICES_DOWNLOAD_URL
 from devservices.exceptions import BinaryInstallError
 from devservices.exceptions import DevservicesUpdateError
+from devservices.utils.console import Console
 from devservices.utils.install_binary import install_binary
 
 
@@ -21,14 +22,16 @@ def is_in_virtualenv() -> bool:
 
 
 def update_version(exec_path: str, latest_version: str) -> None:
+    console = Console()
     system = platform.system().lower()
     url = f"{DEVSERVICES_DOWNLOAD_URL}/{latest_version}/devservices-{system}"
     try:
         install_binary("devservices", exec_path, latest_version, url)
     except BinaryInstallError as e:
-        raise DevservicesUpdateError(f"Failed to update devservices: {e}")
+        console.failure(f"Failed to update devservices: {e}")
+        exit(1)
 
-    print(f"Devservices {latest_version} updated successfully")
+    console.success(f"Devservices {latest_version} updated successfully")
 
 
 def add_parser(subparsers: _SubParsersAction[ArgumentParser]) -> None:
@@ -39,28 +42,29 @@ def add_parser(subparsers: _SubParsersAction[ArgumentParser]) -> None:
 
 
 def update(args: Namespace) -> None:
+    console = Console()
     current_version = metadata.version("devservices")
     latest_version = check_for_update(current_version)
 
     if latest_version is None:
         raise DevservicesUpdateError("Failed to check for updates.")
 
     if latest_version == current_version:
-        print("You're already on the latest version.")
+        console.warning("You're already on the latest version.")
         return
 
-    print(f"A new version of devservices is available: {latest_version}")
+    console.warning(f"A new version of devservices is available: {latest_version}")
 
     if is_in_virtualenv():
-        print("You are running in a virtual environment.")
-        print(
+        console.warning("You are running in a virtual environment.")
+        console.warning(
             "To update, please update your requirements.txt or requirements-dev.txt file with the new version."
         )
-        print(
+        console.warning(
             f"For example, update the line in requirements.txt to: devservices=={latest_version}"
         )
-        print("Then, run: pip install --update -r requirements.txt")
+        console.warning("Then, run: pip install --update -r requirements.txt")
         return
 
-    print("Upgrading to the latest version...")
+    console.info("Upgrading to the latest version...")
     update_version(sys.executable, latest_version)