47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from dataclasses import dataclass
|
|
from typing import Type
|
|
|
|
from app.jobs.base import BaseJob
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HandlerRef:
|
|
module: str
|
|
cls_name: str
|
|
|
|
|
|
def parse_handler_path(handler_path: str) -> HandlerRef:
|
|
"""
|
|
支持两种格式：
|
|
- "pkg.mod:ClassName"（推荐）
|
|
- "pkg.mod.ClassName"
|
|
"""
|
|
if ":" in handler_path:
|
|
module, cls_name = handler_path.split(":", 1)
|
|
return HandlerRef(module=module, cls_name=cls_name)
|
|
if "." not in handler_path:
|
|
raise ValueError(f"Invalid handler_path: {handler_path}")
|
|
module, cls_name = handler_path.rsplit(".", 1)
|
|
return HandlerRef(module=module, cls_name=cls_name)
|
|
|
|
|
|
def load_job_class(handler_path: str) -> Type[BaseJob]:
|
|
ref = parse_handler_path(handler_path)
|
|
mod = importlib.import_module(ref.module)
|
|
cls = getattr(mod, ref.cls_name, None)
|
|
if cls is None:
|
|
raise ImportError(f"Class not found: {ref.module}.{ref.cls_name}")
|
|
if not isinstance(cls, type) or not issubclass(cls, BaseJob):
|
|
raise TypeError(f"Handler is not a BaseJob subclass: {handler_path}")
|
|
return cls
|
|
|
|
|
|
def instantiate(handler_path: str) -> BaseJob:
|
|
cls = load_job_class(handler_path)
|
|
return cls()
|
|
|
|
|