使用Python模仿文件行为-编程知识

使用Python模仿文件行为

在Python中，你可以通过文件操作函数（如open()函数）以及模拟输入输出流的库（如io模块）来模拟文件行为。下面是一些示例，展示了如何使用这些工具在Python中模拟文件行为。

在这里插入图片描述

1、问题背景

在编写一个脚本时，需要将SQL数据库中某些表的列转储到文件，然后通过FTP传输。由于转储的内容可能非常庞大，因此设计了一个方案，即创建一个MysSQLFakeFile，该文件在readline方法中逐行查询光标，并将其传递给ftplib.FTP.storlines。

以下是实现代码：

import ftplib
import MySQLdbdef MySQLFakeFile(object):'''模拟一个只读文件，按需转存储表数据通过将其传递给FTP协议，可使转储更有效率，而无需将其转储到某处并在网络上传输'''def __init__(self, cursor, delimeter, table_name, query):self.cursor = cursorself.delimeter = delimeterself.table_name = table_name# 查询类似于select ... FROM %sself.cursor.execute(query, table_name)self._has_written_index = False# 文件属性self.closed = Falseself.name = table_name + ".csv"self.encoding = "utf-8"self.mode = "r"def close(self):self.cursor.close()self.closed = Truedef flush(self):'''空操作'''passdef read(self, size):passdef readline(self, size):if not self._has_written_index:ret = []for desc in self.cursor.description:ret.append(desc[0])self._has_written_index = Trueelse:ret = self.cursor.fetchone()if not ret:return Nones = ""for col in ret:s += str(col) + self.delimeterreturn s + "\n"def readlines(self, size):ret = []line = self.readline()while line:ret.append(line)line = self.readline()def write(self, string):raise Exception("无法写入MySQLFafeFile")def writelines(self, lines):raise Exception("无法写入MySQLFafeFile")db = MySQLdb("host", "user", "pass", "db")
ftp = ftplib.FTP("host", "user", "pass")
fakeFile = MySQLFakeFile(db.cursor(), ";", "tableName", "SELECT * FROM %s")
ftp.storlines("STOR tableName.csv", fakeFile)

然而，运行这段代码时却产生了以下错误：

Traceback (most recent call last):File "<stdin>", line 1, in <module>File "/usr/lib/python2.7/ftplib.py", line 496, in storlinesif len(buf) > self.maxline:
TypeError: object of type 'NoneType' has no len()

2、解决方案

经过分析，发现问题出在readline方法中，当到达行尾时，它返回None而不是空字符串（“”）。同时，readlines方法也没有返回任何内容。

因此，对readline方法和readlines方法进行了修改，如下：

def readline(self, size):if not self._has_written_index:ret = []for desc in self.cursor.description:ret.append(desc[0])self._has_written_index = Trueelse:ret = self.cursor.fetchone()if not ret:return Nones = ""for col in ret:s += str(col) + self.delimeterreturn s + "\n"def readlines(self, size):ret = []while True:line = self.readline()if not line:breakret.append(line)return ret