feat: update logging
This commit is contained in:
Generated
+16
-1
@@ -129,6 +129,21 @@ files = [
|
||||
docs = ["furo (>=2022.12.7)", "proselint (>=0.13)", "sphinx (>=6.1.3)", "sphinx-autodoc-typehints (>=1.22,!=1.23.4)"]
|
||||
test = ["appdirs (==1.4.4)", "covdefaults (>=2.2.2)", "pytest (>=7.2.1)", "pytest-cov (>=4)", "pytest-mock (>=3.10)"]
|
||||
|
||||
[[package]]
|
||||
name = "termcolor"
|
||||
version = "2.2.0"
|
||||
description = "ANSI color formatting for output in terminal"
|
||||
category = "main"
|
||||
optional = false
|
||||
python-versions = ">=3.7"
|
||||
files = [
|
||||
{file = "termcolor-2.2.0-py3-none-any.whl", hash = "sha256:91ddd848e7251200eac969846cbae2dacd7d71c2871e92733289e7e3666f48e7"},
|
||||
{file = "termcolor-2.2.0.tar.gz", hash = "sha256:dfc8ac3f350788f23b2947b3e6cfa5a53b630b612e6cd8965a015a776020b99a"},
|
||||
]
|
||||
|
||||
[package.extras]
|
||||
tests = ["pytest", "pytest-cov"]
|
||||
|
||||
[[package]]
|
||||
name = "tomli"
|
||||
version = "2.0.1"
|
||||
@@ -156,4 +171,4 @@ files = [
|
||||
[metadata]
|
||||
lock-version = "2.0"
|
||||
python-versions = "^3.8"
|
||||
content-hash = "b6fb61746a881b469139cbab13045cce95037f63ba5474b4a170facd83bb6053"
|
||||
content-hash = "95dd14e87750e4830d773cac5ea2f5b810d406fb34b1913243c8ba6a2145db54"
|
||||
|
||||
@@ -5,7 +5,7 @@ import logging
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger("command")
|
||||
logger = logging.getLogger(">")
|
||||
|
||||
|
||||
class Command:
|
||||
@@ -17,14 +17,14 @@ class Command:
|
||||
|
||||
def run(self, basepath: Path):
|
||||
args = [self._executor, *self._args]
|
||||
self._logger.info(" ".join(args))
|
||||
self._logger.debug(" ".join(args))
|
||||
result = subprocess.run(
|
||||
args,
|
||||
cwd=basepath,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
self._logstd(result.stdout, logging.DEBUG)
|
||||
self._logstd(result.stdout, logging.TRACE)
|
||||
if result.returncode != 0:
|
||||
self._logstd(result.stderr, logging.ERROR)
|
||||
raise RuntimeError(result.stderr)
|
||||
|
||||
@@ -1,4 +1,7 @@
|
||||
__all__ = ("Entry",)
|
||||
__all__ = (
|
||||
"EntryCommandNotImplementedError",
|
||||
"Entry",
|
||||
)
|
||||
|
||||
|
||||
import logging
|
||||
@@ -8,6 +11,10 @@ from typing import Optional
|
||||
logger = logging.getLogger("entry")
|
||||
|
||||
|
||||
class EntryCommandNotImplementedError(NotImplementedError):
|
||||
pass
|
||||
|
||||
|
||||
class Entry:
|
||||
_logger = logger
|
||||
|
||||
@@ -16,7 +23,19 @@ class Entry:
|
||||
self._basepath = basepath
|
||||
|
||||
def setup(self):
|
||||
raise NotImplemented
|
||||
raise EntryCommandNotImplementedError("setup")
|
||||
|
||||
def lint(self):
|
||||
raise EntryCommandNotImplementedError("lint")
|
||||
|
||||
def format(self):
|
||||
raise EntryCommandNotImplementedError("format")
|
||||
|
||||
def test(self):
|
||||
raise EntryCommandNotImplementedError("test")
|
||||
|
||||
def show(self):
|
||||
self._logger.info(f"{self}")
|
||||
|
||||
def __repr__(self):
|
||||
return f"{self._type} {str(self._basepath)}"
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
import logging
|
||||
|
||||
from termcolor import colored
|
||||
|
||||
__all__ = (
|
||||
"addLoggingLevel",
|
||||
"CustomFormatter",
|
||||
)
|
||||
|
||||
|
||||
def addLoggingLevel(levelName, levelNum, methodName=None):
|
||||
"""
|
||||
Comprehensively adds a new logging level to the `logging` module and the
|
||||
currently configured logging class.
|
||||
|
||||
`levelName` becomes an attribute of the `logging` module with the value
|
||||
`levelNum`. `methodName` becomes a convenience method for both `logging`
|
||||
itself and the class returned by `logging.getLoggerClass()` (usually just
|
||||
`logging.Logger`). If `methodName` is not specified, `levelName.lower()` is
|
||||
used.
|
||||
|
||||
To avoid accidental clobberings of existing attributes, this method will
|
||||
raise an `AttributeError` if the level name is already an attribute of the
|
||||
`logging` module or if the method name is already present
|
||||
|
||||
Example
|
||||
-------
|
||||
>>> addLoggingLevel('TRACE', logging.DEBUG - 5)
|
||||
>>> logging.getLogger(__name__).setLevel("TRACE")
|
||||
>>> logging.getLogger(__name__).trace('that worked')
|
||||
>>> logging.trace('so did this')
|
||||
>>> logging.TRACE
|
||||
5
|
||||
|
||||
"""
|
||||
if not methodName:
|
||||
methodName = levelName.lower()
|
||||
|
||||
if hasattr(logging, levelName):
|
||||
raise AttributeError("{} already defined in logging module".format(levelName))
|
||||
if hasattr(logging, methodName):
|
||||
raise AttributeError("{} already defined in logging module".format(methodName))
|
||||
if hasattr(logging.getLoggerClass(), methodName):
|
||||
raise AttributeError("{} already defined in logger class".format(methodName))
|
||||
|
||||
# This method was inspired by the answers to Stack Overflow post
|
||||
# http://stackoverflow.com/q/2183233/2988730, especially
|
||||
# http://stackoverflow.com/a/13638084/2988730
|
||||
def logForLevel(self, message, *args, **kwargs):
|
||||
if self.isEnabledFor(levelNum):
|
||||
self._log(levelNum, message, args, **kwargs)
|
||||
|
||||
def logToRoot(message, *args, **kwargs):
|
||||
logging.log(levelNum, message, *args, **kwargs)
|
||||
|
||||
logging.addLevelName(levelNum, levelName)
|
||||
setattr(logging, levelName, levelNum)
|
||||
setattr(logging.getLoggerClass(), methodName, logForLevel)
|
||||
setattr(logging, methodName, logToRoot)
|
||||
|
||||
|
||||
addLoggingLevel("TRACE", logging.DEBUG - 5)
|
||||
|
||||
|
||||
class CustomFormatter(logging.Formatter):
|
||||
def __init__(self, fmt):
|
||||
super().__init__()
|
||||
self.fmt = fmt
|
||||
self.FORMATS = {
|
||||
logging.TRACE: colored(self.fmt, "dark_grey"),
|
||||
logging.DEBUG: colored(self.fmt, "grey"),
|
||||
logging.INFO: colored(self.fmt, "blue"),
|
||||
logging.WARNING: colored(self.fmt, "yellow"),
|
||||
logging.ERROR: colored(self.fmt, "red"),
|
||||
logging.CRITICAL: colored(self.fmt, "red", attrs=["bold"]),
|
||||
}
|
||||
|
||||
def format(self, record: logging.LogRecord):
|
||||
log_fmt = colored(record.name, "green") + " " + self.FORMATS.get(record.levelno)
|
||||
formatter = logging.Formatter(log_fmt)
|
||||
return formatter.format(record)
|
||||
+24
-10
@@ -1,37 +1,51 @@
|
||||
import logging
|
||||
import os
|
||||
from argparse import ArgumentParser
|
||||
from ast import arg
|
||||
from pathlib import Path
|
||||
|
||||
from project_manager.logging import CustomFormatter
|
||||
from project_manager.manager import ProjectManager
|
||||
|
||||
|
||||
def run():
|
||||
parser = ArgumentParser()
|
||||
parser.add_argument("task", nargs="?", default="show", choices=("show", "setup"))
|
||||
parser.add_argument(
|
||||
"task",
|
||||
nargs="?",
|
||||
default="show",
|
||||
choices=("show", "setup", "lint", "format", "test"),
|
||||
)
|
||||
parser.add_argument("-p", "--path", default=os.getcwd())
|
||||
parser.add_argument("-i", "--ignore-errors", action="store_true")
|
||||
parser.add_argument("-v", "--verbose", action="count", default=1)
|
||||
parser.add_argument("-q", "--quiet", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
logging_level = logging.WARNING
|
||||
if args.verbose > 1:
|
||||
if args.verbose > 2:
|
||||
logging_level = logging.TRACE
|
||||
elif args.verbose > 1:
|
||||
logging_level = logging.DEBUG
|
||||
elif args.verbose > 0:
|
||||
logging_level = logging.INFO
|
||||
if args.quiet:
|
||||
logging_level = logging.CRITICAL
|
||||
|
||||
logging.basicConfig(level=logging_level)
|
||||
if logging_level <= logging.INFO:
|
||||
logging.getLogger(">").setLevel(logging.CRITICAL)
|
||||
|
||||
stdout_handler = logging.StreamHandler()
|
||||
stdout_handler.setFormatter(CustomFormatter("%(message)s"))
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging_level,
|
||||
format="%(message)s",
|
||||
handlers=[stdout_handler],
|
||||
)
|
||||
|
||||
basepath = Path(args.path)
|
||||
manager = ProjectManager(basepath, args.ignore_errors)
|
||||
|
||||
if args.task == "show":
|
||||
manager.show()
|
||||
elif args.task == "setup":
|
||||
manager.setup()
|
||||
manager = ProjectManager(basepath)
|
||||
manager.run_command(args.task)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+20
-13
@@ -1,32 +1,39 @@
|
||||
import logging
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
from typing import Dict, Iterable
|
||||
|
||||
from .entry import Entry
|
||||
from .entry import Entry, EntryCommandNotImplementedError
|
||||
from .nodejs.entry import NodejsEntry
|
||||
from .python.entry import PythonEntry
|
||||
from .resolver import EntryDetector, EntryResolver
|
||||
|
||||
__all__ = ("ProjectManager",)
|
||||
|
||||
logger = logging.getLogger("project")
|
||||
|
||||
|
||||
class ProjectManager:
|
||||
def __init__(self, basepath: Path, ignore_entry_error: bool = False):
|
||||
_logger = logger
|
||||
|
||||
def __init__(self, basepath: Path):
|
||||
self._basepaht = basepath
|
||||
self._resolver = EntryResolver(EntryDetector(PythonEntry, NodejsEntry))
|
||||
self._ignore_entry_error = ignore_entry_error
|
||||
|
||||
def _iter_entries(self) -> Iterable[Entry]:
|
||||
return self._resolver.resolve(self._basepaht)
|
||||
|
||||
def setup(self):
|
||||
def run_command(self, command: str) -> Dict[str, int]:
|
||||
counter = Counter()
|
||||
for entry in self._iter_entries():
|
||||
try:
|
||||
entry.setup()
|
||||
except Exception as exc:
|
||||
if not self._ignore_entry_error:
|
||||
raise exc
|
||||
|
||||
def show(self):
|
||||
for entry in self._iter_entries():
|
||||
print(entry)
|
||||
getattr(entry, command)()
|
||||
counter["success"] += 1
|
||||
except EntryCommandNotImplementedError:
|
||||
self._logger.warn(f"{command} {entry}")
|
||||
counter["skipped"] += 1
|
||||
except Exception:
|
||||
self._logger.exception(entry)
|
||||
counter["failed"] += 1
|
||||
self._logger.info(f"result {dict(counter)}")
|
||||
return counter
|
||||
|
||||
@@ -13,7 +13,7 @@ class NodejsEntry(Entry):
|
||||
super().__init__(self.TYPE, basepath)
|
||||
|
||||
def setup(self):
|
||||
self._logger.info(f"setup:{self}")
|
||||
self._logger.info(f"setup {self}")
|
||||
NpmCommand.install_clean().run(self._basepath)
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -12,10 +12,22 @@ class PythonEntry(Entry):
|
||||
super().__init__(self.TYPE, basepath)
|
||||
|
||||
def setup(self):
|
||||
self._logger.info(f"setup:{self}")
|
||||
self._logger.info(f"setup {self}")
|
||||
PoetryCommand.env_use("3.8").run(self._basepath)
|
||||
PoetryCommand.install().run(self._basepath)
|
||||
|
||||
def lint(self):
|
||||
self._logger.info(f"lint {self}")
|
||||
PoetryCommand("run", "black", "--check", "--diff", "./").run(self._basepath)
|
||||
PoetryCommand("run", "isort", "--profile", "black", "--check-only", "./").run(
|
||||
self._basepath
|
||||
)
|
||||
|
||||
def format(self):
|
||||
self._logger.info(f"format {self}")
|
||||
PoetryCommand("run", "isort", "./").run(self._basepath)
|
||||
PoetryCommand("run", "black", "./").run(self._basepath)
|
||||
|
||||
@classmethod
|
||||
def detect(cls, basepath: Path) -> Optional["PythonEntry"]:
|
||||
if (basepath / "pyproject.toml").exists():
|
||||
|
||||
@@ -8,6 +8,7 @@ packages = [{ include = "project_manager" }]
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python = "^3.8"
|
||||
termcolor = "^2.2.0"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
black = "^23.1.0"
|
||||
|
||||
Reference in New Issue
Block a user