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

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())

if csv then first function to execute and if txt next function to execute.如何使用装饰器实现。我可以放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 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,

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

Answer 2

根据您的代码和 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')

我在 reader 函数中使用了一个有状态的装饰器，在实际执行它之前记住文件名（为了不传递它两次）；并且我使用固定值 None 来表示无效的文件类型。

Answer 3

你可以用装饰器这样做：

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 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

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

How to write two decorator in file operations

python

file-io

file

function

decorator