Skip to content

Commit

Permalink
Add exception that handle error message printing in the cli (temporar…
Browse files Browse the repository at this point in the history
…y fatorization) and use the only existing blitz app if there is only one and no blitz app name is specified
  • Loading branch information
mde-pach committed Feb 17, 2024
1 parent 5e557b0 commit 9a82dd3
Show file tree
Hide file tree
Showing 6 changed files with 135 additions and 50 deletions.
24 changes: 15 additions & 9 deletions blitz/app.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import enum
from pathlib import Path

from blitz.db.errors import NoChangesDetectedError
Expand All @@ -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__(
Expand All @@ -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()

Expand All @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down
46 changes: 31 additions & 15 deletions blitz/cli/commands/release.py
Original file line number Diff line number Diff line change
@@ -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]")
33 changes: 22 additions & 11 deletions blitz/cli/commands/start.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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]
██████╗ ██╗ ██╗████████╗███████╗ ██████╗ ██╗ ██████╗
Expand All @@ -46,17 +54,20 @@ 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)
if version is not None:
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,
Expand Down
38 changes: 24 additions & 14 deletions blitz/cli/commands/swagger.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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()
42 changes: 42 additions & 0 deletions blitz/cli/errors/__init__.py
Original file line number Diff line number Diff line change
@@ -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)
2 changes: 1 addition & 1 deletion blitz/db/migrations.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down

0 comments on commit 9a82dd3

Please sign in to comment.