python中的io流

文章目录读写democsv和jsonbinary二进制示例交互窗口(输入和输出)内存中的 I/O (io.StringIO / io.BytesIO)高性能 I/O: pathlib 和 异步 (Modern Python)高性能io-demo1高性能io-异步文件io常规建议又是一大类记录下。读写demo# --- 写入文件 ---contentHello, Python IO!\n这是第二行内容。\n# w 模式写入模式如果文件存在则覆盖不存在则创建withopen(example.txt,w,encodingutf-8)asf:f.write(content)print(文件写入成功。)# --- 读取文件 ---# r 模式只读模式默认withopen(example.txt,r,encodingutf-8)asf:# 方法 A: 一次性读取所有内容full_contentf.read()print(--- 全部内容 ---)print(full_content)# --- 逐行读取 (适合大文件) ---print(--- 逐行读取 ---)withopen(example.txt,r,encodingutf-8)asf:forlineinf:# strip() 去除每行末尾的换行符print(f当前行:{line.strip()})csv和jsonimportcsvimportjson 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,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent4)# indent 用于格式化输出# 读取 JSONwithopen(data.json,r,encodingutf-8)asf:loaded_datajson.load(f)print(f读取到的名字列表:{[item[name]foriteminloaded_data]})# --- CSV I/O ---# 写入 CSVwithopen(data.csv,w,newline,encodingutf-8)asf:writercsv.DictWriter(f,fieldnames[name,age,city])writer.writeheader()# 写入表头writer.writerows(data)# 读取 CSVwithopen(data.csv,r,encodingutf-8)asf:readercsv.DictReader(f)forrowinreader:print(fCSV 行:{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_size1024# 每次读取 1KBwithopen(dummy_image.bin,rb)assrc,open(copy_image.bin,wb)asdst:whileTrue:chunksrc.read(chunk_size)ifnotchunk:# 读取完毕breakdst.write(chunk)print(二进制文件复制完成。)交互窗口(输入和输出)importsys# --- 基本输入输出 ---user_inputinput(请输入你的名字: )print(f你好,{user_input}!)# --- 重定向示例 (模拟管道) ---# 正常情况 print 输出到屏幕 (sys.stdout)# 我们可以将输出写入文件或变量original_stdoutsys.stdoutwithopen(log_output.txt,w,encodingutf-8)asf:sys.stdoutf# 将标准输出重定向到文件print(这行文字不会出现在屏幕上而是写入 log_output.txt)print(即使是 error 信息如果用 print 也会去这里)sys.stdoutoriginal_stdout# 恢复标准输出print(输出已恢复现在这句话在屏幕上。)# --- 读取多行输入 (直到遇到空行或 CtrlD/Z) ---print(请输入多行文本 (双击回车结束):)lines[]whileTrue:try:lineinput()ifnotline:breaklines.append(line)exceptEOFError:breakprint(f你一共输入了{len(lines)}行。)内存中的 I/O (io.StringIO / io.BytesIO)fromioimportStringIO,BytesIO# --- 字符串流 ---sioStringIO()sio.write(第一行\n)sio.write(第二行\n)# 获取指针位置print(f当前指针位置:{sio.tell()})# 回到开头sio.seek(0)# 像读文件一样读取contentsio.read()print(内存中的内容:)print(content)sio.close()# --- 字节流 (常用于网络传输模拟) ---bioBytesIO()bio.write(bBinary data here)bio.seek(0)databio.read()print(f读取到的字节:{data})高性能 I/O: pathlib 和 异步 (Modern Python)高性能io-demo1frompathlibimportPath pPath(.)# 当前目录# 列出所有 .py 文件print(当前的 Python 文件:)forfileinp.glob(*.py):print(f-{file.name}(大小:{file.stat().st_size}bytes))# 快速读写config_pathPath(config.ini)config_path.write_text([settings]\nmodeprod,encodingutf-8)print(f配置内容:{config_path.read_text(encodingutf-8)})高性能io-异步文件ioimportasyncio# 如果没有安装 aiofiles请注释掉下面这行并跳过此函数演示try:importaiofilesexceptImportError:aiofilesNoneasyncdefasync_file_operation():ifnotaiofiles:print(未安装 aiofiles跳过异步示例。请先运行: pip install aiofiles)returnfilenameasync_test.txt# 异步写入asyncwithaiofiles.open(filename,w,encodingutf-8)asf:awaitf.write(这是异步写入的内容\n)awaitf.write(不会阻塞主线程\n)# 异步读取asyncwithaiofiles.open(filename,r,encodingutf-8)asf:contentawaitf.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。