Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 15 additions & 12 deletions admin/build.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,29 +121,31 @@ def _get_latest_release(dry: bool) -> tuple[str, str, list[dict]]:
"""
import json

release_info_json = (
run('gh', 'release', 'view', '--json', 'name,tagName,assets', dry=dry, capture_output=True)
.stdout.decode() # type: ignore
.strip()
)
release_info_json = run(
'gh', 'release', 'view', '--json', 'name,tagName,assets', dry=dry, capture_output=True
).stdout
release_info = json.loads(release_info_json)
return release_info['name'], release_info['tagName'], release_info['assets']


def _get_branch():
"""Returns the current branch."""
return (
run('git', 'branch', '--show-current', dry=False, capture_output=True)
.stdout.decode() # type: ignore
.strip()
)
return run('git', 'branch', '--show-current', dry=False, capture_output=True).stdout


def _get_default_branch():
"""Returns the default branch (usually ``main``)."""
return run(
False, 'gh', 'repo', 'view', '--json', 'defaultBranchRef', '--jq', '.defaultBranchRef.name'
)
'gh',
'repo',
'view',
'--json',
'defaultBranchRef',
'--jq',
'.defaultBranchRef.name',
dry=False,
capture_output=True,
).stdout


def _commit(message: str, dry: bool):
Expand Down Expand Up @@ -200,6 +202,7 @@ def build_publish(
run('uv', 'build', dry=dry)
if not upload:
return

msg = f'Publishing version `{_get_project_version()}` to PyPI. Press Y to confirm. '
if yes or input(msg).strip().lower() == 'y':
run('uv', 'publish', dry=dry)
Expand Down
69 changes: 63 additions & 6 deletions admin/utils.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,21 @@
import logging
import os
import subprocess
import sys
from dataclasses import dataclass
from enum import Enum
from itertools import chain
from typing import Annotated

import typer
from rich.logging import RichHandler
from rich.text import Text

from admin import PROJECT_ROOT

EMPTY_STR = object()
"""Sentinel object to represent an empty string."""

DryAnnotation = Annotated[
bool,
typer.Option(
Expand All @@ -27,6 +33,23 @@ class OS(str, Enum):
Windows = 'win'


@dataclass
class StripOutput:
strip_ansi: bool = True
normal_strip: bool = True
extra_chars: str | None = None

def strip(self, text: str) -> str:
if self.strip_ansi:
text = strip_ansi(text)
if self.normal_strip:
text = text.strip()
if self.extra_chars:
text = text.strip(self.extra_chars)

return text


def get_os() -> OS:
"""
Similar to ``sys.platform`` and ``platform.system()``, but less ambiguous by returning an Enum
Expand All @@ -41,37 +64,67 @@ def get_os() -> OS:
return OS.Linux


def run(*args, dry: bool = False, **kwargs) -> subprocess.CompletedProcess | None:
def run(
*args,
dry: bool = False,
extra_env: dict[str, str] | None = None,
strip_output: StripOutput | None = StripOutput(),
**kwargs,
) -> subprocess.CompletedProcess | None:
"""
Run a CLI command synchronously (i.e., wait for the command to finish) and return the result.

This function is a wrapper around ``subprocess.run(...)``.

If you need access to the output, add the ``capture_output=True`` argument and do
``.stdout.decode().strip()`` to get the output as a string.
``.stdout`` to get the output as a string.

Notes:

* Args are converted to strings using ``str(...)``.
* Empty strings and ``None`` are removed from the command.
If you want to explicitly include an empty string, use ``EMPTY_STR`` instead.
* ``stdout`` and ``stderr`` will be stripped of ANSI escape sequences by default.
"""
logger.info(' '.join(map(str, args)))
final_args: list[str] = []
for arg in args:
if arg in ['', None]:
continue
if arg == EMPTY_STR:
final_args.append('')
else:
final_args.append(str(arg))
logger.info(' '.join(f'"{a}"' if (not a or ' ' in a) else a for a in final_args))

if dry:
return None

defaults = dict(
cwd=PROJECT_ROOT,
capture_output=False,
text=True,
check=True,
env=os.environ.copy() | (extra_env or {}),
)
final_kwargs = defaults | kwargs

try:
return subprocess.run(args, **(defaults | kwargs)) # type: ignore
result = subprocess.run(final_args, **final_kwargs) # type: ignore
except subprocess.CalledProcessError as e:
msg = str(e)
if e.stdout:
msg += f'\nSTDOUT:\n{e.stdout.decode()}'
msg += f'\nSTDOUT:\n{e.stdout}'
if e.stderr:
msg += f'\nSTDERR:\n{e.stderr.decode()}'
msg += f'\nSTDERR:\n{e.stderr}'
logger.error(msg)
raise typer.Exit(1)

if final_kwargs.get('capture_output') and strip_output:
result.stdout = strip_output.strip(result.stdout)
result.stderr = strip_output.strip(result.stderr)

return result # type: ignore


def run_async(*args, dry: bool = False, **kwargs) -> subprocess.Popen | None:
"""
Expand Down Expand Up @@ -129,6 +182,10 @@ def multiple_parameters(parameter: str, *options) -> list[str]:
return list(chain.from_iterable(zip([parameter] * len(options), map(str, options))))


def strip_ansi(text: str) -> str:
return Text.from_ansi(text).plain


def get_logger(name: str | None = 'typer-invoke', level=logging.DEBUG) -> logging.Logger:
"""Set up logging configuration with Rich handler and custom formatting."""

Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ module-name = 'hd_active'

[project]
name = 'hd_active'
version = '0.1.1'
version = '0.2.0'
description = 'Prevent external HDs from becoming inactive (sleeping).'
readme = 'README.md'
authors = [{name = 'Joao Coelho'}]
Expand Down
Empty file removed src/__init__.py
Empty file.
2 changes: 1 addition & 1 deletion src/hd_active/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +1 @@
__version__ = '0.1.1'
__version__ = '0.2.0'