Python IO 与数据库

Python 处理文件、序列化和数据库的接口简洁一致。从文本读写到 ORM，核心模式只有几个。

---

文件操作

打开与关闭

# 推荐方式：with 自动关闭
with open('file.txt', 'r') as f:
    content = f.read()
 
# 手动方式（需要 try/finally）
f = open('file.txt', 'r')
try:
    content = f.read()
finally:
    f.close()

文件模式

模式	描述
'r'	只读，文件必须存在
'w'	写入，覆盖已有内容
'a'	追加到末尾
'r+'	读写
'rb' / 'wb'	二进制读 / 写
'x'	排他创建，文件存在则失败

读取

with open('file.txt') as f:
    f.read()          # 全部读入字符串
    f.readline()      # 逐行读
    f.readlines()     # 全部读入列表
 
    for line in f:    # 逐行迭代（推荐，省内存）
        process(line)

写入

with open('output.txt', 'w') as f:
    f.write('Hello\n')
    f.writelines(['line1\n', 'line2\n'])

---

数据格式处理

JSON

import json
 
with open('data.json', 'r') as f:
    data = json.load(f)
 
with open('output.json', 'w') as f:
    json.dump({'key': 'value'}, f, indent=2, ensure_ascii=False)

CSV

import csv
 
with open('data.csv', 'r') as f:
    for row in csv.reader(f):
        print(row)
 
with open('output.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['name', 'age'])
    writer.writerow(['Alice', 30])

---

文件系统操作

os 模块

import os
import shutil
 
os.mkdir('dir')              # 创建目录
os.makedirs('a/b/c')         # 递归创建
os.remove('file.txt')        # 删除文件
os.rename('old', 'new')      # 重命名
shutil.move('src', 'dst')    # 移动
shutil.copy('src', 'dst')    # 复制

pathlib（推荐）

from pathlib import Path
 
path = Path('data') / 'file.txt'
path.exists()          # 是否存在
path.suffix            # 后缀名
path.read_text()       # 读文本
path.write_text('hi')  # 写文本
path.parent            # 父目录

---

大文件处理

# 逐行处理，不占内存
with open('large.txt') as f:
    for line in f:
        process(line.strip())
 
# 内存映射（二进制文件）
import mmap
with open('large.bin', 'r+b') as f:
    mm = mmap.mmap(f.fileno(), 0)
    data = mm[:100]
    mm.close()

---

数据库操作

DB-API 2.0 标准

Python 通过 DB-API 2.0（PEP 249）统一了数据库驱动接口：

组件	角色
Connection	数据库连接，管理事务
Cursor	执行 SQL、获取结果

import sqlite3
 
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
 
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users
    (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)
''')
 
cursor.execute(
    "INSERT INTO users VALUES (?, ?, ?)",
    (1, 'Alice', 30)
)
conn.commit()
 
cursor.execute("SELECT * FROM users")
for row in cursor.fetchall():
    print(row)
 
cursor.close()
conn.close()

参数化查询（防 SQL 注入）

# ❌ 拼接——SQL 注入风险
sql = f"SELECT * FROM users WHERE name = '{user_input}'"
 
# ✅ 参数化
cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))

事务管理

try:
    cursor.execute("INSERT INTO accounts VALUES (?, ?)", (1, 100))
    cursor.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
    conn.commit()
except Exception:
    conn.rollback()
    raise

连接池

from sqlalchemy import create_engine
 
engine = create_engine(
    'postgresql://user:pass@localhost/db',
    pool_size=5,
    max_overflow=10
)
 
with engine.connect() as conn:
    result = conn.execute(...)

---

ORM

Python ORM 选型：

库	特点	适用
SQLAlchemy	最强大灵活的 ORM	中大型项目，复杂查询
Django ORM	Django 内置	Django 项目
Peewee	小巧简洁	小型项目
Tortoise ORM	异步 ORM	asyncio 项目
Pony ORM	直观查询语法	快速原型

SQLAlchemy 示例

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, Session
 
class Base(DeclarativeBase):
    pass
 
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
 
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
 
with Session(engine) as session:
    user = User(name='Alice')
    session.add(user)
    session.commit()

---

关联

• meta: I/O 模型 — 跨语言 I/O 策略对比