Python script for ad-hoc continuous integration from Github to publish for a small group of collaborators

Question

I'm doing a little more grad school, and for a group project I wanted us to be able to quickly share and update our project, so I coded up a Python script to handle that. As usual, I'm proud of the work, but I'm here for you to tear it to shreds again.

First, the imports, some globals, and the main:

import time
from subprocess import run
from pathlib import Path
from datetime import datetime
from shlex import split


WD = Path.home() / 'project_name'
SERVEDIR = Path('/var/www/main/project_name')


def main():
    while True:
        just_built = False
        try:
            if git_pull():
                print('pulled at', datetime.now())
                build()
                print('built at', datetime.now())
                move()
                list_index()
                just_built = True
        except Exception as error:
            print(repr(error))
        print('polled at ', datetime.now())
        if not just_built:
            time.sleep(5 * 60)

The main function outlines the work of the script. The just_built variable ensures that if we just built the project (probably more than 5 minutes to do) we don't sleep for another 5 minutes, we first do another git_pull(). The try wasn't really used, but it would keep the script running if there was an problem encountered. The rest is straightforward.

The WD is the working directory where the git repo resides. To make this work I did have to chown the serving subdirectory to my user from root.

As a matter of style, I prefer to put my main function at the top - it's where the outline or table of contents should go, right? It calls the following functions in the rest of the script:

def git_pull():
    proc = run(split("git pull --verbose"), cwd=WD, capture_output=True)
    print(proc.stdout)
    return b"Already up to date." not in proc.stdout


def build():
    run(split('nix-shell --pure --command "make all"'), cwd=WD)


def move():
    timestamp = datetime.now().isoformat(timespec="minutes", sep=" ")
    new_name = f'project{timestamp}.'
    for ext in ('pdf', 'html'):
        new = SERVEDIR / (new_name + ext)
        (WD / f'project.{ext}').rename(new)
        symlink = SERVEDIR / f'project.{ext}'
        symlink.unlink(missing_ok=True)
        symlink.symlink_to(new)


def list_index():
    files = sorted(SERVEDIR.iterdir())
    files = [f'<a href="{f.name}">{f.name}</a>'
             for f in files if 'project' in f.name]
    index = SERVEDIR / 'index.html'
    index.write_text('\n<br>\n'.join(files))       


if __name__ == '__main__':
    main()

To sum up, I poll every 5 minutes with git pull and if we don't pull down anything, we don't build. I do this under the presumption that git has the best API to check to see if there's anything to do. Yes I could have used github webhooks instead of polling, but I'm not set up to accept POSTs yet (and not sure I want to expose that functionality yet...) and besides, github didn't complain.

To build, it calls 'nix-shell --pure --command "make all"'. To sum up, Nix ensures the requirements (via shell.nix, at the bottom) and then make runs the all in my makefile:

.PHONY : all
all:
    Rscript -e 'rmarkdown::render("project.Rmd", "all")'

In spite of calling rmarkdown::render one time, it seems to re-run all the R code twice.

The upsides are all I had to do to kick off a build was

git commit -ac "descriptive comment" && git push

(and then pull, reconcile, merge any changes, and re-push, if it's necessary.)

Other features:

• retain every build (quick output comparisons, see image below), listed in index.html
• canonical link points to the latest build (quick collaboration)
• lots of builds every day (fast iteration)
• updates merge early and often (continuously integrate)
• simple, easy to maintain Python

One downside of this approach versus alternatives is that I had to have a user shell open and running it. I could have detached via tmux, but... I didn't. I'm locked down at home anyways, so no big deal.

Other downsides:

• no unit tests or types checked with mypy
• no style checking
• not represented by any kind of reusable object model, just functions written in a very side-effect-y way - like a script.

Alternatives

I could have written this as a shell script, but I'm not an expert on shell substitution rules (yet). That might be a good response - how to do this with a shell script. I doubt we'd get noticeable improvements in performance or stability with a shell script, though.

I could have used a cron job to run this every 5 minutes (without the while loop) but that seemed like unnecessary configuration fiddling, with the problem of which user to run under as well (a user with minimal perms, naturally).

I could have also used Jenkins (which I will eventually get set up with regardless) but I didn't have the time to set it up, and until I do some version of this script will work fine.

shell.nix

Here's my shell.nix file, which ensures my requirements are in-place in the environment (i.e. in my PATH file) before building the project. I'm using NixOS on this server, so Nix is a natural choice for this purpose:

{ pkgs ? import <nixpkgs> {} }:
with pkgs;
  mkShell {
    buildInputs = [
      texlive.combined.scheme-full
      entr
      ncurses # for tput
      tree
      R
      pandoc
      rPackages.choroplethr
      rPackages.rmarkdown
      rPackages.nycflights13
      rPackages.viridis
      rPackages.tidyverse
      rPackages.ALSM
      rPackages.nortest
      rPackages.alr4
      rPackages.lmtest
      rPackages.EnvStats
      rPackages.GGally
    ];
    shellHook = ''
      source ~/.bashrc || source /etc/bashrc
    '';
  }

This all built an Rmarkdown file that's also the work of others, so we can't show that here.

The question is, how do I improve my code?

Answer 1

I could have used a cron job to run this every 5 minutes (without the while loop) but that seemed like unnecessary configuration fiddling, with the problem of which user to run under as well (a user with minimal perms, naturally).

Instead of cron, you can register this as a systemd service with a timer. This elevates the timing out of the Python script and enhances control over timing tenfold. You can then work with return codes of the Python script for further action (e.g. Restart and RestartSec=300 keywords). This can get rid of the currently awkward try/except blocks and the helper just_built.

NixOS seems to have the capability (under Creating Periodic Services).

At the end, you have a clear and clean Python script that does not have to be kept alive continuously and does one thing well. Further, you leave the timing and success handling to a facility that is much better at it than a time.sleep can hope to be.

The overhead configuration is not a lot; you seem to have more scripting experience than me, and I managed just fine. To get you started, the systemd.service file can be:

[Unit]
Description=Build git project continuously

[Service]
# Type=oneshot is default
Type=oneshot

# User= is required to find ~/.ssh for GitHub.
# Otherwise, User=root is default, which will fail to find keys
User=<user>

WorkingDirectory=/home/<user>/path/to/repo

ExecStart=/usr/bin/python3 -m <your module/package>

You can even play with things like ExecStartPre=/usr/bin/git pull to separate out the git pull part, which seems more natural as a systemd command than in a Python script (since there, it requires run, split, ...).

More info on the .service syntax is found here, and here is more info for the .timer syntax.

---

As a second thought, the git_pull function does not seem terribly robust. A quick check reveals that a git pull when already up-to-date returns 0 (which is fine, but not useful here), which is probably why you implemented the function the way you did. But what if that status message string ever changes?

A different approach is found here, and put into your code, it can look like:

from subprocess import run
from shlex import split
from pathlib import Path

WD = Path.cwd()

def git_pull(work_dir):
    # Instead of lambda, maybe use functools.partial:
    cwd_run = lambda cmd: run(split(cmd), cwd=work_dir, capture_output=True)

    cwd_run("git remote update")
    current_branch_short = "@"
    upstream_branch_short = "@{u}"
    current_branch_hash = cwd_run(f"git rev-parse {current_branch_short}").stdout
    upstream_branch_hash = cwd_run(f"git rev-parse {upstream_branch_short}").stdout

    branches_diverged = current_branch_hash != upstream_branch_hash
    if branches_diverged:
        cwd_run("git pull")
    return branches_diverged

git_pull(work_dir=WD)

This is more robust in the sense that it does not rely on a specific string in stdout. However, it has two distinct disadvantages:

1. It polls the remote twice; once to update, once to actually pull. This overhead is probably not a lot.

2. The test can only check if branches have diverged, but not in which direction. If your local is ahead, the test passes as True and git pull is triggered, which does not make sense. Since this is run on your server that only ever pulls in changes and never has local ones, it is probably fine. In that case, a branch diversion is always equal to a remote change that requires a pull.

   In the link above, this disadvantage is solved using git merge-base @ @{u}, yielding a base at which the branches have diverged. So if implemented correctly/fully (not necessary for your case), it is not really a disadvantage.

See if this can work for you, since it is not a strict (no downsides) improvement.

Answer 2

Great Code & Question

Your codes look great and you seem to be a Python master already, but I would just raise a very minor issue that I'm not even myself good at it (not to mention that I'm not really a code reviewer and here is a good review).

You can certainly improve on naming your variables much better:

• just_built, maybe realtime_built or near_realtime_built could be easier to understand.
• new_name, I guess updated_project_name might be closer.
• index for instance could be index_html, or maybe something better.
• files
• list_index, maybe get_index_htmls might be a bit elaborative.

are some examples.