文件对象关闭的 Python 回调

Question

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

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中可能的行为？

Answer 1

是的，可以使用 python 通过使用 Python 的 函数覆盖，custom context manager和 __ getattr __ 设施。这是基本逻辑:

覆盖 builtins.open() 自定义功能 打开()类(class)。

使其兼容 上下文管理器 使用 __ 输入 __ 和 __ 退出__ 方法。

使其兼容 正常读/写使用 的操作__ getattr __ 方法。

调用 内置 方法来自 类(class)必要时。

调用 自动回调功能当 关闭()方法被调用。

这是示例代码:

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并初始化它是 path和 mode变量。并调用 __ enter __函数(因为上下文管理器(带有块))其中 打开 该文件使用 builtins.open()方法和 返回 _io.TextIOWrapper (打开的文本文件对象)对象。它是一个普通的文件对象，我们可以正常使用它 读/写 操作。在上下文管理器调用 __ exit __ 之后函数末尾 which(__ exit__) 关闭 所需的文件和调用 回调 (这里上传)功能 自动 并通过 文件路径 刚刚关闭。在这个回调函数中，我们可以执行任何操作，例如 复印 .
非上下文管理器案例也类似，但不同之处在于 __ getattr __功能是创造魔法的。
代码执行后文件内容如下:
from_file1.txt

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