Welcome to Starslayerx’ Blog 👋

Engines, Metadata and Sessions 下面是一个异步的数据库链接示例： from sqlalchemy import MetaData from sqlalchemy.orm import DeclarativeBase from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine, async_sessionmaker class Model(DeclarativeBase): metadata = MetaData(naming_convention={ 'ix': 'ix_%(column_0_label)s', 'uq': 'uq_%(table_name)s_%(column_0_name)s', 'ck': 'ck_%(table_name)s_%(constraint_name)s', 'fk': 'fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s', 'pk': 'pk_%(table_name)s', }) pg_dsn: str = 'postgresql+psycopg://postgresql:postgresql@localhost:5432/postgresql' engine = create_async_engine(pg_dsn) session_factory = async_sessionmaker( bind=engine, expire_on_commit=False, # !!! prevent implicit synchronous refresh ) 可以看到与同步代码其实十分类似，一个重要的不同点是 session 的 expire_on_commit 参数。 这会禁止 SQLALchemy 的默认行为：在会话 session 提交后将模型 model 标记为过期。 标记为过期的模型再次访问其任何属性时，会隐式地从数据库查询中刷新。 由于隐式的 implicit 数据库活动不能出现在任何异步应用中，因此不应该使用过期对象。 expire_on_commit=False 选项确保在提交后，不会将任何模型标记为过期。 在异步高并发环境下，仅靠 expire_on_commit 可以保证程序不报错，但是不能保证数据的一致性。 在 long-lived session 中可能会有陈旧模型 stale model，下面是一些解决方法： 手动清理模型 expunge 或显示刷新 refresh，但是会多产生一次网络请求 # refresh example async def create_new_post(db: AsyncSession, title: str, content: str): new_post = Post(title=title, content=content) db.add(new_post) await db.commit() # 提交到数据库 await db.refresh() # 刷新：强制从数据库拉取最新的一行数据 print(f'PostID: {new_post.id}, CreateTime: {new_post.created_at}') return new_post # expunge example async def export_all_posts_title_to_csv(db: AsyncSession): result = await db.execute(select(Post)) posts = result.scalars().all() titles = [] for post in posts: titles.append(post.title) db.expunge(post) # 踢出 Session return titles 这样可以保证数据正确写入，但可能覆盖其他人的数据修改 ...

One-To-Many Relationships 在之前的文章介绍了如何产品表的，有趣的事，有些查询是为了查找产品制造商，而不是产品本身。 这里会介绍如何分组去重。 很多时候，分组都十分有用，尤其是使用聚合函数 aggregation functions 计算为存储在数据库中的分组信息。 当次要数据存储在与主要实体相同的表中时，会出现重复的问题。 因为表可能会变得比实际需要的大得多，而重复数据中的拼写差异可能导致分组结果错误。 One-To-Many Relationships Implementation 关系数据库通过关系 relationships 创建链接，有两种主要的关系： One-to-many: 一对多 Many-to-many: 多对多 一对多是说，有表 A 和 B，A 中的项可以对应 B 中的任意多项，而 B 中的项最多只能对应 A 中的一项。 该模式与电脑制造商和其产品的关系相同，制造商可以制造多种型号产品，而每种产品只能有一个制造商。 这里制造商就是一 “one”，产品就是多 “many”。 定义关系的第一步是先为这两个实体创建数据库表。 因此要定义 products 表和 manufacturer 表。 class Manufacturer(Model): __tablename__ = 'manufacturers' id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(64), index=True, unique=True) def __repr__(self): return f'Manufacturer({self.id}, "{self.name}")' 原始的 Product 表中的 manufacturer 项被移除，并使用 manufacturer_id 替代： from sqlalchemy import ForeignKey class Product(Model): __tablename__ = 'products' id: Mapped[int] = mapped_column(primary_key) name: Mapped[str] = mapped_column(String(64), index=True, unique=True) manufacturer_id: Mapped[int] = mapped_column(ForeignKey('manufacturers.id'), index=True) year: Mapped[int] = mapped_column(index=True) country: Mapped[str] = mapped_column(String(32)) cpu: Mapped[Optional[str]] = mapped_column(String(32)) def __repr__(self): return f'Product({self.id}, "{self.name}")' 引用其他表的主键的列，被称为外键 foreign key，并会给予一个外键约束。 一个好的命名惯例是使用被引用的表，并添加后缀 _id。 ...

这篇文章将通过一个**“为阿里云 Qwen API 添加缓存”**的例子，带你彻底理解 Python 装饰器。 首先，我们定义一个基础的调用函数： import time def call_qwen(messages: list, model: str = 'qwen-max', temperature: float = 0.7): '''调用 Qwen API (模拟)''' print(f'正在请求 API (模型: {model})...') time.sleep(1) # 模拟网络耗时 return {'content': '这是 AI 的回复', 'usage': 100} 1. 函数是一等公民 (First-Class Citizen) 在 Python 中，函数可以像变量一样被传递和赋值。 # 1. 赋值给变量 run_api = call_qwen # 2. 作为参数传递 def logger(func, *args, **kwargs): print('[INFO] Calling qwen ...') return func(*args, **kwargs) logger(call_qwen, [{'role': 'user', 'content': '你好'}]) 2. 闭包 (Closure) 闭包是指函数内部定义了另一个函数，并且内部函数引用了外部函数的变量。它是实现装饰器的基石。 利用闭包，我们可以创建一个带缓存功能的函数： def make_cached_qwen(): cache = {} # 外部函数的变量，会被内部函数“捕获” def wrapped(messages, **kwargs): # 简单起见，用最后一条消息的内容当 Key key = messages[-1]['content'] if key in cache: print(' 命中缓存') return cache[key] result = call_qwen(messages, **kwargs) cache[key] = result return result return wrapped # 此时 cached_call 就是一个带有自己 “私有缓存字典” 的函数 cached_call = make_cached_qwen() 3. 高阶函数 (High-Order Function) 如果我们想让缓存逻辑通用化，不只针对 call_qwen，我们可以写一个接收函数作为参数的高阶函数： ...

SQLALchemy Core and SQLALchemy ORM SQLALchemy 分为两个模块：Core 和 ORM (Object-Relational Mapping)。 Core 模块包含对所有受支持数据库方言的集成逻辑，一组用于描述数据库表的类，用于 Python 语言生成 SQL 语句。 ORM 模块在 Python 应用程序中引入了一层抽象，使得许多数据库操作可以根据对 Python 对象执行的操作自动推导出来。 Database Engine SQLALchemy 使用 engine 对象来管理数据库连接，包含 Core 和 ORM 应用。 create_engine() 函数通过数据库 url 创建一个 engine。 格式为： {dialet}{+driver}://{username}:{password}@{hostname}:{port}/{database} 其中 SQLite 比较特殊，无需 driver。 对于导入 DATABASE_URL 有两种方式，一种是使用 load_dotenv() 加载 .env 文件的环境变量 import os from dotenv import load_dotenv from sqlalchemy import create_engine load_dotenv() engine = create_engine(os.environ['DATABASE_URL']) 另一种是通过 Pydantic BaseSettings import os from pathlib import Path from pydantic_settings import BaseSettings from pydantic impot SecretStr from functools import lru_cache from sqlalchemy import create_engine class Settings(BaseSettings): POSTGRESQL_HOST: str POSTGRESQL_PORT: int POSTGRESQL_USER: str POSTGRESQL_PASSWORD: SecretStr POSTGRESQL_DB: str @property def postgres_db_url(self) -> str: # 通常无需显示地写 driver，只有更换 dirver 时才需要写 return f'postgresql://{self.POSTGRESQL_USER}:{self.POSTGRESQL_PASSWORD.get_secret_value()}@{self.POSTGRESQL_HOST}:{self.POSTGRESQL_PORT}/{self.POSTGRESQL_DB}' class Config: env_file = str(Path(__file__).parent / '.env') case_sensitive = True extra = 'ignore' # 忽略额外字段 @lru_cache def get_settings() -> Settings: return Settings() # 使用示例 settings = get_settings() print('Host:', settings.POSTGRESQL_HOST) print('Password:', settings.PASSWORD.get_secret_value()) engine = create_engine(settings.postgres_db_url) create_engine() 函数有以下配置参数： ...

Python 程序由 modules 和 packages 组成，使用 import 语句导入。 Modules and the import Statement 任何 Python 源文件都可以作为一个模块导入，例如下面 module.py 代码： a = 37 def func(): print(f'func says that a is {a}') class SomeClass: def method(self): print('method says hi') print('loaded module') 改文件包含一些常见的编程元素，包括一个全局变量、一个函数、一个类定义 和 最后的语句。 通过下面方法导入： import module module.a module.func() s = module.SomeClass() s.method() 执行 import 会发送下面这几件事： 加载模块源码，如果找不到抛出 ImportError 创建新模块对象。该对象作为模块内所有全局定义 global defintions 的容器，被称为 “命名空间” namespace 该模块源码在新创建的模块命名空间内执行 如果没有错误发生，调用者会创建一个名称，指向新的模块对象。该名称与模块名称一致，但不包含任何文件后缀。 这些步骤中，第一步是最复杂的。新手容易犯的错误就是使用错误的名称或将代码放到了未知的位置。 且模块文件必须放在 sys.path 所包含的文件路径中，且文件名称要遵循和 python 变量一样的规则。 剩下的步骤都隔离在一个模块中，因此不用担心不同模块间命名冲突的问题。 Python import 会执行所有导入的源码，因此导入上面模块会输出 loaded module。 ...

使用 vars() 方法返回对象的属性和值的字典对象，不带参数时，返回当前局部作用域的变量字典 vars() # 相当于 locals() class Person: name: str age: int def __init__(self, name, age): self.name = name self.age = age self.city = "Beijing" p = Person("Alice", 25) print(vars(p)) Attribute Access 一个示例只有三种基础的方法：getting, setting 和 deleting 属性 class Attribute: owner: str blance: float def __init__(self, owner: str, balance: float): self.owner = owner self.balance = balance def __repr__(self): return f"Account({self.owner!r}, {self.balance!r})" def deposite(self, amount: float): self.balance += amount def withdraw(self, amount: float): self.balance -= amount def inquiry(self) -> float: return self.balance 例如 a = Account("Guido", 1000.0) a.owner # get a.balance = 75 # set del a.balance # delete Python 中的一切都是一个动态过程，几乎没有什么限制。 例如，可以给已创建的对象添加新属性： a = Account("Guido", 1000.0) a.creation_date = "2019-02-14" a.nickname = "Fromer BDFL" 有时候不适用点 . 操作符来执行任务，而是通过将属性名传递给 getattr(), setattr() 和 delattr() 函数来实现。 hasattr() 函数允许你测试一个已存在的属性： a = Account("Guido", 1000.0) getattr(a, "owner") setattr(a, "balance", 750.0) delattr(a, "balance") hasattr(a, "balance") # False getattr(a, "withdraw")(100) # Method Call # a = Account("Guido", 650.0) getattr() 函数可以携带一个默认值，如果想要查看一个可能不存在的属性，可以这样实现： ...

CS 144: Introduction to Computer Networking, Fall 2025 本篇文章对应 check0.pdf 的内容 3 Network by hand 3.1 Fetch a Web page 这个介绍怎么发送一个 GET 请求 telent cs144.keithw.org http 该命令告诉 telnet 打开一个可靠字节流(reliable byte stream)，并在我电脑上运行一个 http 服务 预计会收到这样的内容 Trying 104.196.238.229... Connected to cs144.keithw.org. Escape character is '^]'. ^] telnet> close Connection closed. 该命令不能使用常见的 Ctrl-C 之类方法退出，而是 Ctrl-] 然后输入 close GET /hello HTTP/1.1 该命令告诉服务器 URL 的 path 部分 Host: cs144.keithw.org 该命令告诉服务器 URL 的 host 部分 Connection: close ...

生成器是 Python 中一种强大的特性，其通常被介绍为一种定义新型迭代模式的便捷方式。 但生成器从根本上改变了整个函数执行的模式，本篇文章重点关注：生成器、生成器委托、基于生成器的协程，以及生成器的其他内部机制。 Generators and yield 如果一个函数使用 yield 关键字，这定义了一个生成器。 生成器的主要用户是生成用于迭代的值。 例如： def countdown(n): print("Counting down from", n) while n > 0: yield n n -= 1 # Example use for x in countdown(n): print("T-minus", x) 如果调用该函数则不会开始执行： c = countdown(10) # <generator object countdown at 0x106faa260> 相反，会创建一个生成器对象。 该生成器对象只有在你迭代它的时候才会开始执行，使用的一种方式是调用 next()。 例如： next(c) # Counting down from 10 # 10 next(c) # 9 当调用 next() 时，生成器函数会执行语句直到遇到 yield 语句。 yield 语句会返回一个结果，此时函数的执行被挂起，直到再次调用 next()。 当其暂停的时候，函数会保留所有的本地变量和执行环境。 恢复执行时，程序会从 yield 之后的语句继续运行。 next() 是调用生成器上 __next__() 方法的简写形式。 例如，你可以这样： c.__next__() # 8 c.__next__() # 7 通常不会在生成器直接使用 next()，而是使用 for 或其他一些语句： ...

Taskwarrior 3.x 同步服务器配置指南 本文介绍如何使用 Docker 部署 taskchampion-sync-server，为 Taskwarrior 3.x 提供跨设备同步功能。 背景知识 Taskwarrior 是一款强大的命令行任务管理工具。从 3.0 版本开始，官方不再支持 taskd 服务器，改用新的 taskchampion-sync-server。 与 taskd 相比，新同步服务器的优势： 无需手动配置 SSL 证书 无需预先创建用户账户 客户端数据端到端加密 部署和维护更简单 服务器端配置 使用 Docker 部署 创建数据目录并启动容器： sudo mkdir -p /var/lib/taskchampion-sync-server sudo chmod 777 /var/lib/taskchampion-sync-server docker run -d \ --name taskchampion \ -p 53589:8080 \ -e RUST_LOG=info \ -v taskchampion-data:/var/lib/taskchampion-sync-server \ --restart unless-stopped \ ghcr.io/gothenburgbitfactory/taskchampion-sync-server:main 端口说明：容器内部使用 8080，映射到宿主机的 53589（可自定义）。 配置防火墙 确保服务器防火墙开放相应端口。以常见的云服务器防火墙为例： 类型：入站 行动：允许 协议：TCP 目的端口：53589 验证服务运行 docker ps | grep taskchampion docker logs taskchampion 正常运行时应看到： ...

PEP 683 改变了 Python 原有引用计数的一些逻辑，下面简单介绍一下。 CPython 的“引用计数可变性”已经成为并发、性能和未来发展的系统性障碍。 引用对象导致 “逻辑不可变对象” ≠ “物理不可变对象” 在 Cpython 中 None True/False int, str, list 等内建对象 在运行时引用计数会频繁变动，这意味着内存内容在不断被写入，在底层并非真正的 immutable 引用计数写操作降低并发性能 CPU Cache Line 失效 Py_INCREF / Py_DECREF 会写内存 -> cache line invalidation 在多线程 / 多核环境中，同一个全局对象被频繁引用，会造成严重的缓存抖动 fork + Copy-on-Write 失效 父子进程共享内存页 只要引用计数一变 -> 页面被写 -> 触发 COW 只是“多拿了个引用”，却导致整页内存复制 为 free-threading (no GIL) 清扫道路 CPython 的引用计数本质是全局共享的可变状态，在无 GIL 下会产生高频数据竞争。 要么给 refcount 加锁（性能太差），要么让一部分的 refcount 不再变化。 该提案将“对象生命周期模型”划分成了两类对象 对象类型 生命周期 refcount 行为 普通对象 动态 正常增减 不朽对象 解释器级 固定，不参与 gc 这让后续优化和推理都更清晰，也会导致 sys.getrefcount() 不再具有语义价值，测试默认返回 2 的 23 次方减 1。