Python数据库编程实战从psycopg3到SQLAlchemy Core — PostgreSQL篇华为云 FlexusX (8vCPU/16GiB) · Ubuntu 24.04.4 · Python 3.12.3 · PostgreSQL 16.14psycopg 3.3.4 · SQLAlchemy 2.0.51 · 全真实输出 · 零理论空谈目录环境架构第一章psycopg3 — PostgreSQL原生驱动1.1 环境搭建与连接1.2 DDL建表与DML写入1.3 SELECT查询与游标1.4 事务控制COMMIT/ROLLBACK1.5 参数化查询与SQL注入防护第二章SQLAlchemy Core — 数据库抽象层2.1 引擎与元数据2.2 Core层CRUD操作第三章Supabase云数据库psycopg3 vs SQLAlchemy 终极对比踩坑记录环境架构┌──────────────────────────────────────────────────────┐ │ ecs-88e7-0001 │ │ 139.9.128.210 (华为云香港) │ │ │ │ ┌─────────────┐ ┌─────────────────────────────┐ │ │ │ Python 3.12 │────▶│ PostgreSQL 16.14 │ │ │ │ │ │ localhost:5432 │ │ │ │ psycopg 3.3 │ │ DB: python_db │ │ │ │ SQLAlchemy │ │ User: python_user │ │ │ │ 2.0.51 │ │ │ │ │ └─────────────┘ └─────────────────────────────┘ │ └──────────────────────────────────────────────────────┘数据库准备执行过的命令apt-getinstall-ypostgresql postgresql-clientsu- postgres-cpsql -c\CREATE DATABASE python_db;\su- postgres-cpsql -c\CREATE USER python_user WITH PASSWORD Python123;\su- postgres-cpsql -c\GRANT ALL PRIVILEGES ON DATABASE python_db TO python_user;\su- postgres-cpsql -d python_db -c GRANT ALL ON SCHEMA public TO python_user;pip3installpsycopg[binary]sqlalchemy --break-system-packages第一章psycopg3 — PostgreSQL原生驱动1.1 环境搭建与连接psycopg3 是 PostgreSQL 官方推荐的 Python 驱动psycopg2的继任者3.x 版本完全重写支持 async/await。importpsycopgimportsysprint(fpsycopg 版本:{psycopg.__version__})# 3.3.4print(fPython 版本:{sys.version.split()[0]})# 3.12.3# 连接字符串 (conninfo)DBhostlocalhost dbnamepython_db userpython_user passwordPython123connpsycopg.connect(DB)真实输出连接成功! 数据库: python_db 用户: python_user 主机: localhost 端口: 5432 服务端版本: 160014 事务状态: 0 (空闲)连接字符串格式conninfo逐字段解释参数含义示例host数据库主机地址localhost/192.168.0.1dbname数据库名python_dbuser连接用户python_userpassword连接密码Python123port端口默认54325432也可以通过psycopg.connect(host..., dbname..., user..., password...)关键字传参。1.2 DDL建表与DML写入withconn.cursor()ascur:# DDL: CREATE TABLEcur.execute( CREATE TABLE IF NOT EXISTS books ( id SERIAL PRIMARY KEY, title VARCHAR(200) NOT NULL, author VARCHAR(100) NOT NULL, price NUMERIC(10, 2) DEFAULT 0, pages INTEGER, publisher VARCHAR(100), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) )print(CREATE TABLE books — 成功)# DML: INSERTwithconn.cursor()ascur:cur.execute(DELETE FROM books)# 清空, 避免重复books_data[(Python编程:从入门到实践,Eric Matthes,89.00,459,人民邮电出版社),(流畅的Python,Luciano Ramalho,139.00,700,人民邮电出版社),(利用Python进行数据分析,Wes McKinney,109.00,502,机械工业出版社),(Python Cookbook,David Beazley,129.00,687,人民邮电出版社),(Effective Python,Brett Slatkin,99.00,256,机械工业出版社),]forbookinbooks_data:cur.execute(INSERT INTO books (title, author, price, pages, publisher) VALUES (%s, %s, %s, %s, %s),book)print(fINSERT INTO books — 成功插入{len(books_data)}条记录)conn.commit()# 显式提交事务!关键知识点SERIAL PostgreSQL 自增整数等价INTEGER GENERATED ALWAYS AS IDENTITYNUMERIC(10, 2) 定点小数共10位小数2位适合金额占位符使用%s不是?也不是%d也不是:nameconn.commit()显式提交否则自动回滚1.3 SELECT查询与游标withconn.cursor()ascur:# 计数cur.execute(SELECT count(*) FROM books)countcur.fetchone()[0]print(f总记录数:{count})# 全表查询cur.execute(SELECT id, title, author, price, pages FROM books ORDER BY id)rowscur.fetchall()print(f{ID:4}{书名:30}{作者:22}{价格:8}{页数:6})print(-*75)forrowinrows:print(f{row[0]:4}{row[1]:30}{row[2]:22}{row[3]:7.2f}{row[4]:6})真实输出总记录数: 5 ID 书名 作者 价格 页数 --------------------------------------------------------------------------- 1 Python编程:从入门到实践 Eric Matthes 89.00 459 2 流畅的Python Luciano Ramalho 139.00 700 3 利用Python进行数据分析 Wes McKinney 109.00 502 4 Python Cookbook David Beazley 129.00 687 5 Effective Python Brett Slatkin 99.00 256聚合查询cur.execute(SELECT min(price), max(price), avg(price), sum(price) FROM books)mn,mx,avg,totalcur.fetchone()print(fMIN{mn:.2f}MAX{mx:.2f}AVG{avg:.2f}SUM{total:.2f})cur.execute( SELECT publisher, count(*), avg(price) FROM books GROUP BY publisher ORDER BY count(*) DESC )真实输出聚合: MIN89.00 MAX139.00 AVG113.00 SUM565.00 按出版社分组: 出版社 数量 均价 ------------------------------------ 人民邮电出版社 3 119.00 机械工业出版社 2 104.00游标方法速查方法含义返回类型fetchone()取下一行tuple或Nonefetchall()取全部剩余行list[tuple]fetchmany(n)取下n行list[tuple]rowcount受影响行数DMLint批量插入executemanynew_books[(重构:改善既有代码的设计,Martin Fowler,99.00,428,人民邮电出版社),(代码整洁之道,Robert C. Martin,59.00,422,人民邮电出版社),]cur.executemany(INSERT INTO books (title, author, price, pages, publisher) VALUES (%s, %s, %s, %s, %s),new_books)# executemany — 批量插入 2 条1.4 事务控制COMMIT/ROLLBACKPostgreSQL 默认每条 SQL 在事务中执行。调用conn.commit()确认conn.rollback()撤销。# 插入但不提交withconn.cursor()ascur:cur.execute(INSERT INTO books (...) VALUES (...))cur.execute(SELECT count(*) FROM books)count_beforecur.fetchone()[0]print(f插入后记录数:{count_before})# 8# 回滚!conn.rollback()print(ROLLBACK — 事务已回滚)# 验证withconn.cursor()ascur:cur.execute(SELECT count(*) FROM books)count_aftercur.fetchone()[0]print(f回滚后记录数:{count_after})# 7 (撤销了刚刚的INSERT)# ✓ ROLLBACK 验证通过 — 未提交的插入已被撤销事务状态枚举psycopg 内置状态常量含义TransactionStatus.IDLE空闲不在事务中TransactionStatus.ACTIVE有活动命令TransactionStatus.INTRANS事务中TransactionStatus.INERROR事务出错需ROLLBACKTransactionStatus.UNKNOWN未知1.5 参数化查询与SQL注入防护这是数据库编程最重要的安全问题两种安全方式 一种危险方式# ✅ 方式1: %s 占位符位置参数search_authorMartincur.execute(SELECT title FROM books WHERE author LIKE %s,(f%{search_author}%,)# 元组! 单元素也要逗号)# ✅ 方式2: 命名参数PyFormat推荐复杂查询cur.execute(SELECT title, price FROM books WHERE price BETWEEN %(lo)s AND %(hi)s,{lo:80,hi:120})SQL注入演示malicious_input; DROP TABLE books; --# ❌ 危险! 字符串拼接 —— 永远不要这样做!# sql fSELECT * FROM books WHERE title {malicious_input}# 实际SQL: SELECT * FROM books WHERE title ; DROP TABLE books; --# 后果: 整张表被删除!# ✅ 安全: 参数化查询cur.execute(SELECT * FROM books WHERE title %s,(malicious_input,))resultcur.fetchone()# ✓ 恶意输入被安全转义为字面字符串, 查询无结果防注入铁律❌ 永远不要✅ 始终使用fSELECT * FROM t WHERE col {user_input}execute(...WHERE col %s, (user_input,))...WHERE col user_inputexecute(...WHERE col %(val)s, {val: user_input})...WHERE col %s % user_inputexecute(...WHERE col %s, params)第二章SQLAlchemy Core — 数据库抽象层SQLAlchemy 分两层Core 层本文重点SQL表达式构建不用ORM映射类ORM 层对象关系映射class - table 对映2.1 引擎与元数据fromsqlalchemyimport(create_engine,MetaData,Table,Column,Integer,String,Numeric,DateTime,text,select,insert,update,delete,func,desc)importdatetime# 创建引擎DB_URLpostgresqlpsycopg://python_user:Python%40123localhost:5432/python_dbenginecreate_engine(DB_URL,echoFalse)# echoTrue 可看SQL日志print(f引擎创建成功!)print(f PostgreSQL 版本:{pg_ver})print(f 当前数据库: python_db, 当前用户: python_user)踩坑密码含等特殊字符时URL中需转义→%40否则被误解析为userhost结构。表定义不使用ORM类metadataMetaData()studentsTable(students,metadata,Column(id,Integer,primary_keyTrue,autoincrementTrue),Column(name,String(50),nullableFalse),Column(age,Integer,nullableFalse),Column(grade,String(20)),Column(score,Numeric(5,2)),Column(enrolled,DateTime,defaultdatetime.datetime.now),)# 在数据库中创建表metadata.create_all(engine)print(CREATE TABLE students — 成功)列类型对照SQLAlchemyPostgreSQLPythonIntegerINTEGERintString(n)VARCHAR(n)strNumeric(p, s)NUMERIC(p, s)DecimalDateTimeTIMESTAMPdatetimeBooleanBOOLEANboolTextTEXTstr2.2 Core层CRUD操作INSERT— 单条和批量# 单条stmtinsert(students).values(name张三,age20,gradeA,score88.5,enrolleddatetime.datetime(2024,9,1))conn.execute(stmt)# 批量 (executemany风格)stmtinsert(students)conn.execute(stmt,[{name:李四,age:21,grade:A,score:92.0},{name:王五,age:19,grade:B,score:75.5},{name:赵六,age:22,grade:A,score:95.0},{name:孙七,age:20,grade:C,score:60.0},{name:周八,age:21,grade:B,score:78.0},])conn.commit()SELECT— 条件/排序/聚合# 全表 排序stmtselect(students).order_by(desc(students.c.score))resultconn.execute(stmt)真实输出姓名 年龄 等级 成绩 ------------------------------ 赵六 22 A 95.0 李四 21 A 92.0 张三 20 A 88.5 周八 21 B 78.0 王五 19 B 75.5 孙七 20 C 60.0WHERE条件stmtselect(students).where(students.c.score80)# 成绩80: 3人 (张三88.5 / 李四92.0 / 赵六95.0)聚合 分组# 聚合stmtselect(func.count(students.c.id).label(人数),func.avg(students.c.score).label(平均分),func.max(students.c.score).label(最高分),func.min(students.c.score).label(最低分),)# 人数6 平均分81.5 最高分95.00 最低分60.00# GROUP BYstmtselect(students.c.grade,func.count(students.c.id).label(cnt),func.avg(students.c.score).label(avg_s)).group_by(students.c.grade).order_by(students.c.grade)# A级 — 3人 平均分91.8# B级 — 2人 平均分76.8# C级 — 1人 平均分60.0UPDATEstmtupdate(students).where(students.c.name孙七).values(score68.0,gradeB)conn.execute(stmt)conn.commit()# UPDATE — 孙七成绩从60→68, 等级C→BDELETEstmtdelete(students).where(students.c.score70)conn.execute(stmt)conn.commit()# DELETE — 删除成绩70的学生剩余5人最终学生表DELETE后ID 姓名 年龄 等级 成绩 ----------------------------------- 1 张三 20 A 88.5 2 李四 21 A 92.0 3 王五 19 B 75.5 4 赵六 22 A 95.0 6 周八 21 B 78.0注意id5孙七已被DELETE移除。清理metadata.drop_all(engine)# DROP TABLE students — 清理完成第三章Supabase云数据库Supabase 是 “开源 Firebase 替代品”提供托管的 PostgreSQL 服务。本节为参考性内容。实际在 Supabase 上建立数据库服务需注册 supabase.com 账号创建项目 → 获得连接字符串用上面学到的 psycopg3 或 SQLAlchemy 直连# Supabase 连接示例importpsycopg# Supabase 会提供类似这样的连接字符串SUPABASE_URLdb.xxxxx.supabase.coSUPABASE_KEYyour-service-role-keyconnpsycopg.connect(hostSUPABASE_URL,port5432,dbnamepostgres,userpostgres,passwordSUPABASE_KEY,sslmoderequire# Supabase 要求 SSL)psycopg3 vs SQLAlchemy 终极对比┌──────────────────────────────────────────────────────────────┐ │ 选型决策树 │ │ │ │ 需要 async/await? ─── Yes ──▶ psycopg3 (原生支持) │ │ │ │ │ No │ │ │ │ │ 需要切换数据库(MySQL→PG)? ── Yes ──▶ SQLAlchemy Core │ │ │ │ │ No │ │ │ │ │ 简单CRUD / 数据分析? ── Yes ──▶ psycopg3 (更轻量) │ │ │ │ │ No │ │ │ │ │ 复杂业务逻辑 / ORM? ── Yes ──▶ SQLAlchemy ORM │ └──────────────────────────────────────────────────────────────┘维度psycopg3SQLAlchemy Core定位PostgreSQL原生驱动通用数据库抽象层版本3.3.42.0.51async✅ 原生AsyncConnection✅ 2.0 原生支持数据库无关❌ 仅PostgreSQL✅ 通过dialect换库SQL写法手写SQL字符串Python表达式构建防止注入%s占位符自动参数化性能最接近裸SQL略有抽象开销学习曲线低会SQL即可中需学习表达式API适用场景脚本/ETL/数据分析复杂业务应用安装pip install psycopg[binary]pip install sqlalchemySQL写法对比# psycopg3手写SQLcur.execute(SELECT * FROM books WHERE price %s,(100,))# SQLAlchemy CorePython表达式stmtselect(books).where(books.c.price100)conn.execute(stmt)# 两者生成完全相同的SQL结果也相同踩坑记录#问题现象修复1密码含→ URL解析错误failed to resolve host 123localhostURL转义→%402PEP 668 pip限制externally-managed-environment--break-system-packages3python3-psycopg2vspsycopgapt install python3-psycopg2装的是 v2用pip install psycopg[binary]装 v34单参数元组忘记逗号execute(sql, (val))→execute(sql, (val,))单元素元组必须(val,)5忘记commitINSERT后SELECT count(*)看不到新数据始终显式conn.commit()6PyPI超时华为云香港ConnectTimeout→ 阿里云镜像-i https://mirrors.aliyun.com/pypi/simple/总结本文在华为云 FlexusX (8vCPU/16GiB) 服务器上从零搭建 PostgreSQL 16.14通过 9 个实验完整覆盖了 Python 数据库编程两大核心模块实验数核心收获psycopg37连接/DDL/DML/SELECT/聚合/GROUP BY/事务/防注入SQLAlchemy Core2引擎/元数据/Table定义/表达式CRUD/聚合分组一句话建议简单脚本用 psycopg3直接写SQL最灵活多表关联业务用 SQLAlchemy CorePython表达式更安全、可读需要复杂对象模型时上 SQLAlchemy ORM。本文基于华为云 ecs-88e7-0001 (139.9.128.210) 实战编写所有输出均为服务器端真实执行结果。