89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Run cleanup_files workflow once for manual testing.
|
|
|
|
This script starts the Temporal workflow `cleanup_files` a single time,
|
|
using the same Temporal namespace and task queue as the main worker.
|
|
|
|
It is intended only for local/manual testing; scheduling (cron) must be
|
|
configured separately in Temporal.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from datetime import timedelta
|
|
from typing import Any
|
|
from dotenv import load_dotenv
|
|
from pathlib import Path
|
|
from temporalio.client import Client
|
|
|
|
# Ensure project root is on PYTHONPATH when running directly
|
|
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if ROOT_DIR not in sys.path:
|
|
sys.path.insert(0, ROOT_DIR)
|
|
|
|
from model_manager.workflows.cleanup_files import CleanupFiles
|
|
|
|
|
|
# Carrega variáveis de ambiente do arquivo .env na raiz do projeto
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
ENV_PATH = PROJECT_ROOT / '.env'
|
|
if ENV_PATH.exists():
|
|
load_dotenv(dotenv_path=ENV_PATH)
|
|
|
|
|
|
async def main(argv: list[str]) -> None:
|
|
"""Entry point for manual cleanup workflow execution.
|
|
|
|
Args:
|
|
argv: Command-line arguments (excluding program name).
|
|
"""
|
|
|
|
# Config from environment / defaults
|
|
temporal_host = os.getenv('TEMPORAL_HOST')
|
|
temporal_namespace = os.getenv('TEMPORAL_NAMESPACE')
|
|
task_queue = os.getenv('CLEANUP_TASK_QUEUE')
|
|
default_bucket = os.getenv('DEFAULT_CLEANUP_BUCKET')
|
|
use_tls = os.getenv('TEMPORAL_USE_TLS', 'false').lower() == 'true'
|
|
|
|
# Optional CLI: bucket name override
|
|
bucket_name = default_bucket
|
|
if argv:
|
|
bucket_name = argv[0]
|
|
|
|
print(f"Connecting to Temporal at {temporal_host} (namespace={temporal_namespace})...")
|
|
client = await Client.connect(
|
|
target_host=temporal_host,
|
|
namespace=temporal_namespace,
|
|
tls=use_tls,
|
|
)
|
|
|
|
input_data: dict[str, Any] = {
|
|
'bucket_name': bucket_name,
|
|
}
|
|
|
|
workflow_id = f"cleanup-files-manual-{int(asyncio.get_event_loop().time())}"
|
|
|
|
print(f"Starting cleanup_files workflow once...\n"
|
|
f" workflow_id = {workflow_id}\n"
|
|
f" task_queue = {task_queue}\n"
|
|
f" bucket_name = {bucket_name}")
|
|
|
|
handle = await client.start_workflow(
|
|
CleanupFiles.run,
|
|
input_data,
|
|
id=workflow_id,
|
|
task_queue=task_queue,
|
|
run_timeout=timedelta(minutes=10),
|
|
)
|
|
|
|
print("Workflow started, waiting for completion...")
|
|
await handle.result()
|
|
print("cleanup_files workflow completed successfully.")
|
|
|
|
|
|
if __name__ == '__main__': # pragma: no cover - manual utility script
|
|
asyncio.run(main(sys.argv[1:]))
|