基础与中间件

从 cqorm 导入 基类, 列, 整数, 文本, 数据库

类 用户(基类):
    __tablename__ = "用户表"
    编号 = 列(整数, 主键=True)
    姓名 = 列(文本(50), 空=False, 唯一=True)
    年龄 = 列(整数, 默认=18)

数据库对象 = 数据库("sqlite:///app.db")
数据库对象.创建表()

with 数据库对象.会话() as 会话:
    会话.添加(用户(姓名="小明"))
    会话.提交()

    # 查询（链式）
    全部 = 会话.查询(用户).全部()
    筛选 = 会话.查询(用户).过滤(用户.年龄 > 18).排序(用户.年龄.desc()).全部()
    分页 = 会话.查询(用户).分页(页码=1, 每页数量=10).全部()

    # 按主键取
    某人 = 会话.获取(用户, 1)

    # 更新 / 删除
    某人.年龄 = 30
    会话.提交()
    会话.删除(某人)
    会话.提交()

从 cqdatetime 导入 现在, 解析, 时间, 计算年龄

当前 = 现在()
print(当前.格式化("YYYY年MM月DD日 HH:mm:ss"))   # 2026年08月01日 10:30:00
print(当前.人类化())                            # 刚刚 / 3 小时前

明天 = 当前.偏移(天=1)
print(明天.星期名称())                          # 星期日

生日 = 解析("2000-01-01")
print(计算年龄(生日))                           # 26

# 属性与运算
print(当前.年(), 当前.月(), 当前.日())          # 2026 8 1
print(当前.当天开始(), 当前.当天结束())
print(当前.转时间戳(), 当前.转ISO())
print(当前.转换为时区("Asia/Shanghai"))

从 cqredis 导入 客户端

缓存 = 客户端()                    # 默认 127.0.0.1:6379

# 字符串
缓存.设置("名称", "小明", 过期秒=3600)
print(缓存.获取("名称"))
缓存.递增("计数器")

# 哈希
缓存.哈希批量设置("用户:1", {"姓名": "小红", "年龄": 20})
print(缓存.哈希全部("用户:1"))     # {'姓名': '小红', '年龄': '20'}

# 列表
缓存.列表右推("队列", 1, 2, 3)
print(缓存.列表范围("队列"))       # ['1', '2', '3']

# 发布订阅
缓存.发布("通知", "有新消息")
订阅者 = 缓存.订阅("通知")

# 管道
管道 = 缓存.管道()
管道.设置("a", 1).设置("b", 2).递增("a")
结果 = 管道.执行()

缓存.关闭()

从 cqcli 导入 命令, 选项, 参数, 运行, 打印, 提示, 确认, 进度条, 表

@命令(名称="greet")                # 名称 为终端调用名（中文函数名无法直接输入）
def 打招呼(名字: str = 参数("世界", 帮助="你的名字"),
           大声: bool = 选项(标记=True, 帮助="大声输出")):
    文本 = "你好，{}！".format(名字)
    if 大声:
        文本 = 文本.upper()
    打印(文本, 前景="green")

@命令(名称="统计")
def 统计():
    姓名 = 提示("请输入姓名", 默认="小明")
    if 确认("继续吗？"):
        表(["姓名", "状态"], [[姓名, "完成"]])
    with 进度条(范围(100), 标签="处理中") as 条:
        for 项 in 条:
            跳过

if __name__ == "__main__":
    运行()