参考

pyape.config

解析 config.toml 配置文件提供配置文件相关的读取和写入方法

class pyape.config.Config(*, work_dir: Path = None, cfg: dict | str = 'config.toml')

提供配置文件的获取与读取功能。

cfg_data: dict = None: 全局变量，用于保存 config.toml/config.json 中载入的配置。

getcfg(*args, default_value: Any = None, data: str | dict = 'cfg_file') → Any

递归获取 conf 中的值。getcfg 不仅可用于读取 config.toml 的值，还可以通过传递 data 用于读取任何字典的值。

参数:

args -- 需要读取的参数，支持多级调用，若级不存在，不会报错。
default_value -- 找不到这个键就提供一个默认值。
data -- 提供一个 dict，否则使用 cfg_data。

返回:

获取的配置值

getdir(*args, work_dir: Path = None) → Path

基于当前项目的运行文件夹，返回一个 pathlib.Path 对象如果传递 basedir，就基于这个 basedir 创建路径

参数:

args -- 传递路径
work_dir (Path) -- 工作文件夹

read(filename: str, work_dir: Path = None, throw_error: bool = False) → list | dict

读取一个配置文件，支持 .json 和 .toml 扩展名。

参数:

filename -- 文件名
work_dir -- str
throw_error -- boolean 若值为 True，则当文件不存在的时候抛出异常

返回:

解析后的 dict

返回类型:

dict

setcfg(*args, value: Any, data: str | dict = 'cfg_file') → None

递归设置 conf 中的值。setcfg 不仅可用于设置 config.toml 的值，还可以通过传递 data 用于读取任何字典的值。

参数:

args -- 需要设置的参数，支持多级调用，若级不存在，会自动创建一个内缪的 dict。
data -- 提供一个 dict，否则使用 cfg_data。
value -- 需要设置的值。

write(filename: str, data_dict: list | dict, work_dir: Path = None) → None

将一个 dict 写入成为配置文件，支持 .toml 和 .json 后缀。

参数:: data_dict -- 要写入的配置信息

class pyape.config.Dicto

基于 Python dict 的配置对象。Dicto 意味着 dict object。

dict 默认不适合当作配置对象使用。如要有下面几点不便：

对于不存在的 key，会 raise KeyError 错误；
dict不能使用 . 语法访问。

Dicto 解决了这些问题，还另外提供了一些方法在使用上更加方便。

copy_from_dict(adict: dict, parent=None) → None

从一个已经存在的 dict 中复制所有的值。

参数:

adict (dict。) -- 被复制的 dict。
parent -- 复制到哪个父对象。若为 None 则复制到 self 。

class pyape.config.GlobalConfig(work_dir: Path = None, cfg: dict | str = 'config.toml')

全局配置文件，一般对应 config.toml。

decode_token(token: str) → Dicto

解密使用 encode_token 加密的字符串。

参数:: token -- 需要解密的 token 字符串。

>>> decode_token('b929a9f08a7ba1a01578ec5a8ecd75b7a06431b18866f1132a56aca667c3b33c')
{'r': 0, 'uid': 0, 'usertype': 50, 'status': 1, 'expire_at': unix时间戳单位秒, 'expires': False}

encode_token(expire_in: int = 86400, expire_at: int = None, **kwargs: dict) → str

使用 Fernet 算法加密一组值为 token 用于鉴权。

参数:

expire_in -- 从现在开始，多少秒内过期。 3600*24 = 86400。
expire_at -- 过期时间戳，超过这个时间代表过期。单位秒。若不提供则使用 now + expire_in。

>>> encode_token(r=0, uid=1, nickname='超级管理员', usertype=50)
'b929a9f08a7ba1a01578ec5a8ecd75b7a06431b18866f1132a56aca667c3b33c'

encrypter: Encrypt = None: 用于 Fernet 加解密对象。

getdburi(*, r: int = None, bind_key: str = None): 获取配置文件中保存的数据库配置。提供解析数据库路径的功能，方便直接使用 pymysql 的方法调用数据库。

regional: RegionalConfig = None: 如果 Pyape 框架启用了 Regional 机制，则保存 Regional 配置实例。

class pyape.config.RegionalConfig(rlist: list)

为 Regional 机制提供的配置文件，用于解析 Regional 配置。

check_regional(r: int, ignore_zero: bool = False): 检查 regional 是否有效 :param ignore_zero: 值为真，则允许 r 值为 0。0 是一个特殊的 r 值，代表全局 r :return: 已经转换成整数的 regional 值

get_regional(r: int): 获取一个 regional 配置 :param int r: r 值 :rtype: dict

get_value(r: int, *, key: str, ignore_zero: bool = False): 获取 regional[key] 的值。 :param r: :param key: regional 的键名。 :param ignore_zero: 同 check_regional。

pyape.config.merge_dict(x: dict, y: dict, z: dict = None) → dict

合并 x 和 y 两个 dict。

用 y 的同 key 值覆盖 x 的值。
y 中的新键名（x 中同级不存在）增加到 x 中。

返回一个新的 dict，不修改 x 和 y。

参数:

x -- x 被 y 覆盖。
y -- y 覆盖 x。

返回:

dict

pyape.db

处理 pyape 数据库功能

解决 SQLAlchemy 线程问题
提供多数据库绑定支持

class pyape.db.DBManager(URI: dict | str, ENGINE_OPTIONS: dict = None, **kwargs: dict)

管理 SQL 连接，创建和管理数据库 Engine，Session，Model。

参数:

URI -- 提供数据库地址。
kwargs (dict) -- 提供数据库连接参数。

ENGINE_OPTIONS: dict = None: 从配置文件中解析出的 ENGINE_OPTION 值。用于数据库引擎的配置。

property Models: Iterable: 使用 set 形式返回所有的 Model。

Session_Factory = None: Session 工厂类，使用 orm.sessionmaker 生成。

URI: dict | str = None: 从配置文件中解析出的 URI 值，可能是 str 或者 dict。

property bind_keys: Iterable: 使用 set 形式返回所有的 bind_key。若只有一个数据库，返回的是 (None,) 。

create_new_session() → Session: 创建一个 Session 对象。

create_scoped_session(use_greenlet: bool = False) → scoped_session

创建一个 orm.scoped_session 代理。

参数:: use_greenlet -- 是否使用 greenlet。

default_bind_key: str = None: 默认的 bind_key 必须存在，它的默认值就是 None，这是一个有效的 bind_key。

get_engine(bind_key: str = None) → Engine

获取一个 Engine 对象。

参数:: bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

set_Model(bind_key: str = None)

设置并保存一个 Model。

参数:: bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

set_bind(bind_key: str, uri: str)

利用 SQLAlchemy 提供的 binds 机制，将 bind_key 与数据库连接绑定起来。

参数:: bind_key -- 数据库的 bind_key，在 pyape.toml 中定义，值为 [SQLALCHEMY.URI] 中的键名。若 URI 为 str 而非 dict，则 bind_key 值为 None 。

class pyape.db.SQLAlchemy(dbm: DBManager = None, URI: dict | str = None, ENGINE_OPTIONS: dict = None, is_scoped: bool = True, use_greenlet: bool = False, **kwargs: dict)

创建一个用 sqlalchemy 管理数据库的对象。封装常用的高级功能，例如 table 和 query 操作。

参数:

dbm -- DBManager 的实例。
URI -- 若不提供 dbm 则使用 URI 数据新建 DBManager。
is_scoped -- 为线程安全，使用 scoped session。
in_flask -- 是否在 Flask 框架内部。若在 Flask 内部使用，创建 Session 实例的时候会使用 scoped_session。

Model(bind_key: str = None)

获取对应的 Model Factory class。

参数:: bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

Session: sessionmaker | scoped_session = None: 根据 is_scoped 的值，保存 orm.sessionmaker 或者 orm.scoped_session 的结果对象。

connection(bind_key: str = None) → Connection: 调用 Engine 的 connect 放来了获取一个 connection 对象。

create_all() → None: 创建所有数据库中的所有表。

create_tables(table_names: list[str] = None, bind_key: str = None) → None

创建 table。

参数:

table_names -- 提供 table 名称列表。
bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

drop_all() → None: 删除所有数据库中的所有表。

drop_tables(table_names: list[str] = None, bind_key: str = None) → None

移除 table。

参数:

table_names -- 提供 table 名称列表。
bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

engine(bind_key: str = None) → Engine

从 DBManager 中获取对应的 engine 实例。

参数:: bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

get_table(name: str, bind_key: str = None) → Table

获取一个 Table。

参数:

name -- Table 名称。
bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

isModel(instance, bind_key: str = None)

判断一个实例是否是 Model 的实例。

参数:: bind_key -- 详见 pyape.db.DBManager.set_bind 中的说明。

metadata(bind_key: str = None) → MetaData: 获取对应 Model 的 metadata 实例

recreate_table(*table_names: str, bind_key: str = None) → None: 重建 table，支持单个或者多个名称。

session() → Session: 获取一个 Session 对象。

pyape.app

所有位于 pyape.app 包中的内容，都可能会引用下面的 6 个全局变量中的一个或多个。所有引用了下面任何全局变量的代码，都应该放在 pyape.app 包中。

pyape.app.gapp: PyapeApp = None: 进程唯一的 PyapeApp 实例。

pyape.app.gcache: GlobalCache = None: 全局缓存实例，pyape.cache.GlobalCache 的实例。

pyape.app.gconf: GlobalConfig = None: 全局配置对象。 pyape.config.GlobalConfig 的实例。

pyape.app.gdb: PyapeDB = None: 全局的数据库引用，PyapeDB 的实例。

pyape.app.grc: PyapeRedis = None: grc 就是 Global Redis Client 的缩写，PyapeRedis 的实例。

pyape.app.logger: Logger = None: 全局 logger 对象。

pyape.logging

提供 logging 支持

class pyape.logging.RedisHandler(url, channel, **kwargs)

基于 Redis 的 publish 命令来发布 log

emit(record): Emit a log message on redis.

class pyape.logging.ZeroMQHandler(interface_or_socket, context=None, socket_type=SocketType.DEALER)

基于 ZeroMQ 的 ROUTER-DEALER 模式来发布 log

范例:

sock = context.socket(zmq.DEALER)
sock.connect('tcp://192.168.0.1:5050')
handler = ZeroMQHandler(sock)

或者:

handler = ZeroMQHandler('tcp://192.168.1:5050')

These are equivalent.

emit(record): Emit a log message on my socket.

pyape.logging.get_logger(name: str, target: Path | str, type_: str = 'file', fmt: str = 'text', level: int = 20)

基于 target 创建一个 logger

参数:

name -- logger 的名称，不要带扩展名
target -- 项目主目录的的 path 字符串或者 Path 对象，也可以是 tcp://127.0.0.1:8334 这样的地址
type -- stream/file/zmq/pyzog 若使用 pyzog ，则调用 get_pyzog_handler，fmt 参数必须为 config_dict
fmt -- text/json/config_dict 如果 type_ 参数为 pyzog，则必须为 config_dict
level -- logging 的 level 级别

返回:

一个 Logger 对象

pyape.logging.get_logging_handler(type_, fmt, level=20, target=None, name=None)

获取一个 logger handler

参数:

type (str) -- stream/file/zmq
fmt (str) -- text/json
level -- log 的 level 级别
target -- 项目主目录的的 path 字符串或者 Path 对象，也可以是 tcp://127.0.0.1:8334 这样的地址
name -- logger 的名称，不要带扩展名

pyape.logging.get_pyzog_handler(name, logger_config, target_dir, level=20)

获取一个 pyzog handler 如果不存在 pyzog 配置，那么会返回一个 file handler

参数:

name -- logger 的名称
config_dict -- config.json 配置文件的 dict
target_dir -- file handler 的目标文件夹
level -- handler 级别

pyape.util

封装的小工具

pyape.util.md5base64(txt: str | bytes) → bytes: md5(base64) 算法验证工具： http://www.cmd5.com/hash.aspx

pyape.util.md5txt(txt: str | bytes) → str: 计算 MD5 字符串散列 :param txt: :return:

pyape.util.func

工具类，可独立使用。放置和请求完全无关的，不依赖任何 flask 框架内容的工具。这个类中的方法可以被其它模块导入而不发生冲突。

pyape.util.func.between_date(from_date, to_date) → list[date]: 传递一个 from_date 和一个 to_date，返回中间所有的 date（包括 from_date 和 to_date）。

pyape.util.func.daydt(day: date | datetime = None, default_initday: bool = False, first_day_of_month: bool = False) → datetime

获取一个hour为0的日期。

参数:

day -- 时间戳或者基准 date/datetime 对象。
default_initday -- 仅当 day 参数为 None 的时候有效，True 代表使用初始日期，否则使用今天。
first_day_of_month -- 值为 True 则返回提供的 day 所在月的第一天。

返回:

pyape.util.func.get_cur_and_next_month(day: datetime = None)

获取代表本月到下月的日期对象。

参数:: day -- 一个 datetime 对象，基于这个对象作为基准来创建 cur_month。

pyape.util.func.in_time_range(timestart: str | time, timeend: str | time, dt: datetime = None) → bool

判断给定的时间 dt 是否在两个时间中间。

参数:

timestart -- 开始的时间，支持字符串和 time 对象，字符串使用本模块中的 strptime 解析。
timeend -- 结束的时间，支持字符串和 time 对象，字符串使用本模块中的 strptime 解析。
dt -- 给定的 datetime 对象，如果不提供则使用 datetime.now()。

pyape.util.func.issimple(value): 判断一个值是否为简单类型，包括 None

pyape.util.func.jsondumps(obj) → str

dump 一个 json 字符串，处理 datetime。

参数:: obj -- 需要被 dump 的对象。

pyape.util.func.last_month_dt(day: datetime = None) → datetime

返回上个月的 datetime 对象。

参数:: day -- 基准 datetime 对象。

pyape.util.func.next_month_dt(day: datetime = None) → datetime

返回下一个月的 datetime 对象。

参数:: day -- 基准 datetime 对象

pyape.util.func.parse_date(value, default_value=None, to_day_end=False) → datetime

将一个日期字符串解析成为 datetime 对象。

参数:

value
default_value
to_day_end -- 值为True 则自动加上23小时59分59秒999999微秒（当天的最后时刻），用来判断包含关系。

返回:

pyape.util.func.strfdate(dtobj): 将一个 datetime 转换成成 YYYY-MM-DD 形式的字符串 :param dtobj: datetime 实例 :return:

pyape.util.func.strptime(timestr): 将一个 time 字符串按照 HH:MM:SS 形式的字符串转换成 datetime.time 对象 :param timestr: 时间字符串 :return: datetime.time

pyape.util.func.vername2code(vername) → int

转换 vername 到 vercode，每个版本位允许3位。

参数:: vername -- 形如 3.1.455 的版本号。

pyape.util.encrypt

用于加密的封装

class pyape.util.encrypt.AES_CBC(key, iv)

使用 AES 进行对称加解密。

参数:

key -- 密钥。
iv -- iv。

class pyape.util.encrypt.Encrypt(key: str)

使用 Fernet 模块进行对称加解密。

参数:: key -- 密钥。

decrypt(cipher: str | bytes) → bytes: 对提供的 cipher_text 进行解密。

encrypt(plain: str | bytes) → bytes: 对提供的 plain_text 进行加密。

static fernet_key() → bytes: 生成一个 fernet 密钥。

pyape.util.gen

生成器

pyape.util.gen.gen_nonce(k=8, chars='abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'): 生成一段加密用的字符串字符串可以被用于密码、兑换码等等

pyape.util.gen.gen_password(password: str, salt: str): 计算密码散列后的值 :param password: :param salt: :return:

pyape.util.gen.gen_random_name(value=None, prefix=''): 提供一个整数和一个前缀，获取一个随机的昵称

pyape.util.gen.gen_redis_key_name(r, name, prefix=None, postfix=None, data_type=None, extra=[])

生成一个可用的名称，这个名称可能用于配置的键名，或者是排行榜的键名

参数:

r -- regional
name -- 名称
prefix -- 前缀
postfix -- 后缀
date_type -- 数据类型。数据类型位于后缀之后
extra -- 附加数据

pyape.util.gen.gen_secret_key() → str: 生成 FLASK 可以使用的 SECRET_KEY

pyape.util.gen.gen_sign(**kwargs): 根据传递来的参数顺序生成校验码