123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262 |
- """Object specification."""
- from collections.abc import Iterator
- from typing import TYPE_CHECKING, Optional, Union
- from .objects import Commit, ShaFile, Tree
- if TYPE_CHECKING:
- from .refs import Ref, RefsContainer
- from .repo import Repo
- def to_bytes(text: Union[str, bytes]) -> bytes:
- if getattr(text, "encode", None) is not None:
- text = text.encode("ascii")
- return text
- def parse_object(repo: "Repo", objectish: Union[bytes, str]) -> "ShaFile":
- """Parse a string referring to an object.
- Args:
- repo: A `Repo` object
- objectish: A string referring to an object
- Returns: A git object
- Raises:
- KeyError: If the object can not be found
- """
- objectish = to_bytes(objectish)
- return repo[objectish]
- def parse_tree(repo: "Repo", treeish: Union[bytes, str]) -> "Tree":
- """Parse a string referring to a tree.
- Args:
- repo: A `Repo` object
- treeish: A string referring to a tree
- Returns: A git object
- Raises:
- KeyError: If the object can not be found
- """
- treeish = to_bytes(treeish)
- try:
- treeish = parse_ref(repo, treeish)
- except KeyError:
- pass
- o = repo[treeish]
- if o.type_name == b"commit":
- return repo[o.tree]
- return o
- def parse_ref(
- container: Union["Repo", "RefsContainer"], refspec: Union[str, bytes]
- ) -> "Ref":
- """Parse a string referring to a reference.
- Args:
- container: A RefsContainer object
- refspec: A string referring to a ref
- Returns: A ref
- Raises:
- KeyError: If the ref can not be found
- """
- refspec = to_bytes(refspec)
- possible_refs = [
- refspec,
- b"refs/" + refspec,
- b"refs/tags/" + refspec,
- b"refs/heads/" + refspec,
- b"refs/remotes/" + refspec,
- b"refs/remotes/" + refspec + b"/HEAD",
- ]
- for ref in possible_refs:
- if ref in container:
- return ref
- raise KeyError(refspec)
- def parse_reftuple(
- lh_container: Union["Repo", "RefsContainer"],
- rh_container: Union["Repo", "RefsContainer"],
- refspec: Union[str, bytes],
- force: bool = False,
- ) -> tuple[Optional["Ref"], Optional["Ref"], bool]:
- """Parse a reftuple spec.
- Args:
- lh_container: A RefsContainer object
- rh_container: A RefsContainer object
- refspec: A string
- Returns: A tuple with left and right ref
- Raises:
- KeyError: If one of the refs can not be found
- """
- refspec = to_bytes(refspec)
- if refspec.startswith(b"+"):
- force = True
- refspec = refspec[1:]
- lh: Optional[bytes]
- rh: Optional[bytes]
- if b":" in refspec:
- (lh, rh) = refspec.split(b":")
- else:
- lh = rh = refspec
- if lh == b"":
- lh = None
- else:
- lh = parse_ref(lh_container, lh)
- if rh == b"":
- rh = None
- else:
- try:
- rh = parse_ref(rh_container, rh)
- except KeyError:
-
- if b"/" not in rh:
- rh = b"refs/heads/" + rh
- return (lh, rh, force)
- def parse_reftuples(
- lh_container: Union["Repo", "RefsContainer"],
- rh_container: Union["Repo", "RefsContainer"],
- refspecs: Union[bytes, list[bytes]],
- force: bool = False,
- ):
- """Parse a list of reftuple specs to a list of reftuples.
- Args:
- lh_container: A RefsContainer object
- rh_container: A RefsContainer object
- refspecs: A list of refspecs or a string
- force: Force overwriting for all reftuples
- Returns: A list of refs
- Raises:
- KeyError: If one of the refs can not be found
- """
- if not isinstance(refspecs, list):
- refspecs = [refspecs]
- ret = []
-
- for refspec in refspecs:
- ret.append(parse_reftuple(lh_container, rh_container, refspec, force=force))
- return ret
- def parse_refs(container, refspecs):
- """Parse a list of refspecs to a list of refs.
- Args:
- container: A RefsContainer object
- refspecs: A list of refspecs or a string
- Returns: A list of refs
- Raises:
- KeyError: If one of the refs can not be found
- """
-
- if not isinstance(refspecs, list):
- refspecs = [refspecs]
- ret = []
- for refspec in refspecs:
- ret.append(parse_ref(container, refspec))
- return ret
- def parse_commit_range(
- repo: "Repo", committishs: Union[str, bytes]
- ) -> Iterator["Commit"]:
- """Parse a string referring to a range of commits.
- Args:
- repo: A `Repo` object
- committishs: A string referring to a range of commits.
- Returns: An iterator over `Commit` objects
- Raises:
- KeyError: When the reference commits can not be found
- ValueError: If the range can not be parsed
- """
- committishs = to_bytes(committishs)
-
- return iter([parse_commit(repo, committishs)])
- class AmbiguousShortId(Exception):
- """The short id is ambiguous."""
- def __init__(self, prefix, options) -> None:
- self.prefix = prefix
- self.options = options
- def scan_for_short_id(object_store, prefix, tp):
- """Scan an object store for a short id."""
- ret = []
- for object_id in object_store.iter_prefix(prefix):
- o = object_store[object_id]
- if isinstance(o, tp):
- ret.append(o)
- if not ret:
- raise KeyError(prefix)
- if len(ret) == 1:
- return ret[0]
- raise AmbiguousShortId(prefix, ret)
- def parse_commit(repo: "Repo", committish: Union[str, bytes]) -> "Commit":
- """Parse a string referring to a single commit.
- Args:
- repo: A` Repo` object
- committish: A string referring to a single commit.
- Returns: A Commit object
- Raises:
- KeyError: When the reference commits can not be found
- ValueError: If the range can not be parsed
- """
- committish = to_bytes(committish)
- try:
- return repo[committish]
- except KeyError:
- pass
- try:
- return repo[parse_ref(repo, committish)]
- except KeyError:
- pass
- if len(committish) >= 4 and len(committish) < 40:
- try:
- int(committish, 16)
- except ValueError:
- pass
- else:
- try:
- return scan_for_short_id(repo.object_store, committish, Commit)
- except KeyError:
- pass
- raise KeyError(committish)
|