add SlobsCliProtocolError for wrapping ProtocolError

handle ProtocolError(s) and reraise as SlobsCliProtocolError. This has the following benefits:
A user friendly error message
A non-zero exit code
This commit is contained in:
onyx-and-iris 2025-06-13 14:10:54 +01:00
parent f4421b3351
commit 09a44b2dea
2 changed files with 55 additions and 11 deletions

View File

@ -1,5 +1,7 @@
"""module for custom exceptions in Slobs CLI.""" """module for custom exceptions in Slobs CLI."""
import json
import asyncclick as click import asyncclick as click
@ -14,3 +16,31 @@ class SlobsCliError(click.ClickException):
def show(self): def show(self):
"""Display the error message in red.""" """Display the error message in red."""
click.secho(f'Error: {self.message}', fg='red', err=True) click.secho(f'Error: {self.message}', fg='red', err=True)
class SlobsCliProtocolError(SlobsCliError):
"""Converts pyslobs ProtocolError to a SlobsCliProtocolError."""
def __init__(self, message: str):
"""Initialize the SlobsCliProtocolError with a message."""
protocol_message_to_dict = json.loads(
str(message).replace('"', '\\"').replace("'", '"')
)
super().__init__(
protocol_message_to_dict.get('message', 'Unable to parse error message')
)
self.exit_code = 2
self.protocol_code = protocol_message_to_dict.get('code', 'Unknown error code')
def show(self):
"""Display the protocol error message in red."""
match self.protocol_code:
case -32600:
click.secho(
'Oops! Looks like we hit a rate limit for this command. Please try again later.',
fg='red',
err=True,
)
case _:
# Fall back to the base error display for unknown protocol codes
super().show()

View File

@ -2,11 +2,11 @@
import asyncclick as click import asyncclick as click
from anyio import create_task_group from anyio import create_task_group
from pyslobs import ScenesService, TransitionsService from pyslobs import ProtocolError, ScenesService, TransitionsService
from terminaltables3 import AsciiTable from terminaltables3 import AsciiTable
from .cli import cli from .cli import cli
from .errors import SlobsCliError from .errors import SlobsCliError, SlobsCliProtocolError
@cli.group() @cli.group()
@ -56,9 +56,14 @@ async def list(ctx: click.Context, id: bool = False):
conn.close() conn.close()
try:
async with create_task_group() as tg: async with create_task_group() as tg:
tg.start_soon(conn.background_processing) tg.start_soon(conn.background_processing)
tg.start_soon(_run) tg.start_soon(_run)
except* ProtocolError as excgroup:
raisable = next(iter(excgroup.exceptions))
e = SlobsCliProtocolError(str(raisable))
raise e
@scene.command() @scene.command()
@ -77,9 +82,14 @@ async def current(ctx: click.Context, id: bool = False):
) )
conn.close() conn.close()
try:
async with create_task_group() as tg: async with create_task_group() as tg:
tg.start_soon(conn.background_processing) tg.start_soon(conn.background_processing)
tg.start_soon(_run) tg.start_soon(_run)
except* ProtocolError as excgroup:
raisable = next(iter(excgroup.exceptions))
e = SlobsCliProtocolError(str(raisable))
raise e
@scene.command() @scene.command()
@ -138,12 +148,16 @@ async def switch(
break break
else: # If no scene by the given name was found else: # If no scene by the given name was found
conn.close() conn.close()
raise SlobsCliError(f"Scene '{scene_name}' not found.") raise SlobsCliError(f'Scene "{scene_name}" not found.')
try: try:
async with create_task_group() as tg: async with create_task_group() as tg:
tg.start_soon(conn.background_processing) tg.start_soon(conn.background_processing)
tg.start_soon(_run) tg.start_soon(_run)
except* SlobsCliError as excgroup: except* SlobsCliError as excgroup:
for e in excgroup.exceptions: raisable = next(iter(excgroup.exceptions))
raise e raise raisable
except* ProtocolError as excgroup:
p_error = next(iter(excgroup.exceptions))
raisable = SlobsCliProtocolError(str(p_error))
raise raisable