「学习笔记」Python 进阶(三)：工程实战 — 从脚本到项目

import asyncio

async def fetch_data(url):
    print(f"开始请求: {url}")
    await asyncio.sleep(2)  # 模拟耗时 I/O，期间控制权交还给事件循环
    print(f"请求完成: {url}")
    return f"Data from {url}"

async def main():
    # 必须使用 await 获取结果，否则只会得到协程对象
    result = await fetch_data("https://api.example.com")
    print(result)

# Python 3.7+ 启动事件循环的标准方式
asyncio.run(main())

import asyncio, time

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 完成"

async def main():
    start = time.time()
    results = await asyncio.gather(
        task("任务A", 3),
        task("任务B", 1),
        task("任务C", 2)
    )
    print(results)  # 输出: ['任务A 完成', '任务B 完成', '任务C 完成']
    print(f"并发执行，总耗时: {time.time() - start:.1f}s")  # ~3s（取最大值(最慢的任务)），不是 6s

asyncio.run(main())

import asyncio

# 限制同时最多 3 个并发请求
semaphore = asyncio.Semaphore(3)

async def rate_limited_request(url):
    async with semaphore:              # 获取信号量（最多 3 个并发）
        return await fetch_data(url)

async def main():
    urls = [f"url_{i}" for i in range(20)]
    results = await asyncio.gather(*[
        rate_limited_request(u) for u in urls
    ])

asyncio.run(main())

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task("A", 2))
        t2 = tg.create_task(task("B", 1))
        t3 = tg.create_task(task("C", 3))
    # 离开 async with 时自动等待所有任务
    print(t1.result(), t2.result(), t3.result())

# 推荐：asyncio.run()（Python 3.7+）
result = asyncio.run(async_fetch("http://example.com"))

# 备选：手动管理事件循环
loop = asyncio.new_event_loop()
try:
    result = loop.run_until_complete(async_fetch("http://example.com"))
finally:
    loop.close()

import trio

async def main():
    # Nursery 模型：所有子任务在 with 块结束时自动等待
    async with trio.open_nursery() as nursery:
        nursery.start_soon(child_task, "A")
        nursery.start_soon(child_task, "B")
        nursery.start_soon(child_task, "C")
    # 离开 with 块时，自动等待所有子任务完成——不会遗漏

# trio 核心工具：
# trio.Semaphore(n)           — 并发数限制
# trio.CapacityLimiter(n)     — 容量限制
# trio.fail_after(timeout)    — 超时控制
# trio.to_thread.run_sync     — 在线程池中运行同步函数

async def with_timeout():
    with trio.fail_after(10):        # 10 秒超时
        result = await trio.to_thread.run_sync(cpu_intensive_task)

from peewee import (
    Model, MySQLDatabase, CharField, TextField,
    IntegerField, BigIntegerField, DateTimeField,
    AutoField, fn
)

# 数据库连接
db = MySQLDatabase("ragflow", host="localhost", user="root", password="123456", port=3306)

# 基类模型
class BaseModel(Model):
    create_time = BigIntegerField(null=True, index=True)
    create_date = DateTimeField(null=True, index=True)

    class Meta:
        database = db

# 具体模型
class User(BaseModel):
    id = AutoField()
    name = CharField(max_length=100)
    email = CharField(max_length=255, unique=True)
    status = CharField(max_length=10, default="active")

class Knowledgebase(BaseModel):
    id = AutoField()
    name = CharField(max_length=100)
    description = TextField(null=True)
    tenant_id = IntegerField()     # 手动管理外键

    class Meta:
        table_name = "knowledgebase"

# Create
user = User.create(name="Tom", email="tom@test.com")
# 或：user = User(name="Tom", email="tom@test.com"); user.save()

# Read — 单条
user = User.get(User.email == "tom@test.com")

# Read — 列表 + 条件 + 排序 + 分页
users = (User.select()
    .where((User.status == "active") & (User.name.contains("Tom")))
    .order_by(User.create_time.desc())
    .limit(10))

# 聚合
count = User.select().where(User.status == "active").count()
max_id = User.select(fn.MAX(User.id)).scalar()

# Update
(User.update(status="inactive")
     .where(User.last_login < "2024-01-01")
     .execute())

# Delete
User.delete().where(User.status == "inactive").execute()

import json
from peewee import TextField

class JSONField(TextField):
    """Python 对象 ↔ JSON 字符串自动转换"""
    default_value = {}

    def db_value(self, value):
        """存储时：Python dict → JSON 字符串"""
        if value is None:
            value = self.default_value
        return json.dumps(value, ensure_ascii=False)

    def python_value(self, value):
        """读取时：JSON 字符串 → Python dict"""
        if not value:
            return self.default_value
        return json.loads(value)

# 使用：像操作 dict 一样操作数据库 JSON 字段
class Task(BaseModel):
    id = AutoField()
    config = JSONField()
    tags = JSONField(default_value=[])

task = Task.create(config={"model": "gpt-4"}, tags=["urgent"])
print(task.config)  # {"model": "gpt-4"} — 自动反序列化

# 上下文管理器事务（推荐）
with db.atomic():
    user = User.create(name="Tom", email="tom@test.com")
    kb = Knowledgebase.create(name="KB1", tenant_id=user.id)
    # 任何一句失败，全部回滚

# 装饰器事务
@db.atomic()
def create_user_with_kb(name, email, kb_name):
    user = User.create(name=name, email=email)
    kb = Knowledgebase.create(name=kb_name, tenant_id=user.id)
    return user, kb

# 连接池（生产环境推荐）
from playhouse.pool import PooledMySQLDatabase

db = PooledMySQLDatabase(
    "ragflow",
    max_connections=20,
    stale_timeout=300,
    host="localhost", user="root", password="123456",
)

from flask import Flask, request, jsonify

app = Flask(__name__) # 创建Flask应用实例，__name__表示当前模块名，Flask用它来确定资源位置

@app.route("/api/hello", methods=["GET"]) # 路由装饰器，将URL映射到函数
def hello():                              # 视图函数，处理请求并返回响应
    name = request.args.get("name", "World")
    return jsonify({"message": f"Hello, {name}!"})

@app.route("/api/user", methods=["POST"])
def create_user():
    data = request.json                 # JSON body
    # data = request.form.to_dict()     # 表单数据
    # file = request.files.get("file")  # 上传文件
    return jsonify({"id": 1, "name": data["name"]}), 201
    
@app.route('/post/<int:post_id>') # 动态路由（string(默认)/int/float/path）
def show_post(post_id):
    return f'文章ID: {post_id}'

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=9380, debug=True) # 启动开发服务器

<!DOCTYPE html>
<html>
<head>
    <title>{{ title }}</title>
</head>
<body>
    <h1>欢迎来到 {{ title }}</h1>
    {% if user %}
        <p>用户: {{ user }}</p>
    {% else %}
        <p>请登录</p>
    {% endif %}
</body>
</html>

from flask import render_template

@app.route('/')
def index():
    return render_template('index.html', title='首页', user='admin')

<!DOCTYPE html>
<html>
<head>
    <title>{% block title %}{% endblock %}</title>
</head>
<body>
    <div class="content">
        {% block content %}{% endblock %}
    </div>
</body>
</html>

{% extends "base.html" %}

{% block title %}
首页
{% endblock %}

{% block content %}
    <h1>欢迎来到首页</h1>
{% endblock %}

<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
<script src="{{ url_for('static', filename='script.js') }}"></script>
<img src="{{ url_for('static', filename='logo.png') }}" alt="Logo">

from flask import Blueprint

# 创建蓝图（类似 Spring 的 @RestController 分包）
manager = Blueprint("conversation", __name__)

@manager.route("/set", methods=["POST"])
def set_conversation():
    return jsonify({"success": True})

@manager.route("/get/<int:conv_id>", methods=["GET"])
def get_conversation(conv_id):
    return jsonify({"id": conv_id})

# 注册蓝图到 app
app = Flask(__name__)
app.register_blueprint(manager, url_prefix="/api/v1/conversation")
# 访问: POST /api/v1/conversation/set
#      GET  /api/v1/conversation/get/123

from flask import g, request
from functools import wraps

# 请求前钩子
@app.before_request
def before_request():
    g.start_time = time.time()             # g 是请求级上下文

# 请求后钩子
@app.after_request
def after_request(response):
    elapsed = time.time() - g.start_time
    response.headers["X-Response-Time"] = f"{elapsed:.3f}s"
    return response

# 自定义鉴权装饰器
def login_required(func):
    @wraps(func)
    def decorated_function(*args, **kwargs):
        token = request.headers.get("Authorization")
        if not token:
            return jsonify({"error": "Unauthorized"}), 401
        kwargs["tenant_id"] = get_tenant_from_token(token)
        return func(*args, **kwargs)
    return decorated_function

from flask import Response

@manager.route("/chat", methods=["POST"])
@login_required
def chat(tenant_id):
    """SSE 流式聊天"""
    def generate():
        for chunk in llm_stream(query):
            yield f"data: {json.dumps(chunk)}\n\n"
    return Response(generate(), mimetype="text/event-stream")

# 客户端（JavaScript）: new EventSource("/api/chat")
# 客户端（Python）: requests.get(url, stream=True).iter_lines()

from pathlib import Path
from importlib.util import module_from_spec, spec_from_file_location
from flask import Blueprint

def register_page(app, page_path):
    """动态注册一个路由模块 — Flask 项目的核心架构模式"""
    page_name = page_path.stem.removesuffix("_app")
    module_name = f"api.apps.{page_name}"

    # 动态导入模块
    spec = spec_from_file_location(module_name, page_path)
    page = module_from_spec(spec)
    page.app = app
    page.manager = Blueprint(page_name, module_name)
    sys.modules[module_name] = page
    spec.loader.exec_module(page)         # 执行模块，注册路由

    app.register_blueprint(page.manager, url_prefix=f"/api/v1/{page_name}")
    return f"/api/v1/{page_name}"

# 批量注册所有 *_app.py
pages_dir = Path(__file__).parent / "apps"
for path in pages_dir.glob("*_app.py"):
    if not path.name.startswith("."):
        register_page(app, path)

from pydantic import BaseModel, ValidationError

class User(BaseModel):
    id: int
    name: str
    age: int

# 1. 自动类型转换（宽松模式）
user = User(id="1", name="Alice", age="30") 
print(user.id, type(user.id))  # 输出: 1 <class 'int'>

# 2. 捕获验证错误
try:
    User(id="invalid", name="Bob", age="thirty")
except ValidationError as e:
    print(e)

from pydantic import BaseModel, Field

class Product(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    price: float = Field(..., gt=0, le=9999.99)
    stock: int = Field(default=0, ge=0)

from pydantic import BaseModel, field_validator

class Account(BaseModel):
    username: str = Field(..., min_length=2, max_length=50)

    @field_validator('username')
    @classmethod
    def check_username_length(cls, v: str) -> str:
        not_allowed = {"python", "javascript", "go", "java"}
        if v in not_allowed:
            raise ValueError(f"Must be not in {not_allowed}")
        return v

# test_example.py
def test_add():
    assert add(1, 2) == 3
    assert add(-1, 1) == 0

def test_divide_by_zero():
    import pytest
    with pytest.raises(ZeroDivisionError):
        1 / 0

import pytest
from peewee import SqliteDatabase

@pytest.fixture
def db():
    """创建内存数据库，测试结束后自动销毁"""
    database = SqliteDatabase(":memory:")
    database.connect()
    database.create_tables([User, Knowledgebase])
    yield database       # yield 之前 = setup
    database.close()     # yield 之后 = teardown

def test_create_user(db):
    user = User.create(name="Tom", email="tom@test.com")
    assert user.id is not None
    assert User.select().count() == 1

import pytest

@pytest.mark.parametrize("input,expected", [
    ("hello", "HELLO"),
    ("World", "WORLD"),
    ("", ""),
])
def test_upper(input, expected):
    assert input.upper() == expected

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3), (-1, 1, 0), (100, 200, 300),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

pytest                       # 运行所有测试
pytest -v                    # 详细输出
pytest -s                    # 显示 print 输出
pytest test_file.py::test_fn # 运行指定测试
pytest --cov=rag             # 生成覆盖率报告

from concurrent.futures import ThreadPoolExecutor, as_completed

def download(url):
    return f"Downloaded {url}"

urls = [f"url_{i}" for i in range(10)]

# 方式 1：批量提交
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(download, url): url for url in urls}
    for future in as_completed(futures):
        result = future.result()

# 方式 2：map（保持顺序）
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(download, urls))

import threading

# Lock — 互斥锁
lock = threading.Lock()
with lock:
    shared_resource.update()

# RLock — 可重入锁（同一线程可多次获取）
rlock = threading.RLock()

# Queue — 线程安全队列
from queue import Queue
q = Queue()
q.put(item)
item = q.get()

# httpx（推荐，支持 async）
import httpx

# 同步
response = httpx.get("http://localhost:8080/api/data")
data = response.json()

# 异步
async with httpx.AsyncClient() as client:
    responses = await asyncio.gather(*[
        client.get(f"http://localhost:8080/api/item/{i}")
        for i in range(10)
    ])

# requests（老牌）
import requests
resp = requests.get("http://localhost:8080/api/data")
print(resp.json())
print(resp.status_code)

python -m venv .venv
source .venv/bin/activate   # Linux/Mac
.venv\Scripts\activate      # Windows PowerShell
pip install flask peewee

uv venv                              # 创建虚拟环境
uv pip install flask peewee          # 安装包（比 pip 快 10-100 倍）
uv pip install -e ".[full]"          # 安装项目 + 可选依赖

[project]
name = "ragflow"
version = "0.16.0"
requires-python = ">=3.10"
dependencies = [
    "flask>=3.0.0",
    "peewee>=3.17.0",
    "elasticsearch>=8.0.0",
]

[project.optional-dependencies]
full = [
    "torch>=2.0.0",
    "transformers>=4.40.0",
    "paddleocr>=2.7.0",
]

[tool.ruff]                  # 代码检查（替代 flake8）
line-length = 120

[tool.pytest.ini_options]    # pytest 配置
testpaths = ["test"]

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger(__name__)

# 使用（5 个级别）
logger.debug("调试信息")       # 通常不显示
logger.info("普通信息")        # 记录关键流程
logger.warning("警告")         # 可能的问题
logger.error("错误")           # 需要关注的错误
logger.critical("严重错误")

# 最佳实践：用 % 格式化延迟求值（不要用 f-string！）
logger.info("Processing doc %s, chunks: %d", doc_name, len(chunks))  # ✅
# logger.info(f"Processing doc {doc_name}")  # ❌ 即使不输出也会先格式化

# 记录异常 + 完整堆栈
try:
    1 / 0
except Exception:
    logger.exception("计算失败")  # 自动包含 traceback