Coverage for yuio/git.py: 97%

1# Yuio project, MIT license.

3# https://github.com/taminomara/yuio/

5# You're free to copy this file to your project and edit it for your needs,

6# just keep this copyright line please :3

8"""

9This module provides basic functionality to interact with git.

10It comes in handy when writing deployment scripts.

13Interacting with a repository

14-----------------------------

16All repository interactions are done through the :class:`Repo` class

17and its methods. If an interaction fails, a :class:`GitError` is raised.

19.. autoclass:: Repo

20 :members:

22.. autoclass:: GitError

24.. autoclass:: GitExecError

26.. autoclass:: GitUnavailableError

28.. autoclass:: NotARepositoryError

31Status objects

32--------------

34:meth:`Repo.status` returns repository status parsed from :flag:`git status` command.

35It can show changed and unmerged files and submodules. See details about change

36representation in `git status`__ manual.

38__ https://git-scm.com/docs/git-status#_output

40Yuio represents :flag:`git status` output as close to the original as possible,

41but makes some convenience renames. This results in somewhat unexpected

42class structure:

44.. raw:: html

46

47 <pre class="mermaid">

48 ---

49 config:

50 class:

51 hideEmptyMembersBox: true

52 ---

53 classDiagram

55 class PathStatus

56 click PathStatus href "#yuio.git.PathStatus" "yuio.git.PathStatus"

58 class FileStatus

59 click FileStatus href "#yuio.git.FileStatus" "yuio.git.FileStatus"

60 PathStatus <|-- FileStatus

62 class SubmoduleStatus

63 click SubmoduleStatus href "#yuio.git.SubmoduleStatus" "yuio.git.SubmoduleStatus"

64 FileStatus <|-- SubmoduleStatus

66 class UnmergedFileStatus

67 click UnmergedFileStatus href "#yuio.git.UnmergedFileStatus" "yuio.git.UnmergedFileStatus"

68 PathStatus <|-- UnmergedFileStatus

70 class UnmergedSubmoduleStatus

71 click UnmergedSubmoduleStatus href "#yuio.git.UnmergedSubmoduleStatus" "yuio.git.UnmergedSubmoduleStatus"

72 UnmergedFileStatus <|-- UnmergedSubmoduleStatus

73 </pre>

74

76.. autoclass:: Status

77 :members:

79.. autoclass:: PathStatus

80 :members:

82.. autoclass:: FileStatus

83 :members:

85.. autoclass:: SubmoduleStatus

86 :members:

88.. autoclass:: UnmergedFileStatus

89 :members:

91.. autoclass:: UnmergedSubmoduleStatus

92 :members:

94.. autoclass:: Modification

95 :members:

98Commit objects

99--------------

100

101.. autoclass:: Commit

102 :members:

103

104.. autoclass:: CommitTrailers

105 :members:

106

107

108Parsing git refs

109----------------

110

111When you need to query a git ref from a user, :class:`RefParser` will check

112that the given ref is formatted correctly. Use :class:`Ref` in your type hints

113to help Yuio detect that you want to parse it as a git reference:

114

115.. autoclass:: Ref

116

117.. autoclass:: Tag

118

119.. autoclass:: Branch

120

121.. autoclass:: Remote

122

123.. autoclass:: RefParser

124

125.. autoclass:: TagParser

126

127.. autoclass:: BranchParser

128

129.. autoclass:: RemoteParser

130

131If you know path to your repository before hand, and want to make sure that the user

132supplies a valid ref that points to an existing git object, use :class:`CommitParser`:

133

134.. autoclass:: CommitParser

135

136

137Autocompleting git refs

138-----------------------

139

140.. autoclass:: RefCompleter

141

142.. autoclass:: RefCompleterMode

143 :members:

144

145"""

146

147from __future__ import annotations

148

149import dataclasses

150import enum

151import functools

152import logging

153import pathlib

154import re

155from dataclasses import dataclass

156from datetime import datetime

157

158import yuio.complete

159import yuio.exec

160import yuio.parse

161from yuio.util import dedent as _dedent

162

163import yuio._typing_ext as _tx

164from typing import TYPE_CHECKING

165

166if TYPE_CHECKING:

167 import typing_extensions as _t

168else:

169 from yuio import _typing as _t

170

171__all__ = [

172 "Branch",

173 "BranchParser",

174 "Commit",

175 "CommitParser",

176 "CommitTrailers",

177 "FileStatus",

178 "GitError",

179 "GitExecError",

180 "GitUnavailableError",

181 "Modification",

182 "NotARepositoryError",

183 "PathStatus",

184 "Ref",

185 "RefCompleter",

186 "RefCompleterMode",

187 "RefParser",

188 "Remote",

189 "RemoteParser",

190 "Repo",

191 "Status",

192 "SubmoduleStatus",

193 "Tag",

194 "TagParser",

195 "UnmergedFileStatus",

196 "UnmergedSubmoduleStatus",

197]

198

199_logger = logging.getLogger(__name__)

200

201

202class GitError(Exception):

203 """

204 Raised when interaction with git fails.

205

206 """

207

208

209class GitExecError(GitError, yuio.exec.ExecError):

210 """

211 Raised when git returns a non-zero exit code.

212

213 """

214

215

216class GitUnavailableError(GitError, FileNotFoundError):

217 """

218 Raised when git executable can't be found.

219

220 """

221

222

223class NotARepositoryError(GitError, FileNotFoundError):

224 """

225 Raised when given path is not in git repository.

226

227 """

228

229

230Ref = _t.NewType("Ref", str)

231"""

232A special kind of string that contains a git object reference.

233

234Ref is not guaranteed to be valid; this type is used in type hints

235to make use of the :class:`RefParser`.

236

237"""

238

239Tag = _t.NewType("Tag", str)

240"""

241A special kind of string that contains a tag name.

242

243Ref is not guaranteed to be valid; this type is used in type hints

244to make use of the :class:`TagParser`.

245

246"""

247

248Branch = _t.NewType("Branch", str)

249"""

250A special kind of string that contains a branch name.

251

252Ref is not guaranteed to be valid; this type is used in type hints

253to make use of the :class:`BranchParser`.

254

255"""

256

257Remote = _t.NewType("Remote", str)

258"""

259A special kind of string that contains a remote branch name.

260

261Ref is not guaranteed to be valid; this type is used in type hints

262to make use of the :class:`RemoteParser`.

263

264"""

265

266

267# See https://git-scm.com/docs/git-log#_pretty_formats

268# for explanation of these incantations.

269_LOG_FMT = "%H%n%aN%n%aE%n%aI%n%cN%n%cE%n%cI%n%(decorate:prefix=,suffix=,tag=,separator= )%n%w(0,0,1)%B%w(0,0)%n-"

270_LOG_TRAILERS_FMT = "%H%n%w(0,1,1)%(trailers:only=true)%w(0,0)%n-"

271_LOG_TRAILER_KEY_RE = re.compile(r"^(?P<key>\S+):\s")

272

273

274class Repo:

275 """

276 A class that allows interactions with a git repository.

277

278 :param path:

279 path to the repo root dir.

280 :param env:

281 environment variables for the git executable.

282 :raises:

283 constructor of this class may raise :class:`GitError` if git isn't available

284 or if the given part is not inside of a git repository.

285

286 """

287

288 def __init__(

289 self,

290 path: pathlib.Path | str,

291 /,

292 env: dict[str, str] | None = None,

293 ):

294 self.__path = pathlib.Path(path)

295 self.__env = env

296 self.__git_is_available = None

297 self.__is_repo = None

298

299 try:

300 version = self.git("--version")

301 except GitExecError:

302 raise GitUnavailableError("git executable is not available")

303

304 _logger.debug("%s", version.decode(errors="replace").strip())

305

306 try:

307 self.git("rev-parse", "--is-inside-work-tree")

308 except GitExecError:

309 raise NotARepositoryError(f"{self.__path} is not a git repository")

310

311 @property

312 def path(self) -> pathlib.Path:

313 """

314 Path to the repo, as was passed to the constructor.

315

316 """

317

318 return self.__path

319

320 @functools.cached_property

321 def root(self) -> pathlib.Path:

322 """

323 The root directory of the repo.

324

325 """

326

327 return pathlib.Path(

328 self.git("rev-parse", "--path-format=absolute", "--show-toplevel")

329 .decode()

330 .strip()

331 ).resolve()

332

333 @functools.cached_property

334 def git_dir(self) -> pathlib.Path:

335 """

336 Get path to the ``.git`` directory of the repo.

337

338 """

339

340 return pathlib.Path(

341 self.git("rev-parse", "--path-format=absolute", "--git-dir")

342 .decode()

343 .strip()

344 ).resolve()

345

346 @_t.overload

347 def git(self, *args: str | pathlib.Path) -> bytes: ...

348 @_t.overload

349 def git(self, *args: str | pathlib.Path, capture_io: _t.Literal[False]) -> None: ...

350 @_t.overload

351 def git(self, *args: str | pathlib.Path, capture_io: bool) -> bytes | None: ...

352 def git(self, *args: str | pathlib.Path, capture_io: bool = True):

353 """

354 Call git and return its stdout.

355

356 :param args:

357 arguments for the :flag:`git` command.

358 :param capture_io:

359 If set to :data:`False`, command's stderr and stdout are not captured.

360 :returns:

361 output of the git command.

362 :raises:

363 :class:`GitError`, :class:`OSError`.

364

365 """

366

367 try:

368 return yuio.exec.exec(

369 "git",

370 *args,

371 cwd=self.__path,

372 env=self.__env,

373 capture_io=capture_io,

374 text=False,

375 )

376 except yuio.exec.ExecError as e:

377 raise GitExecError(e.returncode, e.cmd, e.output, e.stderr)

378 except FileNotFoundError:

379 raise GitUnavailableError("git executable not found")

380

381 def status(

382 self, /, include_ignored: bool = False, include_submodules: bool = True

383 ) -> Status:

384 """

385 Query the current repository status.

386

387 :param include_ignored:

388 include ignored status in the list of changes. Disable by default.

389 :param include_submodules:

390 include status of submodules in the list of changes. Enabled by default.

391 :returns:

392 current repository status.

393 :raises:

394 :class:`GitError`, :class:`OSError`.

395

396 """

397

398 text = self.git(

399 "status",

400 "--porcelain=v2",

401 "-z",

402 "--ahead-behind",

403 "--branch",

404 "--renames",

405 "--untracked-files=normal",

406 "--ignore-submodules=" + ("none" if include_submodules else "all"),

407 "--ignored=" + ("matching" if include_ignored else "no"),

408 )

409 lines = iter(text.split(b"\0"))

410

411 status = Status(commit=None)

412

413 for line_b in lines:

414 line = line_b.decode()

415 if line.startswith("# branch.oid"):

416 if line[13:] != "(initial)":

417 status.commit = line[13:]

418 elif line.startswith("# branch.head"):

419 if line[14:] != "(detached)":

420 status.branch = line[14:]

421 elif line.startswith("# branch.upstream"):

422 status.upstream = line[18:]

423 elif line.startswith("# branch.ab"):

424 match = re.match(r"^\+(\d+) -(\d+)$", line[12:])

425 assert match is not None

426 status.ahead = int(match.group(1))

427 status.behind = int(match.group(2))

428 elif line.startswith("1"):

429 match = re.match(

430 r"^(?P<X>.)(?P<Y>.) (?P.{4}) (?:[^ ]+ ){5}(?P<path>.*)$",

431 line[2:],

432 )

433 assert match is not None

434 sub = match.group("sub")

435 if sub[0] == "S":

436 path_status = SubmoduleStatus(

437 path=pathlib.Path(match.group("path")),

438 path_from=None,

439 staged=Modification(match.group("X")),

440 tree=Modification(match.group("Y")),

441 commit_changed=sub[1] != ".",

442 has_tracked_changes=sub[2] != ".",

443 has_untracked_changes=sub[3] != ".",

444 )

445 else:

446 path_status = FileStatus(

447 path=pathlib.Path(match.group("path")),

448 path_from=None,

449 staged=Modification(match.group("X")),

450 tree=Modification(match.group("Y")),

451 )

452 status.changes.append(path_status)

453 elif line.startswith("2"):

454 match = re.match(

455 r"^(?P<X>.)(?P<Y>.) (?P.{4}) (?:[^ ]+ ){6}(?P<path>.*)$",

456 line[2:],

457 )

458 assert match is not None

459 path_from = pathlib.Path(next(lines).decode())

460 sub = match.group("sub")

461 if sub[0] == "S":

462 path_status = SubmoduleStatus(

463 path=pathlib.Path(match.group("path")),

464 path_from=path_from,

465 staged=Modification(match.group("X")),

466 tree=Modification(match.group("Y")),

467 commit_changed=sub[1] != ".",

468 has_tracked_changes=sub[2] != ".",

469 has_untracked_changes=sub[3] != ".",

470 )

471 else:

472 path_status = FileStatus(

473 path=pathlib.Path(match.group("path")),

474 path_from=path_from,

475 staged=Modification(match.group("X")),

476 tree=Modification(match.group("Y")),

477 )

478 status.changes.append(path_status)

479 elif line.startswith("u"):

480 match = re.match(

481 r"^(?P<X>.)(?P<Y>.) (?P.{4}) (?:[^ ]+ ){7}(?P<path>.*)$",

482 line[2:],

483 )

484 assert match is not None

485 sub = match.group("sub")

486 if sub[0] == "S":

487 path_status = UnmergedSubmoduleStatus(

488 path=pathlib.Path(match.group("path")),

489 us=Modification(match.group("X")),

490 them=Modification(match.group("Y")),

491 commit_changed=sub[1] != ".",

492 has_tracked_changes=sub[2] != ".",

493 has_untracked_changes=sub[3] != ".",

494 )

495 else:

496 path_status = UnmergedFileStatus(

497 path=pathlib.Path(match.group("path")),

498 us=Modification(match.group("X")),

499 them=Modification(match.group("Y")),

500 )

501 status.changes.append(path_status)

502 elif line.startswith("?"):

503 status.changes.append(

504 FileStatus(

505 path=pathlib.Path(line[2:]),

506 path_from=None,

507 staged=Modification.UNTRACKED,

508 tree=Modification.UNTRACKED,

509 )

510 )

511 elif line.startswith("!"):

512 status.changes.append(

513 FileStatus(

514 path=pathlib.Path(line[2:]),

515 path_from=None,

516 staged=Modification.IGNORED,

517 tree=Modification.IGNORED,

518 )

519 )

520

521 try:

522 status.cherry_pick_head = (

523 self.git("rev-parse", "--verify", "CHERRY_PICK_HEAD").decode().strip()

524 )

525 except GitError:

526 pass

527 try:

528 status.merge_head = (

529 self.git("rev-parse", "--verify", "MERGE_HEAD").decode().strip()

530 )

531 except GitError:

532 pass

533 try:

534 status.rebase_head = (

535 self.git("rev-parse", "--verify", "REBASE_HEAD").decode().strip()

536 )

537 except GitError:

538 pass

539 try:

540 status.revert_head = (

541 self.git("rev-parse", "--verify", "REVERT_HEAD").decode().strip()

542 )

543 except GitError:

544 pass

545 try:

546 status.bisect_start = (

547 self.git("rev-parse", "--verify", "BISECT_START").decode().strip()

548 )

549 except GitError:

550 pass

551

552 return status

553

554 def print_status(self):

555 """

556 Run :flag:`git status` and show its output to the user.

557

558 """

559

560 self.git("status", capture_io=False)

561

562 def log(self, *refs: str, max_entries: int | None = None) -> list[Commit]:

563 """

564 Query the log for given git objects.

565

566 :param refs:

567 git references that will be passed to :flag:`git log`.

568 :param max_entries:

569 maximum number of returned references.

570 :returns:

571 list of found commits.

572 :raises:

573 :class:`GitError`, :class:`OSError`.

574

575 """

576

577 args = [

578 f"--pretty=format:{_LOG_FMT}",

579 "--decorate-refs=refs/tags",

580 "--decorate=short",

581 ]

582

583 if max_entries is not None:

584 args += ["-n", str(max_entries)]

585

586 args += list(refs)

587

588 text = self.git("log", *args)

589 lines = iter(text.decode().split("\n"))

590

591 commits = []

592

593 while commit := self.__parse_single_log_entry(lines):

594 commits.append(commit)

595

596 return commits

597

598 def trailers(

599 self, *refs: str, max_entries: int | None = None

600 ) -> list[CommitTrailers]:

601 """

602 Query trailer lines for given git objects.

603

604 Trailers are lines at the end of a commit message formatted as ``key: value``

605 pairs. See `git-interpret-trailers`__ for further description.

606

607 __ https://git-scm.com/docs/git-interpret-trailers

608

609 :param refs:

610 git references that will be passed to :flag:`git log`.

611 :param max_entries:

612 maximum number of checked commits.

613

614 .. warning::

615

616 This option limits number of checked commits, not the number

617 of trailers.

618 :returns:

619 list of found commits and their trailers.

620 :raises:

621 :class:`GitError`, :class:`OSError`.

622

623 """

624

625 args = [f"--pretty=format:{_LOG_TRAILERS_FMT}"]

626

627 if max_entries is not None:

628 args += ["-n", str(max_entries)]

629

630 args += list(refs)

631

632 text = self.git("log", *args)

633 lines = iter(text.decode().split("\n"))

634

635 trailers = []

636

637 while commit := self.__parse_single_trailer_entry(lines):

638 trailers.append(commit)

639

640 return trailers

641

642 def show(self, ref: str, /) -> Commit | None:

643 """

644 Query information for the given git object.

645

646 Return :data:`None` if object is not found.

647

648 :param ref:

649 git reference that will be passed to :flag:`git log`.

650 :returns:

651 found commit or :data:`None`.

652 :raises:

653 :class:`OSError`.

654

655 """

656

657 try:

658 self.git("rev-parse", "--verify", ref)

659 except GitError:

660 return None

661

662 log = self.log(ref, max_entries=1)

663 if not log:

664 return None

665 else:

666 commit = log[0]

667 commit.orig_ref = ref

668 return commit

669

670 @staticmethod

671 def __parse_single_log_entry(lines) -> Commit | None:

672 try:

673 commit = next(lines)

674 author = next(lines)

675 author_email = next(lines)

676 author_datetime = datetime.fromisoformat(next(lines).replace("Z", "+00:00"))

677 committer = next(lines)

678 committer_email = next(lines)

679 committer_datetime = datetime.fromisoformat(

680 next(lines).replace("Z", "+00:00")

681 )

682 tags = next(lines).split()

683 title = next(lines)

684 body = ""

685

686 while True:

687 line = next(lines)

688 if not line or line.startswith(" "):

689 body += line[1:] + "\n"

690 else:

691 break

692

693 body = body.strip("\n")

694 if body:

695 body += "\n"

696

697 return Commit(

698 hash=commit,

699 tags=tags,

700 author=author,

701 author_email=author_email,

702 author_datetime=author_datetime,

703 committer=committer,

704 committer_email=committer_email,

705 committer_datetime=committer_datetime,

706 title=title,

707 body=body,

708 )

709 except StopIteration:

710 return None

711

712 @staticmethod

713 def __parse_single_trailer_entry(

714 lines,

715 ) -> CommitTrailers | None:

716 try:

717 commit = next(lines)

718 trailers = []

719 current_key = None

720 current_value = ""

721

722 while True:

723 line = next(lines)

724 if not line or line.startswith(" "):

725 line = line[1:] + "\n"

726 if match := _LOG_TRAILER_KEY_RE.match(line):

727 if current_key:

728 current_value = _dedent(current_value)

729 trailers.append((current_key, current_value))

730 current_key = match.group("key")

731 current_value = line[match.end() :]

732 else:

733 current_value += line

734 else:

735 break

736 if current_key:

737 current_value = _dedent(current_value)

738 trailers.append((current_key, current_value))

739

740 return CommitTrailers(hash=commit, trailers=trailers)

741 except StopIteration:

742 return None

743

744 def tags(self) -> list[str]:

745 """

746 List all tags in this repository.

747

748 :returns:

749 list of strings representing tags, without ``refs/tags`` prefix,

750 sorted lexicographically as strings.

751 :raises:

752 :class:`GitError`, :class:`OSError`.

753

754 """

755

756 return (

757 self.git("for-each-ref", "--format=%(refname:short)", "refs/tags")

758 .decode()

759 .splitlines()

760 )

761

762 def branches(self) -> list[str]:

763 """

764 List all branches in this repository.

765

766 :returns:

767 list of strings representing branch names, without ``refs/heads`` prefix,

768 sorted lexicographically as strings.

769 :raises:

770 :class:`GitError`, :class:`OSError`.

771

772 """

773

774 return (

775 self.git("for-each-ref", "--format=%(refname:short)", "refs/heads")

776 .decode()

777 .splitlines()

778 )

779

780 def remotes(self) -> list[str]:

781 """

782 List all remote branches in this repository.

783

784 :returns:

785 list of strings representing remote branches, without

786 ``refs/remotes`` prefix, sorted lexicographically as strings.

787 :raises:

788 :class:`GitError`, :class:`OSError`.

789

790 """

791

792 return [

793 remote

794 for remote in self.git(

795 "for-each-ref", "--format=%(refname:short)", "refs/remotes"

796 )

797 .decode()

798 .splitlines()

799 if "/" in remote

800 ]

801

802

803@dataclass(kw_only=True, slots=True)

804class Commit:

805 """

806 Commit description.

807

808 """

809

810 hash: str

811 """

812 Commit hash.

813

814 """

815

816 tags: list[str]

817 """

818 Tags attached to this commit.

819

820 """

821

822 author: str

823 """

824 Author name.

825

826 """

827

828 author_email: str

829 """

830 Author email.

831

832 """

833

834 author_datetime: datetime

835 """

836 Author time.

837

838 """

839

840 committer: str

841 """

842 Committer name.

843

844 """

845

846 committer_email: str

847 """

848 Committer email.

849

850 """

851

852 committer_datetime: datetime

853 """

854 Committer time.

855

856 """

857

858 title: str

859 """

860 Commit title, i.e. first line of the message.

861

862 """

863

864 body: str

865 """

866 Commit body, i.e. the rest of the message.

867

868 """

869

870 orig_ref: str | None = None

871 """

872 If commit was parsed from a user input, this field will contain

873 original input. I.e. if a user enters ``HEAD`` and it gets resolved

874 into a commit, `orig_ref` will contain string ``"HEAD"``.

875

876 See also :class:`CommitParser`.

877

878 """

879

880 @property

881 def short_hash(self):

882 """

883 First seven characters of the commit hash.

884

885 """

886

887 return self.hash[:7]

888

889 def __str__(self):

890 if self.orig_ref:

891 return self.orig_ref

892 else:

893 return self.short_hash

894

895

896@dataclass(kw_only=True, slots=True)

897class CommitTrailers:

898 """

899 Commit trailers.

900

901 """

902

903 hash: str

904 """

905 Commit hash.

906

907 """

908

909 trailers: list[tuple[str, str]]

910 """

911 Key-value pairs for commit trailers.

912

913 """

914

915

916class Modification(enum.Enum):

917 """

918 For changed file or submodule, what modification was applied to it.

919

920 """

921

922 UNMODIFIED = "."

923 """

924 File wasn't changed.

925

926 """

927

928 MODIFIED = "M"

929 """

930 File was changed.

931

932 """

933

934 SUBMODULE_MODIFIED = "m"

935 """

936 Contents of submodule were modified.

937

938 """

939

940 TYPE_CHANGED = "T"

941 """

942 File type changed.

943

944 """

945

946 ADDED = "A"

947 """

948 File was created.

949

950 """

951

952 DELETED = "D"

953 """

954 File was deleted.

955

956 """

957

958 RENAMED = "R"

959 """

960 File was renamed (and possibly changed).

961

962 """

963

964 COPIED = "C"

965 """

966 File was copied (and possibly changed).

967

968 """

969

970 UPDATED = "U"

971 """

972 File was updated but unmerged.

973

974 """

975

976 UNTRACKED = "?"

977 """

978 File is untracked, i.e. not yet staged or committed.

979

980 """

981

982 IGNORED = "!"

983 """

984 File is in ``.gitignore``.

985

986 """

987

988

989@dataclass(kw_only=True, slots=True)

990class PathStatus:

991 """

992 Status of a changed path.

993

994 """

995

996 path: pathlib.Path

997 """

998 Path of the file.

999

1000 """

1001

1002

1003@dataclass(kw_only=True, slots=True)

1004class FileStatus(PathStatus):

1005 """

1006 Status of a changed file.

1007

1008 """

1009

1010 path_from: pathlib.Path | None

1011 """

1012 If file was moved, contains path where it was moved from.

1013

1014 """

1015

1016 staged: Modification

1017 """

1018 File modification in the index (staged).

1019

1020 """

1021

1022 tree: Modification

1023 """

1024 File modification in the tree (unstaged).

1025

1026 """

1027

1028

1029@dataclass(kw_only=True, slots=True)

1030class SubmoduleStatus(FileStatus):

1031 """

1032 Status of a submodule.

1033

1034 """

1035

1036 commit_changed: bool

1037 """

1038 The submodule has a different HEAD than recorded in the index.

1039

1040 """

1041

1042 has_tracked_changes: bool

1043 """

1044 Tracked files were changed in the submodule.

1045

1046 """

1047

1048 has_untracked_changes: bool

1049 """

1050 Untracked files were changed in the submodule.

1051

1052 """

1053

1054

1055@dataclass(kw_only=True, slots=True)

1056class UnmergedFileStatus(PathStatus):

1057 """

1058 Status of an unmerged file.

1059

1060 """

1061

1062 us: Modification

1063 """

1064 File modification that has happened at the head.

1065

1066 """

1067

1068 them: Modification

1069 """

1070 File modification that has happened at the merge head.

1071

1072 """

1073

1074

1075@dataclass(kw_only=True, slots=True)

1076class UnmergedSubmoduleStatus(UnmergedFileStatus):

1077 """

1078 Status of an unmerged submodule.

1079

1080 """

1081

1082 commit_changed: bool

1083 """

1084 The submodule has a different HEAD than recorded in the index.

1085

1086 """

1087

1088 has_tracked_changes: bool

1089 """

1090 Tracked files were changed in the submodule.

1091

1092 """

1093

1094 has_untracked_changes: bool

1095 """

1096 Untracked files were changed in the submodule.

1097

1098 """

1099

1100

1101@dataclass(kw_only=True, slots=True)

1102class Status:

1103 """

1104 Status of a working copy.

1105

1106 """

1107

1108 commit: str | None

1109 """

1110 Current commit hash. Can be absent if current branch is orphaned and doesn't have

1111 any commits yet.

1112

1113 """

1114

1115 branch: str | None = None

1116 """

1117 Name of the current branch.

1118

1119 """

1120

1121 upstream: str | None = None

1122 """

1123 Name of the upstream branch.

1124

1125 """

1126

1127 ahead: int | None = None

1128 """

1129 Number of commits the branch is ahead of upstream.

1130

1131 """

1132

1133 behind: int | None = None

1134 """

1135 Number of commits the branch is behind of upstream.

1136

1137 """

1138

1139 changes: list[PathStatus] = dataclasses.field(default_factory=list)

1140 """

1141 List of changed files, both tracked and untracked.

1142

1143 See details about change representation in `git status`__ manual.

1144

1145 __ https://git-scm.com/docs/git-status#_output

1146

1147 """

1148

1149 cherry_pick_head: str | None = None

1150 """

1151 Position of the ``CHERRY_PICK_HEAD``.

1152

1153 If this field is not :data:`None`, cherry pick is in progress.

1154

1155 """

1156

1157 merge_head: str | None = None

1158 """

1159 Position of the ``MERGE_HEAD``.

1160

1161 If this field is not :data:`None`, merge is in progress.

1162

1163 """

1164

1165 rebase_head: str | None = None

1166 """

1167 Position of the ``REBASE_HEAD``.

1168

1169 If this field is not :data:`None`, rebase is in progress.

1170

1171 """

1172

1173 revert_head: str | None = None

1174 """

1175 Position of the ``REVERT_HEAD``.

1176

1177 If this field is not :data:`None`, revert is in progress.

1178

1179 """

1180

1181 bisect_start: str | None = None

1182 """

1183 Position of the ``BISECT_START``.

1184

1185 If this field is not :data:`None`, bisect is in progress.

1186

1187 """

1188

1189 def has_staged_changes(self) -> bool:

1190 """

1191 Return :data:`True` if there are unstaged changes in this repository.

1192

1193 """

1194

1195 return next(self.get_staged_changes(), None) is not None

1196

1197 def get_staged_changes(self) -> _t.Iterator[PathStatus]:

1198 """

1199 Return iterator over all unstaged changes in this repository.

1200

1201 """

1202

1203 return (

1204 change

1205 for change in self.changes

1206 if isinstance(change, FileStatus)

1207 and change.staged

1208 not in [

1209 Modification.UNMODIFIED,

1210 Modification.IGNORED,

1211 Modification.UNTRACKED,

1212 ]

1213 )

1214

1215 def has_unstaged_changes(self) -> bool:

1216 """

1217 Return :data:`True` if there are unstaged changes in this repository.

1218

1219 """

1220

1221 return next(self.get_unstaged_changes(), None) is not None

1222

1223 def get_unstaged_changes(self) -> _t.Iterator[PathStatus]:

1224 """

1225 Return iterator over all staged changes in this repository.

1226

1227 """

1228

1229 return (

1230 change

1231 for change in self.changes

1232 if isinstance(change, UnmergedFileStatus)

1233 or (

1234 isinstance(change, FileStatus)

1235 and change.tree

1236 not in [

1237 Modification.UNMODIFIED,

1238 Modification.IGNORED,

1239 ]

1240 )

1241 )

1242

1243 def has_ongoing_operation(self) -> bool:

1244 """

1245 Return :data:`True` if there is an ongoing operation such as merge or rebase.

1246

1247 """

1248

1249 return (

1250 self.cherry_pick_head is not None

1251 or self.merge_head is not None

1252 or self.rebase_head is not None

1253 or self.revert_head is not None

1254 or self.bisect_start is not None

1255 )

1256

1257

1258class RefCompleterMode(enum.Enum):

1259 """

1260 Specifies operation modes for :class:`RefCompleter`.

1261

1262 """

1263

1264 BRANCH = "b"

1265 """

1266 Completes branches.

1267

1268 """

1269

1270 REMOTE = "r"

1271 """

1272 Completes remote branches.

1273

1274 """

1275

1276 TAG = "t"

1277 """

1278 Completes tags.

1279

1280 """

1281

1282 HEAD = "h"

1283 """

1284 Completes ``HEAD`` and ``ORIG_HEAD``.

1285

1286 """

1287

1288

1289class RefCompleter(yuio.complete.Completer):

1290 """

1291 Completes git refs.

1292

1293 :param repo:

1294 source of completions. If not given, this completer will try to use current

1295 directory as a repo root, and fail silently if it's not a repo.

1296 :param modes:

1297 which objects to complete.

1298

1299 """

1300

1301 def __init__(self, repo: Repo | None, modes: set[RefCompleterMode] | None = None):

1302 super().__init__()

1303

1304 self._repo: Repo | None | _t.Literal[False] = repo

1305 self._modes = modes or {

1306 RefCompleterMode.BRANCH,

1307 RefCompleterMode.TAG,

1308 RefCompleterMode.HEAD,

1309 }

1310

1311 def _process(self, collector: yuio.complete.CompletionCollector, /):

1312 if self._repo is None:

1313 try:

1314 self._repo = Repo(pathlib.Path.cwd())

1315 except (GitError, OSError):

1316 self._repo = False

1317 if not self._repo:

1318 return

1319 try:

1320 if RefCompleterMode.HEAD in self._modes:

1321 collector.add_group()

1322 git_dir = self._repo.git_dir

1323 for head in ["HEAD", "ORIG_HEAD"]:

1324 if (git_dir / head).exists():

1325 collector.add(head)

1326 if RefCompleterMode.BRANCH in self._modes:

1327 collector.add_group()

1328 for branch in self._repo.branches():

1329 collector.add(branch, comment="branch")

1330 if RefCompleterMode.REMOTE in self._modes:

1331 collector.add_group()

1332 for remote in self._repo.remotes():

1333 collector.add(remote, comment="remote")

1334 if RefCompleterMode.TAG in self._modes:

1335 collector.add_group()

1336 for tag in self._repo.tags():

1337 collector.add(tag, comment="tag")

1338 except GitError:

1339 pass

1340

1341 def _get_completion_model(

1342 self, *, is_many: bool = False

1343 ) -> yuio.complete._OptionSerializer.Model:

1344 return yuio.complete._OptionSerializer.Git(

1345 "".join(sorted(list(m.value for m in self._modes)))

1346 )

1347

1348

1349def CommitParser(*, repo: Repo) -> yuio.parse.Parser[Commit]:

1350 """

1351 A parser for git refs (commits, tags, branches, and so on).

1352

1353 This parser validates that the given ref exists in the given repository,

1354 parses it and returns a commit data associated with this ref.

1355

1356 If you need a simple string without additional validation,

1357 use :class:`RefParser`.

1358

1359 :param repo:

1360 initialized repository is required to ensure that commit is valid.

1361

1362 """

1363

1364 def map(value: str, /) -> Commit:

1365 commit = repo.show(value)

1366 if commit is None:

1367 raise yuio.parse.ParsingError("invalid git ref `%s`", value)

1368 return commit

1369

1370 def rev(value: Commit | object) -> str:

1371 if isinstance(value, Commit):

1372 return str(value)

1373 else:

1374 raise TypeError(

1375 f"parser Commit can't handle value of type {_tx.type_repr(type(value))}"

1376 )

1377

1378 return yuio.parse.WithMeta(

1379 yuio.parse.Map(yuio.parse.Str(), map, rev),

1380 desc="<commit>",

1381 completer=RefCompleter(repo),

1382 )

1383

1384

1385T = _t.TypeVar("T")

1386

1387

1388class _RefParserImpl(yuio.parse.Str, _t.Generic[T]):

1389 @functools.cached_property

1390 def _description(self):

1391 return "<" + self.__class__.__name__.removesuffix("Parser").lower() + ">"

1392

1393 def describe(self) -> str | None:

1394 return self._description

1395

1396 def describe_or_def(self) -> str:

1397 return self._description

1398

1399 def describe_many(self) -> str | tuple[str, ...]:

1400 return self._description

1401

1402

1403class RefParser(_RefParserImpl[Ref]):

1404 """

1405 A parser that provides autocompletion for git refs, but doesn't verify

1406 anything else.

1407

1408 """

1409

1410 def completer(self) -> yuio.complete.Completer:

1411 return RefCompleter(None)

1412

1413

1414class TagParser(_RefParserImpl[Tag]):

1415 """

1416 A parser that provides autocompletion for git tag, but doesn't verify

1417 anything else.

1418

1419 """

1420

1421 def completer(self) -> yuio.complete.Completer:

1422 return RefCompleter(None, {RefCompleterMode.TAG})

1423

1424

1425class BranchParser(_RefParserImpl[Branch]):

1426 """

1427 A parser that provides autocompletion for git branches, but doesn't verify

1428 anything else.

1429

1430 """

1431

1432 def completer(self) -> yuio.complete.Completer:

1433 return RefCompleter(None, {RefCompleterMode.BRANCH})

1434

1435

1436class RemoteParser(_RefParserImpl[Remote]):

1437 """

1438 A parser that provides autocompletion for git remotes, but doesn't verify

1439 anything else.

1440

1441 """

1442

1443 def completer(self) -> yuio.complete.Completer:

1444 return RefCompleter(None, {RefCompleterMode.REMOTE})

1445

1446

1447yuio.parse.register_type_hint_conversion(

1448 lambda ty, origin, args: RefParser() if ty is Ref else None

1449)

1450

1451yuio.parse.register_type_hint_conversion(

1452 lambda ty, origin, args: TagParser() if ty is Tag else None

1453)

1454

1455yuio.parse.register_type_hint_conversion(

1456 lambda ty, origin, args: BranchParser() if ty is Branch else None

1457)

1458

1459yuio.parse.register_type_hint_conversion(

1460 lambda ty, origin, args: RemoteParser() if ty is Remote else None

1461)

Coverage for yuio / git.py: 97%

434 statements