codekingpro/portable-devtools

1from __future__ import annotations2 3from contextlib import contextmanager4import datetime5import os6import re7import shutil8import sys9from types import ModuleType10from typing import Any11from typing import cast12from typing import Iterator13from typing import List14from typing import Mapping15from typing import Optional16from typing import Sequence17from typing import Set18from typing import Tuple19from typing import TYPE_CHECKING20from typing import Union21 22from . import revision23from . import write_hooks24from .. import util25from ..runtime import migration26from ..util import compat27from ..util import not_none28 29if TYPE_CHECKING:30    from .revision import _GetRevArg31    from .revision import _RevIdType32    from .revision import Revision33    from ..config import Config34    from ..config import MessagingOptions35    from ..runtime.migration import RevisionStep36    from ..runtime.migration import StampStep37 38try:39    if compat.py39:40        from zoneinfo import ZoneInfo41        from zoneinfo import ZoneInfoNotFoundError42    else:43        from backports.zoneinfo import ZoneInfo  # type: ignore[import-not-found,no-redef] # noqa: E50144        from backports.zoneinfo import ZoneInfoNotFoundError  # type: ignore[no-redef] # noqa: E50145except ImportError:46    ZoneInfo = None  # type: ignore[assignment, misc]47 48_sourceless_rev_file = re.compile(r"(?!\.\#|__init__)(.*\.py)(c|o)?$")49_only_source_rev_file = re.compile(r"(?!\.\#|__init__)(.*\.py)$")50_legacy_rev = re.compile(r"([a-f0-9]+)\.py$")51_slug_re = re.compile(r"\w+")52_default_file_template = "%(rev)s_%(slug)s"53_split_on_space_comma = re.compile(r", *|(?: +)")54 55_split_on_space_comma_colon = re.compile(r", *|(?: +)|\:")56 57 58class ScriptDirectory:59 60    """Provides operations upon an Alembic script directory.61 62    This object is useful to get information as to current revisions,63    most notably being able to get at the "head" revision, for schemes64    that want to test if the current revision in the database is the most65    recent::66 67        from alembic.script import ScriptDirectory68        from alembic.config import Config69        config = Config()70        config.set_main_option("script_location", "myapp:migrations")71        script = ScriptDirectory.from_config(config)72 73        head_revision = script.get_current_head()74 75 76 77    """78 79    def __init__(80        self,81        dir: str,  # noqa82        file_template: str = _default_file_template,83        truncate_slug_length: Optional[int] = 40,84        version_locations: Optional[List[str]] = None,85        sourceless: bool = False,86        output_encoding: str = "utf-8",87        timezone: Optional[str] = None,88        hook_config: Optional[Mapping[str, str]] = None,89        recursive_version_locations: bool = False,90        messaging_opts: MessagingOptions = cast(91            "MessagingOptions", util.EMPTY_DICT92        ),93    ) -> None:94        self.dir = dir95        self.file_template = file_template96        self.version_locations = version_locations97        self.truncate_slug_length = truncate_slug_length or 4098        self.sourceless = sourceless99        self.output_encoding = output_encoding100        self.revision_map = revision.RevisionMap(self._load_revisions)101        self.timezone = timezone102        self.hook_config = hook_config103        self.recursive_version_locations = recursive_version_locations104        self.messaging_opts = messaging_opts105 106        if not os.access(dir, os.F_OK):107            raise util.CommandError(108                "Path doesn't exist: %r.  Please use "109                "the 'init' command to create a new "110                "scripts folder." % os.path.abspath(dir)111            )112 113    @property114    def versions(self) -> str:115        loc = self._version_locations116        if len(loc) > 1:117            raise util.CommandError("Multiple version_locations present")118        else:119            return loc[0]120 121    @util.memoized_property122    def _version_locations(self) -> Sequence[str]:123        if self.version_locations:124            return [125                os.path.abspath(util.coerce_resource_to_filename(location))126                for location in self.version_locations127            ]128        else:129            return (os.path.abspath(os.path.join(self.dir, "versions")),)130 131    def _load_revisions(self) -> Iterator[Script]:132        if self.version_locations:133            paths = [134                vers135                for vers in self._version_locations136                if os.path.exists(vers)137            ]138        else:139            paths = [self.versions]140 141        dupes = set()142        for vers in paths:143            for file_path in Script._list_py_dir(self, vers):144                real_path = os.path.realpath(file_path)145                if real_path in dupes:146                    util.warn(147                        "File %s loaded twice! ignoring. Please ensure "148                        "version_locations is unique." % real_path149                    )150                    continue151                dupes.add(real_path)152 153                filename = os.path.basename(real_path)154                dir_name = os.path.dirname(real_path)155                script = Script._from_filename(self, dir_name, filename)156                if script is None:157                    continue158                yield script159 160    @classmethod161    def from_config(cls, config: Config) -> ScriptDirectory:162        """Produce a new :class:`.ScriptDirectory` given a :class:`.Config`163        instance.164 165        The :class:`.Config` need only have the ``script_location`` key166        present.167 168        """169        script_location = config.get_main_option("script_location")170        if script_location is None:171            raise util.CommandError(172                "No 'script_location' key " "found in configuration."173            )174        truncate_slug_length: Optional[int]175        tsl = config.get_main_option("truncate_slug_length")176        if tsl is not None:177            truncate_slug_length = int(tsl)178        else:179            truncate_slug_length = None180 181        version_locations_str = config.get_main_option("version_locations")182        version_locations: Optional[List[str]]183        if version_locations_str:184            version_path_separator = config.get_main_option(185                "version_path_separator"186            )187 188            split_on_path = {189                None: None,190                "space": " ",191                "os": os.pathsep,192                ":": ":",193                ";": ";",194            }195 196            try:197                split_char: Optional[str] = split_on_path[198                    version_path_separator199                ]200            except KeyError as ke:201                raise ValueError(202                    "'%s' is not a valid value for "203                    "version_path_separator; "204                    "expected 'space', 'os', ':', ';'" % version_path_separator205                ) from ke206            else:207                if split_char is None:208                    # legacy behaviour for backwards compatibility209                    version_locations = _split_on_space_comma.split(210                        version_locations_str211                    )212                else:213                    version_locations = [214                        x for x in version_locations_str.split(split_char) if x215                    ]216        else:217            version_locations = None218 219        prepend_sys_path = config.get_main_option("prepend_sys_path")220        if prepend_sys_path:221            sys.path[:0] = list(222                _split_on_space_comma_colon.split(prepend_sys_path)223            )224 225        rvl = config.get_main_option("recursive_version_locations") == "true"226        return ScriptDirectory(227            util.coerce_resource_to_filename(script_location),228            file_template=config.get_main_option(229                "file_template", _default_file_template230            ),231            truncate_slug_length=truncate_slug_length,232            sourceless=config.get_main_option("sourceless") == "true",233            output_encoding=config.get_main_option("output_encoding", "utf-8"),234            version_locations=version_locations,235            timezone=config.get_main_option("timezone"),236            hook_config=config.get_section("post_write_hooks", {}),237            recursive_version_locations=rvl,238            messaging_opts=config.messaging_opts,239        )240 241    @contextmanager242    def _catch_revision_errors(243        self,244        ancestor: Optional[str] = None,245        multiple_heads: Optional[str] = None,246        start: Optional[str] = None,247        end: Optional[str] = None,248        resolution: Optional[str] = None,249    ) -> Iterator[None]:250        try:251            yield252        except revision.RangeNotAncestorError as rna:253            if start is None:254                start = cast(Any, rna.lower)255            if end is None:256                end = cast(Any, rna.upper)257            if not ancestor:258                ancestor = (259                    "Requested range %(start)s:%(end)s does not refer to "260                    "ancestor/descendant revisions along the same branch"261                )262            ancestor = ancestor % {"start": start, "end": end}263            raise util.CommandError(ancestor) from rna264        except revision.MultipleHeads as mh:265            if not multiple_heads:266                multiple_heads = (267                    "Multiple head revisions are present for given "268                    "argument '%(head_arg)s'; please "269                    "specify a specific target revision, "270                    "'<branchname>@%(head_arg)s' to "271                    "narrow to a specific head, or 'heads' for all heads"272                )273            multiple_heads = multiple_heads % {274                "head_arg": end or mh.argument,275                "heads": util.format_as_comma(mh.heads),276            }277            raise util.CommandError(multiple_heads) from mh278        except revision.ResolutionError as re:279            if resolution is None:280                resolution = "Can't locate revision identified by '%s'" % (281                    re.argument282                )283            raise util.CommandError(resolution) from re284        except revision.RevisionError as err:285            raise util.CommandError(err.args[0]) from err286 287    def walk_revisions(288        self, base: str = "base", head: str = "heads"289    ) -> Iterator[Script]:290        """Iterate through all revisions.291 292        :param base: the base revision, or "base" to start from the293         empty revision.294 295        :param head: the head revision; defaults to "heads" to indicate296         all head revisions.  May also be "head" to indicate a single297         head revision.298 299        """300        with self._catch_revision_errors(start=base, end=head):301            for rev in self.revision_map.iterate_revisions(302                head, base, inclusive=True, assert_relative_length=False303            ):304                yield cast(Script, rev)305 306    def get_revisions(self, id_: _GetRevArg) -> Tuple[Script, ...]:307        """Return the :class:`.Script` instance with the given rev identifier,308        symbolic name, or sequence of identifiers.309 310        """311        with self._catch_revision_errors():312            return cast(313                Tuple[Script, ...],314                self.revision_map.get_revisions(id_),315            )316 317    def get_all_current(self, id_: Tuple[str, ...]) -> Set[Script]:318        with self._catch_revision_errors():319            return cast(Set[Script], self.revision_map._get_all_current(id_))320 321    def get_revision(self, id_: str) -> Script:322        """Return the :class:`.Script` instance with the given rev id.323 324        .. seealso::325 326            :meth:`.ScriptDirectory.get_revisions`327 328        """329 330        with self._catch_revision_errors():331            return cast(Script, self.revision_map.get_revision(id_))332 333    def as_revision_number(334        self, id_: Optional[str]335    ) -> Optional[Union[str, Tuple[str, ...]]]:336        """Convert a symbolic revision, i.e. 'head' or 'base', into337        an actual revision number."""338 339        with self._catch_revision_errors():340            rev, branch_name = self.revision_map._resolve_revision_number(id_)341 342        if not rev:343            # convert () to None344            return None345        elif id_ == "heads":346            return rev347        else:348            return rev[0]349 350    def iterate_revisions(351        self,352        upper: Union[str, Tuple[str, ...], None],353        lower: Union[str, Tuple[str, ...], None],354        **kw: Any,355    ) -> Iterator[Script]:356        """Iterate through script revisions, starting at the given357        upper revision identifier and ending at the lower.358 359        The traversal uses strictly the `down_revision`360        marker inside each migration script, so361        it is a requirement that upper >= lower,362        else you'll get nothing back.363 364        The iterator yields :class:`.Script` objects.365 366        .. seealso::367 368            :meth:`.RevisionMap.iterate_revisions`369 370        """371        return cast(372            Iterator[Script],373            self.revision_map.iterate_revisions(upper, lower, **kw),374        )375 376    def get_current_head(self) -> Optional[str]:377        """Return the current head revision.378 379        If the script directory has multiple heads380        due to branching, an error is raised;381        :meth:`.ScriptDirectory.get_heads` should be382        preferred.383 384        :return: a string revision number.385 386        .. seealso::387 388            :meth:`.ScriptDirectory.get_heads`389 390        """391        with self._catch_revision_errors(392            multiple_heads=(393                "The script directory has multiple heads (due to branching)."394                "Please use get_heads(), or merge the branches using "395                "alembic merge."396            )397        ):398            return self.revision_map.get_current_head()399 400    def get_heads(self) -> List[str]:401        """Return all "versioned head" revisions as strings.402 403        This is normally a list of length one,404        unless branches are present.  The405        :meth:`.ScriptDirectory.get_current_head()` method406        can be used normally when a script directory407        has only one head.408 409        :return: a tuple of string revision numbers.410        """411        return list(self.revision_map.heads)412 413    def get_base(self) -> Optional[str]:414        """Return the "base" revision as a string.415 416        This is the revision number of the script that417        has a ``down_revision`` of None.418 419        If the script directory has multiple bases, an error is raised;420        :meth:`.ScriptDirectory.get_bases` should be421        preferred.422 423        """424        bases = self.get_bases()425        if len(bases) > 1:426            raise util.CommandError(427                "The script directory has multiple bases. "428                "Please use get_bases()."429            )430        elif bases:431            return bases[0]432        else:433            return None434 435    def get_bases(self) -> List[str]:436        """return all "base" revisions as strings.437 438        This is the revision number of all scripts that439        have a ``down_revision`` of None.440 441        """442        return list(self.revision_map.bases)443 444    def _upgrade_revs(445        self, destination: str, current_rev: str446    ) -> List[RevisionStep]:447        with self._catch_revision_errors(448            ancestor="Destination %(end)s is not a valid upgrade "449            "target from current head(s)",450            end=destination,451        ):452            revs = self.iterate_revisions(453                destination, current_rev, implicit_base=True454            )455            return [456                migration.MigrationStep.upgrade_from_script(457                    self.revision_map, script458                )459                for script in reversed(list(revs))460            ]461 462    def _downgrade_revs(463        self, destination: str, current_rev: Optional[str]464    ) -> List[RevisionStep]:465        with self._catch_revision_errors(466            ancestor="Destination %(end)s is not a valid downgrade "467            "target from current head(s)",468            end=destination,469        ):470            revs = self.iterate_revisions(471                current_rev, destination, select_for_downgrade=True472            )473            return [474                migration.MigrationStep.downgrade_from_script(475                    self.revision_map, script476                )477                for script in revs478            ]479 480    def _stamp_revs(481        self, revision: _RevIdType, heads: _RevIdType482    ) -> List[StampStep]:483        with self._catch_revision_errors(484            multiple_heads="Multiple heads are present; please specify a "485            "single target revision"486        ):487            heads_revs = self.get_revisions(heads)488 489            steps = []490 491            if not revision:492                revision = "base"493 494            filtered_heads: List[Script] = []495            for rev in util.to_tuple(revision):496                if rev:497                    filtered_heads.extend(498                        self.revision_map.filter_for_lineage(499                            cast(Sequence[Script], heads_revs),500                            rev,501                            include_dependencies=True,502                        )503                    )504            filtered_heads = util.unique_list(filtered_heads)505 506            dests = self.get_revisions(revision) or [None]507 508            for dest in dests:509                if dest is None:510                    # dest is 'base'.  Return a "delete branch" migration511                    # for all applicable heads.512                    steps.extend(513                        [514                            migration.StampStep(515                                head.revision,516                                None,517                                False,518                                True,519                                self.revision_map,520                            )521                            for head in filtered_heads522                        ]523                    )524                    continue525                elif dest in filtered_heads:526                    # the dest is already in the version table, do nothing.527                    continue528 529                # figure out if the dest is a descendant or an530                # ancestor of the selected nodes531                descendants = set(532                    self.revision_map._get_descendant_nodes([dest])533                )534                ancestors = set(self.revision_map._get_ancestor_nodes([dest]))535 536                if descendants.intersection(filtered_heads):537                    # heads are above the target, so this is a downgrade.538                    # we can treat them as a "merge", single step.539                    assert not ancestors.intersection(filtered_heads)540                    todo_heads = [head.revision for head in filtered_heads]541                    step = migration.StampStep(542                        todo_heads,543                        dest.revision,544                        False,545                        False,546                        self.revision_map,547                    )548                    steps.append(step)549                    continue550                elif ancestors.intersection(filtered_heads):551                    # heads are below the target, so this is an upgrade.552                    # we can treat them as a "merge", single step.553                    todo_heads = [head.revision for head in filtered_heads]554                    step = migration.StampStep(555                        todo_heads,556                        dest.revision,557                        True,558                        False,559                        self.revision_map,560                    )561                    steps.append(step)562                    continue563                else:564                    # destination is in a branch not represented,565                    # treat it as new branch566                    step = migration.StampStep(567                        (), dest.revision, True, True, self.revision_map568                    )569                    steps.append(step)570                    continue571 572            return steps573 574    def run_env(self) -> None:575        """Run the script environment.576 577        This basically runs the ``env.py`` script present578        in the migration environment.   It is called exclusively579        by the command functions in :mod:`alembic.command`.580 581 582        """583        util.load_python_file(self.dir, "env.py")584 585    @property586    def env_py_location(self) -> str:587        return os.path.abspath(os.path.join(self.dir, "env.py"))588 589    def _generate_template(self, src: str, dest: str, **kw: Any) -> None:590        with util.status(591            f"Generating {os.path.abspath(dest)}", **self.messaging_opts592        ):593            util.template_to_file(src, dest, self.output_encoding, **kw)594 595    def _copy_file(self, src: str, dest: str) -> None:596        with util.status(597            f"Generating {os.path.abspath(dest)}", **self.messaging_opts598        ):599            shutil.copy(src, dest)600 601    def _ensure_directory(self, path: str) -> None:602        path = os.path.abspath(path)603        if not os.path.exists(path):604            with util.status(605                f"Creating directory {path}", **self.messaging_opts606            ):607                os.makedirs(path)608 609    def _generate_create_date(self) -> datetime.datetime:610        if self.timezone is not None:611            if ZoneInfo is None:612                raise util.CommandError(613                    "Python >= 3.9 is required for timezone support or"614                    "the 'backports.zoneinfo' package must be installed."615                )616            # First, assume correct capitalization617            try:618                tzinfo = ZoneInfo(self.timezone)619            except ZoneInfoNotFoundError:620                tzinfo = None621            if tzinfo is None:622                try:623                    tzinfo = ZoneInfo(self.timezone.upper())624                except ZoneInfoNotFoundError:625                    raise util.CommandError(626                        "Can't locate timezone: %s" % self.timezone627                    ) from None628            create_date = (629                datetime.datetime.utcnow()630                .replace(tzinfo=datetime.timezone.utc)631                .astimezone(tzinfo)632            )633        else:634            create_date = datetime.datetime.now()635        return create_date636 637    def generate_revision(638        self,639        revid: str,640        message: Optional[str],641        head: Optional[_RevIdType] = None,642        splice: Optional[bool] = False,643        branch_labels: Optional[_RevIdType] = None,644        version_path: Optional[str] = None,645        depends_on: Optional[_RevIdType] = None,646        **kw: Any,647    ) -> Optional[Script]:648        """Generate a new revision file.649 650        This runs the ``script.py.mako`` template, given651        template arguments, and creates a new file.652 653        :param revid: String revision id.  Typically this654         comes from ``alembic.util.rev_id()``.655        :param message: the revision message, the one passed656         by the -m argument to the ``revision`` command.657        :param head: the head revision to generate against.  Defaults658         to the current "head" if no branches are present, else raises659         an exception.660        :param splice: if True, allow the "head" version to not be an661         actual head; otherwise, the selected head must be a head662         (e.g. endpoint) revision.663 664        """665        if head is None:666            head = "head"667 668        try:669            Script.verify_rev_id(revid)670        except revision.RevisionError as err:671            raise util.CommandError(err.args[0]) from err672 673        with self._catch_revision_errors(674            multiple_heads=(675                "Multiple heads are present; please specify the head "676                "revision on which the new revision should be based, "677                "or perform a merge."678            )679        ):680            heads = cast(681                Tuple[Optional["Revision"], ...],682                self.revision_map.get_revisions(head),683            )684            for h in heads:685                assert h != "base"  # type: ignore[comparison-overlap]686 687        if len(set(heads)) != len(heads):688            raise util.CommandError("Duplicate head revisions specified")689 690        create_date = self._generate_create_date()691 692        if version_path is None:693            if len(self._version_locations) > 1:694                for head_ in heads:695                    if head_ is not None:696                        assert isinstance(head_, Script)697                        version_path = os.path.dirname(head_.path)698                        break699                else:700                    raise util.CommandError(701                        "Multiple version locations present, "702                        "please specify --version-path"703                    )704            else:705                version_path = self.versions706 707        norm_path = os.path.normpath(os.path.abspath(version_path))708        for vers_path in self._version_locations:709            if os.path.normpath(vers_path) == norm_path:710                break711        else:712            raise util.CommandError(713                "Path %s is not represented in current "714                "version locations" % version_path715            )716 717        if self.version_locations:718            self._ensure_directory(version_path)719 720        path = self._rev_path(version_path, revid, message, create_date)721 722        if not splice:723            for head_ in heads:724                if head_ is not None and not head_.is_head:725                    raise util.CommandError(726                        "Revision %s is not a head revision; please specify "727                        "--splice to create a new branch from this revision"728                        % head_.revision729                    )730 731        resolved_depends_on: Optional[List[str]]732        if depends_on:733            with self._catch_revision_errors():734                resolved_depends_on = [735                    dep736                    if dep in rev.branch_labels  # maintain branch labels737                    else rev.revision  # resolve partial revision identifiers738                    for rev, dep in [739                        (not_none(self.revision_map.get_revision(dep)), dep)740                        for dep in util.to_list(depends_on)741                    ]742                ]743        else:744            resolved_depends_on = None745 746        self._generate_template(747            os.path.join(self.dir, "script.py.mako"),748            path,749            up_revision=str(revid),750            down_revision=revision.tuple_rev_as_scalar(751                tuple(h.revision if h is not None else None for h in heads)752            ),753            branch_labels=util.to_tuple(branch_labels),754            depends_on=revision.tuple_rev_as_scalar(resolved_depends_on),755            create_date=create_date,756            comma=util.format_as_comma,757            message=message if message is not None else ("empty message"),758            **kw,759        )760 761        post_write_hooks = self.hook_config762        if post_write_hooks:763            write_hooks._run_hooks(path, post_write_hooks)764 765        try:766            script = Script._from_path(self, path)767        except revision.RevisionError as err:768            raise util.CommandError(err.args[0]) from err769        if script is None:770            return None771        if branch_labels and not script.branch_labels:772            raise util.CommandError(773                "Version %s specified branch_labels %s, however the "774                "migration file %s does not have them; have you upgraded "775                "your script.py.mako to include the "776                "'branch_labels' section?"777                % (script.revision, branch_labels, script.path)778            )779        self.revision_map.add_revision(script)780        return script781 782    def _rev_path(783        self,784        path: str,785        rev_id: str,786        message: Optional[str],787        create_date: datetime.datetime,788    ) -> str:789        epoch = int(create_date.timestamp())790        slug = "_".join(_slug_re.findall(message or "")).lower()791        if len(slug) > self.truncate_slug_length:792            slug = slug[: self.truncate_slug_length].rsplit("_", 1)[0] + "_"793        filename = "%s.py" % (794            self.file_template795            % {796                "rev": rev_id,797                "slug": slug,798                "epoch": epoch,799                "year": create_date.year,800                "month": create_date.month,801                "day": create_date.day,802                "hour": create_date.hour,803                "minute": create_date.minute,804                "second": create_date.second,805            }806        )807        return os.path.join(path, filename)808 809 810class Script(revision.Revision):811 812    """Represent a single revision file in a ``versions/`` directory.813 814    The :class:`.Script` instance is returned by methods815    such as :meth:`.ScriptDirectory.iterate_revisions`.816 817    """818 819    def __init__(self, module: ModuleType, rev_id: str, path: str):820        self.module = module821        self.path = path822        super().__init__(823            rev_id,824            module.down_revision,825            branch_labels=util.to_tuple(826                getattr(module, "branch_labels", None), default=()827            ),828            dependencies=util.to_tuple(829                getattr(module, "depends_on", None), default=()830            ),831        )832 833    module: ModuleType834    """The Python module representing the actual script itself."""835 836    path: str837    """Filesystem path of the script."""838 839    _db_current_indicator: Optional[bool] = None840    """Utility variable which when set will cause string output to indicate841    this is a "current" version in some database"""842 843    @property844    def doc(self) -> str:845        """Return the docstring given in the script."""846 847        return re.split("\n\n", self.longdoc)[0]848 849    @property850    def longdoc(self) -> str:851        """Return the docstring given in the script."""852 853        doc = self.module.__doc__854        if doc:855            if hasattr(self.module, "_alembic_source_encoding"):856                doc = doc.decode(  # type: ignore[attr-defined]857                    self.module._alembic_source_encoding858                )859            return doc.strip()  # type: ignore[union-attr]860        else:861            return ""862 863    @property864    def log_entry(self) -> str:865        entry = "Rev: %s%s%s%s%s\n" % (866            self.revision,867            " (head)" if self.is_head else "",868            " (branchpoint)" if self.is_branch_point else "",869            " (mergepoint)" if self.is_merge_point else "",870            " (current)" if self._db_current_indicator else "",871        )872        if self.is_merge_point:873            entry += "Merges: %s\n" % (self._format_down_revision(),)874        else:875            entry += "Parent: %s\n" % (self._format_down_revision(),)876 877        if self.dependencies:878            entry += "Also depends on: %s\n" % (879                util.format_as_comma(self.dependencies)880            )881 882        if self.is_branch_point:883            entry += "Branches into: %s\n" % (884                util.format_as_comma(self.nextrev)885            )886 887        if self.branch_labels:888            entry += "Branch names: %s\n" % (889                util.format_as_comma(self.branch_labels),890            )891 892        entry += "Path: %s\n" % (self.path,)893 894        entry += "\n%s\n" % (895            "\n".join("    %s" % para for para in self.longdoc.splitlines())896        )897        return entry898 899    def __str__(self) -> str:900        return "%s -> %s%s%s%s, %s" % (901            self._format_down_revision(),902            self.revision,903            " (head)" if self.is_head else "",904            " (branchpoint)" if self.is_branch_point else "",905            " (mergepoint)" if self.is_merge_point else "",906            self.doc,907        )908 909    def _head_only(910        self,911        include_branches: bool = False,912        include_doc: bool = False,913        include_parents: bool = False,914        tree_indicators: bool = True,915        head_indicators: bool = True,916    ) -> str:917        text = self.revision918        if include_parents:919            if self.dependencies:920                text = "%s (%s) -> %s" % (921                    self._format_down_revision(),922                    util.format_as_comma(self.dependencies),923                    text,924                )925            else:926                text = "%s -> %s" % (self._format_down_revision(), text)927        assert text is not None928        if include_branches and self.branch_labels:929            text += " (%s)" % util.format_as_comma(self.branch_labels)930        if head_indicators or tree_indicators:931            text += "%s%s%s" % (932                " (head)" if self._is_real_head else "",933                " (effective head)"934                if self.is_head and not self._is_real_head935                else "",936                " (current)" if self._db_current_indicator else "",937            )938        if tree_indicators:939            text += "%s%s" % (940                " (branchpoint)" if self.is_branch_point else "",941                " (mergepoint)" if self.is_merge_point else "",942            )943        if include_doc:944            text += ", %s" % self.doc945        return text946 947    def cmd_format(948        self,949        verbose: bool,950        include_branches: bool = False,951        include_doc: bool = False,952        include_parents: bool = False,953        tree_indicators: bool = True,954    ) -> str:955        if verbose:956            return self.log_entry957        else:958            return self._head_only(959                include_branches, include_doc, include_parents, tree_indicators960            )961 962    def _format_down_revision(self) -> str:963        if not self.down_revision:964            return "<base>"965        else:966            return util.format_as_comma(self._versioned_down_revisions)967 968    @classmethod969    def _from_path(970        cls, scriptdir: ScriptDirectory, path: str971    ) -> Optional[Script]:972        dir_, filename = os.path.split(path)973        return cls._from_filename(scriptdir, dir_, filename)974 975    @classmethod976    def _list_py_dir(cls, scriptdir: ScriptDirectory, path: str) -> List[str]:977        paths = []978        for root, dirs, files in os.walk(path, topdown=True):979            if root.endswith("__pycache__"):980                # a special case - we may include these files981                # if a `sourceless` option is specified982                continue983 984            for filename in sorted(files):985                paths.append(os.path.join(root, filename))986 987            if scriptdir.sourceless:988                # look for __pycache__989                py_cache_path = os.path.join(root, "__pycache__")990                if os.path.exists(py_cache_path):991                    # add all files from __pycache__ whose filename is not992                    # already in the names we got from the version directory.993                    # add as relative paths including __pycache__ token994                    names = {filename.split(".")[0] for filename in files}995                    paths.extend(996                        os.path.join(py_cache_path, pyc)997                        for pyc in os.listdir(py_cache_path)998                        if pyc.split(".")[0] not in names999                    )1000 1001            if not scriptdir.recursive_version_locations:1002                break1003 1004            # the real script order is defined by revision,1005            # but it may be undefined if there are many files with a same1006            # `down_revision`, for a better user experience (ex. debugging),1007            # we use a deterministic order1008            dirs.sort()1009 1010        return paths1011 1012    @classmethod1013    def _from_filename(1014        cls, scriptdir: ScriptDirectory, dir_: str, filename: str1015    ) -> Optional[Script]:1016        if scriptdir.sourceless:1017            py_match = _sourceless_rev_file.match(filename)1018        else:1019            py_match = _only_source_rev_file.match(filename)1020 1021        if not py_match:1022            return None1023 1024        py_filename = py_match.group(1)1025 1026        if scriptdir.sourceless:1027            is_c = py_match.group(2) == "c"1028            is_o = py_match.group(2) == "o"1029        else:1030            is_c = is_o = False1031 1032        if is_o or is_c:1033            py_exists = os.path.exists(os.path.join(dir_, py_filename))1034            pyc_exists = os.path.exists(os.path.join(dir_, py_filename + "c"))1035 1036            # prefer .py over .pyc because we'd like to get the1037            # source encoding; prefer .pyc over .pyo because we'd like to1038            # have the docstrings which a -OO file would not have1039            if py_exists or is_o and pyc_exists:1040                return None1041 1042        module = util.load_python_file(dir_, filename)1043 1044        if not hasattr(module, "revision"):1045            # attempt to get the revision id from the script name,1046            # this for legacy only1047            m = _legacy_rev.match(filename)1048            if not m:1049                raise util.CommandError(1050                    "Could not determine revision id from filename %s. "1051                    "Be sure the 'revision' variable is "1052                    "declared inside the script (please see 'Upgrading "1053                    "from Alembic 0.1 to 0.2' in the documentation)."1054                    % filename1055                )1056            else:1057                revision = m.group(1)1058        else:1059            revision = module.revision1060        return Script(module, revision, os.path.join(dir_, filename))1061