Database

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 这样可以保证数据正确写入，但可能覆盖其他人的数据修改 ...

多对多类型，何其名称暗示的一样，当无法认定任何一方为 “一” 的时候使用。 Many-To-Many Relationships 在标准的 one-to-many 一对多关系中，“多” 的一方会有一个 foreign key 外键指向 “一” 的一方。 但是在尝试建立 Product 和 Country 之前的关系时，在产品中添加指向国家的外键不行，因为这样一个产品就只能来自一个国家。 在国家中添加外键指向产品也不行，这样一个国家就只能对应一种产品。 How Many-To-Many Relationships Work 为了实现多对多关系，需要两个一对多关系来表示这种复杂的关系。 由于无法在两个表之间建立直接的多对多关系，因此需要添加一个称为连接表 join table 的第三个表。 每一边都和 join table 维护一个一对多关系，这意味着连接表有两个外键，指向两个表。 例如：products 表的 id 和 products_countries 的 products_id 形成 1:N 关系，countries 表的 id 和 cuontry_id 形成 1:N 关系。 这里的 products_countries 表就是 join table 连接表，常见的命名规范就是使用构成关系的两个实体的名称。 如果一个产品在 3 个国家生产，那么 join table 连接表就会有 3 对实例。 从另一个方面来看，对于一个已经创建了 7 个产品的国家，将会有对应数量的条目。 其 country_id 被设置为该国家，每个条目都将其与其中一个产品关联起来。 ...

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。 ...

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() 函数有以下配置参数： ...

psql 介绍 psql 是 PostgreSQL 中的一个交互式命令行工具，类似 Oracle 的 sqlplus，它允许用户交互式输入 SQL 语句或命令，然后发送给 PostgreSQL 服务器，再显示结果。 此外，psql 还有大量类似 shell 的特性来编写脚本，实现自动化操作。 在安装 postgresql 的时候，会创建一个与初始化数据库时的操作系统同名的数据库用户，这个用户是这个数据库的超级用户。 因此，在 OS 用户下登陆数据库时，会执行操作系统认证，因此无需用户名和密码，也可以通过修改 pg_hba.conf 文件来要求用户输入密码。 psql 也支持使用命令行参数来查询信息和执行 SQL，这种非交互模式与 linux 命令就没有太大区别了。 例如使用 psql -l 查看数据库，当然也可以通过命令进入数据库后，输出 \l 命令查看有哪些数据库。 默认会有一个叫 postgres 的数据库，这是默认安装后就有的一个数据库。 还有两个模板数据库，template0 和 template1。 当用户创建创建数据库时，是从模板数据库 template1 克隆来的，因此通常可以订制 template1 中的内容，例如添加一些表和函数，这样后续创建的数据库也会有这些表和函数。 template0 是一个最简化的数据库，如果指明从此数据库克隆，将创建出一个最简化的数据库。 使用 \d 查看有哪些表，使用 \c 连接到某数据库。连接命令格式如下： psql -h <hostname or ip> -p <port> [数据库名称] [用户名称] 也可以通过环境变量指定 export PGDATABASE=db_name export PGHOST=ip export PGPORT=port export PGUSER=postgres psql 常用命令 \h 命令查询 SQL 语法 ...

SQL 入门 SQL (Structured Query Language) 是关系数据库最重要的操作语言，且影响已经超出了数据库领域，这篇文章介绍基础部分。 语句分类 SQL 命令一般分为 DQL、DML、DDL 三类： DQL: Data Query Language 数据查询语句，基本就是 SELECT 查询命令，用于数据查询。 DML: Data Manipulation Language 数据操纵语言，主要用于插入、更新、删除数据，即 INSERT、UPDATE、DELETE 三类。 DDL: Data Definition Language 数据定义语言，用于创建、删除、修改表、索引等数据库对象的语言。 词法结构 每次执行的 SQL 语句可以由多条 SQL 命令组成，多条 SQL 语句命令之间由分号 (;) 分隔。 SQL 命令由一系列记号组成，这些记号可以由关键字、标识符、双引号包围的标识符、常量和单引号包围的常量组成。 SQL 命令中可以有注释，这些注释在 PostgreSQL 中等同于空白。 例如下面这些 SQL 命令 SELECT * FROM OSDBA_TABLE01; UPDATE OSDBA_TABLE SET COL1 = 64; INSERT INTO OSDBA_TABLE VALUES (232, 'hello osdba'); DDL 语句 使用 psql 会默认连接到和用户名一样的数据库中，也可以使用命令 psql postgres 来连接到默认数据库，或者 creatdb db_name 创建新数据库。 ...

PostgreSQL 的安装与配置 安装 brew 安装后 brew services start postgresql@17 可以启动服务，使用 brew services list 查看服务状态。 也可以通过设定环境变量 export PGDATA=/opt/homebrew/var/postgresql@17 后使用命令 initdb 创建数据库簇，然后 pg_ctl start -D $PGDATA 启动。 停止命令为 pg_ctl stop -D $PGDATA [-m SHOWDOWN-MODE]，其中 -m 是服务器停止方法： smart: 等所有连接终止后，关闭数据库。如果客户端连不上，则无法关闭数据库。 fast: 快速关闭数据库，断开客户端连接，让已有的事务回滚，然后关闭数据库。相当于 Oracle 关闭时的 immediate 模式。 immediate: 立即关闭数据库，相当于数据库进程立即停止，直接退出，下次启动数据库需要恢复。相当于 Oracle 关闭时的 abort 模式。 PostgreSQL 数据库中的 immediate 关机模式相当于 Oracle 数据库中的 abort 关机模式，而 Oracle 中的 immediate 关机模式实际上对应的是 PostgreSQL 中的 fast 模式。 配置 PostgreSQL 数据库的配置主要通过修改数据目录下的 postgresql.conf 和 pg_hba.conf 文件来实现的。 ...