Feedback zu Generatorfunktionen und Klassenzusammensetzung in Python 3

Question

Ich habe ein Python-Skript geschrieben, dessen Zweck darin besteht, Protokolle aus CloudWatch zu lesen und sie dann an ElasticSearch zu senden. Es ist noch nicht ganz fertig, aber ich bin so weit fortgeschritten, dass ich von den Rückmeldungen von Python-Experten profitieren kann, insbesondere:

• Ist die Verwendung von Python-Generatoren korrekt und idiomatisch?
• Ist die Verwendung der Klassenzusammensetzung korrekt und idiomatisch?
• Alles andere, was stilistisch für Python 3 ungeeignet ist.

#!/usr/bin/env python3

import json
import time
import uuid
import os
import sys

import boto3
from elasticsearch import Elasticsearch, helpers

client = boto3.client("logs")


def usage() -> None:
    print("Usage: GROUP_NAME=cloudwatch_group ES_HOST=es_host {}".format(
        os.path.basename(__file__)))
    sys.exit(1)

if "GROUP_NAME" not in os.environ:
    usage()
if "ES_HOST" not in os.environ:
    usage()


class CWLogs:
    group_name = os.environ["GROUP_NAME"]

    def events(self) -> None:
        for event in self.__generate_events():
            yield event

    def __generate_streams(self) -> None:
        kwargs = {
            "logGroupName": self.group_name,
        }

        while True:
            stream_batch = client.describe_log_streams(**kwargs)
            yield from stream_batch["logStreams"]
            try:
                kwargs["nextToken"] = stream_batch["nextToken"]
            except KeyError:
                break

    def __generate_events(self) -> None:
        stream_names = \
      [stream["logStreamName"] for stream in self.__generate_streams()]

        for stream_name in stream_names:
            kwargs = {
                "logGroupName":  self.group_name,
                "logStreamName": stream_name,
            }

            while True:
                logs_batch = client.get_log_events(**kwargs)
                yield from logs_batch["events"]
                try:
                    kwargs["nextToken"] = logs_batch["nextToken"]
                except KeyError:
                    break


class ESWriter:
    es_host = os.environ["ES_HOST"]
    elastic = Elasticsearch()

    def post(self, events: object) -> None:
        try:
            response = helpers.bulk(
                self.elastic, self.__transformer(events))
            print("\nRESPONSE:", response)
        except Exception as e:
            print("\nERROR:", e)

    @staticmethod
    def __index_name(timestamp: str) -> str:
        return "eventbridge-auth0-{}".format(
            time.strftime("%Y.%m", time.localtime(timestamp)))

    @staticmethod
    def __normalize(message: str) -> str:
        return message # TODO.

    def __transformer(self, events: object) -> None:
        for event in events:
            yield self.__transform(event)

    def __transform(self, event: dict) -> None:
        timestamp = event["timestamp"]
        index_name = self.__index_name(timestamp)
        message = self.__normalize(event["message"])

        return "\n".join([
            json.dumps({
                "index": {
                    "_id": str(uuid.uuid4()), # TODO. Check
                    "_index": index_name,
                    "_type":  "_doc"}}),
            json.dumps({
                "source": {
                    "@source": "auto-populate script",
                    "@timestamp": timestamp,
                    "@message": message}})])


if __name__ == '__main__':
    ESWriter().post(CWLogs().events())

Answer 1

Code Organisation

Ihre Code-Organisation scheint nicht vorhanden zu sein. Du hast:

• Importe
• Code
• Funktionsdefinition
• Code
• Klassendefinitionen
• Hauptschutzcode

Der Code sollte in einer einheitlicheren Struktur organisiert sein, wie z.

• Importe
• Klassendefinitionen
• Funktionsdefinitionen
• Hauptschutzcode

Der Zweck der Verwendung eines Hauptschutzes besteht darin, zu verhindern, dass Code ausgeführt wird, wenn die Datei in eine andere Datei importiert wird. Hier haben Sie zwei separate Codeblöcke, die unbedingt ausgeführt werden. Dies begrenzt die Wiederverwendung von Code. Stellen Sie sich zum Beispiel vor, jemand könnte es CWLogsfür seine eigene Aufgabe verwenden, braucht es aber nicht ESWriter. Sie versuchen es from your_file import CWLogsund finden ihre Programm-Exits, nachdem eine kryptische Fehlermeldung angezeigt wurde, wie ein Programm ausgeführt werden soll, das sie aufgrund einer fehlenden Umgebungsvariablen, die sie nicht tatsächlich verwenden, nicht ausführen.

sys.exit ()

Nenn das nicht. Es beendet den Python-Interpreter.

Ein Debugging, auf das Sie möglicherweise gehofft haben, wenn das Programm beendet ist, ist unmöglich, da die gesamte Python-Umgebung implodiert ist. Es ist unmöglich, Ihre Datei sicher zu importieren, try: import your_file except ImportError:da die Python-Ausführung während des Imports beendet wird, was bedeutet, dass das Programm, das versucht, sie zu importieren, bedingungslos beendet wird. Wenn Sie versuchen, unittestIhr Programm zu testen, oder Sphinx, um Dokumentation für Ihr Programm oder eine Reihe anderer allgemeiner Dinge zu generieren, können Sie dies nicht, da Ihre Datei den Python-Interpreter bedingungslos beendet hat.

Nenn es nicht.

Stattdessen:

if __name__ == '__main__':
    if {'GROUP_NAME', 'ES_HOST'} <= os.environ.keys():
        main()
    else:
        usage()

Sie müssen nicht usage()anrufen sys.exit(). Nachdem der usage()Aufruf aufgerufen wurde und normal zurückkehrt, erreicht die Ausführung das Ende der Datei. Wenn dies die Hauptprogrammdatei ist, wird das Programm natürlich beendet. Wenn dies nicht die Hauptprogrammdatei ist, hätte der Hauptwächter natürlich keine der beiden Methoden ausgeführt. Die Ausführung würde das Ende der Datei erreichen und den Import der Datei als Modul in ein anderes Programm abschließen.

Hör auf, Klassen zu schreiben

Unter " Stop Writing Classes " finden Sie einen PyCon-Vortrag von Jack Diederich.

Eine Klasse ohne Instanzdatenelemente sollte wahrscheinlich keine Klasse sein. Weder ESWriternoch CWLogsInstanzdatenelemente.

Eine Klasse ohne Konstruktor und nur eine öffentliche Methode, die aufgerufen werden soll, sollte keine Klasse sein. Weder ESWriternoch CWLogseinen Konstruktor. Beide haben eine einzige öffentliche Methode, die unmittelbar nach dem Erstellen einer Klasseninstanz aufgerufen wird, sodass die Instanz nicht einmal gespeichert wird.

Dies sollten keine Klassen sein.

Private Name Mangling

Private Name Mangling wird verwendet, um Kollisionen von privaten Mitgliedsnamen zu verhindern, wenn eine Klasse von einer anderen Klasse abgeleitet wird, normalerweise wenn die Basisklasse und die Ableitungsklasse von verschiedenen Entitäten gesteuert werden. Wenn Sie beispielsweise Ihre eigene Klasse von a ableiten tkinter.Frameund eine _validateMethode in Ihrer Klasse erstellen , kann dies dazu führen, dass die Basisklasse nicht mehr ordnungsgemäß funktioniert, wenn sie über eine eigene _validateMethode verfügt, die nur abrupt geändert wurde. Die Basisklasse würde also verwenden __validate, der führende doppelte Unterstrich würde den Namen "Mangling" auslösen und den Namen durch ersetzen _Frame__validate, sodass Kollisionen weniger wahrscheinlich sind.

Es scheint keinen Grund für die Verwendung eines doppelten Unterstrichpräfixes in Ihren Methodennamen zu geben. Ein einziger Unterstrich wäre idiomatischer.

Geben Sie Hinweise ein

Ihre Typhinweise sind falsch.

Zum Beispiel gibt das Folgende eindeutig ein zurück str, nicht None:

    def __transform(self, event: dict) -> None:
        ...
        return "\n".join( ... )

Da __transformeres die Ergebnisse von liefert __transform, kehrt es auch nicht zurück None, sondern sollte deklariert werden als:

from typing import Generator

...

    def __transformer(self, events: object) -> Generator[str, None, None]:
        ...

Oder einfach:

from typing import Iterator

...

    def __transformer(self, events: object) -> Iterator[str]:
        ...

Und events: objectist praktisch bedeutungslos, da alles in Python ein Objekt ist. Verwenden Sie entweder einen geeigneten Typ dafür oder kümmern Sie sich überhaupt nicht um einen Typhinweis.

Generatorausdrücke

Wie yedpodtrziko bemerkte,

def __generate_events(self) -> None:
    stream_names = [stream["logStreamName"] for stream in self.__generate_streams()]

    for stream_name in stream_names:
        ...

Erstellt eine temporäre Liste, um sie sofort zu durchlaufen. Sie haben den Code ziemlich stark geändert, um die temporäre Liste zu vermeiden. Es gibt eine viel kleinere Änderung, die vorgenommen werden kann:

def __generate_events(self) -> None:
    stream_names = (stream["logStreamName"] for stream in self.__generate_streams())

    for stream_name in stream_names:
        ...

Da es schwierig sein kann, die Änderung zu erkennen, werde ich sie verstärken: Die [...]wurde geändert (...). Dies bedeutet stream_names, dass es nicht als speicherinterne Liste realisiert wird, sondern zu einem Generatorausdruck wird, der die Werte nacheinander erzeugt, wenn er dazu aufgefordert wird.

Hier macht es keinen großen Unterschied, aber wenn stream_nameses an eine Funktion übergeben würde, anstatt lokal verwendet zu werden, müsste die von yedpodtrziko vorgeschlagene Änderung den Code viel weiter entfernt überarbeiten, um stream_objdie Stream-Namen in dieser Funktion zu akzeptieren und zu extrahieren.

Answer 2

Codieren Sie keine Umgebungsvariablen innerhalb der Klassen. An Stelle von:

class CWLogs:
    group_name = os.environ["GROUP_NAME"]

mach es so:

class CWLogs:
    group_name = None

    def __init__(self, group_name):
         self.group_name = group_name

if not GROUP_NAME := getenv('GROUP_NAME'):
    usage()
# pass the variable when initializing the class :
CWLogs(GROUP_NAME)

Dadurch wird der Code wartbarer, da er nicht eng an die Umgebung gebunden ist. variabel, sondern an alles, was Sie dort übergeben, und es wird einfacher sein, Tests für solchen Code zu schreiben. Außerdem müssen Sie die Variable nicht an zwei Stellen wiederholen, was die Wahrscheinlichkeit erhöht, dass Sie an einer Stelle einen Tippfehler machen oder vergessen, ihn an beiden Stellen zu ändern, wenn sich die Funktionalität ändert. Gleiches gilt für die ESWriterKlasse.

dann gibt es diese Funktion:

def __generate_events(self) -> None:
    stream_names = [stream["logStreamName"] for stream in self.__generate_streams()]

    for stream_name in stream_names:
        ...

Hier haben Sie eine unnötige zusätzliche Schleife und eine zusätzliche Liste im Speicher zugewiesen. Zuerst durchlaufen Sie die von zurückgegebenen Daten __generate_streams()und dann noch einmal dieselben Daten. Sie können dies stattdessen tun:

def __generate_events(self) -> None:
    for stream_obj in self.__generate_streams():
        stream_name = stream_obj['logStreamName']