easyfilewatcher package

Subpackages

• easyfilewatcher.adapters package
  • Submodules
  • easyfilewatcher.adapters.ORM module
    • init_tables()
  • easyfilewatcher.adapters.Repository module
    • EasyFileWatcherUnitRepository
      • EasyFileWatcherUnitRepository.add()
      • EasyFileWatcherUnitRepository.add_all()
      • EasyFileWatcherUnitRepository.all()
      • EasyFileWatcherUnitRepository.delete()
      • EasyFileWatcherUnitRepository.get_all_by_id()
  • Module contents
• easyfilewatcher.domain package
  • Submodules
  • easyfilewatcher.domain.EasyFileWatcherUnit module
    • EasyFileWatcherUnit
      • EasyFileWatcherUnit.directory_watcher_id
      • EasyFileWatcherUnit.filepath
      • EasyFileWatcherUnit.last_modification
      • EasyFileWatcherUnit.size
  • Module contents
• easyfilewatcher.utils package
  • Submodules
  • easyfilewatcher.utils.config module
  • Module contents

Submodules

easyfilewatcher.EasyFileWatcher module

class easyfilewatcher.EasyFileWatcher.EasyFileWatcher(jobstore: Optional[str] = None)

Bases: object

add_directory_to_watch(directory_path: str, callback: function, start_date: Optional[datetime] = datetime.datetime(2022, 12, 7, 20, 8, 54, 522934), end_date: Optional[datetime] = None, callback_param: Optional[dict] = None, directory_watcher_id: Optional[str] = None, event_on_deletion: Optional[bool] = True, polling_time: Optional[int] = 2) → None

This method registers the directory of interest to watch. It further requires the function signature. Parameters to the function can be passed as a dictionary. Passing a directory_watcher_id is highly recommended as it can be used to delete a certain FileWatcher.

param str directory_path

path of directory to watch

param func callback

custom user to function to be executed at change in directory

param Optional[dict] callback_param

parameters to be passed to callback function

param Optional[str] directory_watcher_id

assigned ID of watcher

param Optional[bool] event_on_deletion

shall an event be triggered at deletion of File

param Optional[datetime] start_date

When shall File Watcher starts tracking

param Optional[datetime] end_date

When shall File Watcher end tracking

param Optional[int] polling_time

Interval Watcher checks on specified directory

returns

List of EasyFileWatcherUnits

rtype

List[EasyFileWatcherUnit]

delete_easy_file_watcher(directory_watcher_id) → bool

This method deletes an EasyFileWatcher. It returns True at success and False at failure. :returns: None :rtype: None

static execute_job(*args)

This method executes the call back of the directory of interest to watch.

get_directory_watcher_jobs() → list

This method return a list of current EasyFileWatcher Jobs. Good way to retrieve DirectoryWatcherID if not initially provided. :returns: list of EasyFileWatcher Jobs :rtype: list

pause_file_watching(directory_watcher_id: str) → bool

This method pauses a FileWatcher. :param str directory_watcher_id: assigned ID of watcher :returns: success :rtype: bool

resume_file_watching(directory_watcher_id: str) → bool

This method resumes a FileWatcher if it was paused before. :param str directory_watcher_id: assigned ID of watcher :returns: success :rtype: bool

easyfilewatcher.UoW module

class easyfilewatcher.UoW.EasyFileWatcherUoW(session_factory: ~sqlalchemy.orm.session.sessionmaker = sessionmaker(class_='Session', bind=Engine(sqlite:///easyfilewatcher.sqlite), autoflush=True, autocommit=False, expire_on_commit=False))

Bases: object

commit()

rollback()

Module contents