关闭文件对象的Python回调



我正在开发一个自定义文件路径类,它应该始终执行一个函数写入相应的系统文件及其文件对象之后关闭该功能将把文件路径的内容上传到远程位置。我希望上传功能完全由用户在幕后进行透视,即用户可以像使用任何其他os.PathLike一样使用该类类,并自动获得上传功能。下面的Psuedo代码引用。

import os
class CustomPath(os.PathLike):

def __init__(self, remote_path: str):
self._local_path = "/some/local/path"
self._remote_path = remote_path
def __fspath__(self) -> str:
return self._local_path

def upload(self):
# Upload local path to remote path.

当用户直接调用任何方法。

然而,我不清楚如何自动调用上传功能,如果有人使用内置的open写入文件,如下所示。

custom_path = CustomPath("some remote location")
with open(custom_path, "w") as handle:
handle.write("Here is some text.")

custom_path = CustomPath("some remote location")
handle = open(custom_path, "w")
handle.write("Here is some text.")
handle.close()

我希望与open函数的调用兼容,以便上传行为将适用于所有第三方文件编写器。这种是Python中的行为可能吗?

是的,通过使用Python的函数重写、自定义上下文管理器和__getattr__功能,Python是可能的。以下是基本逻辑:

  • 使用自定义open((类覆盖内置的函数
  • 使用__enter____ exit__方法使其与上下文管理器兼容
  • 使用__getattr__方法使其与正常读/写操作兼容
  • 必要时从调用内置方法
  • 调用close((方法时,invoke会自动回调函数

以下是示例代码:

import builtins
import os
to_be_monitered = ['from_file1.txt', 'from_file2.txt']   
# callback function (called when file closes)
def upload(content_file):
# check for required file
if content_file in to_be_monitered:
# copy the contents
with builtins.open(content_file, 'r') as ff:
with builtins.open(remote_file, 'a') as tf:
# some logic for writing only new contents can be used here 
tf.write('n'+ff.read())


class open(object):
def __init__(self, path, mode):
self.path = path
self.mode = mode
# called when context manager invokes
def __enter__(self):
self.file = builtins.open(self.path, self.mode)
return  self.file
# called when context manager returns
def __exit__(self, *args):
self.file.close()
# after closing calling upload()
upload(self.path)
return True

# called when normal non context manager invokes the object
def __getattr__(self, item):
self.file = builtins.open(self.path, self.mode)
# if close call upload()
if item == 'close':
upload(self.path)
return getattr(self.file, item)

if __name__ == '__main__':

remote_file  = 'to_file.txt'
local_file1  = 'from_file1.txt' 
local_file2  = 'from_file2.txt' 
# just checks and creates remote file no related to actual problem
if not os.path.isfile(remote_file):
f = builtins.open(remote_file, 'w')
f.close()
# DRIVER CODE
# writing with context manger
with open(local_file1, 'w') as f:
f.write('some text written with context manager to file1')
# writing without context manger
f = open(local_file2, 'w')
f.write('some text written without using context manager to file2')
f.close()
# reading file
with open(remote_file, 'r') as f:
print('remote file contains:n', f.read())

它的作用:

写入">用上下文管理器编写的一些文本到文件1";到local_file1.txt和">在没有上下文管理器的情况下编写的一些文本到文件2";到local_file2.txt,同时自动将这些文本复制到remote_file.txt而不显式复制。

如何操作:(上下文管理器案例(

with open(local_file1, 'w') as f:生成自定义类open对象,并初始化其pathmode变量。并调用__ enter __函数(因为上下文管理器(具有as块((,使用builtins.open()方法打开文件,返回_io.TextIOWrapper(打开的文本文件对象(对象。它是一个普通的文件对象,我们可以正常地将其用于读/写操作。在此之后,上下文管理器在结束时调用__ exit __函数,该函数(__ exit__(关闭文件,并自动调用所需的回调(此处为上传(函数,并传递刚刚关闭的文件路径。在这个回调函数中,我们可以执行任何操作,如复制

非上下文管理器的情况也类似,但不同的是__ getattr __函数是一个神奇的函数。

以下是代码执行后文件的内容:

from_file1.text

some text written with context manager to file1

from_file2.txt

some text written without using context manager to file2

to_file.txt

some text written with context manager to file1
some text written without using context manager to file2

根据您对Girish Dattatray Hegde的评论,您似乎想做以下事情来覆盖open的默认__exit__处理程序:

import io

old_exit = io.FileIO.__exit__ # builtin __exit__ method

def upload(self):
print(self.read()) # just print out contents

def new_exit(self):
try:
upload(self)
finally:
old_exit(self) # invoke the builtin __exit__ method

io.FileIO.__exit__ = new_exit # establish our __exit__ method

with open('test.html') as f:
print(f.closed) # False
print(f.closed) # True

不幸的是,上述代码导致以下错误:

test.py", line 18, in <module>
io.FileIO.__exit__ = new_exit # establish our __exit__ method
TypeError: can't set attributes of built-in/extension type '_io.FileIO'

所以,我认为不可能做你想做的事情。最终,你可以创建自己的子类和重写方法,但你不能替换现有内置open类的方法

最新更新