247 lines
9.2 KiB
Python
247 lines
9.2 KiB
Python
from typing import Any
|
||
|
||
from cli.exit_codes import DATABASE_ERROR, RUNTIME_ERROR
|
||
|
||
from .gateway import JobInfrastructureGateway
|
||
from .support import JobDomainSupport, JobSchedulerSupport
|
||
|
||
|
||
class JobRuntimeService:
|
||
"""
|
||
定时任务运行时服务。
|
||
|
||
该服务作为定时任务运行时 facade，对外统一暴露任务列表、详情、日志、
|
||
执行一次、暂停与恢复入口。
|
||
|
||
:param infrastructure_gateway: 定时任务基础设施网关
|
||
:param domain_support: 定时任务领域支持对象
|
||
:param scheduler_support: 定时任务调度上下文支持对象
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
infrastructure_gateway: JobInfrastructureGateway | None = None,
|
||
domain_support: JobDomainSupport | None = None,
|
||
scheduler_support: JobSchedulerSupport | None = None,
|
||
) -> None:
|
||
"""
|
||
初始化定时任务运行时服务。
|
||
|
||
:param infrastructure_gateway: 定时任务基础设施网关
|
||
:param domain_support: 定时任务领域支持对象
|
||
:param scheduler_support: 定时任务调度上下文支持对象
|
||
:return: None
|
||
"""
|
||
self.infrastructure_gateway = infrastructure_gateway or JobInfrastructureGateway()
|
||
self.domain_support = domain_support or JobDomainSupport(self.infrastructure_gateway)
|
||
self.scheduler_support = scheduler_support or JobSchedulerSupport(self.infrastructure_gateway)
|
||
|
||
async def list_jobs(
|
||
self,
|
||
*,
|
||
job_name: str = '',
|
||
job_group: str = '',
|
||
status: str | None = None,
|
||
paged: bool = False,
|
||
page_num: int = 1,
|
||
page_size: int = 20,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
查询定时任务列表。
|
||
|
||
:param job_name: 任务名称过滤条件
|
||
:param job_group: 任务组过滤条件
|
||
:param status: 状态过滤条件
|
||
:param paged: 是否启用分页
|
||
:param page_num: 页码
|
||
:param page_size: 每页数量
|
||
:return: 定时任务列表结果
|
||
"""
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
job_vo_module = self.infrastructure_gateway.get_job_vo_module()
|
||
job_service = self.infrastructure_gateway.get_job_service()
|
||
query_model = job_vo_module.JobPageQueryModel(
|
||
jobName=job_name or None,
|
||
jobGroup=job_group or None,
|
||
status=status,
|
||
pageNum=page_num,
|
||
pageSize=page_size,
|
||
)
|
||
try:
|
||
async with async_session_local() as session:
|
||
result = await job_service.get_job_list_services(session, query_model, is_page=paged)
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '读取定时任务列表失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
|
||
filters = self.domain_support.build_filters(
|
||
job_name=job_name,
|
||
job_group=job_group,
|
||
status=status,
|
||
paged=paged,
|
||
page_num=page_num,
|
||
page_size=page_size,
|
||
)
|
||
return self.domain_support.build_list_payload(result, filters=filters, paged=paged)
|
||
|
||
async def get_job_detail(self, job_id: int) -> dict[str, Any]:
|
||
"""
|
||
读取单个定时任务详情。
|
||
|
||
:param job_id: 任务 ID
|
||
:return: 定时任务详情结果
|
||
"""
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
job_service = self.infrastructure_gateway.get_job_service()
|
||
try:
|
||
async with async_session_local() as session:
|
||
job_model = await job_service.job_detail_services(session, job_id)
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '读取定时任务详情失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
|
||
job_payload = self.domain_support.serialize_job_item(job_model)
|
||
if not job_payload.get('jobId'):
|
||
return {
|
||
'ok': False,
|
||
'message': f'定时任务不存在：{job_id}',
|
||
'jobId': job_id,
|
||
'exit_code': RUNTIME_ERROR,
|
||
}
|
||
return {'ok': True, 'job': job_payload}
|
||
|
||
async def list_job_logs(
|
||
self,
|
||
*,
|
||
job_name: str = '',
|
||
job_group: str = '',
|
||
status: str | None = None,
|
||
begin_date: str = '',
|
||
end_date: str = '',
|
||
paged: bool = False,
|
||
page_num: int = 1,
|
||
page_size: int = 20,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
查询定时任务执行日志列表。
|
||
|
||
:param job_name: 任务名称过滤条件
|
||
:param job_group: 任务组过滤条件
|
||
:param status: 执行状态过滤条件
|
||
:param begin_date: 开始日期，格式 `YYYY-MM-DD`
|
||
:param end_date: 结束日期，格式 `YYYY-MM-DD`
|
||
:param paged: 是否启用分页
|
||
:param page_num: 页码
|
||
:param page_size: 每页数量
|
||
:return: 定时任务日志列表结果
|
||
"""
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
job_log_service = self.infrastructure_gateway.get_job_log_service()
|
||
job_vo_module = self.infrastructure_gateway.get_job_vo_module()
|
||
query_model = job_vo_module.JobLogPageQueryModel(
|
||
jobName=job_name or None,
|
||
jobGroup=job_group or None,
|
||
status=status,
|
||
beginTime=begin_date or None,
|
||
endTime=end_date or None,
|
||
pageNum=page_num,
|
||
pageSize=page_size,
|
||
)
|
||
try:
|
||
async with async_session_local() as session:
|
||
result = await job_log_service.get_job_log_list_services(session, query_model, is_page=paged)
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '读取定时任务日志列表失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
|
||
filters = self.domain_support.build_filters(
|
||
job_name=job_name,
|
||
job_group=job_group,
|
||
status=status,
|
||
begin_date=begin_date,
|
||
end_date=end_date,
|
||
paged=paged,
|
||
page_num=page_num,
|
||
page_size=page_size,
|
||
)
|
||
return self.domain_support.build_list_payload(result, filters=filters, paged=paged)
|
||
|
||
async def run_with_scheduler_context(
|
||
self,
|
||
operation: str,
|
||
job_id: int,
|
||
*,
|
||
status: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
在调度器上下文中执行任务操作。
|
||
|
||
:param operation: 操作名称
|
||
:param job_id: 任务ID
|
||
:param status: 目标状态
|
||
:return: 任务操作结果
|
||
"""
|
||
operation_metadata = self.domain_support.build_job_operation_metadata(operation)
|
||
redis = None
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
redis_util = self.infrastructure_gateway.get_redis_util()
|
||
scheduler_util = self.infrastructure_gateway.get_scheduler_util()
|
||
job_vo_module = self.infrastructure_gateway.get_job_vo_module()
|
||
job_service = self.infrastructure_gateway.get_job_service()
|
||
try:
|
||
async with async_session_local() as session:
|
||
redis = await redis_util.create_redis_pool(log_enabled=False)
|
||
await scheduler_util.init_system_scheduler(redis)
|
||
|
||
if operation == 'run-once':
|
||
result = await job_service.execute_job_once_services(session, job_vo_module.JobModel(jobId=job_id))
|
||
elif operation in {'pause', 'resume'}:
|
||
result = await job_service.edit_job_services(
|
||
session,
|
||
job_vo_module.EditJobModel(jobId=job_id, status=status, type='status'),
|
||
)
|
||
else:
|
||
raise ValueError(f'不支持的任务操作：{operation}')
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '执行定时任务操作失败', 'error': str(exc), 'exit_code': 22}
|
||
finally:
|
||
await self.scheduler_support.close_scheduler_context(redis)
|
||
|
||
return {
|
||
'ok': bool(result.is_success),
|
||
'jobId': job_id,
|
||
'operation': operation,
|
||
'operationLabel': operation_metadata['operationLabel'],
|
||
'targetStatus': status,
|
||
'message': operation_metadata['successMessage'] if result.is_success else '定时任务操作执行失败',
|
||
'serviceMessage': result.message,
|
||
}
|
||
|
||
async def run_job_once(self, job_id: int) -> dict[str, Any]:
|
||
"""
|
||
执行一次指定定时任务。
|
||
|
||
:param job_id: 任务ID
|
||
:return: 执行结果
|
||
"""
|
||
return await self.run_with_scheduler_context('run-once', job_id)
|
||
|
||
async def pause_job(self, job_id: int) -> dict[str, Any]:
|
||
"""
|
||
暂停指定定时任务。
|
||
|
||
:param job_id: 任务ID
|
||
:return: 暂停结果
|
||
"""
|
||
return await self.run_with_scheduler_context('pause', job_id, status='1')
|
||
|
||
async def resume_job(self, job_id: int) -> dict[str, Any]:
|
||
"""
|
||
恢复指定定时任务。
|
||
|
||
:param job_id: 任务ID
|
||
:return: 恢复结果
|
||
"""
|
||
return await self.run_with_scheduler_context('resume', job_id, status='0')
|
||
|
||
|
||
JOB_RUNTIME = JobRuntimeService()
|