0
点赞
收藏
分享

微信扫一扫

使用Python模仿文件行为

吴陆奇 1天前 阅读 0

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

使用Python模仿文件行为_Python

1、问题背景

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

以下是实现代码:

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)

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

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方法进行了修改,如下:

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中模拟文件的行为,并根据需要进行读写操作。

举报

相关推荐

0 条评论