diff options
Diffstat (limited to 'alembic')
-rw-r--r-- | alembic/__init__.py | 2 | ||||
-rw-r--r-- | alembic/command.py | 37 | ||||
-rw-r--r-- | alembic/config.py | 42 | ||||
-rw-r--r-- | alembic/script/base.py | 34 | ||||
-rw-r--r-- | alembic/script/write_hooks.py | 15 | ||||
-rw-r--r-- | alembic/util/__init__.py | 1 | ||||
-rw-r--r-- | alembic/util/langhelpers.py | 2 | ||||
-rw-r--r-- | alembic/util/messaging.py | 35 |
8 files changed, 107 insertions, 61 deletions
diff --git a/alembic/__init__.py b/alembic/__init__.py index b069e1a..4c6b97b 100644 --- a/alembic/__init__.py +++ b/alembic/__init__.py @@ -3,4 +3,4 @@ import sys from . import context from . import op -__version__ = "1.10.5" +__version__ = "1.11.0" diff --git a/alembic/command.py b/alembic/command.py index a8e5657..a015be3 100644 --- a/alembic/command.py +++ b/alembic/command.py @@ -17,7 +17,7 @@ if TYPE_CHECKING: from .runtime.environment import ProcessRevisionDirectiveFn -def list_templates(config): +def list_templates(config: Config): """List available templates. :param config: a :class:`.Config` object. @@ -69,28 +69,32 @@ def init( raise util.CommandError("No such template %r" % template) if not os.access(directory, os.F_OK): - util.status( - "Creating directory %s" % os.path.abspath(directory), - os.makedirs, - directory, - ) + with util.status( + f"Creating directory {os.path.abspath(directory)!r}", + **config.messaging_opts, + ): + os.makedirs(directory) versions = os.path.join(directory, "versions") - util.status( - "Creating directory %s" % os.path.abspath(versions), - os.makedirs, - versions, - ) + with util.status( + f"Creating directory {os.path.abspath(versions)!r}", + **config.messaging_opts, + ): + os.makedirs(versions) script = ScriptDirectory(directory) + config_file: str | None = None for file_ in os.listdir(template_dir): file_path = os.path.join(template_dir, file_) if file_ == "alembic.ini.mako": assert config.config_file_name is not None config_file = os.path.abspath(config.config_file_name) if os.access(config_file, os.F_OK): - util.msg("File %s already exists, skipping" % config_file) + util.msg( + f"File {config_file!r} already exists, skipping", + **config.messaging_opts, + ) else: script._generate_template( file_path, config_file, script_location=directory @@ -104,12 +108,15 @@ def init( os.path.join(os.path.abspath(directory), "__init__.py"), os.path.join(os.path.abspath(versions), "__init__.py"), ]: - file_ = util.status("Adding %s" % path, open, path, "w") - file_.close() # type:ignore[attr-defined] + with util.status("Adding {path!r}", **config.messaging_opts): + with open(path, "w"): + pass + assert config_file is not None util.msg( "Please edit configuration/connection/logging " - "settings in %r before proceeding." % config_file + f"settings in {config_file!r} before proceeding.", + **config.messaging_opts, ) diff --git a/alembic/config.py b/alembic/config.py index 338769b..2968f0c 100644 --- a/alembic/config.py +++ b/alembic/config.py @@ -6,12 +6,16 @@ from configparser import ConfigParser import inspect import os import sys -from typing import Dict +from typing import Any +from typing import cast +from typing import Mapping from typing import Optional from typing import overload from typing import TextIO from typing import Union +from typing_extensions import TypedDict + from . import __version__ from . import command from . import util @@ -99,7 +103,7 @@ class Config: output_buffer: Optional[TextIO] = None, stdout: TextIO = sys.stdout, cmd_opts: Optional[Namespace] = None, - config_args: util.immutabledict = util.immutabledict(), + config_args: Mapping[str, Any] = util.immutabledict(), attributes: Optional[dict] = None, ) -> None: """Construct a new :class:`.Config`""" @@ -162,6 +166,8 @@ class Config: those arguments will formatted against the provided text, otherwise we simply output the provided text verbatim. + This is a no-op when the``quiet`` messaging option is enabled. + e.g.:: >>> config.print_stdout('Some text %s', 'arg') @@ -174,7 +180,7 @@ class Config: else: output = str(text) - util.write_outstream(self.stdout, output, "\n") + util.write_outstream(self.stdout, output, "\n", **self.messaging_opts) @util.memoized_property def file_config(self): @@ -213,14 +219,14 @@ class Config: @overload def get_section( - self, name: str, default: Dict[str, str] - ) -> Dict[str, str]: + self, name: str, default: Mapping[str, str] + ) -> Mapping[str, str]: ... @overload def get_section( - self, name: str, default: Optional[Dict[str, str]] = ... - ) -> Optional[Dict[str, str]]: + self, name: str, default: Optional[Mapping[str, str]] = ... + ) -> Optional[Mapping[str, str]]: ... def get_section(self, name: str, default=None): @@ -311,6 +317,20 @@ class Config: """ return self.get_section_option(self.config_ini_section, name, default) + @util.memoized_property + def messaging_opts(self) -> MessagingOptions: + """The messaging options.""" + return cast( + MessagingOptions, + util.immutabledict( + {"quiet": getattr(self.cmd_opts, "quiet", False)} + ), + ) + + +class MessagingOptions(TypedDict, total=False): + quiet: bool + class CommandLine: def __init__(self, prog: Optional[str] = None) -> None: @@ -512,6 +532,12 @@ class CommandLine: action="store_true", help="Raise a full stack trace on error", ) + parser.add_argument( + "-q", + "--quiet", + action="store_true", + help="Do not log to std output.", + ) subparsers = parser.add_subparsers() positional_translations = {command.stamp: {"revision": "revisions"}} @@ -568,7 +594,7 @@ class CommandLine: if options.raiseerr: raise else: - util.err(str(e)) + util.err(str(e), **config.messaging_opts) def main(self, argv=None): options = self.parser.parse_args(argv) diff --git a/alembic/script/base.py b/alembic/script/base.py index b6858b5..2440865 100644 --- a/alembic/script/base.py +++ b/alembic/script/base.py @@ -9,9 +9,9 @@ import sys from types import ModuleType from typing import Any from typing import cast -from typing import Dict from typing import Iterator from typing import List +from typing import Mapping from typing import Optional from typing import Sequence from typing import Set @@ -27,6 +27,7 @@ from ..util import not_none if TYPE_CHECKING: from ..config import Config + from ..config import MessagingOptions from ..runtime.migration import RevisionStep from ..runtime.migration import StampStep from ..script.revision import Revision @@ -79,8 +80,11 @@ class ScriptDirectory: sourceless: bool = False, output_encoding: str = "utf-8", timezone: Optional[str] = None, - hook_config: Optional[Dict[str, str]] = None, + hook_config: Optional[Mapping[str, str]] = None, recursive_version_locations: bool = False, + messaging_opts: MessagingOptions = cast( + "MessagingOptions", util.EMPTY_DICT + ), ) -> None: self.dir = dir self.file_template = file_template @@ -92,6 +96,7 @@ class ScriptDirectory: self.timezone = timezone self.hook_config = hook_config self.recursive_version_locations = recursive_version_locations + self.messaging_opts = messaging_opts if not os.access(dir, os.F_OK): raise util.CommandError( @@ -225,6 +230,7 @@ class ScriptDirectory: timezone=config.get_main_option("timezone"), hook_config=config.get_section("post_write_hooks", {}), recursive_version_locations=rvl, + messaging_opts=config.messaging_opts, ) @contextmanager @@ -580,24 +586,24 @@ class ScriptDirectory: return os.path.abspath(os.path.join(self.dir, "env.py")) def _generate_template(self, src: str, dest: str, **kw: Any) -> None: - util.status( - "Generating %s" % os.path.abspath(dest), - util.template_to_file, - src, - dest, - self.output_encoding, - **kw, - ) + with util.status( + f"Generating {os.path.abspath(dest)}", **self.messaging_opts + ): + util.template_to_file(src, dest, self.output_encoding, **kw) def _copy_file(self, src: str, dest: str) -> None: - util.status( - "Generating %s" % os.path.abspath(dest), shutil.copy, src, dest - ) + with util.status( + f"Generating {os.path.abspath(dest)}", **self.messaging_opts + ): + shutil.copy(src, dest) def _ensure_directory(self, path: str) -> None: path = os.path.abspath(path) if not os.path.exists(path): - util.status("Creating directory %s" % path, os.makedirs, path) + with util.status( + f"Creating directory {path}", **self.messaging_opts + ): + os.makedirs(path) def _generate_create_date(self) -> datetime.datetime: if self.timezone is not None: diff --git a/alembic/script/write_hooks.py b/alembic/script/write_hooks.py index 8bc7ac1..d37555d 100644 --- a/alembic/script/write_hooks.py +++ b/alembic/script/write_hooks.py @@ -7,6 +7,7 @@ from typing import Any from typing import Callable from typing import Dict from typing import List +from typing import Mapping from typing import Optional from typing import Union @@ -41,7 +42,7 @@ def register(name: str) -> Callable: def _invoke( - name: str, revision: str, options: Dict[str, Union[str, int]] + name: str, revision: str, options: Mapping[str, Union[str, int]] ) -> Any: """Invokes the formatter registered for the given name. @@ -61,7 +62,7 @@ def _invoke( return hook(revision, options) -def _run_hooks(path: str, hook_config: Dict[str, str]) -> None: +def _run_hooks(path: str, hook_config: Mapping[str, str]) -> None: """Invoke hooks for a generated revision.""" from .base import _split_on_space_comma @@ -84,14 +85,8 @@ def _run_hooks(path: str, hook_config: Dict[str, str]) -> None: "Key %s.type is required for post write hook %r" % (name, name) ) from ke else: - util.status( - 'Running post write hook "%s"' % name, - _invoke, - type_, - path, - opts, - newline=True, - ) + with util.status("Running post write hook {name!r}", newline=True): + _invoke(type_, path, opts) def _parse_cmdline_options(cmdline_options_str: str, path: str) -> List[str]: diff --git a/alembic/util/__init__.py b/alembic/util/__init__.py index c81d431..8f684ab 100644 --- a/alembic/util/__init__.py +++ b/alembic/util/__init__.py @@ -5,6 +5,7 @@ from .langhelpers import _with_legacy_names from .langhelpers import asbool from .langhelpers import dedupe_tuple from .langhelpers import Dispatcher +from .langhelpers import EMPTY_DICT from .langhelpers import immutabledict from .langhelpers import memoized_property from .langhelpers import ModuleClsProxy diff --git a/alembic/util/langhelpers.py b/alembic/util/langhelpers.py index 8203358..f62bc19 100644 --- a/alembic/util/langhelpers.py +++ b/alembic/util/langhelpers.py @@ -7,6 +7,7 @@ from typing import Any from typing import Callable from typing import Dict from typing import List +from typing import Mapping from typing import Optional from typing import overload from typing import Sequence @@ -25,6 +26,7 @@ from sqlalchemy.util import unique_list # noqa from .compat import inspect_getfullargspec +EMPTY_DICT: Mapping[Any, Any] = immutabledict() _T = TypeVar("_T") diff --git a/alembic/util/messaging.py b/alembic/util/messaging.py index 7d9d090..35592c0 100644 --- a/alembic/util/messaging.py +++ b/alembic/util/messaging.py @@ -1,11 +1,10 @@ from __future__ import annotations from collections.abc import Iterable +from contextlib import contextmanager import logging import sys import textwrap -from typing import Any -from typing import Callable from typing import Optional from typing import TextIO from typing import Union @@ -34,7 +33,11 @@ except (ImportError, OSError): TERMWIDTH = None -def write_outstream(stream: TextIO, *text) -> None: +def write_outstream( + stream: TextIO, *text: Union[str, bytes], quiet: bool = False +) -> None: + if quiet: + return encoding = getattr(stream, "encoding", "ascii") or "ascii" for t in text: if not isinstance(t, bytes): @@ -49,21 +52,23 @@ def write_outstream(stream: TextIO, *text) -> None: break -def status(_statmsg: str, fn: Callable, *arg, **kw) -> Any: - newline = kw.pop("newline", False) - msg(_statmsg + " ...", newline, True) +@contextmanager +def status(status_msg: str, newline: bool = False, quiet: bool = False): + msg(status_msg + " ...", newline, flush=True, quiet=quiet) try: - ret = fn(*arg, **kw) - write_outstream(sys.stdout, " done\n") - return ret + yield except: - write_outstream(sys.stdout, " FAILED\n") + if not quiet: + write_outstream(sys.stdout, " FAILED\n") raise + else: + if not quiet: + write_outstream(sys.stdout, " done\n") -def err(message: str): +def err(message: str, quiet: bool = False): log.error(message) - msg("FAILED: %s" % message) + msg(f"FAILED: {message}", quiet=quiet) sys.exit(-1) @@ -76,7 +81,11 @@ def warn(msg: str, stacklevel: int = 2) -> None: warnings.warn(msg, UserWarning, stacklevel=stacklevel) -def msg(msg: str, newline: bool = True, flush: bool = False) -> None: +def msg( + msg: str, newline: bool = True, flush: bool = False, quiet: bool = False +) -> None: + if quiet: + return if TERMWIDTH is None: write_outstream(sys.stdout, msg) if newline: |