Skip to main content

StreamFinalizer

ClassPrivateContextSource

with StreamFinalizer(
fobj: IO[Any]@StreamFinalizer,
truncate: bool = False,
close: bool = False,
callback_on_exit: Callable[[_IO], None] | None = None
) as _fobj:
_fobj: IO[Any]@StreamFinalizer

Finalizer for a stream IO when the stream has been delivered.

This class will be used by ServiceData for safely maintain the file object while serving the file stream.

Arguments

ArgumentTypeRequired
Description
fobjIO[Any]The file-like object to be managed whent the exiting the finalizer environment.
truncateboolA flag. If set, will trucate the file when exiting the context. This flag needs cannot be used when the IO is read-only.
closeboolA flag. If set, will close the file handle when exiting the context
callback_on_exitCallable[[_IO], None] | NoneThe callback function that will be called when exiting from the context. This method will be called before the file is truncated or closed.

Examples

Automatically close the file when exiting from the context

import os
from dash_file_cache.utilities import StreamFinalizer, TempDir


temp_dir = TempDir()
fobj = open(os.path.join(temp_dir.path, "test.txt"), "w")
with StreamFinalizer(fobj, close=True) as _fobj:
_fobj.write("test")

print(fobj.closed)

Automatically cut the file to an empty file when exiting from the context

import os
from dash_file_cache.utilities import StreamFinalizer, TempDir


temp_dir = TempDir()
fobj = open(os.path.join(temp_dir.path, "test.txt"), "w")
with StreamFinalizer(fobj, truncate=True) as _fobj:
_fobj.write("test")

print(fobj.closed)
if not fobj.closed:
fobj.seek(0, os.SEEK_END)
print(fobj.tell())
fobj.close()