记录:将数据写入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
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
封装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
工厂方法(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}")
测试(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()
封装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
封装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
测试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()