如何在文件操作中编写两个装饰器

Question

我的 csv 文件包含以下内容

101,item_1 101,item_1

如果是 csv 我下面的代码将执行

import csv    
fName = input()
def read_csv(fName):
    try:
        with open(fName, 'r') as f:
            reader = csv.reader(f)
            for row in reader:
                print (row)

read_csv(fName)

这里是如何在decorator函数中编写异常并在上面调用。

第一个装饰器

如果fName 不以.txt 或.csv 结尾，则它必须生成输出not accept

第二个装饰器

如果 fName = file.txt 文本文件，则必须注意以下操作

def read_txt(fName):
    f = open(fName, "r")
    print(f.readline())

如果 csv 则执行第一个函数，如果 txt 则执行下一个函数。如何使用装饰器来实现。我可以设置 if 条件来实现这种情况，但事实并非如此

我没有装饰器的整个代码如下

fName = input()
def read_csv(fName):
    if fName.endswith('.csv'):
        #print  ('hi')
        try:
            with open(fName, 'r') as f:
                reader = csv.reader(f)
                for row in reader:
                    print (row)
        except IOError:
            print ("Could not read file:", fName)
    #SECOND DECORATOR
    if fName.endswith('.txt'):
        f = open(fName, "r")
        print(f.readline())
    #FIRST DECORATOR
    if not(fName.endswith('.csv')) and not(fName.endswith('.txt')):
        print ('not accept')
read_csv(fName)

Answer 1

你可以用装饰器这样做：

import functools


def check_arguments(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        fname = kwargs['fname']
        if not fname.endswith('.csv') and not fname.endswith('.txt'):
            print('not accept')
        return func(*args, **kwargs)
    return wrapper


def set_file_processor(func):
    def read_csv(fname):
        print('read_csv', fname)

    def read_txt(fname):
        print('read_txt', fname)


    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        fname = kwargs['fname']
        if fname.endswith('.csv'):
            read_csv(fname)
        elif fname.endswith('.txt'):
            read_txt(fname)
        return func(*args, **kwargs)
    return wrapper


@check_arguments
@set_file_processor
def process(fname):
    pass


process(fname='input.csv')

Answer 2

您的问题似乎不在装饰器下，而是在工厂模式下，即根据输入文件进行不同的处理。

下面的代码是一个非常简单和基本的工厂模式解决你的问题，这应该根据你的需要进行相应的修改，

import os
from abc import ABC, abstractmethod


class FileProcessor(ABC):
    @abstractmethod
    def process():
        pass

class TextFileProcessor(FileProcessor):
    def process(self, file_path):
        print("Text file processing goes here")


class CsvFileProcessor(FileProcessor):
    def process(self, file_path):
        print("CSV file processing goes here")


class DefaultFileProcessor(FileProcessor):
    def process(self, file_path):
        raise ValueError("File %s is not valid" % file_path)


class FileFactory:
    processors = {
        'txt': TextFileProcessor,
        'csv': CsvFileProcessor,
        'default': DefaultFileProcessor
    }

    def __init__(self, file_path):
        if not os.path.exists(file_path):
            raise IOError("File not found")
        self.file_path = file_path

    def process(self):
        dot_splits = self.file_path.split(".")
        ext = dot_splits[-1] if len(dot_splits) > 1 else "default"
        ext = ext if ext in self.processors else "default"
        processor_class = self.processors.get(ext)

        return processor_class().process(self.file_path)


FileFactory(file_path).process()

后期如果你想添加json处理器，也可以通过添加来轻松完成

processors = {
    'txt': TextFileProcessor,
    'csv': CsvFileProcessor,
    'json': JsonFileProcessor,
    'default': DefaultFileProcessor
}

并创建新的 Json 处理器类，

class JsonFileProcessor(FileProcessor):
    def process(self, file_path):
        print("JSON file processing goes here")

Answer 3

根据您的代码和this very useful guide，这是一个可能的解决方案：

def read_file_decorator(fName):
    def read_csv():
        print('read_csv')
        with open(fName, 'r') as f:
            reader = csv.reader(f)
            for row in reader:
                print(row)

    def read_txt():
        print('read_txt')
        f = open(fName, 'r')
        for row in f:
            print(row)

    if fName.endswith('.csv'):
        return read_csv
    elif fName.endswith('.txt'):
        return read_txt
    else:
        return None

reader_function = read_file_decorator(fileName)
if reader_function != None:
    reader_function()
else:
    print('not accept')

我使用有状态装饰器在实际执行之前记住读取器函数中的文件名（为了不传递两次）；我将固定值None 用于无效文件类型。

Answer 4

根据需求使用装饰器将是装饰器的过度杀伤力。但是如果必须使用装饰器来实现它，我们可以这样实现：

• 我们可以创建一个名为read_file 的虚拟函数和一个名为reader 的装饰器函数
• 用户将始终以文件名作为参数调用read_file，装饰器函数reader将检查传递的文件扩展名并调用所需的函数-read_csv或read_text

def reader(fun):
    def wrapper(*args):
        fname = args[0]
        if fname.endswith('.csv'):
            read_csv(fname)
        elif fname.endswith('.txt'):
            read_text(fname)
        else:
            print('not accepted')
    return wrapper

def read_csv(fname):
    print('In read_csv()')

def read_text(fname):
    print('In read_text()')

@reader
def read_file(fname):
    pass


read_file('a.csv')
read_file('a.txt')
read_file('filename.py')

输出

In read_csv()
In read_text()
not accepted