1. import os
  2. import time
  3. import threading
  6. def extend_file_logic(filepath, task_function, timeout=5):
  7. """
  8. Executes a task on a file, handling timeouts and errors with defensive checks.
  10. Args:
  11. filepath (str): The path to the file.
  12. task_function (callable): The function to execute on the file.
  13. timeout (int): Timeout in seconds for the task.
  15. Returns:
  16. bool: True if the task completed successfully, False otherwise.
  17. """
  19. if not os.path.exists(filepath):
  20. print(f"Error: File not found at {filepath}")
  21. return False
  23. if not os.access(filepath, os.R_OK):
  24. print(f"Error: File is not readable at {filepath}")
  25. return False
  27. def worker():
  28. try:
  29. start_time = time.time()
  30. result = task_function(filepath)
  31. end_time = time.time()
  32. if (end_time - start_time) > timeout:
  33. print(f"Warning: Task timed out after {timeout} seconds for {filepath}")
  34. return False
  35. return True
  36. except Exception as e:
  37. print(f"Error executing task on {filepath}: {e}")
  38. return False
  40. # Use a thread to avoid blocking the main program
  41. thread = threading.Thread(target=worker)
  42. thread.start()
  43. thread.join(timeout) # Wait for the thread to complete, with timeout
  45. return thread.is_alive() == False # Check if the thread is still running after the timeout
  47. if __name__ == '__main__':
  48. def my_task(filepath):
  49. """Example task: Adds a line to the file."""
  50. try:
  51. with open(filepath, "a") as f:
  52. f.write(f"Task executed at {time.ctime()}\n")
  53. return True
  54. except Exception as e:
  55. print(f"Error in my_task: {e}")
  56. return False
  58. # Example usage:
  59. filepath = "my_file.txt"
  60. if extend_file_logic(filepath, my_task):
  61. print(f"Task completed successfully for {filepath}")
  62. else:
  63. print(f"Task failed or timed out for {filepath}")

Add your comment