From 9a82dd39799165904ec54f3b5084e171b15a5df1 Mon Sep 17 00:00:00 2001 From: Maxime de Pachtere Date: Sat, 17 Feb 2024 16:48:01 +0100 Subject: [PATCH] Add exception that handle error message printing in the cli (temporary fatorization) and use the only existing blitz app if there is only one and no blitz app name is specified --- blitz/app.py | 24 +++++++++++------- blitz/cli/commands/release.py | 46 +++++++++++++++++++++++------------ blitz/cli/commands/start.py | 33 ++++++++++++++++--------- blitz/cli/commands/swagger.py | 38 ++++++++++++++++++----------- blitz/cli/errors/__init__.py | 42 ++++++++++++++++++++++++++++++++ blitz/db/migrations.py | 2 +- 6 files changed, 135 insertions(+), 50 deletions(-) create mode 100644 blitz/cli/errors/__init__.py diff --git a/blitz/app.py b/blitz/app.py index 6abccdd..bd51bc2 100644 --- a/blitz/app.py +++ b/blitz/app.py @@ -1,3 +1,4 @@ +import enum from pathlib import Path from blitz.db.errors import NoChangesDetectedError @@ -12,6 +13,11 @@ from semver import Version from loguru import logger +class ReleaseLevel(enum.Enum): + PATCH = "PATCH" + MINOR = "MINOR" + MAJOR = "MAJOR" + class BlitzApp: def __init__( @@ -33,7 +39,7 @@ def __init__( self.resources: list[BlitzResource] = [] self._is_loaded = False self._base_resource_model: type[BaseResourceModel] = BaseResourceModel - self._available_version: list[Version] = [] + self.versions: list[Version] = [] self._load_versions() @@ -54,12 +60,12 @@ def _load_versions(self) -> None: raise ValueError( f"Blitz app {self.name} has a version dir '{version}' without a blitz file inside." ) - self._available_version.append(version) + self.versions.append(version) - self._available_version = sorted(self._available_version) + self.versions = sorted(self.versions) def get_version(self, version: Version) -> "BlitzApp": - if version not in self._available_version: + if version not in self.versions: raise ValueError(f"Version {version} not found for Blitz app {self.name}") return BlitzApp( name=self.name, @@ -118,10 +124,10 @@ def load(self) -> None: self._is_loaded = True - def release(self, level: str, force: bool = False) -> Version: + def release(self, level: ReleaseLevel, force: bool = False) -> Version: clear_metadata() - latest_version = self._available_version[-1] if self._available_version else None + latest_version = self.versions[-1] if self.versions else None # If there is already a released version if latest_version is not None: @@ -131,11 +137,11 @@ def release(self, level: str, force: bool = False) -> Version: # We bump the version regarding the release level match level: - case "major": + case ReleaseLevel.MAJOR: new_version = latest_version.bump_major() - case "minor": + case ReleaseLevel.MINOR: new_version = latest_version.bump_minor() - case "patch": + case ReleaseLevel.PATCH: new_version = latest_version.bump_patch() if self.file.path is None: diff --git a/blitz/cli/commands/release.py b/blitz/cli/commands/release.py index be1a73f..9bb9bca 100644 --- a/blitz/cli/commands/release.py +++ b/blitz/cli/commands/release.py @@ -1,31 +1,47 @@ -from typing import Annotated +from typing import Annotated, Optional from blitz import BlitzCore +from blitz.app import ReleaseLevel import typer from rich import print import enum -from blitz.db.errors import NoChangesDetectedError +from blitz.db.errors import NoChangesDetectedError as MigrationNoChangesDetectedError +from blitz.cli.errors import BlitzAppNotFoundError, MissingBlitzAppNameError, NoChangesDetectedError + -class ReleaseLevel(enum.Enum): - major = "major" - minor = "minor" - patch = "patch" def release_blitz( - blitz_app_name: Annotated[str, typer.Argument(..., help="Blitz app to release")], - level: Annotated[ReleaseLevel, typer.Argument(..., help="Release level")], + blitz_app_name: Annotated[Optional[str], typer.Argument(help="Blitz app to release")] = None, force: Annotated[bool, typer.Option(help="Force the release even if no changes are detected")] = False, + patch: Annotated[bool, typer.Option(help="Release level patch")] = False, + minor: Annotated[bool, typer.Option(help="Release level minor")] = False, + major: Annotated[bool, typer.Option(help="Release level major")] = False, ) -> None: + # Can't have both --patch and --minor or --major + if sum([patch, minor, major]) > 1: + raise typer.BadParameter("You can't use more than one release level") + release_level = ReleaseLevel.MAJOR if major else ReleaseLevel.MINOR if minor else ReleaseLevel.PATCH + blitz = BlitzCore() - app = blitz.get_app(blitz_app_name) + if blitz_app_name is None: + if len(blitz.apps) == 1: + blitz_app = blitz.apps[0] + else: + raise MissingBlitzAppNameError() + else: + try: + blitz_app = blitz.get_app(blitz_app_name) + except Exception: + raise BlitzAppNotFoundError(blitz_app_name) + try: - new_version = app.release(level.value, force=force) - except NoChangesDetectedError: - typer.echo("No changes detected since the latest version. Use --force to release anyway.") - raise typer.Exit(code=1) - typer.echo(f"Blitz app {blitz_app_name} released at version {new_version}") - typer.echo("You can now start your versioned blitz app by running:") + new_version = blitz_app.release(release_level, force=force) + except MigrationNoChangesDetectedError: + raise NoChangesDetectedError() + + print(f"Blitz app {blitz_app_name} released at version {new_version}") + print("You can now start your versioned blitz app by running:") print(f" [bold medium_purple1]blitz start {blitz_app_name} --version {new_version}[/bold medium_purple1]") diff --git a/blitz/cli/commands/start.py b/blitz/cli/commands/start.py index 80dff42..ffc599f 100644 --- a/blitz/cli/commands/start.py +++ b/blitz/cli/commands/start.py @@ -13,10 +13,11 @@ from blitz.settings import get_settings from rich import print +from blitz.cli.errors import BlitzAppNotFoundError, BlitzAppVersionNotFoundError, MissingBlitzAppNameError def start_blitz( - blitz_app_name: Annotated[str, typer.Argument(..., help="Blitz app name")], + blitz_app_name: Annotated[Optional[str], typer.Argument(help="Blitz app name")] = None, admin: Annotated[bool, typer.Option(help="Don't create admin.")] = True, port: Annotated[int, typer.Option(help="Define the port of the server")] = get_settings().BLITZ_PORT, config_route: Annotated[bool, typer.Option(help="Enable the blitz config route.")] = True, @@ -25,17 +26,24 @@ def start_blitz( ) -> None: blitz = BlitzCore() - try: - blitz_app = blitz.get_app(blitz_app_name) - if version is not None: + if blitz_app_name is None: + if len(blitz.apps) == 1: + blitz_app = blitz.apps[0] + else: + raise MissingBlitzAppNameError() + else: + try: + blitz_app = blitz.get_app(blitz_app_name) + except Exception: + raise BlitzAppNotFoundError(blitz_app_name) + + if version is not None: + try: blitz_app = blitz_app.get_version(Version.parse(version)) - except Exception as exc: - print(f"[red bold]There is no blitz app named {blitz_app_name}[/red bold]") - print("To list the available blitz apps run:") - print("[bold] blitz list[bold]") - print(f"Error: {exc}") - raise typer.Exit() + except Exception: + raise BlitzAppVersionNotFoundError(blitz_app, version) + # https://patorjk.com/software/taag/#p=display&f=ANSI%20Shadow&t=BLITZ%200.1.0 print( """[bold medium_purple1] ██████╗ ██╗ ██╗████████╗███████╗ ██████╗ ██╗ ██████╗ @@ -46,7 +54,8 @@ def start_blitz( ╚═════╝ ╚══════╝╚═╝ ╚═╝ ╚══════╝ ╚═════╝ ╚═╝╚═╝╚═╝ ╚═════╝ [/bold medium_purple1]""" ) - time.sleep(1) + time.sleep(0.5) + if hot_reload: # Need to be refacto os.environ["BLITZ_APP"] = str(blitz_app.name) @@ -54,9 +63,11 @@ def start_blitz( os.environ["BLITZ_VERSION"] = str(version) os.environ["BLITZ_ADMIN"] = str(admin).lower() os.environ["BLITZ_CONFIG_ROUTE"] = str(config_route).lower() + if blitz_app.file.path is None: # TODO: handle error raise Exception + server_config = uvicorn.Config( "blitz.api:create_blitz_api", factory=True, diff --git a/blitz/cli/commands/swagger.py b/blitz/cli/commands/swagger.py index bc69502..d9fada3 100644 --- a/blitz/cli/commands/swagger.py +++ b/blitz/cli/commands/swagger.py @@ -6,6 +6,7 @@ from blitz.models import BlitzResource from rich.style import Style from rich.panel import Panel +from blitz.cli.errors import BlitzAppNotFoundError, MissingBlitzAppNameError, BlitzAppVersionNotFoundError class SwaggerPrinter: @@ -72,26 +73,35 @@ def print(self) -> None: def list_routes( - blitz_app_name: Annotated[str, typer.Argument(..., help="Blitz app name")], - model: Annotated[Optional[str], typer.Option()] = None, + blitz_app_name: Annotated[Optional[str], typer.Argument(..., help="Blitz app name")] = None, + resource_name: Annotated[Optional[str], typer.Option(help="The resource name")] = None, version: Annotated[Optional[str], typer.Option(help="Define the version of the app.")] = None, ) -> None: blitz = BlitzCore() - try: - blitz_app = blitz.get_app(blitz_app_name) - if version is not None: + if blitz_app_name is None: + if len(blitz.apps) == 1: + blitz_app = blitz.apps[0] + else: + raise MissingBlitzAppNameError() + else: + try: + blitz_app = blitz.get_app(blitz_app_name) + except Exception: + raise BlitzAppNotFoundError(blitz_app_name) + + if version is not None: + try: blitz_app = blitz_app.get_version(Version.parse(version)) - blitz_app.load() - except Exception as exc: - print(f"[red bold]There is no blitz app named {blitz_app_name}[/red bold]") - print("To list the available blitz apps run:") - print("[bold] blitz list[bold]") - print(f"Error: {exc}") - raise typer.Exit() - if model: + except Exception: + raise BlitzAppVersionNotFoundError(blitz_app=blitz_app, version=version) + + blitz_app.load() + + if resource_name: for resource in blitz_app.resources: - if resource.config.name.lower() == model.lower(): + if resource.config.name == resource_name: resources = [resource] else: resources = blitz_app.resources + SwaggerPrinter(resources).print() diff --git a/blitz/cli/errors/__init__.py b/blitz/cli/errors/__init__.py new file mode 100644 index 0000000..39e4f35 --- /dev/null +++ b/blitz/cli/errors/__init__.py @@ -0,0 +1,42 @@ +import typer +from rich import print + +from blitz.app import BlitzApp + + +class BlitzCLIError(typer.Exit): + CODE = 1 + + +class BlitzAppNotFoundError(BlitzCLIError): + def __init__(self, blitz_app_name: str) -> None: + self.blitz_app_name = blitz_app_name + print(f"[red bold]There is no blitz app named {blitz_app_name}[/red bold]") + print("To list the available blitz apps run:") + print("[bold] blitz list[bold]") + super().__init__(code=self.CODE) + + +class BlitzAppVersionNotFoundError(BlitzCLIError): + def __init__(self, blitz_app: BlitzApp, version: str) -> None: + self.blitz_app = blitz_app + self.version = version + print(f"[red bold]There is no version {version} for the blitz app {blitz_app.name}[/red bold]") + print("Available versions are:") + for blitz_app_version in blitz_app.versions: + print(f" {blitz_app_version}") + super().__init__(code=self.CODE) + + +class MissingBlitzAppNameError(BlitzCLIError): + def __init__(self) -> None: + print("You need to specify a blitz app name.") + print("To list the available blitz apps run:") + print("[bold] blitz list[bold]") + super().__init__(code=self.CODE) + + +class NoChangesDetectedError(BlitzCLIError): + def __init__(self) -> None: + print("No changes detected since the latest version. Use --force to release anyway.") + super().__init__(code=self.CODE) \ No newline at end of file diff --git a/blitz/db/migrations.py b/blitz/db/migrations.py index 0f2c888..e72b7e0 100644 --- a/blitz/db/migrations.py +++ b/blitz/db/migrations.py @@ -42,7 +42,7 @@ def get_alembic_config( if is_release: file_name_template = RELEASE_MIGRATION_FILE_NAME_TEMPLATE migrations_paths = set( - [str(blitz_app.path / str(version)) for version in blitz_app._available_version] + [str(blitz_app.path / str(version)) for version in blitz_app.versions] + [str(blitz_app.path / str(blitz_app.file.config.version))] ) else: