和Python使用有关的一些教程，按类别分为不同文件

_intro.md

Python教程

Python是一个新手友好的语言，并且现在机器学习社区深度依赖于Python，C++, Cuda C, R等语言，使得Python的热度稳居第一。本Gist提供Python相关的一些教程，可以直接在Jupyter Notebook中运行。

1. 语言级教程，一般不涉及初级主题；
2. 标准库教程，最常见的标准库基本用法；
3. 第三方库教程，主要是常见的库如numpy，pytorch诸如此类，只涉及基本用法，不考虑新特性

其他内容就不往这个Gist里放了，注意Gist依旧由git进行版本控制，所以可以git clone 到本地，或者直接Google Colab\ Kaggle打开相应的ipynb文件

直接在网页浏览时，由于没有文件列表，可以按Ctrl + F来检索相应的目录，或者点击下面的超链接。

想要参与贡献的直接在评论区留言，有什么问题的也在评论区说 ^.^

目录-语言部分

• asyncio_tutorial.ipynb 协程
• c_extensions_tutorial.ipynb C语言扩展
• concurrency_tutorial.ipynb 并发
• context_managers_tutorial.ipynb 上下文管理器
• design_patterns_tutorial.ipynb 设计模式
• dunder_methods_tutorial.ipynb 内置方法（双下划线方法）
• generics_tutorial.ipynb 泛型和类型注解
• iterators_generators_tutorial.ipynb 生成器与迭代器
• metaprogramming_tutorial.ipynb 元编程
• pattern_matching_tutorial.ipynb 模式匹配
• python_internals_performance_tutorial.ipynb 性能分析
• snippets外链 个人日常开发中有用的一些代码片段

目录-库部分

• standard_library_tutorial.ipynb 标准库简易教程
• numpy_tutorial.ipynb 数学运算基石numpy
• pandas_tutorial.ipynb 结构化数据处理pandas
• matplotlib_tutorial.ipynb 绘图库matplotlib
• seaborn_tutorial.ipynb 绘图库seaborn

目录-具体业务库部分-本教程更多关注机器学习深度学习内容

• scikit_learn_tutorial.ipynb 机器学习库scikit_learn
• boosting_libs_tutorial.ipynb 梯度提升库XGBoost LightGBM CatBoost
• pytorch_tutorial.ipynb 深度学习库pytorch
• gymnasium_tutorial.ipynb 强化学习环境gymnasium
• stable_baselines3_tutorial.ipynb 强化学习算法实现SB3
• transformers_tutorial.ipynb 便利的预训练模型库Transformers
• opencv_tutorial.ipynb 经典的计算机视觉库opencv
• experiment_tracking_tutorial.ipynb 实验跟踪与可视化TensorBoard, WanDB, MLfow
• hpo_tutorial.ipynb 超参调优Optuna Raytune
• app_deploy_tutorial.ipynb 便捷的部署框架FastAPI Gradio Streamlit
• shap_tutorial.ipynb 模型可解释性SHAP
• langchain_tutorial.ipynb 构建完整AI应用LangChain
• llamaindex_tutorial.ipynb 另一个强大工具LlamaIndex
• faiss_tutorial.ipynb 向量数据库faiss
• MCP外链 MCP协议的使用

目录-附录

• sigh.md个人对于Python动态语言的看法

app_deploy_tutorial.ipynb

asyncio_tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Python 异步编程 (asyncio) 教程\n",
    "\n",
    "欢迎来到 Python `asyncio` 教程！本教程将引导你了解异步编程的核心概念，以及如何使用 `async` 和 `await` 关键字在 Python 中编写高效的并发代码，特别适用于 I/O 密集型任务。\n",
    "\n",
    "**什么是异步编程？**\n",
    "\n",
    "传统的同步编程中，当一个任务（例如，发出网络请求或读取文件）需要等待 I/O 操作完成时，整个程序会阻塞，直到该操作完成才能继续执行。这会导致资源浪费，尤其是在需要处理多个此类任务时。\n",
    "\n",
    "异步编程允许程序在等待一个 I/O 操作完成时，切换去执行其他任务，而不是空等。当等待的 I/O 操作完成后，程序可以回来继续处理它。\n",
    "\n",
    "**为什么使用 `asyncio`？**\n",
    "\n",
    "1.  **高并发**：`asyncio` 可以在单线程内处理成千上万的并发连接，非常适合网络服务器、爬虫、聊天应用等。\n",
    "2.  **I/O 密集型任务**：对于那些大部分时间花在等待网络、文件系统或其他外部资源响应的任务，`asyncio` 能显著提高效率。\n",
    "3.  **避免线程开销**：相比多线程，`asyncio` 使用协程，其上下文切换开销通常更小。\n",
    "\n",
    "**核心概念：**\n",
    "\n",
    "*   **事件循环 (Event Loop)**：`asyncio` 的核心，负责调度和执行协程。\n",
    "*   **协程 (Coroutine)**：使用 `async def` 定义的特殊函数。它们可以暂停执行并将控制权交还给事件循环，稍后再从暂停处恢复。\n",
    "*   **`await` 关键字**：用于暂停协程的执行，等待一个 `awaitable` 对象（通常是另一个协程或 Future）完成。\n",
    "*   **`async` 关键字**：用于声明一个函数为协程 (`async def`)，或用于 `async for` 和 `async with`。\n",
    "*   **Task**：一个被事件循环调度的协程。可以使用 `asyncio.create_task()` 创建。\n",
    "*   **Future**：一个代表异步操作最终结果的低级对象。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. 基础：`async def` 和 `await`\n",
    "\n",
    "让我们从最简单的协程开始。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time\n",
    "\n",
    "async def say_hello(delay: int, message: str):\n",
    "    \"\"\"一个简单的协程，会等待指定的秒数然后打印消息\"\"\"\n",
    "    print(f\"[{time.strftime('%X')}] Starting '{message}' (will wait {delay}s)\")\n",
    "    await asyncio.sleep(delay) # asyncio.sleep 是一个 awaitable，它会暂停协程，但不会阻塞事件循环\n",
    "    print(f\"[{time.strftime('%X')}] Finished '{message}' after {delay}s\")\n",
    "    return f\"Message from {message}: Done\"\n",
    "\n",
    "async def main_sequential():\n",
    "    \"\"\"顺序执行协程\"\"\"\n",
    "    print(f\"[{time.strftime('%X')}] --- Running main_sequential ---\")\n",
    "    result1 = await say_hello(2, \"Task 1\") # 等待 Task 1 完成\n",
    "    result2 = await say_hello(1, \"Task 2\") # 然后等待 Task 2 完成\n",
    "    print(f\"Result 1: {result1}\")\n",
    "    print(f\"Result 2: {result2}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_sequential ---\")\n",
    "\n",
    "# 运行主协程 (Python 3.7+)\n",
    "# asyncio.run() 会创建一个事件循环，运行传入的协程，然后关闭事件循环。\n",
    "asyncio.run(main_sequential())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在上面的 `main_sequential` 中，`await say_hello(...)` 会使 `main_sequential` 暂停，直到 `say_hello` 完成。这和普通的同步调用看起来类似，但关键在于 `asyncio.sleep()` 并不会阻塞整个 Python 进程，它只是将控制权交还给事件循环，允许事件循环处理其他任务（如果有的话）。\n",
    "\n",
    "注意 `say_hello` 函数是用 `async def` 定义的，这意味着它是一个协程函数。调用它会返回一个协程对象，这个对象本身不做任何事情，直到你用 `await` 它或者用 `asyncio.create_task()` 把它包装成一个任务。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. 并发执行协程：`asyncio.create_task()` 和 `asyncio.gather()`\n",
    "\n",
    "为了真正实现并发，我们需要同时启动多个协程，而不是一个接一个地等待。\n",
    "\n",
    "-   `asyncio.create_task(coro)`：将协程 `coro` 包装成一个 `Task` 对象，并安排其在事件循环中执行。它会立即返回 `Task` 对象，不会等待协程完成。\n",
    "-   `asyncio.gather(*aws)`：接收一个或多个 `awaitable` 对象（协程或 Task），并发运行它们，并等待所有对象完成。它会按输入顺序返回所有结果。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "async def main_concurrent_tasks():\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_concurrent_tasks ---\")\n",
    "    \n",
    "    # 创建任务，它们会立即开始在事件循环中被调度执行\n",
    "    task1 = asyncio.create_task(say_hello(2, \"Task A (created)\"))\n",
    "    task2 = asyncio.create_task(say_hello(1, \"Task B (created)\"))\n",
    "\n",
    "    print(f\"[{time.strftime('%X')}] Tasks created. Now awaiting task1...\")\n",
    "    result_a = await task1 # 等待 task1 完成\n",
    "    print(f\"[{time.strftime('%X')}] Task A finished. Now awaiting task2...\")\n",
    "    result_b = await task2 # 等待 task2 完成 (如果它在task1完成前就结束了，这里会立即返回)\n",
    "    \n",
    "    print(f\"Result A: {result_a}\")\n",
    "    print(f\"Result B: {result_b}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_concurrent_tasks ---\")\n",
    "\n",
    "asyncio.run(main_concurrent_tasks())\n",
    "\n",
    "print(\"\\n-------------------------------------\\n\")\n",
    "\n",
    "async def main_concurrent_gather():\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_concurrent_gather ---\")\n",
    "    \n",
    "    # 使用 asyncio.gather 并发运行协程\n",
    "    # gather 会自动将协程包装成 Task\n",
    "    results = await asyncio.gather(\n",
    "        say_hello(3, \"Task X (gather)\"),\n",
    "        say_hello(1, \"Task Y (gather)\"),\n",
    "        say_hello(2, \"Task Z (gather)\")\n",
    "    )\n",
    "    \n",
    "    print(f\"[{time.strftime('%X')}] All tasks in gather finished.\")\n",
    "    for i, res in enumerate(results):\n",
    "        print(f\"Result from gather task {i+1}: {res}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_concurrent_gather ---\")\n",
    "\n",
    "asyncio.run(main_concurrent_gather())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "观察 `main_concurrent_tasks` 的输出：\n",
    "*   Task A 和 Task B 几乎同时开始（因为 `create_task` 是非阻塞的）。\n",
    "*   Task B (等待1秒) 会在 Task A (等待2秒) 之前完成。\n",
    "*   尽管我们先 `await task1`，但事件循环在 `task1` 等待时会运行 `task2`。\n",
    "\n",
    "观察 `main_concurrent_gather` 的输出：\n",
    "*   所有任务 (X, Y, Z) 都被提交给 `gather` 并发执行。\n",
    "*   Task Y (1s) 最先完成，然后是 Task Z (2s)，最后是 Task X (3s)。\n",
    "*   `gather` 会等待所有任务都完成后才返回一个包含所有结果的列表，结果的顺序与传入 `gather` 的协程顺序一致。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. 实际应用：异步获取网页内容\n",
    "\n",
    "让我们用一个更实际的例子：使用 `aiohttp` 库异步获取多个网页的内容。\n",
    "首先，你需要安装 `aiohttp`：\n",
    "```bash\n",
    "pip install aiohttp\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import aiohttp\n",
    "\n",
    "async def fetch_url(session: aiohttp.ClientSession, url: str):\n",
    "    print(f\"[{time.strftime('%X')}] Fetching {url}...\")\n",
    "    try:\n",
    "        async with session.get(url) as response: # 异步上下文管理器\n",
    "            # response.raise_for_status() # 如果需要，检查HTTP错误\n",
    "            content = await response.text() # 异步读取响应体\n",
    "            print(f\"[{time.strftime('%X')}] Fetched {url}, length: {len(content)}\")\n",
    "            return url, len(content)\n",
    "    except Exception as e:\n",
    "        print(f\"[{time.strftime('%X')}] Error fetching {url}: {e}\")\n",
    "        return url, f\"Error: {e}\"\n",
    "\n",
    "async def main_fetch_urls():\n",
    "    urls = [\n",
    "        \"https://www.python.org\",\n",
    "        \"https://aiohttp.readthedocs.io\",\n",
    "        \"https://github.com\",\n",
    "        \"http://httpbin.org/delay/2\", # 这个URL会延迟2秒响应\n",
    "        \"https://example.com/nonexistent\" # 一个不存在的页面\n",
    "    ]\n",
    "    \n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_fetch_urls ---\")\n",
    "    # 创建一个 aiohttp.ClientSession，最好在 async with 语句中使用以确保正确关闭\n",
    "    async with aiohttp.ClientSession() as session:\n",
    "        tasks = []\n",
    "        for url in urls:\n",
    "            tasks.append(fetch_url(session, url)) # 注意：这里是协程对象，不是直接创建Task\n",
    "        \n",
    "        # asyncio.gather 会自动将协程包装成 Task 并并发执行\n",
    "        results = await asyncio.gather(*tasks, return_exceptions=False) # return_exceptions=True 会收集异常而不是立即抛出\n",
    "        \n",
    "    print(f\"\\n[{time.strftime('%X')}] All URLs fetched. Results:\")\n",
    "    for url, length_or_error in results:\n",
    "        print(f\"  {url}: {length_or_error}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_fetch_urls ---\")\n",
    "\n",
    "# 如果在Jupyter Notebook中运行，可能需要特殊处理事件循环\n",
    "# 但通常 asyncio.run() 应该可以工作\n",
    "asyncio.run(main_fetch_urls())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在这个例子中：\n",
    "*   `aiohttp.ClientSession()` 用于管理HTTP连接池。\n",
    "*   `async with session.get(url)` 是一个异步上下文管理器，确保请求完成后资源被正确释放。\n",
    "*   `await response.text()` 异步读取响应内容。\n",
    "*   所有 `fetch_url` 协程通过 `asyncio.gather` 并发执行，显著减少了总的等待时间。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. 错误处理\n",
    "\n",
    "在异步代码中，错误处理与同步代码类似，使用 `try...except` 块。\n",
    "\n",
    "当使用 `asyncio.gather` 时：\n",
    "*   默认情况下（`return_exceptions=False`），如果任何一个被 `gather` 的任务抛出异常，`gather` 会立即将这个异常传播出去，其他未完成的任务会被取消。\n",
    "*   如果设置 `return_exceptions=True`，`gather` 会等待所有任务完成（无论成功或失败），并将异常对象作为结果返回，而不是抛出它们。这允许你分别处理每个任务的结果/异常。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "async def faulty_coroutine(delay, should_fail=False):\n",
    "    print(f\"[{time.strftime('%X')}] Starting faulty_coroutine (delay={delay}, fail={should_fail})\")\n",
    "    await asyncio.sleep(delay)\n",
    "    if should_fail:\n",
    "        raise ValueError(f\"Intentional error from faulty_coroutine after {delay}s\")\n",
    "    result = f\"Success from faulty_coroutine after {delay}s\"\n",
    "    print(f\"[{time.strftime('%X')}] {result}\")\n",
    "    return result\n",
    "\n",
    "async def main_error_handling():\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_error_handling (default gather) ---\")\n",
    "    try:\n",
    "        results = await asyncio.gather(\n",
    "            faulty_coroutine(1, should_fail=False),\n",
    "            faulty_coroutine(2, should_fail=True), # 这个会失败\n",
    "            faulty_coroutine(3, should_fail=False) # 这个可能不会运行或被取消\n",
    "        )\n",
    "        print(\"Gather results (default):\", results)\n",
    "    except ValueError as e:\n",
    "        print(f\"[{time.strftime('%X')}] Gather caught an error: {e}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_error_handling (default gather) ---\")\n",
    "\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_error_handling (return_exceptions=True) ---\")\n",
    "    results_with_exceptions = await asyncio.gather(\n",
    "        faulty_coroutine(1, should_fail=False),\n",
    "        faulty_coroutine(2, should_fail=True),\n",
    "        faulty_coroutine(1.5, should_fail=False), # 确保这个有机会在错误发生前或后完成\n",
    "        return_exceptions=True\n",
    "    )\n",
    "    print(f\"[{time.strftime('%X')}] Gather results (return_exceptions=True):\")\n",
    "    for res in results_with_exceptions:\n",
    "        if isinstance(res, Exception):\n",
    "            print(f\"  Error: {res}\")\n",
    "        else:\n",
    "            print(f\"  Success: {res}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_error_handling (return_exceptions=True) ---\")\n",
    "\n",
    "asyncio.run(main_error_handling())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. 与阻塞代码交互: `loop.run_in_executor()`\n",
    "\n",
    "有时你需要在异步代码中运行无法轻易转换为异步的阻塞 I/O 或 CPU 密集型代码。直接在协程中调用它们会阻塞整个事件循环。\n",
    "\n",
    "`loop.run_in_executor(executor, func, *args)` 可以在一个单独的线程（或进程，如果 `executor` 是 `ProcessPoolExecutor`）中运行 `func(*args)`，并返回一个 `Future` 对象，你可以 `await` 它。\n",
    "\n",
    "默认情况下，`executor=None` 会使用事件循环的默认线程池执行器。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def blocking_io_operation(duration):\n",
    "    \"\"\"模拟一个阻塞的I/O操作\"\"\"\n",
    "    print(f\"[{time.strftime('%X')}] Enter blocking_io_operation (sleeps {duration}s)\")\n",
    "    time.sleep(duration) # 这是真正的 time.sleep()，会阻塞当前线程\n",
    "    result = f\"Blocking I/O completed after {duration}s\"\n",
    "    print(f\"[{time.strftime('%X')}] Exit blocking_io_operation with: {result}\")\n",
    "    return result\n",
    "\n",
    "async def main_with_blocking_code():\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_with_blocking_code ---\")\n",
    "    \n",
    "    loop = asyncio.get_running_loop() # 获取当前事件循环\n",
    "    \n",
    "    # 同时运行一个异步任务和一个在线程池中运行的阻塞任务\n",
    "    # 注意：不能直接 await blocking_io_operation(2)，因为它不是协程\n",
    "    blocking_task_future = loop.run_in_executor(None, blocking_io_operation, 2) # None uses default ThreadPoolExecutor\n",
    "    async_task = say_hello(1, \"Concurrent Async Task\")\n",
    "    \n",
    "    print(f\"[{time.strftime('%X')}] Tasks launched. Now gathering...\")\n",
    "    \n",
    "    # 等待两个任务完成\n",
    "    blocking_result, async_result = await asyncio.gather(\n",
    "        blocking_task_future,\n",
    "        async_task\n",
    "    )\n",
    "    \n",
    "    print(f\"\\n[{time.strftime('%X')}] Results:\")\n",
    "    print(f\"  Blocking task: {blocking_result}\")\n",
    "    print(f\"  Async task: {async_result}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_with_blocking_code ---\")\n",
    "\n",
    "asyncio.run(main_with_blocking_code())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "观察输出，你会看到 `Concurrent Async Task` (1s) 和 `blocking_io_operation` (2s) 是并发执行的，即使 `blocking_io_operation` 内部使用了阻塞的 `time.sleep()`。这是因为 `run_in_executor` 将其移到了另一个线程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. 异步上下文管理器 (`async with`)\n",
    "\n",
    "类似于同步代码中的 `with` 语句，`async with` 用于处理异步资源，确保它们在使用后被正确清理。\n",
    "一个异步上下文管理器需要实现 `__aenter__` 和 `__aexit__` 两个异步方法。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class AsyncResource:\n",
    "    def __init__(self, name):\n",
    "        self.name = name\n",
    "\n",
    "    async def __aenter__(self):\n",
    "        print(f\"[{time.strftime('%X')}] Entering context for {self.name}\")\n",
    "        await asyncio.sleep(0.5) # 模拟异步获取资源\n",
    "        print(f\"[{time.strftime('%X')}] Resource {self.name} acquired\")\n",
    "        return self # 返回的值会赋给 as 后面的变量\n",
    "\n",
    "    async def __aexit__(self, exc_type, exc_val, exc_tb):\n",
    "        print(f\"[{time.strftime('%X')}] Exiting context for {self.name}\")\n",
    "        await asyncio.sleep(0.5) # 模拟异步释放资源\n",
    "        print(f\"[{time.strftime('%X')}] Resource {self.name} released\")\n",
    "        if exc_type:\n",
    "            print(f\"  Exception occurred in {self.name}: {exc_val}\")\n",
    "        return False # 返回 False 表示异常会继续传播，True 表示异常被处理\n",
    "\n",
    "    async def use(self):\n",
    "        print(f\"[{time.strftime('%X')}] Using resource {self.name}\")\n",
    "        await asyncio.sleep(1)\n",
    "\n",
    "async def main_async_with():\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_async_with ---\")\n",
    "    async with AsyncResource(\"DB Connection\") as res:\n",
    "        await res.use()\n",
    "        # res.use() could raise an exception here, __aexit__ would still be called.\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_async_with ---\")\n",
    "\n",
    "asyncio.run(main_async_with())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. 异步迭代器 (`async for`)\n",
    "\n",
    "类似于同步迭代器，异步迭代器用于在 `async for` 循环中迭代异步生成的数据。\n",
    "一个异步可迭代对象需要实现 `__aiter__` 方法，该方法返回一个异步迭代器对象。\n",
    "异步迭代器对象需要实现 `__anext__` 异步方法，该方法返回下一个值，或者在没有更多值时引发 `StopAsyncIteration` 异常。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class AsyncCounter:\n",
    "    def __init__(self, limit):\n",
    "        self.limit = limit\n",
    "        self.current = 0\n",
    "\n",
    "    def __aiter__(self):\n",
    "        return self # 通常 __aiter__ 返回 self 如果类本身就是迭代器\n",
    "\n",
    "    async def __anext__(self):\n",
    "        if self.current < self.limit:\n",
    "            await asyncio.sleep(0.3) # 模拟异步获取数据\n",
    "            value = self.current\n",
    "            self.current += 1\n",
    "            return value\n",
    "        else:\n",
    "            raise StopAsyncIteration\n",
    "\n",
    "# 也可以使用异步生成器函数，更简洁\n",
    "async def async_generator_counter(limit):\n",
    "    for i in range(limit):\n",
    "        await asyncio.sleep(0.3)\n",
    "        yield i\n",
    "\n",
    "async def main_async_for():\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_async_for (class-based) ---\")\n",
    "    async for number in AsyncCounter(3):\n",
    "        print(f\"[{time.strftime('%X')}] Got number: {number}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_async_for (class-based) ---\")\n",
    "\n",
    "    print(f\"\\n[{time.strftime('%X')}] --- Running main_async_for (generator-based) ---\")\n",
    "    async for number in async_generator_counter(3):\n",
    "        print(f\"[{time.strftime('%X')}] Got number from generator: {number}\")\n",
    "    print(f\"[{time.strftime('%X')}] --- Finished main_async_for (generator-based) ---\")\n",
    "\n",
    "asyncio.run(main_async_for())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8. 关键点与最佳实践\n",
    "\n",
    "1.  **`await` 只能用在 `async def` 函数内部。**\n",
    "2.  **只能 `await` `awaitable` 对象**：这包括协程、Task、Future，以及实现了 `__await__` 方法的对象。\n",
    "3.  **不要阻塞事件循环**：避免在协程中直接调用长时间运行的同步阻塞代码 (如 `time.sleep()`, 大量 CPU 计算, 阻塞文件 I/O)。如果必须，请使用 `loop.run_in_executor()`。\n",
    "4.  **\"Async all the way\"**：如果一个函数执行异步操作，它应该是 `async def`。调用它的函数也应该是 `async def`，依此类推，直到程序的异步入口点 (通常是 `asyncio.run()` 中的主协程)。\n",
    "5.  **谨慎使用 `asyncio.create_task()` 而不 `await`**：如果你创建了一个任务但从不 `await` 它（或 `gather` 它），那么它的异常可能会被静默地忽略（取决于 Python 版本和配置），或者任务可能在程序退出前未完成。总是确保任务被适当地等待或管理。\n",
    "6.  **理解 `asyncio.gather()` 的行为**：特别是关于错误处理 (`return_exceptions`) 和结果顺序。\n",
    "7.  **使用异步库**：对于网络、数据库等操作，使用专门的异步库 (如 `aiohttp`, `httpx`, `asyncpg`, `aiomysql`)，它们提供了非阻塞的 API。\n",
    "\n",
    "## 总结\n",
    "\n",
    "`asyncio` 是 Python 中实现高并发 I/O 密集型应用的强大工具。通过理解协程、事件循环以及 `async`/`await` 语法，你可以编写出性能更高、响应更快的应用程序。\n",
    "\n",
    "这只是 `asyncio` 的入门。更高级的主题包括：\n",
    "*   同步原语 (`asyncio.Lock`, `asyncio.Semaphore`, `asyncio.Event`, `asyncio.Condition`)\n",
    "*   队列 (`asyncio.Queue`)\n",
    "*   子进程 (`asyncio.create_subprocess_exec`)\n",
    "*   流 (`asyncio.start_server`, `asyncio.open_connection`)\n",
    "\n",
    "希望本教程对你有所帮助！"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}

boosting_libs_tutorial.ipynb

c_extensions_tutorial.ipynb

concurrency_tutorial.ipynb

context_managers_tutorial.ipynb

design_patterns_tutorial.ipynb

dunder_methods_tutorial.ipynb

experiment_tracking_tutorial.ipynb

faiss_tutorial.ipynb

generics_tutorial.ipynb

gymnasium_tutorial.ipynb

hpo_tutorial.ipynb

iterators_generators_tutorial.ipynb

langchain_tutorial.ipynb

llamaindex_tutorial.ipynb

matplotlib_tutorial.ipynb

metaprogramming_tutorial.ipynb

numpy_tutorial.ipynb

opencv_tutorial.ipynb

pandas_tutorial.ipynb

pattern_matching_tutorial.ipynb

python_internals_performance_tutorial.ipynb

pytorch_tutorial.ipynb

scikit_learn_tutorial.ipynb

seaborn_tutorial.ipynb

shap_tutorial.ipynb

sigh.md

对动态语言Python的一些感慨

众所周知Python是完全动态的语言，体现在

1. 类型动态绑定
2. 运行时检查
3. 对象结构内容可动态修改（而不仅仅是值）
4. 反射
5. 一切皆对象（instance, class, method）
6. 可动态执行代码(eval, exec)
7. 鸭子类型支持

动态语言的约束更少，对使用者来说更易于入门，但相应的也会有代价就是运行时开销很大，和底层汇编执行逻辑完全解耦不知道代码到底是怎么执行的。

而且还有几点是我认为较为严重的缺陷。下面进行梳理。

破坏了OOP的语义

较为流行的编程语言大多支持OOP编程范式。即继承和多态。同样，Python在执行简单任务时候可以纯命令式(Imperative Programming)，也可以使用复杂的面向对象OOP。

但是，其动态特性破环了OOP的结构：

1. 类型模糊：任何类型实例，都可以在运行时添加或者删除属性或者方法（相比之下静态语言只能在运行时修改它们的值）。经此修改的实例，按理说不再属于原来的类型，毕竟和原类型已经有了明显的区别。但是该实例的内建__class__属性依旧会指向原类型，这会给类型的认知造成困惑。符合一个class不应该只是名义上符合，而是内容上也应该符合。
2. 破坏继承：体现在以下两个方面
   1. 大部分实践没有虚接口继承。abc模块提供了虚接口的基类ABC，经典的做法是让自己的抽象类继承自ABC，然后具体类继承自自己的抽象类，然后去实现抽象方法。但PEP提案认为Pythonic的做法是用typing.Protocol来取代ABC，具体类完全不继承任何虚类，只要实现相应的方法，那么就可以被静态检查器认为是符合Protocol的。
   2. 不需要继承自具体父类。和上一条一样，即使一个类没有任何父类（除了object类），它依旧可以生成同名的方法，以实现和父类方法相同的调用接口。这样在语义逻辑上，类的定义完全看不出和其他类有何种关系。完全可以是一种松散的组织结构，任何两个类之间都没继承关系。
3. 破坏多态：任何一个入参出参，天然不限制类型。这使得要求父类型的参数处，传入子类型显得没有意义，依旧是因为任何类型都能动态修改满足要求。

破坏了设计模式

经典的模式诸如工厂模式，抽象工厂，访问者模式，都严重依赖于继承和多态的性质。但是在python的设计中，其动态能力使得设计模式形同虚设。 大家常见的库中使用设计模式的有transformers库，其中的from_pretrained系列则是工厂模式，通过字符串名称确定了具体的构造器得到具体的子类。而工厂构造器的输出类型是一个所有模型的基类。

安全性问题

Python在代码层面一般不直接管理指针，所以指针越界，野指针，悬空指针等问题一般不存在。而gc机制也能自动处理垃圾回收使得编码过程不必关注这类安全性问题。但与之相对的，Python也有自己的安全性问题。以往非托管形式的代码的攻击难度较大，注入代码想要稳定执行需要避免破坏原来的结构导致程序直接崩溃（段错误）。 Python却可以直接注入任何代码修改原本的逻辑，并且由于不是在code段固定的内容，攻击时候也无需有额外考虑。运行时可以手动修改globals() locals()内容，亦有一定风险。 另一个危险则是类型不匹配导致的代码执行问题，因为只有在运行时才确定类型，无法提前做出保证，可能会产生类型错误的异常，造成程序崩溃。

总结

我出身于C++。但是近年来一直在用python编程。而且python的市场占有率已经多年第一，且遥遥领先。这和其灵活性分不开关系。对于一个面向大众的编程语言，这样的特性是必要的。即使以上说了诸多python的不严谨之处，但是对于程序员依旧可以选择严谨的面向对象写法。所以，程序的优劣不在于语言怎么样，而在于程序员本身。程序员有责任写出易于维护，清晰，规范的代码~

stable_baselines3_tutorial.ipynb

standard_library_tutorial.ipynb

transformers_tutorial.ipynb

返回顶部

有见解，有问题，或者单纯想盖楼灌水，都可以在这里发表！

因为文档比较多，有时候渲染不出来ipynb是浏览器性能的问题，刷新即可

或者git clone到本地来阅读