—— 数据持久化的第一步
# with 语句：自动关闭文件（推荐）
with open("hello.txt", "w", encoding="utf-8") as f:
f.write("Hello, Python!\n")
f.write("你好，世界！\n")
# 读取整个文件
with open("hello.txt", "r", encoding="utf-8") as f:
content = f.read()
print(content)
# 逐行读取
with open("hello.txt", "r", encoding="utf-8") as f:
for line in f:
print(line.rstrip()) # rstrip去掉末尾换行
# 读取所有行为列表
with open("hello.txt", "r", encoding="utf-8") as f:
lines = f.readlines()| 模式 | 说明 | 文件不存在 |
|---|---|---|
"r" | 只读（默认） | 报错 |
"w" | 只写（覆盖） | 创建 |
"a" | 追加 | 创建 |
"x" | 独占创建 | 报错 |
"rb" / "wb" | 二进制读/写 | — |
"r+" | 读写 | 报错 |
import csv
# 写入 CSV
data = [
["姓名", "年龄", "城市"],
["张三", 28, "北京"],
["李四", 32, "上海"],
["王五", 25, "深圳"],
]
with open("users.csv", "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerows(data)
# 读取 CSV
with open("users.csv", "r", encoding="utf-8-sig") as f:
reader = csv.reader(f)
header = next(reader) # 读取表头
for row in reader:
print(row)
# DictReader / DictWriter（推荐！）
with open("users.csv", "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
print(f"{row['姓名']}, {row['年龄']}岁, {row['城市']}")
# DictWriter
fields = ["name", "score"]
with open("scores.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fields)
writer.writeheader()
writer.writerow({"name": "张三", "score": 95})
writer.writerow({"name": "李四", "score": 88})import json
# Python 对象 → JSON 字符串
data = {
"name": "张三",
"age": 28,
"skills": ["Python", "SQL"],
"active": True,
}
# 写入 JSON 文件
with open("data.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 读取 JSON 文件
with open("data.json", "r", encoding="utf-8") as f:
loaded = json.load(f)
print(loaded["name"]) # 张三
# JSON 字符串 ↔ Python 对象
json_str = json.dumps(data, ensure_ascii=False, indent=2)
parsed = json.loads(json_str)
# 处理自定义类型
from datetime import datetime
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
# 编码
now = datetime.now()
encoded = json.dumps({"timestamp": now}, cls=DateTimeEncoder)
# 解码
def datetime_hook(d):
for k, v in d.items():
try:
d[k] = datetime.fromisoformat(v)
except (TypeError, ValueError):
pass
return d
decoded = json.loads(encoded, object_hook=datetime_hook)import configparser
# 写入 INI
config = configparser.ConfigParser()
config["database"] = {
"host": "localhost",
"port": "3306",
"name": "mydb",
}
config["logging"] = {
"level": "INFO",
"file": "app.log",
}
with open("config.ini", "w") as f:
config.write(f)
# 读取 INI
config = configparser.ConfigParser()
config.read("config.ini")
# 获取值
host = config.get("database", "host")
port = config.getint("database", "port") # 自动转 int
level = config.get("logging", "level")
# 带默认值
timeout = config.getint("database", "timeout", fallback=30)
# 检查 section 是否存在
if config.has_section("database"):
print("数据库配置存在")
# 遍历所有配置
for section in config.sections():
print(f"[{section}]")
for key, value in config.items(section):
print(f" {key} = {value}")# 自定义上下文管理器
class Timer:
def __init__(self, name="操作"):
self.name = name
def __enter__(self):
import time
self.start = time.perf_counter()
return self
def __exit__(self, *exc):
import time
elapsed = time.perf_counter() - self.start
print(f"[{self.name}] 耗时: {elapsed:.4f}s")
return False # 不吞异常
with Timer("数据处理"):
total = sum(range(1000000))
# [数据处理] 耗时: 0.0312s
# 使用 contextlib 简化
from contextlib import contextmanager
@contextmanager
def temp_dir():
"""创建临时目录，用完自动删除"""
import tempfile, shutil
d = tempfile.mkdtemp()
try:
yield d
finally:
shutil.rmtree(d)
with temp_dir() as td:
print(f"临时目录: {td}")
# 在这里操作临时文件...#!/usr/bin/env python3
"""第05课 文件操作验证"""
import csv
import json
import configparser
import tempfile
import os
def test_text_file():
"""文本文件读写测试"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt",
delete=False, encoding="utf-8") as f:
path = f.name
f.write("Hello\n")
f.write("世界\n")
with open(path, "r", encoding="utf-8") as f:
lines = f.readlines()
assert lines == ["Hello\n", "世界\n"]
os.unlink(path)
print("✅ 文本文件读写测试通过")
def test_csv():
"""CSV 读写测试"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv",
delete=False, newline="", encoding="utf-8") as f:
path = f.name
# 写入
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["name", "age"])
writer.writeheader()
writer.writerow({"name": "张三", "age": "28"})
writer.writerow({"name": "李四", "age": "32"})
# 读取
with open(path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
rows = list(reader)
assert len(rows) == 2
assert rows[0]["name"] == "张三"
os.unlink(path)
print("✅ CSV 读写测试通过")
def test_json():
"""JSON 读写测试"""
data = {"name": "张三", "scores": [95, 88], "active": True}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json",
delete=False, encoding="utf-8") as f:
path = f.name
json.dump(data, f, ensure_ascii=False)
with open(path, "r", encoding="utf-8") as f:
loaded = json.load(f)
assert loaded["name"] == "张三"
assert loaded["scores"] == [95, 88]
os.unlink(path)
print("✅ JSON 读写测试通过")
def test_ini():
"""INI 配置文件测试"""
config = configparser.ConfigParser()
config["db"] = {"host": "localhost", "port": "3306"}
with tempfile.NamedTemporaryFile(mode="w", suffix=".ini",
delete=False) as f:
path = f.name
config.write(f)
config2 = configparser.ConfigParser()
config2.read(path)
assert config2.get("db", "host") == "localhost"
assert config2.getint("db", "port") == 3306
os.unlink(path)
print("✅ INI 配置文件测试通过")
if __name__ == "__main__":
test_text_file()
test_csv()
test_json()
test_ini()
print("\n🎉 第05课全部验证通过！")