在Python中,你可以通过文件操作函数(如open()
函数)以及模拟输入输出流的库(如io
模块)来模拟文件行为。下面是一些示例,展示了如何使用这些工具在Python中模拟文件行为。
1、问题背景
在编写一个脚本时,需要将SQL数据库中某些表的列转储到文件,然后通过FTP传输。由于转储的内容可能非常庞大,因此设计了一个方案,即创建一个MysSQLFakeFile,该文件在readline方法中逐行查询光标,并将其传递给ftplib.FTP.storlines。
以下是实现代码:
代码语言:javascript复制import ftplib
import MySQLdb
def MySQLFakeFile(object):
'''
模拟一个只读文件,按需转存储表数据
通过将其传递给FTP协议,可使转储更有效率,而无需将其转储到某处并在网络上传输
'''
def __init__(self, cursor, delimeter, table_name, query):
self.cursor = cursor
self.delimeter = delimeter
self.table_name = table_name
# 查询类似于select ... FROM %s
self.cursor.execute(query, table_name)
self._has_written_index = False
# 文件属性
self.closed = False
self.name = table_name ".csv"
self.encoding = "utf-8"
self.mode = "r"
def close(self):
self.cursor.close()
self.closed = True
def flush(self):
'''空操作'''
pass
def read(self, size):
pass
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 = True
else:
ret = self.cursor.fetchone()
if not ret:
return None
s = ""
for col in ret:
s = str(col) self.delimeter
return 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)
然而,运行这段代码时却产生了以下错误:
代码语言:javascript复制Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python2.7/ftplib.py", line 496, in storlines
if len(buf) > self.maxline:
TypeError: object of type 'NoneType' has no len()
2、解决方案
经过分析,发现问题出在readline方法中,当到达行尾时,它返回None而不是空字符串("")。同时,readlines方法也没有返回任何内容。
因此,对readline方法和readlines方法进行了修改,如下:
代码语言:javascript复制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 = True
else:
ret = self.cursor.fetchone()
if not ret:
return None
s = ""
for col in ret:
s = str(col) self.delimeter
return s "n"
def readlines(self, size):
ret = []
while True:
line = self.readline()
if not line:
break
ret.append(line)
return ret
修改后的代码运行正常,可以将表数据通过FTP传输到指定文件中。
在这个示例中,我在使用io.StringIO
创建了一个内存中的文件对象,并向其中写入了一些文本。然后我们将文件指针移动到开头,读取内容并打印出来。最后,我们关闭内存中的文件对象。
使用这些方法,我们可以在Python中模拟文件的行为,并根据需要进行读写操作。