异步数据库与任务队列

# 安装
# pip install asyncpg

import asyncio
import asyncpg


async def basic_connection():
    """基本连接与查询"""
    # 创建连接
    conn = await asyncpg.connect(
        host="localhost",
        port=5432,
        user="postgres",
        password="secret",
        database="mydb",
    )

    try:
        # 执行查询 — 返回 Record 对象列表
        rows = await conn.fetch("SELECT id, name, email FROM users WHERE age > $1", 18)

        for row in rows:
            print(f"ID: {row['id']}, Name: {row['name']}, Email: {row['email']}")

        # 查询单行
        user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", 42)
        if user:
            print(f"Found: {user['name']}")

        # 查询单值
        count = await conn.fetchval("SELECT COUNT(*) FROM users")
        print(f"Total users: {count}")

        # 执行写操作 — 返回影响行数
        result = await conn.execute(
            "INSERT INTO users(name, email, age) VALUES($1, $2, $3)",
            "张三",
            "zhangsan@example.com",
            25,
        )
        print(f"Inserted: {result}")
    finally:
        await conn.close()


asyncio.run(basic_connection())

import asyncio
import asyncpg


class DatabasePool:
    """异步数据库连接池管理器"""

    def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self._pool: asyncpg.Pool | None = None

    async def initialize(self):
        """初始化连接池"""
        self._pool = await asyncpg.create_pool(
            dsn=self.dsn,
            min_size=self.min_size,
            max_size=self.max_size,
            # 连接超时
            command_timeout=60,
            # 连接初始化 SQL（设置时区等）
            setup=self._setup_connection,
        )
        print(f"连接池已初始化: min={self.min_size}, max={self.max_size}")

    @staticmethod
    async def _setup_connection(conn: asyncpg.Connection):
        """每个新连接创建时执行"""
        await conn.execute("SET timezone = 'Asia/Shanghai'")

    async def close(self):
        """关闭连接池"""
        if self._pool:
            await self._pool.close()
            print("连接池已关闭")

    @property
    def pool(self) -> asyncpg.Pool:
        if self._pool is None:
            raise RuntimeError("连接池未初始化，请先调用 initialize()")
        return self._pool

    async def fetch(self, query: str, *args):
        return await self.pool.fetch(query, *args)

    async def fetchrow(self, query: str, *args):
        return await self.pool.fetchrow(query, *args)

    async def fetchval(self, query: str, *args):
        return await self.pool.fetchval(query, *args)

    async def execute(self, query: str, *args):
        return await self.pool.execute(query, *args)


# 使用示例
async def main():
    db = DatabasePool("postgresql://postgres:secret@localhost/mydb")

    try:
        await db.initialize()

        # 并发查询 — 连接池自动分配连接
        results = await asyncio.gather(
            db.fetch("SELECT * FROM users LIMIT 10"),
            db.fetchval("SELECT COUNT(*) FROM orders"),
            db.fetch("SELECT * FROM products WHERE price > $1", 100),
        )

        print(f"用户列表: {len(results[0])} 条")
        print(f"订单总数: {results[1]}")
        print(f"高价商品: {len(results[2])} 条")
    finally:
        await db.close()


asyncio.run(main())

import asyncpg


async def transaction_example():
    conn = await asyncpg.connect("postgresql://postgres:secret@localhost/mydb")

    try:
        # 方式一：async with 自动提交/回滚
        async with conn.transaction():
            await conn.execute(
                "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
                100, "user_a",
            )
            await conn.execute(
                "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
                100, "user_b",
            )
            await conn.execute(
                "INSERT INTO transfers(from_id, to_id, amount) VALUES($1, $2, $3)",
                "user_a", "user_b", 100,
            )
        # 自动提交 — 如果中间任何一步出错，自动回滚

        # 方式二：手动控制事务
        tr = conn.transaction()
        await tr.start()
        try:
            await conn.execute("DELETE FROM temp_data WHERE expired_at < NOW()")
            await conn.execute("INSERT INTO cleanup_log(table_name, rows_deleted) VALUES($1, $2)", "temp_data", 50)
            await tr.commit()
        except Exception:
            await tr.rollback()
            raise

        # 方式三：SAVEPOINT 嵌套事务
        async with conn.transaction():
            await conn.execute("INSERT INTO orders(user_id, total) VALUES($1, $2)", 1, 500)
            async with conn.transaction():  # 自动创建 SAVEPOINT
                await conn.execute("INSERT INTO order_items(order_id, product_id, qty) VALUES($1, $2, $3)", 101, 5, 2)
                # 即使这里出错，外层事务也会回滚
    finally:
        await conn.close()


asyncio.run(transaction_example())

import asyncio
import asyncpg


async def bulk_operations():
    conn = await asyncpg.connect("postgresql://postgres:secret@localhost/mydb")

    try:
        # 方式一：executemany — 高效批量插入/更新
        users = [
            ("张三", "zhangsan@example.com", 25),
            ("李四", "lisi@example.com", 30),
            ("王五", "wangwu@example.com", 28),
        ]
        await conn.executemany(
            "INSERT INTO users(name, email, age) VALUES($1, $2, $3)",
            users,
        )

        # 方式二：copy_records_to_table — 最快的大批量导入
        # 适合万级以上的数据量
        records = [(i, f"product_{i}", i * 10.0) for i in range(1, 10001)]
        await conn.copy_records_to_table(
            "products",
            records=records,
            columns=["id", "name", "price"],
        )
        print("批量导入 10000 条记录完成")

        # 方式三：UNNEST — PostgreSQL 特有的高效批量操作
        names = ["产品A", "产品B", "产品C"]
        prices = [99.9, 199.9, 299.9]
        await conn.execute(
            """
            INSERT INTO products(name, price)
            SELECT * FROM unnest($1::text[], $2::float8[])
            """,
            names,
            prices,
        )
    finally:
        await conn.close()


asyncio.run(bulk_operations())

# 安装
# pip install aiomysql

import asyncio
import aiomysql


async def basic_mysql():
    """aiomysql 基本连接与查询"""
    conn = await aiomysql.connect(
        host="localhost",
        port=3306,
        user="root",
        password="secret",
        db="mydb",
        charset="utf8mb4",
        autocommit=False,  # 手动控制事务
    )

    try:
        async with conn.cursor() as cur:
            # 查询
            await cur.execute("SELECT id, name, email FROM users WHERE age > %s", (18,))
            rows = await cur.fetchall()
            for row in rows:
                print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")

            # 单行查询
            await cur.execute("SELECT * FROM users WHERE id = %s", (42,))
            user = await cur.fetchone()
            if user:
                print(f"Found: {user[1]}")

            # 写操作
            await cur.execute(
                "INSERT INTO users(name, email, age) VALUES(%s, %s, %s)",
                ("张三", "zhangsan@example.com", 25),
            )
            await conn.commit()  # 显式提交
    finally:
        conn.close()


asyncio.run(basic_mysql())

import asyncio
import aiomysql


async def dict_cursor_example():
    """使用字典游标，通过列名访问字段"""
    conn = await aiomysql.connect(
        host="localhost",
        port=3306,
        user="root",
        password="secret",
        db="mydb",
        charset="utf8mb4",
    )

    try:
        # 使用 DictCursor
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute("SELECT id, name, age FROM users LIMIT 5")
            rows = await cur.fetchall()

            for row in rows:
                print(f"用户: {row['name']}, 年龄: {row['age']}")

        # 也可以在创建连接时全局指定
        conn2 = await aiomysql.connect(
            host="localhost",
            port=3306,
            user="root",
            password="secret",
            db="mydb",
            charset="utf8mb4",
            cursorclass=aiomysql.DictCursor,
        )
        await conn2.ensure_closed()
    finally:
        conn.close()


asyncio.run(dict_cursor_example())

import asyncio
import aiomysql


async def pool_example():
    """aiomysql 连接池"""
    pool = await aiomysql.create_pool(
        host="localhost",
        port=3306,
        user="root",
        password="secret",
        db="mydb",
        charset="utf8mb4",
        minsize=5,
        maxsize=20,
        autocommit=True,
    )

    try:
        # 从连接池获取连接
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT COUNT(*) FROM users")
                count = await cur.fetchone()
                print(f"用户总数: {count[0]}")

        # 并发查询
        async def query_user(user_id: int):
            async with pool.acquire() as conn:
                async with conn.cursor(aiomysql.DictCursor) as cur:
                    await cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
                    return await cur.fetchone()

        results = await asyncio.gather(
            query_user(1), query_user(2), query_user(3),
        )
        for r in results:
            if r:
                print(f"查询结果: {r['name']}")
    finally:
        pool.close()
        await pool.wait_closed()


asyncio.run(pool_example())

# 安装
# pip install "redis[hiredis]"

import asyncio
import redis.asyncio as aioredis


async def basic_redis():
    """Redis 异步基本操作"""
    # 创建连接
    redis = await aioredis.from_url(
        "redis://localhost:6379",
        encoding="utf-8",
        decode_responses=True,  # 自动解码为字符串
    )

    try:
        # String 操作
        await redis.set("user:1:name", "张三")
        name = await redis.get("user:1:name")
        print(f"用户名: {name}")

        # 带过期时间
        await redis.set("session:abc123", "user_data", ex=3600)  # 1小时过期
        ttl = await redis.ttl("session:abc123")
        print(f"剩余 TTL: {ttl}秒")

        # Hash 操作
        await redis.hset("user:1", mapping={
            "name": "张三",
            "email": "zhangsan@example.com",
            "age": "25",
        })
        user = await redis.hgetall("user:1")
        print(f"用户信息: {user}")

        # List 操作
        await redis.lpush("task_queue", "task1", "task2", "task3")
        task = await redis.brpop("task_queue", timeout=5)  # 阻塞式弹出
        print(f"取出任务: {task}")

        # Set 操作
        await redis.sadd("tags:python", "asyncio", "fastapi", "web")
        tags = await redis.smembers("tags:python")
        print(f"标签: {tags}")

        # Sorted Set 操作
        await redis.zadd("leaderboard", {"玩家A": 9500, "玩家B": 8700, "玩家C": 9200})
        top3 = await redis.zrevrange("leaderboard", 0, 2, withscores=True)
        print(f"排行榜: {top3}")
    finally:
        await redis.close()


asyncio.run(basic_redis())

import asyncio
import redis.asyncio as aioredis


async def pipeline_example():
    """Redis Pipeline 批量操作"""
    redis = await aioredis.from_url(
        "redis://localhost:6379",
        decode_responses=True,
    )

    try:
        # Pipeline — 一次性发送多条命令
        async with redis.pipeline(transaction=True) as pipe:
            # 链式调用
            results = (
                await pipe.set("key1", "value1")
                .set("key2", "value2")
                .set("key3", "value3")
                .get("key1")
                .get("key2")
                .execute()
            )
            # results = [True, True, True, "value1", "value2"]
            print(f"Pipeline 结果: {results}")

        # 批量设置 Hash
        users = {
            "user:1": {"name": "张三", "age": "25"},
            "user:2": {"name": "李四", "age": "30"},
            "user:3": {"name": "王五", "age": "28"},
        }

        async with redis.pipeline() as pipe:
            for key, data in users.items():
                pipe.hset(key, mapping=data)
            await pipe.execute()
            print("批量 Hash 设置完成")

        # 批量读取
        async with redis.pipeline() as pipe:
            for key in users:
                pipe.hgetall(key)
            results = await pipe.execute()
            print(f"批量读取: {results}")
    finally:
        await redis.close()


asyncio.run(pipeline_example())

import asyncio
import redis.asyncio as aioredis


async def publisher():
    """发布者"""
    redis = await aioredis.from_url("redis://localhost:6379", decode_responses=True)
    try:
        for i in range(5):
            await redis.publish("news_channel", f"新闻 #{i}: 今日要闻")
            print(f"[发布] 新闻 #{i}")
            await asyncio.sleep(1)
    finally:
        await redis.close()


async def subscriber():
    """订阅者"""
    redis = await aioredis.from_url("redis://localhost:6379", decode_responses=True)
    try:
        pubsub = redis.pubsub()
        await pubsub.subscribe("news_channel")
        print("[订阅] 等待消息...")

        async for message in pubsub.listen():
            if message["type"] == "message":
                print(f"[收到] {message['data']}")
    finally:
        await redis.close()


async def pubsub_demo():
    """同时运行发布者和订阅者"""
    await asyncio.gather(
        subscriber(),
        publisher(),
    )


# 实际使用时，发布者和订阅者通常在不同进程中运行
# asyncio.run(pubsub_demo())

import asyncio


async def queue_basics():
    """Queue 基本操作"""
    # 创建无界队列
    q: asyncio.Queue[str] = asyncio.Queue()

    # 放入元素
    await q.put("任务A")
    await q.put("任务B")
    await q.put("任务C")

    # 查看队列大小
    print(f"队列大小: {q.qsize()}")  # 3

    # 取出元素（FIFO）
    item = await q.get()
    print(f"取出: {item}")  # 任务A

    # 标记任务完成（配合 join 使用）
    q.task_done()

    # 等待队列清空
    await q.join()
    print("所有任务已完成")


async def bounded_queue():
    """有界队列 — 背压控制"""
    # maxsize 限制队列最大容量
    q: asyncio.Queue[int] = asyncio.Queue(maxsize=3)

    # 队列满时 put 会阻塞，直到有消费者取走元素
    async def producer():
        for i in range(10):
            await q.put(i)
            print(f"[生产] 放入 {i}, 队列大小: {q.qsize()}")

    async def consumer():
        for _ in range(10):
            item = await q.get()
            print(f"[消费] 取出 {item}")
            await asyncio.sleep(0.1)  # 模拟处理时间
            q.task_done()

    await asyncio.gather(producer(), consumer())


asyncio.run(bounded_queue())

import asyncio


async def priority_queue_example():
    """优先级队列 — 按优先级处理任务"""
    q: asyncio.PriorityQueue[tuple[int, str]] = asyncio.PriorityQueue()

    # 放入 (优先级, 数据) 元组 — 数字越小优先级越高
    await q.put((3, "普通任务"))
    await q.put((1, "紧急任务"))
    await q.put((2, "重要任务"))
    await q.put((1, "另一个紧急任务"))

    # 按优先级取出
    while not q.empty():
        priority, task = await q.get()
        print(f"优先级 {priority}: {task}")
        q.task_done()

    # 输出:
    # 优先级 1: 紧急任务
    # 优先级 1: 另一个紧急任务
    # 优先级 2: 重要任务
    # 优先级 3: 普通任务


asyncio.run(priority_queue_example())

import asyncio
import random


async def producer(queue: asyncio.Queue, producer_id: int):
    """生产者 — 产生数据"""
    for i in range(5):
        item = f"产品-{producer_id}-{i}"
        await queue.put(item)
        print(f"[生产者 {producer_id}] 放入: {item}, 队列大小: {queue.qsize()}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
    print(f"[生产者 {producer_id}] 完成")


async def consumer(queue: asyncio.Queue, consumer_id: int):
    """消费者 — 处理数据"""
    while True:
        item = await queue.get()
        try:
            print(f"[消费者 {consumer_id}] 处理: {item}")
            await asyncio.sleep(random.uniform(0.2, 0.5))  # 模拟处理
        finally:
            queue.task_done()  # 无论成功失败都标记完成


async def producer_consumer():
    """生产者-消费者编排"""
    queue: asyncio.Queue[str] = asyncio.Queue(maxsize=10)

    # 启动多个生产者
    producers = [asyncio.create_task(producer(queue, i)) for i in range(3)]

    # 启动多个消费者
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(2)]

    # 等待所有生产者完成
    await asyncio.gather(*producers)

    # 等待队列清空
    await queue.join()

    # 取消消费者（它们在无限循环中等待）
    for c in consumers:
        c.cancel()
    await asyncio.gather(*consumers, return_exceptions=True)

    print("全部完成")


asyncio.run(producer_consumer())

import asyncio


class PoisonPillConsumer:
    """使用毒丸（Poison Pill）模式优雅退出消费者"""

    POISON = None  # 特殊标记，消费者收到后退出

    def __init__(self, num_producers: int = 3, num_consumers: int = 2):
        self.queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=20)
        self.num_producers = num_producers
        self.num_consumers = num_consumers

    async def producer(self, pid: int):
        for i in range(3):
            item = f"P{pid}-item{i}"
            await self.queue.put(item)
            print(f"[生产者 {pid}] 产出: {item}")
            await asyncio.sleep(0.1)
        print(f"[生产者 {pid}] 结束")

    async def consumer(self, cid: int):
        while True:
            item = await self.queue.get()
            try:
                if item is self.POISON:
                    print(f"[消费者 {cid}] 收到终止信号，退出")
                    break
                print(f"[消费者 {cid}] 处理: {item}")
                await asyncio.sleep(0.15)
            finally:
                self.queue.task_done()

    async def run(self):
        producers = [asyncio.create_task(self.producer(i)) for i in range(self.num_producers)]
        consumers = [asyncio.create_task(self.consumer(i)) for i in range(self.num_consumers)]

        # 等待生产者完成
        await asyncio.gather(*producers)

        # 向每个消费者发送一个毒丸
        for _ in consumers:
            await self.queue.put(self.POISON)

        # 等待消费者收到毒丸后自行退出
        await asyncio.gather(*consumers)
        print("所有参与者已优雅退出")


asyncio.run(PoisonPillConsumer().run())

import asyncio


async def task_scheduling():
    """asyncio 任务调度基础"""

    # 1. create_task — 立即调度协程
    task = asyncio.create_task(some_io_work("任务A"), name="task-a")

    # 2. gather — 并发执行，等待全部完成
    results = await asyncio.gather(
        some_io_work("任务1"),
        some_io_work("任务2"),
        some_io_work("任务3"),
        return_exceptions=True,  # 异常不抛出，作为结果返回
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"任务失败: {r}")
        else:
            print(f"任务结果: {r}")

    # 3. wait — 更精细的控制
    tasks = [asyncio.create_task(some_io_work(f"T{i}")) for i in range(5)]
    done, pending = await asyncio.wait(
        tasks,
        timeout=2.0,          # 最长等待2秒
        return_when=asyncio.FIRST_COMPLETED,  # 第一个完成就返回
    )
    print(f"已完成: {len(done)}, 进行中: {len(pending)}")

    # 取消未完成的任务
    for t in pending:
        t.cancel()

    # 4. TaskGroup (3.11+) — 结构化并发
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(some_io_work("TG-1"))
        t2 = tg.create_task(some_io_work("TG-2"))
        t3 = tg.create_task(some_io_work("TG-3"))
    # 退出 with 块时，所有任务保证完成；任一异常会取消其他任务


async def some_io_work(name: str) -> str:
    await asyncio.sleep(0.5)
    return f"{name} 完成"


asyncio.run(task_scheduling())

import asyncio


async def semaphore_control():
    """使用信号量限制并发数"""
    sem = asyncio.Semaphore(5)  # 最多 5 个并发

    async def limited_task(task_id: int):
        async with sem:  # 获取许可
            print(f"[任务 {task_id}] 开始")
            await asyncio.sleep(1)
            print(f"[任务 {task_id}] 完成")
        # 离开 with 块自动释放许可

    # 20 个任务，但同一时刻最多只有 5 个在执行
    await asyncio.gather(*[limited_task(i) for i in range(20)])


asyncio.run(semaphore_control())

import asyncio
import time
from datetime import datetime


class AsyncScheduler:
    """轻量级异步定时任务调度器"""

    def __init__(self):
        self._tasks: list[asyncio.Task] = []

    def schedule_interval(self, coro_factory, interval: float, name: str = ""):
        """按固定间隔执行"""
        async def wrapper():
            while True:
                start = time.monotonic()
                try:
                    await coro_factory()
                except Exception as e:
                    print(f"[{name}] 执行出错: {e}")
                elapsed = time.monotonic() - start
                remaining = interval - elapsed
                if remaining > 0:
                    await asyncio.sleep(remaining)
                else:
                    print(f"[{name}] 执行超时 {elapsed:.2f}s > 间隔 {interval}s")

        task = asyncio.create_task(wrapper(), name=name)
        self._tasks.append(task)
        return task

    def schedule_daily(self, coro_factory, hour: int, minute: int = 0, name: str = ""):
        """每天定时执行"""
        async def wrapper():
            while True:
                now = datetime.now()
                target = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
                if now >= target:
                    # 今天的时间已过，等明天
                    target = target.replace(day=target.day + 1)
                delay = (target - now).total_seconds()
                print(f"[{name}] 下次执行: {target}, 等待 {delay:.0f}秒")
                await asyncio.sleep(delay)
                try:
                    await coro_factory()
                except Exception as e:
                    print(f"[{name}] 执行出错: {e}")

        task = asyncio.create_task(wrapper(), name=name)
        self._tasks.append(task)
        return task

    async def stop(self):
        """停止所有定时任务"""
        for task in self._tasks:
            task.cancel()
        await asyncio.gather(*self._tasks, return_exceptions=True)
        print("调度器已停止")


# 使用示例
async def main():
    scheduler = AsyncScheduler()

    # 每 5 秒清理过期缓存
    scheduler.schedule_interval(
        lambda: clean_cache(),
        interval=5,
        name="缓存清理",
    )

    # 每 30 秒同步数据
    scheduler.schedule_interval(
        lambda: sync_data(),
        interval=30,
        name="数据同步",
    )

    # 每天凌晨 2 点生成报表
    scheduler.schedule_daily(
        lambda: generate_report(),
        hour=2,
        minute=0,
        name="日报生成",
    )

    try:
        await asyncio.sleep(120)  # 运行2分钟演示
    finally:
        await scheduler.stop()


async def clean_cache():
    print(f"[{datetime.now()}] 清理过期缓存...")


async def sync_data():
    print(f"[{datetime.now()}] 同步数据...")


async def generate_report():
    print(f"[{datetime.now()}] 生成日报...")


# asyncio.run(main())

# pip install apscheduler

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
import asyncio


scheduler = AsyncScheduler()


async def cleanup_job():
    print("执行清理任务...")


async def report_job():
    print("生成报表...")


async def health_check():
    print("健康检查...")


async def main():
    # 间隔触发
    scheduler.add_job(cleanup_job, IntervalTrigger(seconds=30), id="cleanup")

    # Cron 表达式触发
    scheduler.add_job(
        report_job,
        CronTrigger(hour=2, minute=0),  # 每天凌晨2点
        id="daily_report",
    )

    # 每5分钟健康检查
    scheduler.add_job(
        health_check,
        IntervalTrigger(minutes=5),
        id="health_check",
    )

    scheduler.start()
    print("调度器已启动")

    try:
        await asyncio.sleep(300)  # 运行5分钟
    finally:
        scheduler.shutdown()


# asyncio.run(main())

import asyncio
import asyncpg
import aiomysql
import redis.asyncio as aioredis
from contextlib import asynccontextmanager
from dataclasses import dataclass


@dataclass
class PostgresConfig:
    dsn: str = "postgresql://postgres:secret@localhost:5432/app_db"
    min_size: int = 5
    max_size: int = 20


@dataclass
class MySQLConfig:
    host: str = "localhost"
    port: int = 3306
    user: str = "root"
    password: str = "secret"
    db: str = "app_db"
    minsize: int = 3
    maxsize: int = 15


@dataclass
class RedisConfig:
    url: str = "redis://localhost:6379/0"
    max_connections: int = 20


class DatabaseManager:
    """多数据源连接池管理器 — 单例模式"""

    _instance: "DatabaseManager | None" = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if hasattr(self, "_initialized"):
            return
        self._initialized = True
        self._pg_pool: asyncpg.Pool | None = None
        self._mysql_pool: aiomysql.Pool | None = None
        self._redis: aioredis.Redis | None = None

    async def init(
        self,
        pg_config: PostgresConfig = PostgresConfig(),
        mysql_config: MySQLConfig = MySQLConfig(),
        redis_config: RedisConfig = RedisConfig(),
    ):
        """初始化所有连接池"""
        # PostgreSQL
        self._pg_pool = await asyncpg.create_pool(
            dsn=pg_config.dsn,
            min_size=pg_config.min_size,
            max_size=pg_config.max_size,
        )

        # MySQL
        self._mysql_pool = await aiomysql.create_pool(
            host=mysql_config.host,
            port=mysql_config.port,
            user=mysql_config.user,
            password=mysql_config.password,
            db=mysql_config.db,
            minsize=mysql_config.minsize,
            maxsize=mysql_config.maxsize,
            charset="utf8mb4",
            autocommit=True,
        )

        # Redis
        self._redis = await aioredis.from_url(
            redis_config.url,
            max_connections=redis_config.max_connections,
            decode_responses=True,
        )

        print("所有数据库连接池已初始化")

    @property
    def pg(self) -> asyncpg.Pool:
        if self._pg_pool is None:
            raise RuntimeError("PostgreSQL 连接池未初始化")
        return self._pg_pool

    @property
    def mysql(self) -> aiomysql.Pool:
        if self._mysql_pool is None:
            raise RuntimeError("MySQL 连接池未初始化")
        return self._mysql_pool

    @property
    def redis(self) -> aioredis.Redis:
        if self._redis is None:
            raise RuntimeError("Redis 连接未初始化")
        return self._redis

    @asynccontextmanager
    async def pg_transaction(self):
        """PostgreSQL 事务上下文管理器"""
        async with self.pg.acquire() as conn:
            async with conn.transaction():
                yield conn

    @asynccontextmanager
    async def mysql_transaction(self):
        """MySQL 事务上下文管理器"""
        async with self.mysql.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                try:
                    yield cur
                    await conn.commit()
                except Exception:
                    await conn.rollback()
                    raise

    async def close(self):
        """关闭所有连接"""
        if self._pg_pool:
            await self._pg_pool.close()
        if self._mysql_pool:
            self._mysql_pool.close()
            await self._mysql_pool.wait_closed()
        if self._redis:
            await self._redis.close()
        print("所有数据库连接已关闭")


# 使用示例 — 模拟 FastAPI 生命周期
async def app_lifecycle():
    db = DatabaseManager()

    # 启动时初始化
    await db.init()

    try:
        # 业务操作
        async with db.pg_transaction() as conn:
            users = await conn.fetch("SELECT * FROM users LIMIT 5")
            print(f"PG 用户: {len(users)} 条")

        async with db.mysql_transaction() as cur:
            await cur.execute("SELECT COUNT(*) FROM orders")
            count = await cur.fetchone()
            print(f"MySQL 订单数: {count[0]}")

        cached = await db.redis.get("app:config")
        print(f"Redis 缓存: {cached}")
    finally:
        # 关闭时清理
        await db.close()


asyncio.run(app_lifecycle())

import asyncio
import random
import re
import json
from dataclasses import dataclass, asdict
from datetime import datetime


@dataclass
class RawItem:
    """原始爬取数据"""
    url: str
    title: str
    content: str
    fetched_at: str


@dataclass
class CleanItem:
    """清洗后数据"""
    url: str
    title: str
    content: str
    word_count: int
    cleaned_at: str


class AsyncDataPipeline:
    """异步数据管道：爬取 → 清洗 → 入库"""

    def __init__(
        self,
        num_fetchers: int = 3,
        num_cleaners: int = 2,
        num_writers: int = 1,
        batch_size: int = 20,
    ):
        self.num_fetchers = num_fetchers
        self.num_cleaners = num_cleaners
        self.num_writers = num_writers
        self.batch_size = batch_size

        self.raw_queue: asyncio.Queue[RawItem | None] = asyncio.Queue(maxsize=50)
        self.clean_queue: asyncio.Queue[CleanItem | None] = asyncio.Queue(maxsize=100)

        self.stats = {"fetched": 0, "cleaned": 0, "written": 0}

    async def fetcher(self, fetcher_id: int):
        """爬取层 — 从数据源获取原始数据"""
        urls = [
            "https://example.com/article/1",
            "https://example.com/article/2",
            "https://example.com/article/3",
            "https://example.com/article/4",
            "https://example.com/article/5",
        ]

        for url in urls:
            # 模拟网络请求
            await asyncio.sleep(random.uniform(0.1, 0.3))

            item = RawItem(
                url=url,
                title=f"  文章标题: 示例{fetcher_id}  ",
                content=f"这是从 {url} 爬取的内容，包含 <b>HTML标签</b> 和  多余空格  ",
                fetched_at=datetime.now().isoformat(),
            )
            await self.raw_queue.put(item)
            self.stats["fetched"] += 1
            print(f"[爬虫 {fetcher_id}] 爬取: {url}")

        print(f"[爬虫 {fetcher_id}] 完成")

    async def cleaner(self, cleaner_id: int):
        """清洗层 — 数据清洗与转换"""
        while True:
            item = await self.raw_queue.get()
            try:
                if item is None:  # 毒丸信号
                    await self.clean_queue.put(None)
                    print(f"[清洗器 {cleaner_id}] 收到终止信号")
                    break

                # 清洗逻辑
                clean_title = item.title.strip()
                clean_content = re.sub(r"<[^>]+>", "", item.content)  # 去HTML标签
                clean_content = re.sub(r"\s+", " ", clean_content).strip()  # 去多余空格

                clean_item = CleanItem(
                    url=item.url,
                    title=clean_title,
                    content=clean_content,
                    word_count=len(clean_content),
                    cleaned_at=datetime.now().isoformat(),
                )

                await self.clean_queue.put(clean_item)
                self.stats["cleaned"] += 1
                print(f"[清洗器 {cleaner_id}] 清洗: {clean_title}")
            finally:
                self.raw_queue.task_done()

    async def writer(self, writer_id: int):
        """入库层 — 批量写入数据库"""
        batch: list[CleanItem] = []

        while True:
            # 使用 wait_for 避免无限等待
            try:
                item = await asyncio.wait_for(self.clean_queue.get(), timeout=2.0)
            except asyncio.TimeoutError:
                # 超时，先写入已积累的数据
                if batch:
                    await self._write_batch(batch)
                    batch.clear()
                continue

            if item is None:  # 毒丸信号
                self.clean_queue.task_done()
                break

            batch.append(item)
            self.clean_queue.task_done()

            # 达到批次大小，执行写入
            if len(batch) >= self.batch_size:
                await self._write_batch(batch)
                batch.clear()

        # 写入剩余数据
        if batch:
            await self._write_batch(batch)

        print(f"[写入器 {writer_id}] 完成")

    async def _write_batch(self, batch: list[CleanItem]):
        """批量写入数据库（模拟）"""
        # 实际项目中: await conn.executemany(INSERT_SQL, [(item.url, item.title, ...) for item in batch])
        await asyncio.sleep(0.1)  # 模拟数据库写入
        self.stats["written"] += len(batch)
        print(f"[写入] 批量写入 {len(batch)} 条，累计 {self.stats['written']} 条")

    async def run(self):
        """启动管道"""
        print("=" * 60)
        print("异步数据管道启动")
        print("=" * 60)

        # 启动所有协程
        fetchers = [asyncio.create_task(self.fetcher(i)) for i in range(self.num_fetchers)]
        cleaners = [asyncio.create_task(self.cleaner(i)) for i in range(self.num_cleaners)]
        writers = [asyncio.create_task(self.writer(i)) for i in range(self.num_writers)]

        # 等待爬取完成
        await asyncio.gather(*fetchers)
        print("\n--- 爬取阶段完成 ---\n")

        # 等待原始队列清空
        await self.raw_queue.join()

        # 发送毒丸给清洗器
        for _ in cleaners:
            await self.raw_queue.put(None)

        # 等待清洗器完成
        await asyncio.gather(*cleaners)
        print("\n--- 清洗阶段完成 ---\n")

        # 等待清洗队列清空
        await self.clean_queue.join()

        # 发送毒丸给写入器
        for _ in writers:
            await self.clean_queue.put(None)

        # 等待写入器完成
        await asyncio.gather(*writers)

        print("\n" + "=" * 60)
        print(f"管道完成! 统计: 爬取={self.stats['fetched']}, "
              f"清洗={self.stats['cleaned']}, 写入={self.stats['written']}")
        print("=" * 60)


asyncio.run(AsyncDataPipeline().run())

import asyncio
import time
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Any, Callable, Coroutine


class Priority(IntEnum):
    """任务优先级"""
    CRITICAL = 0  # 紧急
    HIGH = 1      # 高
    NORMAL = 2    # 普通
    LOW = 3       # 低


@dataclass(order=True)
class Task:
    """调度任务"""
    priority: int
    sort_index: float = field(compare=True)  # 用于同优先级内的排序
    task_id: str = field(compare=False)
    coro_factory: Callable[[], Coroutine] = field(compare=False)
    retries: int = field(compare=False, default=0)
    max_retries: int = field(compare=False, default=3)
    result: Any = field(compare=False, default=None)
    error: Exception | None = field(compare=False, default=None)


class TaskScheduler:
    """异步任务调度器 — 支持优先级、并发控制、重试"""

    def __init__(self, max_concurrency: int = 5):
        self.queue: asyncio.PriorityQueue[Task] = asyncio.PriorityQueue()
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self._running = False
        self._workers: list[asyncio.Task] = []
        self._counter = 0
        self._results: dict[str, Any] = {}

    def submit(
        self,
        coro_factory: Callable[[], Coroutine],
        priority: Priority = Priority.NORMAL,
        task_id: str | None = None,
        max_retries: int = 3,
    ):
        """提交任务"""
        if task_id is None:
            self._counter += 1
            task_id = f"task-{self._counter}"

        task = Task(
            priority=priority,
            sort_index=time.monotonic(),
            task_id=task_id,
            coro_factory=coro_factory,
            max_retries=max_retries,
        )
        self.queue.put_nowait(task)
        print(f"[调度器] 提交任务: {task_id}, 优先级: {priority.name}")
        return task_id

    async def _worker(self, worker_id: int):
        """工作协程"""
        while self._running:
            try:
                task = await asyncio.wait_for(self.queue.get(), timeout=1.0)
            except asyncio.TimeoutError:
                continue

            async with self.semaphore:
                for attempt in range(task.max_retries + 1):
                    try:
                        result = await task.coro_factory()
                        task.result = result
                        self._results[task.task_id] = result
                        print(f"[Worker {worker_id}] 完成: {task.task_id} = {result}")
                        break
                    except Exception as e:
                        task.error = e
                        if attempt < task.max_retries:
                            print(f"[Worker {worker_id}] 重试 ({attempt + 1}/{task.max_retries}): {task.task_id}")
                            await asyncio.sleep(2 ** attempt)  # 指数退避
                        else:
                            print(f"[Worker {worker_id}] 失败: {task.task_id}, 错误: {e}")

            self.queue.task_done()

    async def start(self, num_workers: int = 3):
        """启动调度器"""
        self._running = True
        self._workers = [
            asyncio.create_task(self._worker(i), name=f"worker-{i}")
            for i in range(num_workers)
        ]
        print(f"[调度器] 启动，{num_workers} 个 Worker")

    async def stop(self):
        """停止调度器"""
        self._running = False
        for w in self._workers:
            w.cancel()
        await asyncio.gather(*self._workers, return_exceptions=True)
        print(f"[调度器] 已停止，完成 {len(self._results)} 个任务")

    async def wait_all(self):
        """等待所有任务完成"""
        await self.queue.join()


# 使用示例
async def sample_task(name: str, duration: float = 0.5) -> str:
    """模拟任务"""
    await asyncio.sleep(duration)
    return f"{name} 完成"


async def failing_task():
    """模拟失败任务"""
    await asyncio.sleep(0.1)
    raise ValueError("故意失败")


async def main():
    scheduler = TaskScheduler(max_concurrency=5)
    await scheduler.start(num_workers=3)

    # 提交不同优先级的任务
    scheduler.submit(lambda: sample_task("紧急任务", 0.3), Priority.CRITICAL, "urgent-1")
    scheduler.submit(lambda: sample_task("普通任务1", 0.5), Priority.NORMAL, "normal-1")
    scheduler.submit(lambda: sample_task("普通任务2", 0.4), Priority.NORMAL, "normal-2")
    scheduler.submit(lambda: sample_task("低优任务", 0.6), Priority.LOW, "low-1")
    scheduler.submit(lambda: sample_task("高优任务", 0.2), Priority.HIGH, "high-1")
    scheduler.submit(failing_task, Priority.NORMAL, "fail-1", max_retries=2)

    # 等待所有任务完成
    await scheduler.wait_all()
    await scheduler.stop()

    print(f"\n结果: {scheduler._results}")


# asyncio.run(main())