Reduzir o pico e a retenção de memória

def ids_ativos(registros: list[dict[str, object]]) -> list[int]:
    trabalho = registros[:]  # novo contêiner com as mesmas referências
    return [registro["id"] for registro in trabalho if registro["ativo"]]

def ids_ativos(registros: list[dict[str, object]]) -> list[int]:
    return [registro["id"] for registro in registros if registro["ativo"]]

import tracemalloc

QUANTIDADE = 30_000


def criar_registro(indice: int) -> dict[str, object]:
    return {
        "id": indice,
        "ativo": indice % 3 == 0,
        "texto": f"registro-{indice:06d}-" + ("x" * 240),
    }


def criar_lista(quantidade: int):
    return [criar_registro(indice) for indice in range(quantidade)]


def gerar_registros(quantidade: int):
    for indice in range(quantidade):
        yield criar_registro(indice)


def normalizar(registro: dict[str, object]) -> dict[str, int]:
    return {
        "id": int(registro["id"]),
        "tamanho": len(str(registro["texto"])),
    }


def resumir(registros) -> dict[str, int]:
    quantidade = 0
    total_caracteres = 0
    for registro in registros:
        quantidade += 1
        total_caracteres += registro["tamanho"]
    return {
        "quantidade": quantidade,
        "total_caracteres": total_caracteres,
    }


# Fonte, seleção e saída transformada coexistem.
def versao_materializada(quantidade: int) -> dict[str, int]:
    entrada = criar_lista(quantidade)
    selecionados = [registro for registro in entrada if registro["ativo"]]
    normalizados = [normalizar(registro) for registro in selecionados]
    return resumir(normalizados)


# Não há listas intermediárias, mas entrada permanece viva.
def versao_gerador_sobre_lista(quantidade: int) -> dict[str, int]:
    entrada = criar_lista(quantidade)
    normalizados = (
        normalizar(registro)
        for registro in entrada
        if registro["ativo"]
    )
    return resumir(normalizados)


# A própria fonte é incremental; só o resumo é retido.
def versao_totalmente_incremental(quantidade: int) -> dict[str, int]:
    normalizados = (
        normalizar(registro)
        for registro in gerar_registros(quantidade)
        if registro["ativo"]
    )
    return resumir(normalizados)


def medir(nome: str, funcao, quantidade: int) -> dict[str, int]:
    tracemalloc.start()
    try:
        # O rastreamento já está ativo; a fonte será criada dentro da função.
        tracemalloc.reset_peak()
        resultado = funcao(quantidade)
        atual, pico = tracemalloc.get_traced_memory()
    finally:
        tracemalloc.stop()

    print(f"{nome:28} atual={atual / 1024 / 1024:6.2f} MiB  "
          f"pico={pico / 1024 / 1024:6.2f} MiB  resumo={resultado}")
    return resultado


resultados = [
    medir("materializada", versao_materializada, QUANTIDADE),
    medir("gerador sobre lista", versao_gerador_sobre_lista, QUANTIDADE),
    medir("totalmente incremental", versao_totalmente_incremental, QUANTIDADE),
]

assert resultados[0] == resultados[1] == resultados[2]
print("Resumos equivalentes.")

import tracemalloc

MIB = 1024 * 1024


def criar_registros(quantidade: int) -> list[dict[str, object]]:
    return [
        {
            "id": indice,
            "categoria": f"grupo-{indice % 20}",
            "payload": "x" * (300 + (indice % 5) * 200),
        }
        for indice in range(quantidade)
    ]


def transformar(registros: list[dict[str, object]]) -> int:
    # A lista transformada é uma estrutura temporária que coexiste com a entrada.
    saida = [
        f"{registro['id']}|{registro['categoria']}|{registro['payload']}"
        for registro in registros
    ]
    return sum(len(item) for item in saida)  # resumo compacto


def pico_do_lote(quantidade: int) -> int:
    tracemalloc.start()
    registros = criar_registros(quantidade)
    transformar(registros)
    _, pico = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return pico


amostra = 1_000
pico_vazio = pico_do_lote(0)
pico_amostra = pico_do_lote(amostra)
custo_por_registro = (pico_amostra - pico_vazio) / amostra

print(f"Pico vazio: {pico_vazio / MIB:.2f} MiB")
print(f"Pico da amostra: {pico_amostra / MIB:.2f} MiB")
print(f"Custo efetivo estimado: {custo_por_registro:.0f} bytes/registro")

import tracemalloc
from itertools import islice

MIB = 1024 * 1024
ORCAMENTO = 12 * MIB


def registros_sinteticos(quantidade: int):
    for indice in range(quantidade):
        yield {
            "id": indice,
            "categoria": f"grupo-{indice % 20}",
            "payload": "x" * (300 + (indice % 5) * 200),
        }


def em_lotes(fonte, capacidade: int):
    while lote := list(islice(fonte, capacidade)):
        yield lote


def resumir_lote(lote: list[dict[str, object]]) -> int:
    # Esta saída temporária pode ser maior que a entrada do lote.
    codificados = [
        f"{registro['id']}|{registro['categoria']}|{registro['payload']}"
        for registro in lote
    ]
    return sum(len(item) for item in codificados)


def medir(capacidade: int, quantidade_total: int = 10_000) -> tuple[int, int]:
    tracemalloc.start()
    total = 0

    for lote in em_lotes(registros_sinteticos(quantidade_total), capacidade):
        total += resumir_lote(lote)  # consome o resumo antes do próximo lote
        del lote

    atual, pico = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return total, pico


for capacidade in (1536, 2048, 2560):
    total, pico = medir(capacidade)
    situacao = "dentro" if pico <= ORCAMENTO else "acima"
    print(
        f"lote={capacidade:4d} | total={total} | "
        f"pico={pico / MIB:.2f} MiB | {situacao} do orçamento"
    )

import tracemalloc


def processar(quantidade: int):
    registros = [
        {"id": i, "valor": i % 97, "texto": f"registro-{i}"}
        for i in range(quantidade)
    ]
    return (registro["valor"] * 2 for registro in registros)


def executar_ciclo(quantidade: int) -> int:
    # sum consome toda a saída; nenhum lote é devolvido ao chamador.
    return sum(processar(quantidade))


quantidade = 30_000
ciclos_medidos = 6
medidas = []  # apenas (ciclo, atual, pico), não os registros

tracemalloc.start()

# Aquecimento: mesmo trabalho, mas fora da série comparada.
executar_ciclo(quantidade)
tracemalloc.reset_peak()

for ciclo in range(1, ciclos_medidos + 1):
    resultado = executar_ciclo(quantidade)
    atual, pico = tracemalloc.get_traced_memory()
    medidas.append((ciclo, atual, pico))
    print(f"ciclo={ciclo} resultado={resultado} atual={atual:,} pico={pico:,}")
    tracemalloc.reset_peak()

tracemalloc.stop()

class Processador:
    def __init__(self) -> None:
        self.historico: list[dict[str, int]] = []
        self.ultimo_lote: list[dict[str, int]] | None = None

    def processar_antes(self, lote: list[dict[str, int]]) -> dict[str, int]:
        self.ultimo_lote = lote  # mantém todos os registros após o retorno
        resumo = {
            "quantidade": len(lote),
            "total": sum(registro["valor"] for registro in lote),
        }
        self.historico.append(resumo)
        return resumo


class ProcessadorAjustado:
    def __init__(self) -> None:
        self.historico: list[dict[str, int]] = []

    def processar(self, lote: list[dict[str, int]]) -> dict[str, int]:
        resumo = {
            "quantidade": len(lote),
            "total": sum(registro["valor"] for registro in lote),
        }
        self.historico.append(resumo)  # persiste somente o contrato de histórico
        return resumo

def resumir_em_lotes(registros, tamanho_lote):
    """Produz lotes de resumos sem alterar os registros de origem."""
    if tamanho_lote <= 0:
        raise ValueError("tamanho_lote deve ser positivo")

    lote = []
    for registro in registros:
        # KeyError é preservado se um registro não tiver "valor".
        resumo = {
            "id": registro["id"],
            "dobro": registro["valor"] * 2,
        }
        lote.append(resumo)

        if len(lote) == tamanho_lote:
            yield lote
            lote = []

    if lote:
        yield lote

import pytest
from processamento import resumir_em_lotes


def consumir(registros, tamanho_lote):
    return [item for lote in resumir_em_lotes(registros, tamanho_lote)
            for item in lote]


def test_resultado_e_entrada_preservada():
    origem = [{"id": 1, "valor": 3}, {"id": 2, "valor": 5}]
    antes = [registro.copy() for registro in origem]

    assert consumir(origem, 2) == [
        {"id": 1, "dobro": 6},
        {"id": 2, "dobro": 10},
    ]
    assert origem == antes


def test_saidas_mutaveis_nao_compartilham_estado():
    saida = consumir([{"id": 1, "valor": 7}, {"id": 2, "valor": 9}], 1)

    saida[0]["dobro"] = 999
    assert saida[1]["dobro"] == 18
    assert saida[0] is not saida[1]


def test_vazio_e_lote_incompleto():
    assert list(resumir_em_lotes([], 3)) == []
    assert list(resumir_em_lotes([{"id": 1, "valor": 4}], 3)) == [
        [{"id": 1, "dobro": 8}]
    ]


def test_registro_grande_e_dado_invalido():
    grande = {"id": 1, "valor": 10**1000}
    assert consumir([grande], 1)[0]["dobro"] == 2 * 10**1000

    fluxo = resumir_em_lotes([{"id": 1, "valor": 2}, {"id": 2}], 1)
    assert next(fluxo) == [{"id": 1, "dobro": 4}]
    with pytest.raises(KeyError):
        next(fluxo)


def test_tamanho_invalido():
    with pytest.raises(ValueError):
        list(resumir_em_lotes([], 0))

import timeit
import tracemalloc
from processamento import resumir_em_lotes


def fonte(quantidade=20_000):
    for indice in range(quantidade):
        yield {"id": indice, "valor": indice}


def consumir_compacto(funcao):
    quantidade = 0
    total = 0
    for lote in funcao(fonte(), 200):
        for resumo in lote:
            quantidade += 1
            total += resumo["dobro"]
    return quantidade, total


def medir_memoria(funcao):
    tracemalloc.start()
    tracemalloc.reset_peak()
    resultado = consumir_compacto(funcao)
    atual, pico = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return resultado, atual, pico


def medir_tempo(funcao):
    # Sem tracemalloc: a instrumentação alteraria a comparação de tempo.
    temporizador = timeit.Timer(lambda: consumir_compacto(funcao))
    return min(temporizador.repeat(repeat=5, number=3)) / 3

resultado, atual, pico = medir_memoria(resumir_em_lotes)
tempo = medir_tempo(resumir_em_lotes)
print(f"resultado={resultado}, atual={atual}, pico={pico}, tempo={tempo:.6f}s")

from itertools import islice
import tracemalloc

N_REGISTROS = 40_000
TAMANHO_LOTE = 200
CICLOS = 3


def gerar_registros(quantidade: int):
    for indice in range(quantidade):
        yield {
            "id": indice,
            "valor": (indice * 17) % 1_000,
            "ativo": indice % 3 != 0,
        }


def transformar(registro: dict) -> int:
    # Leitura בלבד: não altera o registro recebido.
    return registro["valor"] * 2 if registro["ativo"] else 0


class ProcessadorInicial:
    def __init__(self, tamanho_lote: int):
        if tamanho_lote <= 0:
            raise ValueError("tamanho_lote deve ser positivo")
        self.tamanho_lote = tamanho_lote
        self.last_batch = None  # vínculo persistente desnecessário

    def processar(self, registros: list[dict]) -> dict:
        # Fonte completa + todas as fatias coexistem.
        lotes = [
            registros[inicio : inicio + self.tamanho_lote]
            for inicio in range(0, len(registros), self.tamanho_lote)
        ]
        total = 0
        quantidade = 0
        for lote in lotes:
            transformados = [transformar(registro) for registro in lote]
            total += sum(transformados)
            quantidade += len(transformados)
            self.last_batch = transformados
        return {"quantidade": quantidade, "total": total}


class ProcessadorAjustado:
    def __init__(self, tamanho_lote: int):
        if tamanho_lote <= 0:
            raise ValueError("tamanho_lote deve ser positivo")
        self.tamanho_lote = tamanho_lote

    def processar(self, registros) -> dict:
        total = 0
        quantidade = 0
        fonte = iter(registros)
        while lote := list(islice(fonte, self.tamanho_lote)):
            # O lote é consumido antes de obter o próximo.
            for registro in lote:
                total += transformar(registro)
                quantidade += 1
        return {"quantidade": quantidade, "total": total}


def medir(processador, criar_fonte) -> tuple[dict, int, int]:
    tracemalloc.start()
    try:
        resultado = None
        for _ in range(CICLOS):
            # A fonte é recriada e consumida dentro da fronteira medida.
            resultado = processador.processar(criar_fonte())
        atual, pico = tracemalloc.get_traced_memory()
        return resultado, atual, pico
    finally:
        tracemalloc.stop()


def verificar_contrato() -> None:
    registros = list(gerar_registros(20))
    entrada_antes = [registro.copy() for registro in registros]

    inicial = ProcessadorInicial(4)
    ajustado = ProcessadorAjustado(4)
    resumo_inicial = inicial.processar(registros)
    resumo_ajustado = ajustado.processar(gerar_registros(20))

    assert resumo_inicial == resumo_ajustado
    assert registros == entrada_antes, "a entrada não pode ser alterada"
    assert resumo_inicial is not resumo_ajustado, "cada chamada retorna seu próprio resumo"

    for classe in (ProcessadorInicial, ProcessadorAjustado):
        try:
            classe(0)
        except ValueError:
            pass
        else:
            raise AssertionError("lote inválido deve continuar falhando")


verificar_contrato()

inicial = ProcessadorInicial(TAMANHO_LOTE)
ajustado = ProcessadorAjustado(TAMANHO_LOTE)

resumo_inicial, atual_inicial, pico_inicial = medir(
    inicial,
    lambda: list(gerar_registros(N_REGISTROS)),
)
resumo_ajustado, atual_ajustado, pico_ajustado = medir(
    ajustado,
    lambda: gerar_registros(N_REGISTROS),
)

assert resumo_inicial == resumo_ajustado
assert inicial.last_batch is not None
assert not hasattr(ajustado, "last_batch")

print("Testes de contrato: OK")
print(f"Resumo equivalente: {resumo_ajustado}")
print(f"Inicial  — atual: {atual_inicial / 1024:.1f} KiB | pico: {pico_inicial / 1024:.1f} KiB")
print(f"Ajustado — atual: {atual_ajustado / 1024:.1f} KiB | pico: {pico_ajustado / 1024:.1f} KiB")
print(f"Lote escolhido: {TAMANHO_LOTE} registros")