Madtornado4¶
Madtornado4是一个构建MVC Tornado项目的工具。
安装¶
pip install madtornado4
用法¶
madtornado4通过 mad
命令提供构建操作,在控制台中键入mad即可获得帮助
- mad install [version]: 指定madtornado的版本号在当前目录下初始化项目。
- mad list: 查看所有可用的版本号。
License¶
The MIT License (MIT)
Copyright (c) 2019, 2020 madtornado4 contributors
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
核心内容
¶
madtornado4核心包¶
startup
¶
启动器¶
-
class
startup.
MVCApplication
(handlers=None, default_host=None, transforms=None, **settings)¶ 基类:
tornado.web.Application
-
destroy
()¶ 销毁注册的单例服务
返回:
-
-
class
startup.
Startup
¶ -
-
add_scoped
(service: Type[core.interface.IDispose]) → core.interface.IStartup¶ 添加会话对象,该对象必须继承自IDispose抽象类,注意service要保证类名唯一, 请不要注册同类名的service
参数: service -- 返回: IStartup
-
add_singleton
(service: Union[Type[core.interface.IDispose], Type[Awaitable]]) → core.interface.IStartup¶ 添加单例对象,该对象必须继承自IDispose抽象类,注意service要保证类名唯一, 请不要注册同类名的service
参数: service -- 返回: IStartup
-
build
() → tornado.web.Application¶ 实现应用构建逻辑
返回: Application
-
config_debug
() → core.interface.IStartup¶ 实现debug参数配置
返回: NoReturn
-
config_route
() → core.interface.IStartup¶ 实现路由配置
返回: NoReturn
-
config_static_file
() → core.interface.IStartup¶ 实现静态文件参数配置
返回: NoReturn
-
static
generate_static_route
()¶ 根据静态文件配置项生成路由表
参数: setting -- static_file_setting 字典 返回: 路由数组
-
-
startup.
build_host
(stp: core.interface.IStartup) → tornado.httpserver.HTTPServer¶ 构建主机服务对象
参数: stp -- startup.Startup 启动配置对象,用于启动之前对应用进行构建生成 返回: HTTPServer
core.client
--- 客户端请求工具模块¶
-
class
core.client.parser.
FormData
(boundary: str = '------SystemLightBoundaryqLpw6MGCDYFr9bn')¶ 基类:
object
form-data 请求体构造器
-
add_data
(name: str, filename: str, data: bytes, mime: Optional[str] = None) → NoReturn¶ 添加数据流
参数: - name -- 字段名称
- filename -- 文件名称
- data -- 数据二进制对象
- mime -- 文件Content-Type类型,可选
返回: NoReturn
-
add_field
(name: str, value: str) → NoReturn¶ 添加字段
参数: - name -- 字段名称
- value -- 字段值
返回: NoReturn
-
add_file
(name: str, path: str, mime: str = None) → NoReturn¶ 从文件中添加数据流
参数: - name -- 字段名称
- path -- 文件路径
- mime -- 文件Content-Type类型,可选
返回: NoReturn
-
destroy
() → NoReturn¶ 销毁对象
返回: NoReturn
-
get_body
() → bytes¶ 返回请求体,用于填充到body当中
返回: body
-
get_content_type
() → str¶ 需要将请求头的Content-Type设置为该方法返回值
返回: Content-Type
-
-
core.client.parser.
generate_boundary
(size: int = 15) → str¶ 生成form-data报文字符串分割符号
返回: Boundary分割符号
core.db
--- 数据库内置服务¶
core.ui
--- 模版引擎自定义渲染¶
-
exception
core.ui.method.
NotFoundStatic
¶ 基类:
Exception
-
core.ui.method.
static_url
(self, path: str, name: Optional[str] = None, **kwargs) → str¶ 生成静态目录路径
参数: - self -- ApiController
- path -- 传进来的文件位置参数
- name -- 静态文件寻址注册名称
返回: 合成的文件路径
core.form
--- 表单模型验证器¶
-
class
core.form.
PropertyType
¶ 基类:
object
定义模型属性类型,用于验证映射,注意装饰器必须加在property.setter下面
-
static
array
(modal_type: Type[Model]) → Callable¶ 数组类实例模型
返回: Callable
-
static
model
(modal_type: Type[Model]) → Callable¶ 嵌套子模型类型
返回: Callable
-
static
-
class
core.form.
Rule
¶ 基类:
object
定义规则限制,用于验证属性,注意装饰器必须加在property.setter下面
-
static
length
(min_len: int, max_len: int) → Callable¶ 字段长度限制规则
参数: - min_len -- 字符串最小长度
- max_len -- 字符串最大长度
返回: Callable
-
static
scope
(min_size: int, max_size: int, number=<class 'int'>) → Callable¶ 参数最小到最大的范围
参数: - min_size -- 字符串最小长度
- max_size -- 字符串最大长度
- number -- 数字类型,整型或者浮点类型
返回: Callable
-
static
-
core.form.
verify
(model_type: Type[Model], catch_method: Callable[str, Any]) → Model¶ 验证参数是否符合模型需求
参数: - model_type -- 继承IModal的模型对象
- catch_method -- 捕获参数的方法,例如self.get_argument,dict().get等等,你甚至可以自定义,始终会传递一个value参数
返回: 验证后的模型实例对象
-
core.form.
weak_verify
(pool: List[str], catch_method: Callable[str, Any]) → Dict¶ 对类型验证更随意,主要用来收集数据,无须建立model实例对象
返回: 数据对象Dict
core.fs
--- 文件快捷处理器¶
-
exception
core.fs.
InvalidPath
¶ 基类:
tornado.web.HTTPError
抛出非法Http异常错误
-
core.fs.
check_join
(root_path: str, *args) → str¶ 检查合并后的路径是否在root_path当中,如果超出抛出异常
参数: - root_path -- 根路径
- args -- 路径块集合
返回: 合并后的绝对路径
-
core.fs.
read
(path: str, encoding: str = 'utf-8') → str¶ 读取文件返回字符串
参数: - path -- 文件路径
- encoding -- 编码方式
返回: 读取所有字符串
-
core.fs.
read_bytes
(path: str) → bytes¶ 读取文件返回bytes
参数: path -- 文件路径 返回: 读取所有字符串
-
core.fs.
require
(path: str, encoding: str = 'utf-8') → Dict[str, Any]¶ 有时你可能只是需要从文件中读取到json数据,这是require函数将根据 获取到的path,返回dict对象,相当方便,该函数同样类似于json.load
参数: - path -- json文件路径
- encoding -- 编码方式
返回: dict
-
core.fs.
safe_join
(*args) → str¶ 合并给定路径成为一个绝对路径,如果某个子路径块超出父路径就会抛出异常
参数: args -- 路径块集合 返回: 合并后的绝对路径
-
core.fs.
write
(path: str, data: str, encoding: str = 'utf-8') → NoReturn¶ 将字符串写入文件当中
参数: - path -- 文件路径
- data -- 写入的字符串数据
- encoding -- 编码方式
返回:
-
core.fs.
write_bytes
(path: str, data: bytes) → NoReturn¶ 将bytes写入文件当中
参数: - path -- 文件路径
- data -- 写入的字符串数据
返回:
core.interface
--- 接口定义模块¶
-
class
core.interface.
IDispose
(launch)¶ 基类:
abc.ABC
定义可被销毁对象
-
destroy
() → NoReturn¶ 定义如何销毁对象
返回: NoReturn
-
-
class
core.interface.
IStartup
¶ 基类:
abc.ABC
定义web服务启动器
-
add_scoped
(service: Type[core.interface.IDispose]) → core.interface.IStartup¶ 添加会话对象,该对象必须继承自IDispose抽象类,注意service要保证类名唯一, 请不要注册同类名的service
参数: service -- 返回: IStartup
-
add_singleton
(service: Type[core.interface.IDispose]) → core.interface.IStartup¶ 添加单例对象,该对象必须继承自IDispose抽象类,注意service要保证类名唯一, 请不要注册同类名的service
参数: service -- 返回: IStartup
-
build
() → tornado.web.Application¶ 实现应用构建逻辑
返回: Application
-
config_debug
() → core.interface.IStartup¶ 实现debug参数配置
返回: NoReturn
-
config_route
() → core.interface.IStartup¶ 实现路由配置
返回: NoReturn
-
config_static_file
() → core.interface.IStartup¶ 实现静态文件参数配置
返回: NoReturn
-
core.register
--- 路由注册器¶
-
class
core.register.
ArgType
¶ 基类:
object
包含uri可以用到的类型参数
-
Any
= '(.+)'¶
-
Letter
= '([A-Za-z]+)'¶
-
Number
= '(\\d+)'¶
-
Param
= '([^\\\\/]+)'¶
-
Word
= '([\\w-]+)'¶
-
-
class
core.register.
RegisterMeta
(what: str, bases=None, _dict=None)¶ 基类:
type
路由注册器元类
-
route_pool
= {}¶
-
-
core.register.
api_method
(method)¶ api装饰的方法将函数定义为API类型请求方法,可以获取到任意param参数值 同时方法返回的对象值会自动写入到返回请求当中
举例说明:
@cross_domain() @api_method async def get(self): return {}
参数: method -- 异步http方法函数 返回: 装饰包裹后的函数
-
core.register.
create_router
(prefix: str, use_uri: bool = False) → Callable[Union[List[str], str], List]¶ 创键router函数,该函数用于包裹__urls,添加公用前缀
举例:
router = create_router("/api") __urls = router(["/test"]) 或者 __urls = [router("/test")]
参数: - prefix -- 公用url前缀
- use_uri -- 是否使用URI函数解析路径
返回: router函数
-
core.register.
cross_domain
(origin: str = '*', headers: str = '*', methods: str = 'GET, POST, PUT, DELETE, OPTIONS', max_age=600)¶ access为当前请求方法添加跨域属性
举例说明:
@cross_domain() async def get(self): return {}
参数: - origin -- 允许的作用域
- headers -- 允许的请求头
- methods -- 允许的请求方法
- max_age -- 最大非重复预请求时间
返回: 装饰包裹后的函数
-
core.register.
uri
(path: str, *args) → str¶ 快捷的定义url注册,可以根据类型参数生成对应的正则匹配
使用方法:
__urls = [ uri("/home/{Letter}"), "/home", "/index" ] __urls = [ uri("/home/{}", ArgType.Letter), "/home", "/index" ]
参数: - path -- 模板url路径
- args -- 替换类型参数,ArgType的静态属性
返回: 渲染后的URL字符串
core.utils
--- 工具函数¶
-
class
core.utils.
AdvancedJSONEncoder
(skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)¶ 基类:
json.encoder.JSONEncoder
定义ApiController JSON解析器
-
default
(obj)¶ Implement this method in a subclass such that it returns a serializable object for
o
, or calls the base implementation (to raise aTypeError
).For example, to support arbitrary iterators, you could implement default like this:
def default(self, o): try: iterable = iter(o) except TypeError: pass else: return list(iterable) # Let the base class default method raise the TypeError return JSONEncoder.default(self, o)
-
find_dict
= {'Decimal': <function AdvancedJSONEncoder.<lambda>>, 'date': <function AdvancedJSONEncoder.<lambda>>, 'datetime': <function AdvancedJSONEncoder.<lambda>>}¶
-
-
class
core.utils.
UpdateList
(*args, **kwargs)¶ 基类:
list
主要方法update(),该方法是对list类型拓展, 当update的数据对象存在时对其更新,注意请保证UpdateList 的子项是dict类型而不要使用值类型,值类型对于UpdateList毫无意义
on_update hook函数,接收old_val(旧数据), p_object(新数据),需要返回更新数据 on_append hook函数,接收p_object(添加数据),需要返回添加数据 on_fetch_key hook函数,当key属性定义为函数时需要同时定义如何捕获key值
- key 支持字符串,字符串指定子元素中的更新参考值
- 支持函数,接收val(当前数据),key(参考key值)该key值由on_fetch_key返回,函数返回bool值True为更新,False为添加
on_fetch_key作用:
复杂场景下我们可能需要up[("home2", True)]这样来找到响应的item,这样显示传递key值没有什么问题,key函数可以获取到 相应的key数据以供我们处理,但是当我们调用update时,update需要判断该内容是更新还是添加,这时我们传入的内容是数据,显然 update无法知晓如何获取我们想要的类型key值,如("home2", True),所以我们要定义on_fetch_key来告知update如何捕获我们 想要的类型的key值,on_fetch_key只有当key属性定义为函数时才有意义。
-
find
(callback)¶ 返回满足回调函数的内容
参数: callback -- 回调函数,返回布尔类型用于判断是否满足要求 返回: (索引,值)
-
update
(p_object)¶ 类似于append方法,不同的是当内容存在时会对内容进行更新,更新逻辑遵从update_callback 而当内容不存在时与append方法一致进行末尾加入内容
参数: p_object -- 内容对象 返回: None
-
core.utils.
find
(iterable: Iterable[IT], func: Callable[IT, bool]) → Tuple[int, Optional[IT]]¶ 查找可迭代对象的指定项,匹配第一个子项并返回,无匹配项时返回(-1,None)
参数: - func -- 匹配函数
- iterable -- 可迭代对象
返回: 索引,子对象
-
core.utils.
find_key
(obj: Union[Dict, List], key: str)¶ 根据字符串查找对象值,字符串形式如a.b.0, 查找对象,如:
{"a":{"b":["val"]}}
val值将被查出
参数: - obj -- 查找key值的对象
- key -- 查找key
返回: 查找到的值
-
core.utils.
inin
(content: str, pool: List[str]) → Optional[str]¶ 查找指定内容是否存在于列表的字符串中,这种情况content一定要比列表中字符串短
举例:
inin("a",["asdf","fsfsdf"]) 将返回 "asdf"
参数: - content -- 内容
- pool -- 列表
返回: 匹配内容
-
core.utils.
retry
(freq: int = 3, retry_hook: Optional[Callable[int, None]] = None) → Callable¶ 装饰器,为函数添加此装饰器当函数抛出异常时会对函数重新调用,重新调用次数取决于freq指定的参数
参数: - freq -- 重试次数
- retry_hook -- 钩子函数,当函数重调用时回调的函数
返回: 原函数返回值
-
core.utils.
rinin
(content: str, pool: List[str]) → Optional[str]¶ 查找指定内容是否存在于列表的字符串中,这种情况content一定要比列表中字符串长
举例:
inin("asdf",["a","fsfsdf"]) 将返回 "a"
参数: - content -- 内容
- pool -- 列表
返回: 匹配内容