Mudei um pouco a estratégia principal desta atividade para aprimorar o que já sei:
- Utilizei uma classe para trabalhar de forma assíncrona as funções
- Utilizei o método Lock para bloquear quando uma tarefa já estiver sendo utilizada para evitar mudar a váriável utilizada
import asyncio
from typing import List, Dict, Any
class ValidadorAlunosAsync:
def __init__(self, cursos: Dict[str, Dict[str, Any]], alunos: List[Dict[str, str]]):
self.cursos = cursos
self.alunos = alunos
self.lock = asyncio.Lock() # trava para evitar condições de corrida
async def validar_curso_existente(self, aluno: Dict[str, str]) -> bool:
return aluno["curso"] in self.cursos
async def validar_vagas(self, aluno: Dict[str, str]) -> bool:
curso = aluno["curso"]
return len(self.cursos[curso]["inscritos"]) < self.cursos[curso]["vagas"]
async def validar_nao_duplicado(self, aluno: Dict[str, str]) -> bool:
curso = aluno["curso"]
return aluno["nome"] not in self.cursos[curso]["inscritos"]
async def validar_capacidade_zero(self, aluno: Dict[str, str]) -> bool:
curso = aluno["curso"]
return self.cursos[curso]["vagas"] > 0
async def inscrever_aluno(self, aluno: Dict[str, str]):
nome = aluno["nome"]
curso = aluno["curso"]
print(f"Inscrevendo {nome} no curso {curso}...")
async with self.lock:
# Toda a lógica de verificação e inscrição é protegida pela trava
if not await self.validar_curso_existente(aluno):
print(f"O curso {curso} não existe! Inscrição rejeitada.\n")
return
if not await self.validar_capacidade_zero(aluno):
print(f"Turma lotada! {nome} não pôde se inscrever no curso {curso}.\n")
return
if not await self.validar_nao_duplicado(aluno):
print(f"{nome} já está inscrita no curso {curso}! Inscrição rejeitada.\n")
return
if not await self.validar_vagas(aluno):
print(f"Turma lotada! {nome} não pôde se inscrever no curso {curso}.\n")
return
self.cursos[curso]["inscritos"].append(nome)
print(f"Inscrição confirmada para {nome} no curso {curso}!\n")
async def processar_inscricoes(self):
# Executa em paralelo, mas protegendo alterações com Lock
tarefas = [self.inscrever_aluno(aluno) for aluno in self.alunos]
await asyncio.gather(*tarefas)
print("Todas as inscrições foram processadas!")
async def main():
cursos = {
"Python Avançado": {"vagas": 2, "inscritos": []},
"Java para Iniciantes": {"vagas": 1, "inscritos": []},
"Machine Learning": {"vagas": 0, "inscritos": []},
}
alunos = [
{"nome": "Alice", "curso": "Python Avançado"},
{"nome": "Bruno", "curso": "Python Avançado"},
{"nome": "Carlos", "curso": "Java para Iniciantes"},
{"nome": "Daniela", "curso": "Machine Learning"},
{"nome": "Alice", "curso": "Python Avançado"}
]
gerenciador = ValidadorAlunosAsync(cursos, alunos)
await gerenciador.processar_inscricoes()
asyncio.run(main())