git » git-arr » master » tree

[master] / git.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
"""
Python wrapper for git.

This module is a light Python API for interfacing with it. It calls the git
command line tool directly, so please be careful with using untrusted
parameters.
"""

import functools
import sys
import io
import subprocess
from collections import defaultdict
import email.utils
import datetime
import urllib.request, urllib.parse, urllib.error
from html import escape
from typing import Any, Dict, IO, Iterable, List, Optional, Tuple, Union


# Path to the git binary.
GIT_BIN = "git"


def run_git(
    repo_path: str, params, stdin: bytes = None, silent_stderr=False, raw=False
) -> Union[IO[str], IO[bytes]]:
    """Invokes git with the given parameters.

    This function invokes git with the given parameters, and returns a
    file-like object with the output (from a pipe).
    """
    params = [GIT_BIN, "--git-dir=%s" % repo_path] + list(params)

    stderr = None
    if silent_stderr:
        stderr = subprocess.PIPE

    if not stdin:
        p = subprocess.Popen(
            params, stdin=None, stdout=subprocess.PIPE, stderr=stderr
        )
    else:
        p = subprocess.Popen(
            params,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=stderr,
        )

        assert p.stdin is not None
        p.stdin.write(stdin)
        p.stdin.close()

    assert p.stdout is not None

    if raw:
        return p.stdout

    return io.TextIOWrapper(
        p.stdout, encoding="utf8", errors="backslashreplace"
    )


class GitCommand(object):
    """Convenient way of invoking git."""

    def __init__(self, path: str, cmd: str):
        self._override = True
        self._path = path
        self._cmd = cmd
        self._args: List[str] = []
        self._kwargs: Dict[str, str] = {}
        self._stdin_buf: Optional[bytes] = None
        self._raw = False
        self._override = False

    def __setattr__(self, k, v):
        if k == "_override" or self._override:
            self.__dict__[k] = v
            return
        k = k.replace("_", "-")
        self._kwargs[k] = v

    def arg(self, a: str):
        """Adds an argument."""
        self._args.append(a)

    def raw(self, b: bool):
        """Request raw rather than utf8-encoded command output."""
        self._override = True
        self._raw = b
        self._override = False

    def stdin(self, s: bytes):
        """Sets the contents we will send in stdin."""
        self._override = True
        self._stdin_buf = s
        self._override = False

    def run(self):
        """Runs the git command."""
        params = [self._cmd]

        for k, v in list(self._kwargs.items()):
            dash = "--" if len(k) > 1 else "-"
            if v is None:
                params.append("%s%s" % (dash, k))
            else:
                params.append("%s%s=%s" % (dash, k, str(v)))

        params.extend(self._args)

        return run_git(self._path, params, self._stdin_buf, raw=self._raw)


class SimpleNamespace(object):
    """An entirely flexible object, which provides a convenient namespace."""

    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)


class smstr:
    """A "smart" string, containing many representations for ease of use."""

    raw: str  # string, probably utf8-encoded, good enough to show.
    url: str  # escaped for safe embedding in URLs (not human-readable).
    html: str  # HTML-embeddable representation.

    def __init__(self, s: str):
        self.raw = s
        self.url = urllib.request.pathname2url(s)
        self.html = self._to_html()

    # Note we don't define __repr__() or __str__() to prevent accidental
    # misuse. It does mean that some uses become more annoying, so it's a
    # tradeoff that may change in the future.

    @staticmethod
    def from_url(url):
        """Returns an smstr() instance from an url-encoded string."""
        return smstr(urllib.request.url2pathname(url))

    def split(self, sep):
        """Like str.split()."""
        return [smstr(s) for s in self.raw.split(sep)]

    def __add__(self, other):
        if isinstance(other, smstr):
            other = other.raw
        return smstr(self.raw + other)

    def _to_html(self):
        """Returns an html representation of the unicode string."""
        html = ""
        for c in escape(self.raw):
            if c in "\t\r\n\r\f\a\b\v\0":
                esc_c = c.encode("unicode-escape").decode("utf8")
                html += '<span class="ctrlchr">%s</span>' % esc_c
            else:
                html += c

        return html


def unquote(s: str):
    """Git can return quoted file names, unquote them. Always return a str."""
    if not (s[0] == '"' and s[-1] == '"'):
        # Unquoted strings are always safe, no need to mess with them
        return s

    # The string will be of the form `"<escaped>"`, where <escaped> is a
    # backslash-escaped representation of the name of the file.
    # Examples:  "with\ttwo\ttabs" , "\303\261aca-utf8", "\361aca-latin1"

    # Get rid of the quotes, we never want them in the output.
    s = s[1:-1]

    # Un-escape the backslashes.
    # latin1 is ok to use here because in Python it just maps the code points
    # 0-255 to the bytes 0x-0xff, which is what we expect.
    s = s.encode("latin1").decode("unicode-escape")

    # Convert to utf8.
    s = s.encode("latin1").decode("utf8", errors="backslashreplace")

    return s


class Repo:
    """A git repository."""

    def __init__(self, path: str, name=None, info=None):
        self.path = path
        self.name = name
        self.info: Any = info or SimpleNamespace()

    def cmd(self, cmd):
        """Returns a GitCommand() on our path."""
        return GitCommand(self.path, cmd)

    @functools.lru_cache
    def _for_each_ref(self, pattern=None, sort=None, count=None):
        """Returns a list of references."""
        cmd = self.cmd("for-each-ref")
        if sort:
            cmd.sort = sort
        if count:
            cmd.count = count
        if pattern:
            cmd.arg(pattern)

        refs = []
        for l in cmd.run():
            obj_id, obj_type, ref = l.split()
            refs.append((obj_id, obj_type, ref))
        return refs

    @functools.cache
    def branch_names(self):
        """Get the names of the branches."""
        refs = self._for_each_ref(pattern="refs/heads/", sort="-authordate")
        return [ref[len("refs/heads/") :] for _, _, ref in refs]

    @functools.cache
    def tags(self, sort="-taggerdate"):
        """Get the (name, obj_id) of the tags."""
        refs = self._for_each_ref(pattern="refs/tags/", sort=sort)
        return [(ref[len("refs/tags/") :], obj_id) for obj_id, _, ref in refs]

    @functools.lru_cache
    def commit_ids(self, ref, limit=None):
        """Generate commit ids."""
        cmd = self.cmd("rev-list")
        if limit:
            cmd.max_count = limit

        cmd.arg(ref)
        cmd.arg("--")

        return [l.rstrip("\n") for l in cmd.run()]

    @functools.lru_cache
    def commit(self, commit_id):
        """Return a single commit."""
        cs = list(self.commits(commit_id, limit=1))
        if len(cs) != 1:
            return None
        return cs[0]

    @functools.lru_cache
    def commits(self, ref, limit, offset=0):
        """Generate commit objects for the ref."""
        cmd = self.cmd("rev-list")
        cmd.max_count = limit + offset

        cmd.header = None

        cmd.arg(ref)
        cmd.arg("--")

        info_buffer = ""
        count = 0
        commits = []
        for l in cmd.run():
            if "\0" in l:
                pre, post = l.split("\0", 1)
                info_buffer += pre

                count += 1
                if count > offset:
                    commits.append(Commit.from_str(self, info_buffer))

                # Start over.
                info_buffer = post
            else:
                info_buffer += l

        if info_buffer:
            count += 1
            if count > offset:
                commits.append(Commit.from_str(self, info_buffer))

        return commits

    @functools.lru_cache
    def diff(self, ref):
        """Return a Diff object for the ref."""
        cmd = self.cmd("diff-tree")
        cmd.patch = None
        cmd.numstat = None
        cmd.find_renames = None
        if self.info.root_diff:
            cmd.root = None
        # Note we intentionally do not use -z, as the filename is just for
        # reference, and it is safer to let git do the escaping.

        cmd.arg(ref)

        return Diff.from_str(cmd.run())

    @functools.lru_cache
    def refs(self):
        """Return a dict of obj_id -> ref."""
        cmd = self.cmd("show-ref")
        cmd.dereference = None

        r = defaultdict(list)
        for l in cmd.run():
            l = l.strip()
            obj_id, ref = l.split(" ", 1)
            r[obj_id].append(ref)

        return r

    @functools.lru_cache
    def tree(self, ref):
        """Returns a Tree instance for the given ref."""
        return Tree(self, ref)

    @functools.lru_cache
    def blob(self, path, ref):
        """Returns a Blob instance for the given path."""
        cmd = self.cmd("cat-file")
        cmd.raw(True)
        cmd.batch = "%(objectsize)"

        # Format: <ref>:<path>
        # Construct it in binary since the path might not be utf8.
        cmd.stdin(ref.encode("utf8") + b":" + path)

        out = cmd.run()
        head = out.readline()
        if not head or head.strip().endswith(b"missing"):
            return None

        return Blob(out.read()[: int(head)])

    @functools.cache
    def last_commit_timestamp(self):
        """Return the timestamp of the last commit."""
        refs = self._for_each_ref(
            pattern="refs/heads/", sort="-committerdate", count=1
        )
        for obj_id, _, _ in refs:
            commit = self.commit(obj_id)
            return commit.committer_epoch
        return -1


class Commit(object):
    """A git commit."""

    def __init__(
        self,
        repo,
        commit_id,
        parents,
        tree,
        author,
        author_epoch,
        author_tz,
        committer,
        committer_epoch,
        committer_tz,
        message,
    ):
        self._repo = repo
        self.id = commit_id
        self.parents = parents
        self.tree = tree
        self.author = author
        self.author_epoch = author_epoch
        self.author_tz = author_tz
        self.committer = committer
        self.committer_epoch = committer_epoch
        self.committer_tz = committer_tz
        self.message = message

        self.author_name, self.author_email = email.utils.parseaddr(
            self.author
        )

        self.committer_name, self.committer_email = email.utils.parseaddr(
            self.committer
        )

        self.subject, self.body = self.message.split("\n", 1)

        self.author_date = Date(self.author_epoch, self.author_tz)
        self.committer_date = Date(self.committer_epoch, self.committer_tz)

        # Only get this lazily when we need it; most of the time it's not
        # required by the caller.
        self._diff = None

    def __repr__(self):
        return "<C %s p:%s a:%s s:%r>" % (
            self.id[:7],
            ",".join(p[:7] for p in self.parents),
            self.author_email,
            self.subject[:20],
        )

    @property
    def diff(self):
        """Return the diff for this commit, in unified format."""
        if not self._diff:
            self._diff = self._repo.diff(self.id)
        return self._diff

    @staticmethod
    def from_str(repo, buf):
        """Parses git rev-list output, returns a commit object."""
        if "\n\n" in buf:
            # Header, commit message
            header, raw_message = buf.split("\n\n", 1)
        else:
            # Header only, no commit message
            header, raw_message = buf.rstrip(), "    "

        header_lines = header.split("\n")
        commit_id = header_lines.pop(0)

        header_dict = defaultdict(list)
        for line in header_lines:
            k, v = line.split(" ", 1)
            header_dict[k].append(v)

        tree = header_dict["tree"][0]
        parents = set(header_dict["parent"])

        authorhdr = header_dict["author"][0]
        author, author_epoch, author_tz = authorhdr.rsplit(" ", 2)

        committerhdr = header_dict["committer"][0]
        committer, committer_epoch, committer_tz = committerhdr.rsplit(" ", 2)

        # Remove the first four spaces from the message's lines.
        message = ""
        for line in raw_message.split("\n"):
            message += line[4:] + "\n"

        return Commit(
            repo,
            commit_id=commit_id,
            tree=tree,
            parents=parents,
            author=author,
            author_epoch=author_epoch,
            author_tz=author_tz,
            committer=committer,
            committer_epoch=committer_epoch,
            committer_tz=committer_tz,
            message=message,
        )


class Date:
    """Handy representation for a datetime from git."""

    def __init__(self, epoch, tz):
        self.epoch = int(epoch)
        self.tz = tz
        self.utc = datetime.datetime.utcfromtimestamp(self.epoch)

        self.tz_sec_offset_min = int(tz[1:3]) * 60 + int(tz[4:])
        if tz[0] == "-":
            self.tz_sec_offset_min = -self.tz_sec_offset_min

        self.local = self.utc + datetime.timedelta(
            minutes=self.tz_sec_offset_min
        )

        self.str = self.utc.strftime("%a, %d %b %Y %H:%M:%S +0000 ")
        self.str += "(%s %s)" % (self.local.strftime("%H:%M"), self.tz)

    def __str__(self):
        return self.str


class Diff:
    """A diff between two trees."""

    def __init__(self, ref, changes, body):
        """Constructor.

        - ref: reference id the diff refers to.
        - changes: [ (added, deleted, filename), ... ]
        - body: diff body, as text, verbatim.
        """
        self.ref = ref
        self.changes = changes
        self.body = body

    @staticmethod
    def from_str(buf):
        """Parses git diff-tree output, returns a Diff object."""
        lines = iter(buf)
        try:
            ref_id = next(lines)
        except StopIteration:
            # No diff; this can happen in merges without conflicts.
            return Diff(None, [], "")

        # First, --numstat information.
        changes = []
        l = next(lines)
        while l != "\n":
            l = l.rstrip("\n")
            added, deleted, fname = l.split("\t", 2)
            added = added.replace("-", "0")
            deleted = deleted.replace("-", "0")
            fname = smstr(unquote(fname))
            changes.append((int(added), int(deleted), fname))
            l = next(lines)

        # And now the diff body. We just store as-is, we don't really care for
        # the contents.
        body = "".join(lines)

        return Diff(ref_id, changes, body)


class Tree:
    """A git tree."""

    def __init__(self, repo: Repo, ref: str):
        self.repo = repo
        self.ref = ref

    @functools.lru_cache
    def ls(
        self, path, recursive=False
    ) -> Iterable[Tuple[str, smstr, Optional[int]]]:
        """Generates (type, name, size) for each file in path."""
        cmd = self.repo.cmd("ls-tree")
        cmd.long = None
        if recursive:
            cmd.r = None
            cmd.t = None

        cmd.arg(self.ref)
        if not path:
            cmd.arg(".")
        else:
            cmd.arg(path)

        files = []
        for l in cmd.run():
            _mode, otype, _oid, size, name = l.split(None, 4)
            if size == "-":
                size = None
            else:
                size = int(size)

            # Remove the quoting (if any); will always give us a str.
            name = unquote(name.strip("\n"))

            # Strip the leading path, the caller knows it and it's often
            # easier to work with this way.
            name = name[len(path) :]

            # We use a smart string for the name, as it's often tricky to
            # manipulate otherwise.
            files.append((otype, smstr(name), size))

        return files


class Blob:
    """A git blob."""

    def __init__(self, raw_content: bytes):
        self.raw_content = raw_content
        self._utf8_content = None

    @property
    def utf8_content(self):
        if not self._utf8_content:
            self._utf8_content = self.raw_content.decode("utf8", "replace")
        return self._utf8_content