22 lines
No EOL
681 B
Python
22 lines
No EOL
681 B
Python
from typing import Dict, List, Tuple
|
|
|
|
# Directed Acyclic Graph (DAG): task -> [dependencies]
|
|
JOB_GRAPH: Dict[str, Dict[str, List[str]]] = {
|
|
"MailCheck": {
|
|
"MXValidation": [],
|
|
"SPFValidation": [],
|
|
"TLSCheck": ["MXValidation"],
|
|
"CertValidity": ["MXValidation"],
|
|
"MTASTSCheck": [],
|
|
"TLSRPTCheck": [],
|
|
}
|
|
}
|
|
|
|
def routing_key(job_type: str, task_name: str) -> str:
|
|
return f"{job_type}.{task_name}"
|
|
|
|
def list_tasks_for_job(job_type: str) -> List[str]:
|
|
return list(JOB_GRAPH.get(job_type, {}).keys())
|
|
|
|
def list_dependencies(job_type: str, task_name: str) -> List[str]:
|
|
return JOB_GRAPH.get(job_type, {}).get(task_name, []) |