b203b7d22c
Implement workflows for fake data generation, scouter processing, and core scouter operations - Added `FakeData` workflow to generate random data and send it to a Kafka topic. - Implemented `Scouter` workflow to load data from Kafka and trigger the core scouter workflow. - Created `CoreScouter` workflow to process data through quality gates, aggregation, and export to PostgreSQL. - Developed comprehensive unit tests for activities and workflows, ensuring proper functionality and error handling. - Enhanced Redis and Postgres activities with robust testing for data handling and error notifications. - Introduced quality filters for data validation and implemented tests to verify their functionality.
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
from temporalio.client import Client, Schedule, ScheduleActionStartWorkflow, ScheduleSpec, ScheduleIntervalSpec
|
|
import asyncio
|
|
from datetime import timedelta
|
|
|
|
seconds = 5
|
|
|
|
|
|
async def main():
|
|
# Conecta ao servidor Temporal
|
|
client = await Client.connect("http://localhost:7233")
|
|
|
|
for i in range(1, 2):
|
|
# Define o agendamento para rodar a cada 5 segundos
|
|
schedule = Schedule(
|
|
action=ScheduleActionStartWorkflow(
|
|
'fake_data', # Nome da classe do workflow no worker.py
|
|
# Argumento de entrada (ajuste conforme seu workflow)
|
|
{
|
|
'topic': 'fake_data'
|
|
},
|
|
# ID que você define aqui
|
|
id=f"test-{i}-{seconds}",
|
|
task_queue="fake_data-queue", # Deve coincidir com o worker
|
|
),
|
|
spec=ScheduleSpec(
|
|
intervals=[ScheduleIntervalSpec(
|
|
every=timedelta(seconds=seconds))]
|
|
),
|
|
)
|
|
|
|
# Cria ou atualiza o schedule no Temporal
|
|
# ID único para o schedule
|
|
schedule_id = f"test-schedule-c-{i}-every-o-{seconds}s"
|
|
try:
|
|
await client.create_schedule(schedule_id, schedule)
|
|
print(
|
|
f"Schedule '{schedule_id}' criado com sucesso. Workflow rodará a cada {seconds} segundos.")
|
|
except Exception as e:
|
|
print(f"Erro ao criar o schedule: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|