Python 将数据写入csv、xlsx、xls文件中（工厂方法、封装、优雅）

前言

学了一下设计模式之后，就想到之前写爬虫时经常把数据保存到csv（常用）、xlsx文件中，这些反复地操作很是繁琐。所以就想用这个工厂设计的方法来记录一下，向常见的文件格式里写入数据的操作进行封装，方便以后调用！

---

文件目录存放结构

先看看文件存放的结构展示：

---

my_file_save.py

这个模块是用来定义一个MyFileSave父类，并让这个父类具有一些抽象的方法，来让实现它的子类必须重写这些方法，达到约束的效果。

from abc import ABC, abstractmethod
class MyFileSave(ABC):
    @abstractmethod
    def __init__(self):
        pass
    @abstractmethod
    def my_writeheader(self, *args):
        pass
    @abstractmethod
    def my_writerow(self, *args):
        pass
    @abstractmethod
    def my_close(self):
        pass
    @abstractmethod
    def __del__(self):
        pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

---

wrapper_verify_param.py

这个模块是一个装饰器，用来增强my_writerow()，my_writerow()这两个方法里面参数传入形式的功能。

def verify_param(param):
    def wrapper(func):
        def inner(*args, **kwargs):
            if not param: raise Exception("请传入数据！！！")
            if isinstance(param[0],list) or isinstance(param[0],tuple):
                args = param[0]
            else:
                args = args[0]
            res = func(*args, **kwargs)
            return res
        return inner
    return wrapper
1
2
3
4
5
6
7
8
9
10
11
12

---

封装csv（my_csv.py）

采用单行写入的方法。
这里的my_writeheader()里面注入了verify_param这个装饰器，所以我们在外面调用my_writeheader()这个方法时可以这样进行传入数据（my_writerow()与之同理）：

my_writeheader([“id”,“name”,“age”]) #写入表头数据
my_writerow([1,“yyds”,“21”]) #行写入数据
my_writerow((2,“小尹”,“22”))
my_writerow(3, “小兰”, “18”)

如果不加这个装饰器，只能以这种my_writerow(3, “小兰”, “18”)形式传入数据。
后面的xlsx、xls写法都是这样，只是重写的内容有区别而已！
当然还有更多的方法，欢迎大家在评论里留言！！！

import csv
from my_file_save import MyFileSave
from wrapper_verify_param import verify_param
class SaveToCsv(MyFileSave):
    def __init__(self,fileName="Data",filePath=".",writeMode="w+",encoding="utf-8"):
        self._fileName = fileName
        self._filePath = filePath
        self._writeMode = writeMode
        self._encoding = encoding
        self._get_csv_obj()
    def _get_csv_obj(self):
        self._f = open(f"{self._filePath}/{self._fileName}.csv", mode=self._writeMode, encoding=self._encoding,
                       newline='')
        self._csv_obj = csv.writer(self._f)
    # 创建表头
    def my_writeheader(self,*args):
        @verify_param(args)
        def wrapper(*args):
            self._get_csv_obj() #防止多次写入表头，造成重复
            self._csv_obj.writerow(args)
        wrapper(args)
    # 一行一行写入数据
    def my_writerow(self,*args):
        @verify_param(args)
        def wrapper(*args):
            self._csv_obj.writerow(args)
        wrapper(args)
    # 关闭io管道
    def my_close(self):
        self._f.close()
    # 当对象销毁时，关闭io管道（可以防止手动时忘关了！）
    def __del__(self):
        try: self._f.close()
        except: pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

---

工厂方法（savedata_factory.py）

三个方法：在工厂中进行注册，在工厂中进行注销，工厂开始生产。

myfilesave_creation_function这个是字典类型的。
factory_register(“csv”, SaveToCsv)；
factory_register(“xlsx”, SaveToXlsx)；
factory_register(“xls”, SaveToXls)

---

如果注册成功后，print(myfilesave_creation_function) 打印后的效果：

---

在工厂中进行注销： 简单就是从字典里移除myfilesave_creation_function.pop(file_type,None)。

from typing import Callable, Dict
from my_file_save import MyFileSave

myfilesave_creation_function: Dict[str, Callable[..., MyFileSave]] = {}

# 在工厂中进行注册
def factory_register(file_type:str, creation_function:Callable[..., MyFileSave]) -> None:
    myfilesave_creation_function[file_type] = creation_function
# 在工厂中进行注销
def factory_unregister(file_type:str) -> None:
    myfilesave_creation_function.pop(file_type,None)
# 工厂开始生产
def factory_production(**args) -> MyFileSave:
    that_args = args.copy()
    file_type = that_args.pop("file_type",None)
    if not file_type: raise Exception(f"无指定文件类型异常！")
    try:
        creation_function = myfilesave_creation_function[file_type]
        return creation_function(**that_args)
    except:
        raise Exception(f"未知的文件类型异常！！！{file_type}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

---

测试（test.py）

factory_register(“csv”, SaveToCsv)与
factory_production(file_type=“csv”)，file_type=""必须以这样写，且两个csv必须要保持一致！
后面这些：fileName=“Data”,filePath=“.”,writeMode=“w+”,encoding=“utf-8”
是默认的，可自定义配置。

效果：

from savedata_factory import factory_register, factory_unregister, factory_production
from my_csv import SaveToCsv
# 所有格式的文件在工厂中应先进行注册
factory_register("csv", SaveToCsv)
def csv_test():
    # 开始生产一个操作csv文件格式的对象
    # csv = factory_production(file_type="csv",fileName="Data",filePath=".",writeMode="w+",encoding="utf-8")
    csv = factory_production(file_type="csv")
    # 写入表头数据
    csv.my_writeheader(["id", "name", "age"])
    # 行写入内容
    csv.my_writerow([1, "yyds", "21"])
    csv.my_writerow((2, "小尹", "22"))
    csv.my_writerow(3, "小兰", "18")
    # 关闭io流管道
    csv.my_close()
if __name__ == '__main__':
    csv_test()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

---

封装xlsx（my_xlsx.py）

import openpyxl
from my_file_save import MyFileSave
from wrapper_verify_param import verify_param
class SaveToXlsx(MyFileSave):
    def __init__(self, fileName="Data", filePath=".", tableName="", encoding="utf-8"):
        self._fileName = fileName
        self._filePath = filePath
        self._workbook = openpyxl.Workbook()
        self._workbook.encoding = encoding
        self._worksheet = self._workbook.active
        if not tableName: tableName = fileName
        self._worksheet.title = tableName
        self._ROW = 2
    # 创建表头
    def my_writeheader(self,*args):
        @verify_param(args)
        def wrapper(*args):
            for column in range(len(args)):
                self._worksheet.cell(1, column + 1, args[column])
        wrapper(args)
    # 一行一行写入数据
    def my_writerow(self,*args):
        @verify_param(args)
        def wrapper(*args):
            for column in range(len(args)):
                self._worksheet.cell(self._ROW, column + 1, args[column])
        wrapper(args)
        self._ROW +=1
    # 关闭io管道
    def my_close(self):
        self._workbook.save(filename=f"{self._filePath}/{self._fileName}.xlsx")
        self._workbook.close()
    # 当对象销毁时，关闭io管道（可以防止手动时忘关了！）
    def __del__(self):
        try:
            self._workbook.save(filename=f"{self._filePath}/{self._fileName}.xlsx")
            self._workbook.close()
        except:
            pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

---

封装xls（my_xls.py）

import xlwt
from my_file_save import MyFileSave
from wrapper_verify_param import verify_param
class SaveToXls(MyFileSave):
    def __init__(self, fileName="Data", filePath=".", tableName="", encoding="utf-8"):
        self._fileName = fileName
        self._filePath = filePath
        self._workbook = xlwt.Workbook(encoding=encoding, style_compression=0)
        if not tableName: tableName = fileName
        self._worksheet = self._workbook.add_sheet(tableName, cell_overwrite_ok=True)
        self._ROW = 1
    # 创建表头
    def my_writeheader(self, *args):
        @verify_param(args)
        def wrapper(*args):
            for column in range(len(args)):
                self._worksheet.write(0, column, args[column])
        wrapper(args)
    # 一行一行写入数据
    def my_writerow(self, *args):
        @verify_param(args)
        def wrapper(*args):
            for column in range(len(args)):
                self._worksheet.write(self._ROW, column, args[column])
        wrapper(args)
        self._ROW += 1
    # 关闭io管道
    def my_close(self):
        self._workbook.save(f"{self._filePath}/{self._fileName}.xls")
    # 当对象销毁时，关闭io管道（可以防止手动时忘关了！）
    def __del__(self):
        try:
            self._workbook.save(f"{self._filePath}/{self._fileName}.xls")
        except:
            pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

测试xlsx、xls

一定要先在工厂中进行注册了，才能生产！！！

if __name__ == '__main__':
    print("-------------xlsx_test--------------")
    from my_xlsx import SaveToXlsx
    factory_register("xlsx", SaveToXlsx)
    xlsx = factory_production(file_type="xlsx")
    xlsx.my_writeheader(["id", "name", "age"])
    xlsx.my_writerow([1, "yyds", "21"])
    xlsx.my_close()
    print("-------------xls_test--------------")
    from my_xls import SaveToXls
    factory_register("xls", SaveToXls)
    xls = factory_production(file_type="xls")
    xls.my_writeheader(["id", "name", "age"])
    xls.my_close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14