Limitar operações assíncronas com Semaphore

import asyncio

async def processar(item: str, semaforo: asyncio.Semaphore) -> str:
    async with semaforo:
        print(f"início: {item}")
        await asyncio.sleep(1)  # operação assíncrona simulada
        print(f"fim: {item}")
        return item.upper()

import asyncio

async def processar(item: str, semaforo: asyncio.Semaphore) -> str:
    async with semaforo:
        print(f"início: {item}")
        await asyncio.sleep(1)
        print(f"fim: {item}")
        return item.upper()

async def main() -> None:
    itens = ["a", "b", "c", "d"]
    semaforo = asyncio.Semaphore(2)

    async with asyncio.TaskGroup() as grupo:
        tarefas = [
            grupo.create_task(processar(item, semaforo))
            for item in itens
        ]

    resultados = [tarefa.result() for tarefa in tarefas]
    print(resultados)

asyncio.run(main())

import asyncio

async def preparar(item: str) -> str:
    return item.upper()

async def enviar(dado: str) -> str:
    await asyncio.sleep(0.2)  # recurso está ocupado
    return f"resposta para {dado}"

async def fechar() -> None:
    await asyncio.sleep(0)  # encerramento necessário do uso

async def pos_processar(resposta: str) -> str:
    return resposta.replace("resposta", "resultado")

async def processar(item: str, semaforo: asyncio.Semaphore) -> str:
    dado = await preparar(item)  # independente: fora do limite

    async with semaforo:
        resposta = await enviar(dado)  # recurso ocupado
        await fechar()                 # ainda ocupa até encerrar

    return await pos_processar(resposta)  # independente: fora do limite

import asyncio

async def enviar(item: str, semaforo: asyncio.Semaphore) -> str:
    async with semaforo:
        # A permissão já foi adquirida aqui.
        await asyncio.sleep(0.1)  # representa uma operação cooperativa
        if item == "invalido":
            raise ValueError("item recusado")
        return f"enviado: {item}"

# Se sleep, a validação ou outra etapa lançar exceção,
# ou se a tarefa for cancelada, o async with devolve a permissão.
# Não capture CancelledError para ignorá-lo; deixe-o se propagar.

import asyncio

async def usar_recurso(semaforo: asyncio.Semaphore, item: str) -> None:
    try:
        async with asyncio.timeout(5):
            async with semaforo:
                await operacao_externa(item)
    except TimeoutError:
        print(f"{item}: prazo total esgotado")

async def operacao_externa(item: str) -> None:
    await asyncio.sleep(1)
    print(f"{item}: concluído")

import asyncio

async def usar_recurso(semaforo: asyncio.Semaphore, item: str) -> None:
    async with semaforo:
        try:
            async with asyncio.timeout(5):
                await operacao_externa(item)
        except TimeoutError:
            print(f"{item}: operação excedeu o prazo")

async def operacao_externa(item: str) -> None:
    await asyncio.sleep(1)
    print(f"{item}: concluído")

import asyncio
from itertools import batched


async def processar_item(item: str, semaforo: asyncio.Semaphore) -> str:
    # Preparação independente do recurso limitado.
    item_normalizado = item.strip()

    # Somente o uso simulado do recurso consome uma permissão.
    async with semaforo:
        await asyncio.sleep(0.1)
        resposta = item_normalizado.upper()

    # Pós-processamento independente também fica fora do semáforo.
    return f"processado: {resposta}"


async def processar_em_lotes(entradas: list[str]) -> None:
    tamanho_lote = 3
    capacidade = 2
    semaforo = asyncio.Semaphore(capacidade)

    for numero_lote, lote in enumerate(batched(entradas, tamanho_lote), start=1):
        tarefas: list[asyncio.Task[str]] = []

        async with asyncio.TaskGroup() as grupo:
            for item in lote:
                tarefas.append(grupo.create_task(processar_item(item, semaforo)))

        # A saída bem-sucedida do grupo garante que estas tarefas terminaram.
        resultados_lote = [tarefa.result() for tarefa in tarefas]
        print(f"lote {numero_lote}: {resultados_lote}")


async def main() -> None:
    entradas = ["ana", "bia", "caio", "davi", "elis"]
    await processar_em_lotes(entradas)


asyncio.run(main())

import asyncio
from dataclasses import dataclass, field

CAPACIDADE = 2


@dataclass
class Sonda:
    capacidade: int
    semaforo: asyncio.Semaphore = field(init=False)
    liberar: asyncio.Event = field(default_factory=asyncio.Event)
    capacidade_cheia: asyncio.Event = field(default_factory=asyncio.Event)
    ativas: int = 0
    pico: int = 0

    def __post_init__(self) -> None:
        self.semaforo = asyncio.Semaphore(self.capacidade)

    def preparar_onda(self) -> None:
        self.liberar = asyncio.Event()
        self.capacidade_cheia = asyncio.Event()
        self.ativas = 0
        self.pico = 0

    async def operacao(
        self,
        *,
        falhar: bool = False,
        iniciou_tentativa: asyncio.Event | None = None,
    ) -> None:
        # O sinal é emitido antes de tentar adquirir uma permissão.
        if iniciou_tentativa is not None:
            iniciou_tentativa.set()

        async with self.semaforo:
            self.ativas += 1
            self.pico = max(self.pico, self.ativas)
            if self.ativas == self.capacidade:
                self.capacidade_cheia.set()

            try:
                await self.liberar.wait()
                if falhar:
                    raise RuntimeError("falha simulada")
            finally:
                self.ativas -= 1


async def esperar_capacidade_cheia(sonda: Sonda) -> None:
    # Este timeout é apenas uma proteção contra teste travado.
    try:
        async with asyncio.timeout(1):
            await sonda.capacidade_cheia.wait()
    except TimeoutError as erro:
        raise AssertionError("A capacidade não foi ocupada no teste") from erro


def verificar_encerramento(sonda: Sonda) -> None:
    assert sonda.pico <= sonda.capacidade
    assert sonda.pico == sonda.capacidade
    assert sonda.ativas == 0


async def onda_com_sucesso(sonda: Sonda) -> None:
    sonda.preparar_onda()
    tarefas = [
        asyncio.create_task(sonda.operacao())
        for _ in range(sonda.capacidade)
    ]
    await esperar_capacidade_cheia(sonda)
    sonda.liberar.set()
    await asyncio.gather(*tarefas)
    verificar_encerramento(sonda)


async def onda_com_falha(sonda: Sonda) -> None:
    sonda.preparar_onda()
    tarefas = [
        asyncio.create_task(sonda.operacao(falhar=(indice == 0)))
        for indice in range(sonda.capacidade)
    ]
    await esperar_capacidade_cheia(sonda)
    sonda.liberar.set()
    resultados = await asyncio.gather(*tarefas, return_exceptions=True)
    assert any(isinstance(resultado, RuntimeError) for resultado in resultados)
    verificar_encerramento(sonda)


async def cancelamento_durante_espera(sonda: Sonda) -> None:
    sonda.preparar_onda()
    ocupantes = [
        asyncio.create_task(sonda.operacao())
        for _ in range(sonda.capacidade)
    ]
    await esperar_capacidade_cheia(sonda)

    tentativa_iniciada = asyncio.Event()
    aguardando = asyncio.create_task(
        sonda.operacao(iniciou_tentativa=tentativa_iniciada)
    )
    await tentativa_iniciada.wait()
    aguardando.cancel()
    try:
        await aguardando
    except asyncio.CancelledError:
        pass
    else:
        raise AssertionError("A tarefa que aguardava deveria ser cancelada")

    sonda.liberar.set()
    await asyncio.gather(*ocupantes)
    verificar_encerramento(sonda)


async def cancelamento_durante_operacao(sonda: Sonda) -> None:
    sonda.preparar_onda()
    tarefas = [
        asyncio.create_task(sonda.operacao())
        for _ in range(sonda.capacidade)
    ]
    await esperar_capacidade_cheia(sonda)

    tarefas[0].cancel()
    try:
        await tarefas[0]
    except asyncio.CancelledError:
        pass
    else:
        raise AssertionError("A operação admitida deveria ser cancelada")

    sonda.liberar.set()
    await asyncio.gather(*tarefas[1:])
    verificar_encerramento(sonda)


async def main() -> None:
    sonda = Sonda(CAPACIDADE)

    await onda_com_sucesso(sonda)
    await onda_com_falha(sonda)
    await onda_com_sucesso(sonda)  # confirma recuperação após falha

    await cancelamento_durante_espera(sonda)
    await onda_com_sucesso(sonda)  # confirma que espera cancelada não devolveu a mais

    await cancelamento_durante_operacao(sonda)
    await onda_com_sucesso(sonda)  # confirma devolução após cancelamento admitido

    print("Todas as invariantes foram verificadas.")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from itertools import batched

CAPACIDADE = 2       # Máximo de operações usando o recurso.
TAMANHO_LOTE = 3     # Máximo de tarefas de processamento por lote.


async def operacao_simulada(item: str) -> str:
    """Substitua pelo uso real do recurso limitado."""
    await asyncio.sleep(0)
    if item == "falha":
        raise RuntimeError("falha simulada")
    return item.upper()


async def processar(item: str, semaforo: asyncio.Semaphore, operacao) -> str:
    try:
        # Política: os 2 segundos incluem esperar capacidade e usar o recurso.
        async with asyncio.timeout(2):
            async with semaforo:
                return await operacao(item)
    except TimeoutError:
        return f"timeout:{item}"
    except RuntimeError as erro:
        return f"erro:{item}:{erro}"


async def processar_em_lotes(itens: list[str]) -> list[str]:
    semaforo = asyncio.Semaphore(CAPACIDADE)  # Uma instância compartilhada.
    resultados: list[str] = []

    for lote in batched(itens, TAMANHO_LOTE):
        # As referências das tarefas vivem somente durante este lote.
        async with asyncio.TaskGroup() as grupo:
            tarefas = [
                grupo.create_task(processar(item, semaforo, operacao_simulada))
                for item in lote
            ]
        resultados.extend(tarefa.result() for tarefa in tarefas)

    return resultados


class RecursoControlado:
    def __init__(self, capacidade: int) -> None:
        self.capacidade = capacidade
        self.ativas = 0
        self.pico = 0
        self.capacidade_ocupada = asyncio.Event()
        self.liberar = asyncio.Event()

    async def usar(self, item: str) -> str:
        self.ativas += 1
        self.pico = max(self.pico, self.ativas)
        if self.ativas == self.capacidade:
            self.capacidade_ocupada.set()
        try:
            if item == "falha":
                raise RuntimeError("falha controlada")
            await self.liberar.wait()
            return item.upper()
        finally:
            self.ativas -= 1


async def verificar_limites() -> None:
    # Sucesso: há tarefas suficientes para ocupar toda a capacidade.
    semaforo = asyncio.Semaphore(CAPACIDADE)
    recurso = RecursoControlado(CAPACIDADE)
    tarefas = [
        asyncio.create_task(processar(str(indice), semaforo, recurso.usar))
        for indice in range(CAPACIDADE)
    ]
    await asyncio.wait_for(recurso.capacidade_ocupada.wait(), timeout=1)
    assert recurso.pico == CAPACIDADE
    assert recurso.ativas == CAPACIDADE
    recurso.liberar.set()
    assert await asyncio.gather(*tarefas) == ["0", "1"]
    assert recurso.ativas == 0

    # Falha após admissão: async with devolve a permissão automaticamente.
    recurso_falha = RecursoControlado(1)
    resultado = await processar("falha", semaforo, recurso_falha.usar)
    assert resultado.startswith("erro:falha:")
    async with semaforo:  # A permissão continua disponível.
        pass

    # Cancelamento enquanto espera: não há acquire bem-sucedido para devolver.
    recurso_espera = RecursoControlado(1)
    ocupante = asyncio.create_task(processar("ocupante", semaforo, recurso_espera.usar))
    await asyncio.wait_for(recurso_espera.capacidade_ocupada.wait(), timeout=1)
    chegou_para_esperar = asyncio.Event()

    async def aguardar_permissao() -> None:
        chegou_para_esperar.set()
        async with semaforo:
            raise AssertionError("não deveria adquirir durante este cenário")

    aguardando = asyncio.create_task(aguardar_permissao())
    await chegou_para_esperar.wait()
    aguardando.cancel()
    try:
        await aguardando
    except asyncio.CancelledError:
        pass
    recurso_espera.liberar.set()
    await ocupante

    # Cancelamento durante a operação também libera a permissão.
    recurso_cancelado = RecursoControlado(1)
    em_operacao = asyncio.create_task(processar("cancelar", semaforo, recurso_cancelado.usar))
    await asyncio.wait_for(recurso_cancelado.capacidade_ocupada.wait(), timeout=1)
    em_operacao.cancel()
    try:
        await em_operacao
    except asyncio.CancelledError:
        pass
    async with semaforo:
        pass


async def main() -> None:
    resultados = await processar_em_lotes(["ana", "bia", "falha", "davi"])
    print(resultados)
    await verificar_limites()
    print("Verificações concluídas.")


if __name__ == "__main__":
    asyncio.run(main())