os:
- "ubuntu-latest"
python-version:
- - "3.10"
+ - "3.11"
sqlalchemy:
- sqla13
- sqla14
os:
- "ubuntu-latest"
python-version:
- - "3.10"
+ - "3.11"
fail-fast: false
- "3.8"
- "3.9"
- "3.10"
+ - "3.11"
sqlalchemy:
- sqla13
- sqla14
python-version:
- "3.9"
- "3.10"
+ - "3.11"
fail-fast: false
from . import context
from . import op
-__version__ = "1.10.5"
+__version__ = "1.11.0"
from .runtime.environment import ProcessRevisionDirectiveFn
-def list_templates(config):
+def list_templates(config: Config):
"""List available templates.
:param config: a :class:`.Config` object.
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
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,
)
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
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`"""
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')
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):
@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):
"""
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:
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"}}
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)
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
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
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
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(
timezone=config.get_main_option("timezone"),
hook_config=config.get_section("post_write_hooks", {}),
recursive_version_locations=rvl,
+ messaging_opts=config.messaging_opts,
)
@contextmanager
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:
from typing import Callable
from typing import Dict
from typing import List
+from typing import Mapping
from typing import Optional
from typing import Union
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.
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
"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]:
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
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
from .compat import inspect_getfullargspec
+EMPTY_DICT: Mapping[Any, Any] = immutabledict()
_T = TypeVar("_T")
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
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):
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)
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:
--- /dev/null
+.. change::
+ :tags: usecase, commands
+
+ Added quiet option to the command line, using the ``-q/--quiet``
+ option. This flag will prevent alembic from logging anything
+ to stdout.
mock.call(
os.path.abspath(os.path.join(path, "__init__.py")), "w"
),
- mock.call().close(),
+ mock.call().__enter__(),
+ mock.call().__exit__(None, None, None),
mock.call(
os.path.abspath(
os.path.join(path, "versions", "__init__.py")
),
"w",
),
- mock.call().close(),
+ mock.call().__enter__(),
+ mock.call().__exit__(None, None, None),
],
)
import datetime
import os
+from pathlib import Path
import re
from unittest.mock import patch
).replace("/", ":NORM:"),
)
- def test_script_location_muliple(self):
+ def test_script_location_multiple(self):
config = _multi_dir_testing_config()
script = ScriptDirectory.from_config(config)
- def normpath(path):
+ def _normpath(path):
return path.replace("/", ":NORM:")
- normpath = mock.Mock(side_effect=normpath)
+ normpath = mock.Mock(side_effect=_normpath)
with mock.patch("os.path.normpath", normpath):
+ sd = Path(_get_staging_directory()).as_posix()
eq_(
script._version_locations,
[
- os.path.abspath(
- os.path.join(_get_staging_directory(), "model1/")
- ).replace("/", ":NORM:"),
- os.path.abspath(
- os.path.join(_get_staging_directory(), "model2/")
- ).replace("/", ":NORM:"),
- os.path.abspath(
- os.path.join(_get_staging_directory(), "model3/")
- ).replace("/", ":NORM:"),
+ _normpath(os.path.abspath(sd + "/model1/")),
+ _normpath(os.path.abspath(sd + "/model2/")),
+ _normpath(os.path.abspath(sd + "/model3/")),
],
)
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
