Status-Update für Slack basierend auf WLAN-Netzwerk

Code & Zahlen

Auf der Arbeit nutze ich das Team-Chatprogramm Slack. Dort kann man einen Status einstellen. Wenn ich von Zuhause arbeite, möchte ich einen entsprechenden Status haben, damit die Kolleg*innen wissen, dass sie mich nicht im Bürogebäude suchen brauchen. Wenn ich aber im Büro bin, will ich das auch entsprechend ankündigen. Das geht, ich vergesse es allerdings nur immer.

Und daher habe ich ein Programm geschrieben, das die Slack-API nutzt um den Status zu setzen. Welcher Status gesetzt werden soll wird anhand des WLAN-Netzwerkes abgeleitet. So muss ich mir da keine Gedanken mehr machen und der Status wird automatisch korrekt gesetzt.

In diesem Blogeintrag möchte ich ein bisschen etwas zu Softwarearchitektur und Dependency Injection schreiben. Ich habe nämlich recht schnell eine erste Iteration dieses Skripts fertig gehabt und wollte es dann noch mit mehr Architektur erweitern. Einen wirklichen Bedarf für mehr Funktionalität hatte ich nicht. Allerdings wollte ich noch dafür sorgen, dass der Status nicht ständig gesetzt wird. Und das fühlte sich derart nach dem Decorator Pattern an, dass ich umgebaut habe. Aber der Reihe nach.

Das hier ist die Version vom Skript in der rein prozeduralen Form. Es gibt keine Klassen, keine Kapselung in Objekte. Einfach fünf Funktionen für verschiedene Dinge.

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--poll-minutes", type=int, default=5, help="Poll interval in minutes. Default: %(default)d")
    options = parser.parse_args()

    config = get_config()
    while True:
        check_and_update_status(config)
        time.sleep(options.poll_minutes * 60)


def check_and_update_status(config: dict) -> None:
    active_connections = get_active_connections()
    for environment_name, environment in config["environments"].items():
        if environment["network"] in active_connections:
            for slack_name, slack in config["slack"].items():
                print(f"Setting status according to environment “{environment_name}” for Slack workspace “{slack_name}”.")
                set_status(environment["emoji"], environment["text"], slack["token"])
            break


def get_config() -> dict:
    raw_config_path = "~/.config/slack-wlan-status-updater/config.toml"
    config_path = pathlib.Path(raw_config_path).expanduser()
    with open(config_path, 'rb') as f:
        config = tomllib.load(f)
    return config


def set_status(emoji: str, text: str, token: str) -> None:
    midnight = datetime.datetime.combine(datetime.date.today(), datetime.time(23, 59, 59))
    data = {"profile": {"status_emoji": f":{emoji}:", "status_text": text, "status_expiration": midnight.timestamp()}}
    url = "https://slack.com/api/users.profile.set"
    headers = {"Content-type": "application/json", "Authorization": f"Bearer {token}"}
    request = urllib.request.Request(url, data=json.dumps(data).encode(), headers=headers)
    with urllib.request.urlopen(request, timeout=10) as response:
        assert response.status == 200
        response_data = json.loads(response.read().decode())
        assert response_data["ok"], response_data


def get_active_connections() -> str:
    result = subprocess.run(
        ["nmcli", "con", "show", "--active"], capture_output=True, check=True
    )
    output = result.stdout.decode()
    return output

Die main() ließt die Configuration ein und schaut alle paar Minuten mal, in welchem WLAN-Netzwerk wir sind. Die check_and_update_status() schaut nach der aktuellen WLAN-Verbindung und setzt den Status bei Slack. Die get_config() liest die Konfigurationsdatei ein. Die set_status() nutzt die Slack-API zum Setzen des Status. Und die get_active_connections() schaut über Network Manager nach, welche Verbindung wir gerade so alle haben.

Das scheint mir ein einfach verständliches Programm zu sein. Alles ist fest verbunden, es gibt keine Polymorphie, keine Unklarheiten. An sich ist das so vollkommen in Ordnung. Was mir jetzt allerdings noch fehlte war dass der Status nur dann wirklich zu Slack hochgeladen wird, wenn er sich innerhalb des aktuellen Tages auch wirklich verändert hat. Ich wollte unnötige Anfragen an die API sparen. Also musste ich noch irgendwas dazwischenpacken. Und das bietet sich das Decorator Pattern immer an.

So bin ich also losgezogen und habe daraus sozusagen die Enterprise Edition gemacht, angelehnt an die Java EE (Java Enterprise Edition), bei der man irgendwann in einem Luftschloss aus Design Pattern sitzt und eine Dependency Injection Framework das Programm zusammensetzen lässt. Die Konfiguration kann dann über eine XML-Hölle gehen. Das ist also etwas, wovor Python-Entwickler*innen häufig Angst haben. Mir scheint das im richtigen Maß aber durchaus eine Bereicherung zu sein, solange es eben einem Zweck dient.

Status Setter

Angefangen habe ich mit den Status Settern, also einem Interface für etwas, was einen Status setzen kann. Da kann man einen Emoji setzen, einen Text und ein Ablaufdatum:

class StatusSetter(abc.ABC):
    @abc.abstractmethod
    def set_status(self, emoji: str, text: str, expiration: datetime.datetime) -> None:
        raise NotImplementedError()

Und als nächstes habe ich die eigentliche Implementierung mit der Slack-API hinzugefügt:

class SlackStatusSetter(StatusSetter):
    def __init__(self, name: str, token: str):
        self._name = name
        self._token = token

    def set_status(self, emoji: str, text: str, expiration: datetime.datetime) -> None:
        print(f"Setting status for “{self._name}” to “:{emoji}: {text}” …")
        data = {
            "profile": {
                "status_emoji": f":{emoji}:",
                "status_text": text,
                "status_expiration": expiration.timestamp(),
            }
        }
        url = "https://slack.com/api/users.profile.set"
        headers = {
            "Content-type": "application/json",
            "Authorization": f"Bearer {self._token}",
        }
        request = urllib.request.Request(
            url, data=json.dumps(data).encode(), headers=headers
        )
        with urllib.request.urlopen(request, timeout=10) as response:
            assert response.status == 200
            response_data = json.loads(response.read().decode())
            assert response_data["ok"], response_data

Dies ist letztlich der vorherige Code, nur aufgrund der höheren Einrückung jetzt anders formatiert.

Da ich nun das Interface hatte, konnte ich einen Decorator reinbauen, der nur dann weiterreicht, wenn sich der Status auch wirklich verändert hat:

class InterceptingSlackStatusSetterDecorator(StatusSetter):
    def __init__(self, status_setter: StatusSetter):
        self._status_setter = status_setter
        self._last_status = None

    def set_status(self, emoji: str, text: str, expiration: datetime.datetime) -> None:
        new_status = (emoji, text, expiration)
        if new_status != self._last_status:
            self._status_setter.set_status(emoji, text, expiration)
            self._last_status = new_status

Nun ist hier die nötige Logik für das Zwischenspeichern gekapselt und kann transparent herumgewickelt werden.

Dann kann es sein, dass man mehrere Slack-Workspaces hat. Daher habe ich noch ein Aggregate gebaut, mit dem man mehrere Stati setzen kann:

class MultiStatusSetter(StatusSetter):
    def __init__(self, status_setters: Sequence[StatusSetter]):
        self._status_setters = status_setters

    def set_status(self, emoji: str, text: str, expiration: datetime.datetime) -> None:
        for status_setter in self._status_setters:
            status_setter.set_status(emoji, text, expiration)

Status Selector

Zur Auswahl des zu setzenden Statuses habe ich auch noch eine Klasse geschrieben, aber ohne Interface:

Status = collections.namedtuple("Status", ["emoji", "text"])


class StatusSelector:
    def __init__(self, environments):
        self._environments = environments

    def select_status(self) -> Optional[Status]:
        active_connections = _get_active_connections()
        for environment_name, environment in self._environments.items():
            if environment["network"] in active_connections:
                return Status(environment["emoji"], environment["text"])


def _get_active_connections() -> str:
    result = subprocess.run(
        ["nmcli", "con", "show", "--active"], capture_output=True, check=True
    )
    output = result.stdout.decode()
    return output

Die gibt dann einen Tuple zurück, in dem der Emoji und der Text stehen.

Main

Dann gibt es eine Klasse für die Schleife, die jetzt als Abhängikeiten jeweils einen StatusSelector und einen StatusSetter bekommt. Hinter dem StatusSetter kann sich ein MultiStatusSetter verstecken, der mehrere jeweils in InterceptingSlackStatusSetterDecorator verpackte SlackStatusSetter enthält. Oder direkt einen SlackStatusSetter, es macht hier keinen Unterschied.

class MainLoop:
    def __init__(
        self,
        status_selector: StatusSelector,
        status_setter: StatusSetter,
        poll_minutes: int,
    ):
        self._status_selector = status_selector
        self._status_setter = status_setter
        self._poll_minutes = poll_minutes

    def run(self) -> None:
        while True:
            status = self._status_selector.select_status()
            if status is None:
                time.sleep(60)
            else:
                emoji, text = status
                expiration = datetime.datetime.combine(
                    datetime.date.today(), datetime.time(23, 59, 59)
                )
                self._status_setter.set_status(emoji, text, expiration)
                time.sleep(self._poll_minutes * 60)

Dann gibt es eine main(), die die Argumente einliest und dann mit Hilfsfunktionen den Objektgraph zusammenbaut. Dann wird die Schleife gestartet.

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--poll-minutes",
        type=float,
        default=5,
        help="Poll interval in minutes. Default: %(default)d",
    )
    options = parser.parse_args()

    config = get_config()
    status_selector = build_status_selector(config)
    status_setter = build_status_setter(config)
    main_loop = MainLoop(status_selector, status_setter, options.poll_minutes)

    main_loop.run()


def get_config() -> dict:
    raw_config_path = "~/.config/slack-wlan-status-updater/config.toml"
    config_path = pathlib.Path(raw_config_path).expanduser()
    with open(config_path, "rb") as f:
        config = tomllib.load(f)
    return config


def build_status_selector(config: dict) -> StatusSelector:
    return StatusSelector(config["environments"])


def build_status_setter(config: dict) -> StatusSetter:
    slack_status_setters = [
        SlackStatusSetter(name, slack["token"]) for name, slack in config["slack"].items()
    ]
    interceptors = [
        InterceptingSlackStatusSetterDecorator(slack_status_setter)
        for slack_status_setter in slack_status_setters
    ]
    multi = MultiStatusSetter(interceptors)
    return multi

Fazit

Insgesamt ist das deutlich mehr Code, als vorher die einfache Version. Allerdings bietet es auch viel mehr Flexibilität als vorher. Hier muss man schauen, wie viel man noch so erweitern will. Bei diesem einfachen Skript ist es wohl zu viel gewesen, ein einfacher Funktions-Dekorator hätte auch erstmal gereicht.

Auch muss man nun auch noch Design Pattern im Kopf haben, wenn man an diesem Code arbeitet. Kennt man diese aber, so finde ich es einfacher über die Bausteine nachzudenken. Das konnte ich auch so schon auf der Arbeit beobachten. Meine Kollegen haben diverse Hintergründe bezüglich Softwareentwicklung. Manche denken in Pattern, andere nicht. Alle lösen Probleme auf leicht unterschiedliche Arten. Manche kommen mit den Pattern besser klar, andere arbeiten lieber frei von diesen Schablonen und schreiben kompakteren Code. Was besser ist, muss man im Einzelfall entscheiden. Für mich persönlich passen die Pattern besser und ich finde mich besser in Code zurecht, der sie nutzt.