luigi_erroneous_atomiclocalfile_excep_behavior.py

import os
import luigi.target

class ExampleUserError(Exception):
  pass

class TestAtomicLocalFileExcepBehavior(luigi.target.AtomicLocalFile):
  def move_to_final_destination(self):
    print("!!!ERROR File Move Requested. Should not happen this demo. %s" % self.path)

  def __del__(self):
    print("DBG: Subclass __del__() called")
    super(TestAtomicLocalFileExcepBehavior, self).__del__()


    
alf = TestAtomicLocalFileExcepBehavior('finaldest_foo')
tmp_path = alf.tmp_path
try:
  with alf:
    alf.write(b'...')
    raise ExampleUserError("Pre-close excep from user code...")
except ExampleUserError:
  print("Example exception caught outside context manager (expected).")
del alf
if os.path.exists(tmp_path):
  print("!!!ERROR: Tmp file not removed! %s" % tmp_path)
else:
  print("DBG: Tmp file removed. %s" % tmp_path)