参考

pyape.db

处理 pyape 数据库功能

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

class pyape.db.DBManager(URI: dict | str, ENGINE_OPTIONS: dict | None = 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 工厂类，使用 sessionmaker 生成。

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

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

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

create_scoped_session(in_flask: bool = False) → scoped_session

创建一个 scoped_session 代理。

参数:: in_flask – 是否在 Flask 中使用。

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

get_engine(bind_key: str | None = None) → Engine

获取一个 Engine 对象。

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

set_Model(bind_key: str | None = 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 = None, URI: dict | str | None = None, ENGINE_OPTIONS: dict | None = None, is_scoped: bool = True, in_flask: 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 = None)

获取对应的 Model Factory class。

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

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

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

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

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

创建 table。

参数:

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

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

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

移除 table。

参数:

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

engine(bind_key: str | None = None) → Engine

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

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

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

获取一个 Table。

参数:

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

isModel(instance, bind_key: str | None = None)

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

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

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

query(*entities) → Query

获取一个 Query 对象。

参数:: model_cls – 一个 Model对象。

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

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

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, target, type_='file', fmt='text', level=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): md5(base64) 算法验证工具： http://www.cmd5.com/hash.aspx

pyape.util.md5txt(txt): 计算 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 = 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 = None)

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

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

pyape.util.func.in_time_range(timestart: str | time, timeend: str | time, dt: datetime | None = 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 = None) → datetime

返回上个月的 datetime 对象。

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

pyape.util.func.next_month_dt(day: datetime | None = 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) → str: 对提供的 cipher_text 进行解密。

encrypt(plain: str | bytes) → str: 对提供的 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, salt): 计算密码散列后的值 :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): 根据传递来的参数顺序生成校验码