flatpak auto update timer

flatpak-auto-update.py

#!/usr/bin/env python3

from argparse import ArgumentParser
from contextlib import suppress
from functools import cached_property
from sys import argv, exit, stderr
from time import time
import dbus
import os
import shlex
import subprocess as sp

class Updater:
    def __init__(self):
        self.o = None

    def parse_args(self):
        a = ArgumentParser(description='flatpak auto updater')
        a.add_argument('--conf-file', metavar='PATH', action='append', help='load more arguments from file if found')
        a.add_argument('--lockfile', metavar='PATH', help='lockfile path')
        a.add_argument('--interval', metavar='SECONDS', type=int, default=43200, help='minimum update interval')
        a.add_argument('--metered-interval', metavar='SECONDS', type=int, default=86400*7, help='update interval over metered internet')
        a.add_argument('--idle', action='store_true', help='avoid update if not all logind sessions are idle')
        a.add_argument('--no-check-power', action='store_true', help='allow update on battery')
        a.add_argument('--no-check-network', action='store_true', help='allow update without internet')
        a.add_argument('--name', metavar='NAME', default=argv[0], help='program name')
        a.add_argument('update_options', nargs='*', help='options passed to flatpak update')

        self.o = a.parse_args()
        q = self.o.conf_file
        processed = set()
        while q:
            cf = q.pop(0)
            if cf in processed:
                continue
            processed.add(cf)
            with suppress(FileNotFoundError):
                with open(cf) as f:
                    o = a.parse_args(shlex.split(f.read()))
                    self.o.__dict__.update(o.__dict__)
                    if o.conf_file:
                        q.extend(o.conf_file)

        if not self.o.lockfile:
            raise RuntimeError('invalid lockfile')

    def main(self):
        self.parse_args()

        class CondFail(Exception):
            pass

        try:
            lock = None
            newlock = False
            try:
                lock = open(self.o.lockfile, 'r+b')
            except FileNotFoundError:
                lock = open(self.o.lockfile, 'wb')
                newlock = True
            with lock:
                os.lockf(lock.fileno(), os.F_LOCK, 0)
                self.enable_inhibitor()
                if self.o.idle and self.check_idle():
                    raise CondFail('Active session present')
                if not self.o.no_check_power and self.check_power():
                    raise CondFail('Running on battery power')
                if not self.o.no_check_network and self.check_network():
                    raise CondFail('No internet')
                metered = self.is_metered()
                if metered and self.o.metered_interval <= 0:
                    raise CondFail('Metered internet')
                if not newlock and self.check_timestamp(lock, self.o.metered_interval if metered else self.o.interval):
                    raise CondFail(f'Interval ({self.o.interval}) not yet expired')
                self.update_timestamp(lock)
                r = sp.run(['flatpak', 'update', '--noninteractive'] + self.get_update_options(), stdin=sp.DEVNULL, check=False).returncode
                exit(r)
        except CondFail as e:
            print(f'Condition failed: {e}', file=stderr)
            exit(1)

    def get_update_options(self):
        if self.o.update_options:
            return self.o.update_options
        if os.getuid() == 0:
            return ['--system']
        else:
            return ['--user']

    def check_timestamp(self, lock, interval):
        t = os.fstat(lock.fileno()).st_mtime
        now = time()
        return now - t < interval

    def update_timestamp(self, lock):
        lock.truncate(0)

    @cached_property
    def bus(self):
        return dbus.SystemBus()

    @property
    def login1(self):
        return self.bus.get_object('org.freedesktop.login1', '/org/freedesktop/login1')

    def enable_inhibitor(self):
        with suppress(dbus.DBusException):
            m = dbus.Interface(self.login1, 'org.freedesktop.login1.Manager')
            fd = m.Inhibit('sleep', self.o.name, 'Flatpak updating', 'block')

    def check_idle(self):
        with suppress(dbus.DBusException):
            p = dbus.Interface(self.login1, 'org.freedesktop.DBus.Properties')
            return not p.Get('org.freedesktop.login1.Manager', 'IdleHint')
        return False

    def check_power(self):
        with suppress(dbus.DBusException):
            upower = self.bus.get_object('org.freedesktop.UPower', '/org/freedesktop/UPower')
            p = dbus.Interface(upower, 'org.freedesktop.DBus.Properties')
            return p.Get('org.freedesktop.UPower', 'OnBattery')
        return False

    @cached_property
    def nm_props(self):
        nm = self.bus.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager')
        return dbus.Interface(nm, 'org.freedesktop.DBus.Properties')

    def check_network(self):
        with suppress(dbus.DBusException):
            if self.nm_props.Get('org.freedesktop.NetworkManager', 'State') < 60:
                return True
        return False

    def is_metered(self):
        with suppress(dbus.DBusException):
            return self.nm_props.Get('org.freedesktop.NetworkManager', 'Metered') in (1, 3)
        return False

if __name__ == '__main__':
    Updater().main()

flatpak-auto-update.service.in

[Unit]
Description=Flatpak auto update

[Service]
Type=oneshot
ExecStart=@LIBEXEC@/flatpak-auto-update.py --lockfile %C/%N.lock --name %n --conf-file /etc/flatpak-auto-update.conf

flatpak-auto-update.spec.rpkg

Name:    flatpak-auto-update
Version: {{{ git_dir_version }}}
Release: 1%{?dist}
Summary: Flatpak auto update systemd timer
URL:     https://gist.github.com/mentha/5f57451bd93b6193077fce6061f0e703
License: MIT
VCS:     {{{ git_dir_vcs }}}
Source:  {{{ git_dir_pack }}}

BuildArch: noarch
BuildRequires: python3-dbus
BuildRequires: systemd-rpm-macros
%systemd_ordering

%description
Flatpak auto update systemd timer.

%prep
{{{ git_dir_setup_macro }}}

%build
sed "s#@LIBEXEC@#%{_libexecdir}#" < flatpak-auto-update.service.in > flatpak-auto-update.service

%install
install -Dm755 flatpak-auto-update.py %{buildroot}%{_libexecdir}/flatpak-auto-update.py
install -Dm644 flatpak-auto-update.service %{buildroot}%{_unitdir}/flatpak-auto-update.service
install -Dm644 flatpak-auto-update.service %{buildroot}%{_userunitdir}/flatpak-auto-update.service
install -Dm644 flatpak-auto-update.timer %{buildroot}%{_unitdir}/flatpak-auto-update.timer
install -Dm644 flatpak-auto-update.timer %{buildroot}%{_userunitdir}/flatpak-auto-update.timer

%post
%systemd_post flatpak-auto-update.service flatpak-auto-update.timer
%systemd_user_post flatpak-auto-update.service flatpak-auto-update.timer

%preun
%systemd_preun flatpak-auto-update.service flatpak-auto-update.timer
%systemd_user_preun flatpak-auto-update.service flatpak-auto-update.timer

%postun
%systemd_postun flatpak-auto-update.service flatpak-auto-update.timer
%systemd_user_postun flatpak-auto-update.service flatpak-auto-update.timer

%files
%{_libexecdir}/flatpak-auto-update.py
%{_unitdir}/flatpak-auto-update.service
%{_userunitdir}/flatpak-auto-update.service
%{_unitdir}/flatpak-auto-update.timer
%{_userunitdir}/flatpak-auto-update.timer

%changelog
{{{ git_dir_changelog }}}

flatpak-auto-update.timer

[Unit]
Description=Flatpak auto update timer
Wants=network-online.target
After=network-online.target

[Timer]
OnCalendar=hourly

[Install]
WantedBy=timers.target