SpoolWatcher

Announces each document that lands whole in the spool.

A document renamed into place or closed after being written there is whole at that event. One that only appears, moved in from another directory, is announced once it has sat unopened for SETTLE_SECONDS, which is what tells it apart from a file still being written.

Attributes

attributespool

= spool

attributeloop

= loop

attributeon_document

= on_document

attributeobserver

= Observer()

attributeon_modified

= on_opened

Functions

constructor__init__(spool, loop, on_document) -> None

Source Code

def __init__(self, spool: str, loop: asyncio.AbstractEventLoop,
             on_document: Callable[[str, int], Awaitable[None]]) -> None:
    self.spool = spool
    self.loop = loop
    self.on_document = on_document
    self.observer = Observer()
    self._appeared: Dict[str, asyncio.TimerHandle] = {}

paramspoolstr

paramloopasyncio.AbstractEventLoop

paramon_documentCallable[[str, int], Awaitable[None]]

Returns

None

funcstart() -> None

Source Code

def start(self) -> None:
    os.makedirs(self.spool, exist_ok=True)
    self.observer.schedule(self, self.spool)
    self.observer.start()

Returns

None

funcstop() -> None

Source Code

def stop(self) -> None:
    self.observer.stop()
    if self.observer.is_alive():
        self.observer.join(timeout=2)

Returns

None

funcon_moved(event) -> None

Source Code

def on_moved(self, event: Any) -> None:
    if not event.is_directory and os.path.dirname(event.dest_path) == self.spool:
        self.loop.call_soon_threadsafe(self._announce, os.path.basename(event.dest_path))

parameventAny

Returns

None

funcon_closed(event) -> None

Source Code

def on_closed(self, event: Any) -> None:
    if not event.is_directory:
        self.loop.call_soon_threadsafe(self._announce, os.path.basename(event.src_path))

parameventAny

Returns

None

funcon_created(event) -> None

Source Code

def on_created(self, event: Any) -> None:
    if not event.is_directory:
        self.loop.call_soon_threadsafe(self._appeared_whole, os.path.basename(event.src_path))

parameventAny

Returns

None

funcon_opened(event) -> None

Source Code

def on_opened(self, event: Any) -> None:
    if not event.is_directory:
        self.loop.call_soon_threadsafe(self._being_written, os.path.basename(event.src_path))

parameventAny

Returns

None

func_appeared_whole(name) -> None

Source Code

def _appeared_whole(self, name: str) -> None:
    self._being_written(name)
    self._appeared[name] = self.loop.call_later(SETTLE_SECONDS, self._announce, name)

paramnamestr

Returns

None

func_being_written(name) -> None

Source Code

def _being_written(self, name: str) -> None:
    timer = self._appeared.pop(name, None)
    if timer is not None:
        timer.cancel()

paramnamestr

Returns

None

func_announce(name) -> None

Source Code

def _announce(self, name: str) -> None:
    self._being_written(name)
    if document_name(name) is None:
        return
    try:
        size = os.path.getsize(os.path.join(self.spool, name))
    except OSError:
        return
    self.loop.create_task(self.on_document(name, size))

paramnamestr

Returns

None

Attributes

Functions

On this page