python中的io流

读写demo

# --- 写入文件 ---content="Hello, Python IO!\n这是第二行内容。\n"# 'w' 模式：写入模式，如果文件存在则覆盖，不存在则创建withopen('example.txt','w',encoding='utf-8')asf:f.write(content)print("文件写入成功。")# --- 读取文件 ---# 'r' 模式：只读模式（默认）withopen('example.txt','r',encoding='utf-8')asf:# 方法 A: 一次性读取所有内容full_content=f.read()print("--- 全部内容 ---")print(full_content)# --- 逐行读取 (适合大文件) ---print("--- 逐行读取 ---")withopen('example.txt','r',encoding='utf-8')asf:forlineinf:# strip() 去除每行末尾的换行符print(f"当前行:{line.strip()}")

csv和json

importcsvimportjson data=[{"name":"Alice","age":25,"city":"Beijing"},{"name":"Bob","age":30,"city":"Shanghai"},{"name":"Charlie","age":28,"city":"Shenzhen"}]# --- JSON I/O ---# 写入 JSONwithopen('data.json','w',encoding='utf-8')asf:json.dump(data,f,ensure_ascii=False,indent=4)# indent 用于格式化输出# 读取 JSONwithopen('data.json','r',encoding='utf-8')asf:loaded_data=json.load(f)print(f"读取到的名字列表:{[item['name']foriteminloaded_data]}")# --- CSV I/O ---# 写入 CSVwithopen('data.csv','w',newline='',encoding='utf-8')asf:writer=csv.DictWriter(f,fieldnames=["name","age","city"])writer.writeheader()# 写入表头writer.writerows(data)# 读取 CSVwithopen('data.csv','r',encoding='utf-8')asf:reader=csv.DictReader(f)forrowinreader:print(f"CSV 行:{row['name']}-{row['city']}")

binary二进制示例

# 假设我们要复制一张图片 (请先确保当前目录下有一张 test.jpg，或者用下面的代码生成一个假的)# 生成一个假的二进制文件用于演示withopen('dummy_image.bin','wb')asf:f.write(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01')# 伪造的文件头# --- 二进制文件复制 (分块读取，节省内存) ---chunk_size=1024# 每次读取 1KBwithopen('dummy_image.bin','rb')assrc,open('copy_image.bin','wb')asdst:whileTrue:chunk=src.read(chunk_size)ifnotchunk:# 读取完毕breakdst.write(chunk)print("二进制文件复制完成。")

交互窗口(输入和输出)

importsys# --- 基本输入输出 ---user_input=input("请输入你的名字: ")print(f"你好,{user_input}!")# --- 重定向示例 (模拟管道) ---# 正常情况 print 输出到屏幕 (sys.stdout)# 我们可以将输出写入文件或变量original_stdout=sys.stdoutwithopen('log_output.txt','w',encoding='utf-8')asf:sys.stdout=f# 将标准输出重定向到文件print("这行文字不会出现在屏幕上，而是写入 log_output.txt")print("即使是 error 信息，如果用 print 也会去这里")sys.stdout=original_stdout# 恢复标准输出print("输出已恢复，现在这句话在屏幕上。")# --- 读取多行输入 (直到遇到空行或 Ctrl+D/Z) ---print("请输入多行文本 (双击回车结束):")lines=[]whileTrue:try:line=input()ifnotline:breaklines.append(line)exceptEOFError:breakprint(f"你一共输入了{len(lines)}行。")

内存中的 I/O (io.StringIO / io.BytesIO)

fromioimportStringIO,BytesIO# --- 字符串流 ---sio=StringIO()sio.write("第一行\n")sio.write("第二行\n")# 获取指针位置print(f"当前指针位置:{sio.tell()}")# 回到开头sio.seek(0)# 像读文件一样读取content=sio.read()print("内存中的内容:")print(content)sio.close()# --- 字节流 (常用于网络传输模拟) ---bio=BytesIO()bio.write(b"Binary data here")bio.seek(0)data=bio.read()print(f"读取到的字节:{data}")

高性能 I/O: pathlib 和 异步 (Modern Python)

高性能io-demo1

frompathlibimportPath p=Path('.')# 当前目录# 列出所有 .py 文件print("当前的 Python 文件:")forfileinp.glob('*.py'):print(f"-{file.name}(大小:{file.stat().st_size}bytes)")# 快速读写config_path=Path('config.ini')config_path.write_text("[settings]\nmode=prod",encoding='utf-8')print(f"配置内容:{config_path.read_text(encoding='utf-8')}")

高性能io-异步文件io

importasyncio# 如果没有安装 aiofiles，请注释掉下面这行并跳过此函数演示try:importaiofilesexceptImportError:aiofiles=Noneasyncdefasync_file_operation():ifnotaiofiles:print("未安装 aiofiles，跳过异步示例。请先运行: pip install aiofiles")returnfilename='async_test.txt'# 异步写入asyncwithaiofiles.open(filename,'w',encoding='utf-8')asf:awaitf.write("这是异步写入的内容\n")awaitf.write("不会阻塞主线程\n")# 异步读取asyncwithaiofiles.open(filename,'r',encoding='utf-8')asf:content=awaitf.read()print("异步读取结果:")print(content)# 运行异步函数ifaiofiles:asyncio.run(async_file_operation())

常规建议

1、日常脚本：首选 with open(…) 和 pathlib。
2、数据处理：使用 csv 和 json 模块。
3、大文件处理：务必分块读取 (read(chunk_size)) 或 逐行迭代，避免内存溢出。
4、Web 服务/高并发：考虑使用 io.StringIO 减少磁盘 IO，或在异步框架中使用 aiofiles。
5、二进制文件：记得加 ‘b’ 标志，且不要指定 encoding。