Merge "Added quiet option to command line" into main

.github/workflows/run-on-pr.yaml

@@ -25,7 +25,7 @@ jobs:
         os:
           - "ubuntu-latest"
         python-version:
-          - "3.10"
+          - "3.11"
         sqlalchemy:
           - sqla13
           - sqla14
@@ -61,7 +61,7 @@ jobs:
         os:
           - "ubuntu-latest"
         python-version:
-          - "3.10"
+          - "3.11"
 
       fail-fast: false
 

.github/workflows/run-test.yaml

@@ -34,6 +34,7 @@ jobs:
           - "3.8"
           - "3.9"
           - "3.10"
+          - "3.11"
         sqlalchemy:
           - sqla13
           - sqla14
@@ -71,6 +72,7 @@ jobs:
         python-version:
           - "3.9"
           - "3.10"
+          - "3.11"
 
       fail-fast: false
 

alembic/__init__.py

@@ -3,4 +3,4 @@
 from . import context
 from . import op
 
-__version__ = "1.10.5"
+__version__ = "1.11.0"

alembic/command.py

@@ -17,7 +17,7 @@
     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,
     )
 
 

alembic/config.py

@@ -6,12 +6,16 @@
 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 @@ def __init__(
         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 @@ def print_stdout(self, text: str, *arg) -> None:
         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 @@ def print_stdout(self, text: str, *arg) -> None:
         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 @@ def get_template_directory(self) -> str:
 
     @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 @@ def get_main_option(self, name, default=None):
         """
         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 @@ def add_options(fn, parser, positional, kwargs):
             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 @@ def run_cmd(self, config: Config, options: Namespace) -> None:
             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)

alembic/script/base.py

@@ -9,9 +9,9 @@
 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 @@
 
 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 @@ def __init__(
         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 @@ def __init__(
         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 @@ def from_config(cls, config: Config) -> 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 @@ def env_py_location(self):
         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:

alembic/script/write_hooks.py

@@ -7,6 +7,7 @@
 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 decorate(fn):
 
 
 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]:

alembic/util/__init__.py

@@ -5,6 +5,7 @@
 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

alembic/util/langhelpers.py

@@ -7,6 +7,7 @@
 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 .compat import inspect_getfullargspec
 
 
+EMPTY_DICT: Mapping[Any, Any] = immutabledict()
 _T = TypeVar("_T")