Feedback sobre funções geradoras e composição de classes em Python 3

Question

Eu escrevi um script Python cujo objetivo é ler os logs do CloudWatch e depois publicá-los no ElasticSearch. Não está totalmente concluído, mas progredi o suficiente para poder me beneficiar do feedback de especialistas em Python, especificamente:

• O uso de Geradores Python é correto e idiomático
• O uso da composição da classe é correto e idiomático
• Qualquer outra coisa inadequada em termos de estilo para Python 3.

#!/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

Organização de Código

Sua organização de código parece inexistente. Você tem:

• importações
• código
• definição de função
• código
• definições de classe
• código principal guardado

O código deve ser organizado em uma estrutura mais consistente, como:

• importações
• definições de classe
• definições de função
• código principal guardado

O objetivo de usar um protetor principal é evitar que o código seja executado se o arquivo for importado para outro arquivo. Aqui, você tem dois blocos de código separados que são executados incondicionalmente. Isso limita a reutilização do código. Por exemplo, imagine que alguém poderia usar CWLogspara sua própria tarefa, mas não precisa ESWriter. Eles tentam from your_file import CWLogse encontram suas saídas de programa depois de exibir uma mensagem de erro enigmática sobre como executar um programa que não estão realmente executando, devido à falta de uma variável de ambiente que eles não usam de fato.

sys.exit ()

Não chame isso. Ele termina o interpretador Python.

Qualquer depuração que você possa esperar fazer quando o programa terminar será impossível, porque todo o ambiente Python implodiu. É impossível importar seu arquivo com segurança usando try: import your_file except ImportError:porque a execução do Python termina durante a importação, o que significa que o programa que tentava importá-lo foi finalizado incondicionalmente. Se você tentar usar unittestpara testar seu programa, ou o Sphinx para gerar documentação para seu programa, ou qualquer outra coisa comum, você não poderá, porque seu arquivo encerrou incondicionalmente o interpretador Python.

Não chame isso.

Em vez de:

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

Não há necessidade de usage()ligar sys.exit(). Após usage()ser chamado, e retornar normalmente, a execução chega ao final do arquivo, que se este for o arquivo principal do programa, naturalmente encerrará o programa. Claro, se este não for o arquivo do programa principal, o guarda principal não teria executado nenhum dos métodos, a execução chegaria ao final do arquivo, completando a importação do arquivo como um módulo em outro programa.

Pare de escrever aulas

Veja " Pare de Escrever Classes " para uma palestra PyCon de Jack Diederich.

Uma classe sem membros de dados de instância provavelmente não deveria ser uma classe. Nem ESWriternem CWLogstem quaisquer membros instância de dados.

Uma classe sem construtor e apenas um método público para chamar não deve ser uma classe. Nem ESWriternem CWLogstem construtor. Ambos têm um único método público, chamado imediatamente após a construção de uma instância de classe, portanto, a instância nem mesmo é salva.

Estas não devem ser aulas.

Nome privado mutilado

A mutilação de nomes privados é usada para evitar colisões de nomes de membros privados quando uma classe é derivada de outra classe, normalmente quando a classe base e a classe derivada estão sob controle de entidades diferentes. Por exemplo, se você derivar sua própria classe de a tkinter.Framee criar um _validatemétodo em sua classe, poderá fazer com que a classe base pare de funcionar corretamente se tiver seu próprio _validatemétodo que foi alterado abruptamente nela. Assim, a classe base usaria __validate, o sublinhado duplo inicial acionaria o nome "mutilamento" e substituiria o nome por _Frame__validate, portanto, as colisões são menos prováveis.

Parece não haver razão para o uso de um prefixo de sublinhado duplo nos nomes dos métodos; um único sublinhado seria mais idiomático.

Dicas de tipo

Suas dicas de tipo estão erradas.

Por exemplo, o seguinte está claramente retornando a str, não None:

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

Como __transformerestá produzindo os resultados de __transform, também não está retornando None, mas deve ser declarado como:

from typing import Generator

...

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

Ou simplesmente:

from typing import Iterator

...

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

E events: objecté virtualmente sem sentido, já que tudo em Python é um objeto. Use um tipo adequado para ele ou nem se preocupe com uma dica de tipo.

Expressões do gerador

Como yedpodtrziko observou,

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

    for stream_name in stream_names:
        ...

cria uma lista temporária, apenas para iterar imediatamente por ela. Eles fizeram uma mudança bastante grande no código para evitar a lista temporária. Há uma mudança muito menor que pode ser feita:

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

    for stream_name in stream_names:
        ...

Porque pode ser difícil ver a mudança, vou amplificá-la: o [...]mudou para (...). Isso significa que, em vez de stream_namesser realizada como uma lista na memória, ela se torna uma expressão geradora, que produzirá os valores um de cada vez quando solicitado.

Não faz muita diferença aqui, mas se stream_namesestivesse sendo passada para uma função, em vez de ser usada localmente, a mudança proposta por yedpodtrziko exigiria retrabalhar o código muito mais longe para aceitar stream_obje extrair os nomes de fluxo dentro dessa função.

Answer 2

Não codifique variáveis ​​de ambiente dentro das classes. Em vez disso:

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

faça isso deste modo:

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)

Isso tornará o código mais sustentável, pois não o vincula fortemente ao env. variável, mas sim para tudo o que você vai passar lá e será mais fácil escrever testes para esse código. Além disso, você não precisa repetir a variável em dois lugares, o que aumentará a chance de você cometer um erro de digitação em um lugar ou se esquecer de modificá-la em ambos quando a funcionalidade for alterada. O mesmo acontece com a ESWriterclasse.

então há esta função:

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

    for stream_name in stream_names:
        ...

Aqui você tem um loop extra desnecessário e uma lista extra alocada na memória. Primeiro você itera pelos dados retornados de __generate_streams()e, em seguida, itera pelos mesmos dados mais uma vez. Em vez disso, você pode fazer isso:

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