import importlib import pkgutil from types import ModuleType from typing import Optional, Any import tornado from tornado.routing import URLSpec from tornado.web import OutputTransform, _RuleList class Application(tornado.web.Application): """ 从 Tornado 派生的应用程序类。 """ @classmethod def modules_iterator(cls, package: [str, ModuleType]): """ 从 package 包装载所有模块。这里返回模块迭代器。 若为字符串，则直接从目录中载入模块；若为模块，则根据模块的参数装载。 :param package: 包，允许为路径或包（模块对象） :return: 模块迭代器 """ if isinstance(package, str): package = importlib.import_module(package) # 模块迭代器，能够遍历出所有子包中的子模块 return pkgutil.walk_packages(package.__path__, f"{package.__name__}.") @classmethod def fetch_handlers(cls, module: ModuleType): """ 查找模块中所有的请求处理类，即类 RequestHandler 的子类。 :param module: 模块 :return: [(路由模式, 请求处理类)] """ # 判断是否是有效的 RequestHandler 类，且是 RequestHandler 的子类 def is_handler(handler_cls): return isinstance(handler_cls, type) and issubclass(handler_cls, tornado.web.RequestHandler) # 判断是否拥有 route_pattern 模式属性，且该属性值为字符串类型 def has_pattern(handler_cls): return hasattr(handler_cls, 'route_pattern') and isinstance(getattr(handler_cls, 'route_pattern'), str) handlers: list[tuple[str, ModuleType]] = [] # 迭代模块成员 for _n in dir(module): _cls = getattr(module, _n) is_hdl = is_handler(_cls) has_pat = has_pattern(_cls) if is_hdl and has_pat: _route = _cls.route_pattern handlers.append((_route, _cls)) return handlers @classmethod def load_ui_modules(cls, ui_modules_config): """ 将JSON配置中的模块字符串转换为实际的类。 """ loaded_modules = {} for name, path in ui_modules_config.items(): try: module_path, class_name = path.rsplit('.', 1) module = importlib.import_module(module_path) loaded_modules[name] = getattr(module, class_name) except (ImportError, AttributeError) as e: raise RuntimeError(f"Failed to load UIModule {name} from {path}: {str(e)}") return loaded_modules def __init__( self, handlers: Optional[_RuleList] = None, handlers_pkg: [str, ModuleType] = None, uri_prefix: str = "", **settings: Any ) -> None: """ 重写应用程序构造函数，增加自动装载功能。 :param handlers: 请求处理路由配置列表 :param handlers_pkg: 执行自动装载的请求处理类所在包 :param uri_prefix: URI 前缀 :param settings: 其他配置 """ self.routes: list[(URLSpec, _RuleList)] = [] """ 请求处理路由列表。 """ if uri_prefix: uri_prefix = uri_prefix if uri_prefix.startswith('/') else f"/{uri_prefix}" self.uri_prefix = uri_prefix """ 统一资源标识符前缀。仅支持动态加载的请求处理类。 """ # 合并构造参数中的请求处理路由 if handlers: self.routes.extend(handlers) # 动态加载请求处理类，并执行合并 if handlers_pkg: self.routes.extend(self.load_handlers(handlers_pkg=handlers_pkg)) self.before_create() super().__init__(handlers=self.routes, **settings) def before_create(self): """ 在创建应用之前执行。 """ pass def load_handlers(self, handlers_pkg: [str, ModuleType] = None): """ 从 handlers_pkg 指定的包装载所有模块，分析出所有请求处理类和路由路径，并返回。 :param handlers_pkg: 模块根目录，允许为路径或包（模块对象） :return: 动态装载的所有路由配置 """ _routes = [] if handlers_pkg is None: return _routes # 迭代器装载所有子包中的子模块 modules_itr = self.modules_iterator(package=handlers_pkg) for _file_finder, _module_name, _is_package in modules_itr: if _is_package: continue _module = importlib.import_module(_module_name) _handlers = self.fetch_handlers(module=_module) for _hdl in _handlers: _pattern, _cls = str(_hdl[0]), _hdl[1] _pattern = _pattern if _pattern.startswith('/') else f"/{_pattern}" _url_spec = tornado.web.url( pattern=f"{self.uri_prefix}{_pattern}", handler=_cls, name=_cls.__name__ ) _routes.append(_url_spec) return _routes class ApplicationSwagger(Application): """ 从框架 Application 派生，增加对 Swagger 的支持。 """ swagger_schema = "" """ 在 Swagger 注入时，保存 json schema。 """ swagger_home_template = "" """ 在 Swagger 注入时，保存 Ui 页面内容。 """ swagger_url = "/docs" """ Swagger URL。 """ swagger_api_base_url = "/" """ Swagger API base URL。 """ swagger_title = "" """ Swagger 页面标题。 """ swagger_description = "" """ Swagger 页面描述。 """ swagger_api_version = "" """ Swagger 页面版本。 """ swagger_contact = "" """ Swagger 页面联系方式。 """ swagger_schemes = ["http", "https"] """ Swagger 协议方案。 """ def __init__(self, **settings: Any) -> None: self.swagger_schema = settings.get('swagger_schema', self.swagger_schema) self.swagger_home_template = settings.get('swagger_home_template', self.swagger_home_template) self.swagger_url = settings.get('swagger_url', self.swagger_url) self.swagger_url = self.swagger_url if self.swagger_url.startswith('/') else f"/{self.swagger_url}" self.swagger_api_base_url = settings.get('swagger_api_base_url', self.swagger_api_base_url) self.swagger_api_base_url = self.swagger_api_base_url if self.swagger_api_base_url.startswith('/') \ else f"/{self.swagger_api_base_url}" self.swagger_title = settings.get('swagger_title', self.swagger_title) self.swagger_description = settings.get('swagger_description', self.swagger_description) self.swagger_api_version = settings.get('swagger_api_version', self.swagger_api_version) self.swagger_contact = settings.get('swagger_contact', self.swagger_contact) self.swagger_schemes = settings.get('swagger_schemes', self.swagger_schemes) super().__init__(**settings) def before_create(self): _swagger_url = f"{self.uri_prefix}{self.swagger_url}" _swagger_api_base_url = f"{self.uri_prefix}{self.swagger_api_base_url}" from paste.web.swagger import setup_swagger setup_swagger( app=self, routes=self.routes, swagger_url=_swagger_url, api_base_url=_swagger_api_base_url, title=self.swagger_title, description=self.swagger_description, api_version=self.swagger_api_version, contact=self.swagger_contact, schemes=self.swagger_schemes, )