API

Implementation of fastgit

Imports

import io, tempfile
from contextlib import redirect_stdout
from fastcore.test import test_eq,expect_fail

---

source

GitRes

def GitRes(
    *args, **kwargs
):

Git output str, carrying the command’s returncode

Git signals “worked, but the answer is no” with exit 1 on its read-only query commands: no matches for grep, no common ancestor for merge-base, a conflicted (but still written) tree from merge-tree --write-tree. Treating those as errors loses output you need, but treating every exit 1 as success would swallow real failures like a rejected push or an empty commit. So results are GitRes strings carrying the returncode, and exit 1 counts as success only for the commands in _ok1 – or wherever an explicit ok_exit allows it.

---

source

callgit

def callgit(
    path, # Directory to run git in
    *args, # git subcommand and arguments
    uname:NoneType=None, # Deprecated; use `pre`
    pre:NoneType=None, # argv prefix, e.g. a sudo wrapper
    ok_exit:NoneType=None, # Extra accepted exit codes (default: 1 for `_ok1` commands)
    runner:NoneType=None, # Replaces the local subprocess: `(path, args, pre) -> (stdout, stderr, returncode)`
):

Run git in path, returning stripped stdout+stderr as a GitRes

callgit returns git’s stdout and stderr combined as a single str, just as a terminal shows them. Call .splitlines() on the result when you need line-oriented output.

with tempfile.TemporaryDirectory() as td:
    msg = callgit(td, 'init', '-b', 'main')
    assert 'Initialized' in msg
    test_eq(callgit(td, 'rev-parse', '--git-dir'), '.git')
    assert '\n' in callgit(td, 'status')

---

source

get_top

def get_top(
    folder, runner:NoneType=None
):

Call self as a function.

with tempfile.TemporaryDirectory() as td:
    test_eq(get_top(td), None)
    msg = callgit(td, 'init', '-b', 'main')
    assert get_top(td) in msg

Git represents a python wrapper for the git command. It will execute every command as if git were being run with the working directory being the directory d passed in at init time.

As a result, callers should take care not to checkout a branch which does not have the directory used to initialize Git.

Pass pre to prefix every git command with additional arguments, e.g. a sudo invocation to run as another user.

Keyword arguments become git options: single-letter names are short options (n=1 → -n 1) and longer names are long options (pretty='format:%s' → --pretty=format:%s), with underscores converted to dashes. True passes the flag alone, and False omits it. Use the __ parameter to pass path arguments after --.

Failed commands print the error and return None. Pass mute_errors=True to silence the message, or raise_exc=True (per call, or at init to make it the instance default) to raise the CalledProcessError instead. Exit 1 from the _ok1 query commands is not a failure, and is returned normally; pass ok_exit to extend that to other commands (e.g. ok_exit=1 to accept merge’s conflicts), or ok_exit=0 to turn it off.

Pass sync=False at init for an async client: see the Async section below.

---

source

Git

def Git(
    d, pre:NoneType=None, raise_exc:bool=False, sync:bool=True, runner:NoneType=None
):

Run git commands in dir d; sync=False makes every command return an awaitable

with tempfile.TemporaryDirectory() as td:
    g = Git(td)
    assert not g.exists
    msg = g.init(b='main')
    assert g.top() in msg

Options compose naturally: values like n=1 are stringified, and False omits the option entirely:

with tempfile.TemporaryDirectory() as td:
    g = Git(td)
    g.init(b='main')
    (g.d/'a.txt').write_text('a')
    g.add('a.txt'); g.commit(m='first')
    test_eq(g.log(n=1, format='%s'), 'first')
    test_eq(g.log(n=1, format='%s', p=False), 'first')

Errors print as a terse one-line message (like git itself) and return None by default, which suits interactive and quick-tooling use. Library code that consumes a command’s output should pass raise_exc (at init, or per call) so failures raise instead of returning None:

with tempfile.TemporaryDirectory() as td:
    test_eq(Git(td).log(mute_errors=True), None)
    s = io.StringIO()
    with redirect_stdout(s): Git(td).log()
    test_eq(s.getvalue().count('\n'), 1)  # errors print as one terse git-style line
    assert s.getvalue().startswith('ERROR: git log: fatal:')
    with expect_fail(CalledProcessError): Git(td, raise_exc=True).log()
    with expect_fail(CalledProcessError): Git(td).log(raise_exc=True)

For the query commands in _ok1, git documents exit 1 as a “no” answer rather than a failure – no merge base, no grep matches, a conflicted (but still written) merge-tree – so exit 1 returns normally for them, even with raise_exc set. Results are GitRes strings whose returncode distinguishes the two cases; for boolean commands like merge-base --is-ancestor, which print nothing, the returncode is the whole answer:

with tempfile.TemporaryDirectory() as td:
    g = Git(td, raise_exc=True)
    g.init(b='main')
    (g.d/'f.txt').write_text('base')
    g.add('f.txt'); g.commit(m='base')
    g.checkout('-b', 'side'); (g.d/'f.txt').write_text('side'); g.commit('-a', m='side')
    g.checkout('main');       (g.d/'f.txt').write_text('main'); g.commit('-a', m='main')
    test_eq(g.merge_base('--is-ancestor', 'main', 'side').returncode, 1)
    test_eq(g.merge_base('--is-ancestor', 'main~1', 'side').returncode, 0)
    conflicted = g.merge_tree('--write-tree', 'main', 'side')
    test_eq(conflicted.returncode, 1)
    assert 'f.txt' in conflicted

An explicit ok_exit overrides the default for any command: here commit’s exit 1 for “nothing to commit” becomes data instead of an error, and ok_exit=0 makes an _ok1 command strict again:

with tempfile.TemporaryDirectory() as td:
    g = Git(td)
    g.init(b='main')
    res = g.commit(m='x', a=True, ok_exit=1)
    test_eq(res.returncode, 1)
    assert 'nothing' in res
    with expect_fail(CalledProcessError): g.grep('x', ok_exit=0, raise_exc=True)

---

source

Git.last_commit

def last_commit():

Call self as a function.

---

source

Git.commits

def commits():

Call self as a function.

A repo with no commits yet has an empty commits list:

with tempfile.TemporaryDirectory() as td:
    g = Git(td)
    g.init(b='main')
    test_eq(g.commits, [])

---

source

Git.current_branch

def current_branch():

Call self as a function.

current_branch returns the currently checked-out branch. Note that git writes checkout’s status messages to stderr, which fastgit includes in the returned output just as a terminal would show them:

with tempfile.TemporaryDirectory() as td:
    g = Git(td)
    g.init(b='main')
    test_eq(g.current_branch, 'main')
    assert 'Switched' in g.checkout('-b', 'tmp')
    test_eq(g.current_branch, 'tmp')

with tempfile.TemporaryDirectory() as td:
    g = Git(td, pre=['env'])
    g.init()
    (Path(td)/'a.txt').write_text('a')
    g.add('a.txt')
    g.commit(m='first')
    test_eq(len(g.commits), 1)
    assert 'first' in g.commits[0]
    with expect_fail(Exception, 'not both'): callgit(td, 'status', uname='x', pre=['env'])

pre changes how a command is wrapped; runner replaces how it is run. Every command funnels through one runner, (path, args, pre) -> (stdout, stderr, returncode), defaulting to a local subprocess. Passing a different one swaps the transport (for example, POSTing the argv to a gateway that runs git elsewhere) while GitRes, ok_exit, and error semantics stay here. A runner never touches the local filesystem unless it chooses to:

calls = []
def runner(path, args, pre=None):
    calls.append([path, args, pre])
    return 'HEAD is at abc1234', '', 0
g = Git('/nonexistent', runner=runner)
test_eq(g.show(s=True), 'HEAD is at abc1234')
test_eq(get_top('/nonexistent', runner=lambda p,a,pre=None: ('/repo\n','',0)), '/repo')
calls

Async

Pass sync=False for an async client: every command returns an awaitable, and git runs in an asyncio subprocess, so an event loop (e.g. a web server) is never blocked while git works. The command surface, option mapping, and error behavior are all identical, and properties follow the client’s mode too, so on an async client they are awaited like everything else (await g.commits). acallgit is the async twin of callgit, with the same contract:

---

source

acallgit

async def acallgit(
    path, # Directory to run git in
    *args, # git subcommand and arguments
    uname:NoneType=None, # Deprecated; use `pre`
    pre:NoneType=None, # argv prefix, e.g. a sudo wrapper
    ok_exit:NoneType=None, # Extra accepted exit codes (default: 1 for `_ok1` commands)
    runner:NoneType=None, # Async runner replacing the local subprocess: `async (path, args, pre) -> (stdout, stderr, returncode)`
):

Async callgit: identical contract, run in an asyncio subprocess

with tempfile.TemporaryDirectory() as td:
    g = Git(td, sync=False)
    await g.init(b='main')
    (g.d/'a.txt').write_text('a')
    await g.add('a.txt'); await g.commit(m='first')
    test_eq(await g.log(n=1, format='%s'), 'first')
    test_eq(len(await g.commits), 1)
    test_eq(await g.current_branch, 'main')
    assert await g.exists
    test_eq((await g.grep('nope')).returncode, 1)
    test_eq(await g.log('--bad-flag', mute_errors=True), None)
    with expect_fail(CalledProcessError): await g.rev_parse('nope', raise_exc=True)