使用Python模仿文件行为

news2024/11/27 8:39:38

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