coxbuild.events package

Submodules

coxbuild.events.datetime module

async coxbuild.events.datetime.attime(dt: datetime.datetime | datetime.date | datetime.time)

coxbuild.events.datetime.nextWeek()

coxbuild.events.datetime.tomorrow()

async coxbuild.events.datetime.weekdays(*weekdays: int)

coxbuild.events.filesystems module

class coxbuild.events.filesystems.DirectoryEntry(path: pathlib.Path, creation: datetime.datetime, modification: datetime.datetime, access: datetime.datetime)

Bases: coxbuild.events.filesystems.FileSystemEntry

class coxbuild.events.filesystems.FileEntry(path: pathlib.Path, creation: datetime.datetime, modification: datetime.datetime, access: datetime.datetime)

Bases: coxbuild.events.filesystems.FileSystemEntry

class coxbuild.events.filesystems.FileSystemChangeType(value)

Bases: enum.Enum

An enumeration.

Access = 2

Create = 0

Delete = 3

Modify = 1

class coxbuild.events.filesystems.FileSystemEntry(path: pathlib.Path, creation: datetime.datetime, modification: datetime.datetime, access: datetime.datetime)

Bases: object

access: datetime.datetime

creation: datetime.datetime

id()

modification: datetime.datetime

path: pathlib.Path

class coxbuild.events.filesystems.FileSystemSnapshot(path: pathlib.Path, glob: Optional[str] = None)

Bases: object

coxbuild.events.filesystems.access(path: Optional[pathlib.Path] = None, glob: Optional[str] = None, period: Optional[datetime.timedelta] = None)

async coxbuild.events.filesystems.changed(path: Optional[pathlib.Path] = None, glob: Optional[str] = None, type: Optional[set[coxbuild.events.filesystems.FileSystemChangeType]] = None, period: Optional[datetime.timedelta] = None)

Detect file or directory change (create, delete, modify, access).

Provide event context:

type: Type of the change, in FileSystemChangeType entry: Changed entry

path: path to watch glob: glob pattern to watch type: change type to watch

coxbuild.events.filesystems.create(path: Optional[pathlib.Path] = None, glob: Optional[str] = None, period: Optional[datetime.timedelta] = None)

coxbuild.events.filesystems.delete(path: Optional[pathlib.Path] = None, glob: Optional[str] = None, period: Optional[datetime.timedelta] = None)

coxbuild.events.filesystems.diff(old: coxbuild.events.filesystems.FileSystemSnapshot, new: coxbuild.events.filesystems.FileSystemSnapshot)

coxbuild.events.filesystems.modify(path: Optional[pathlib.Path] = None, glob: Optional[str] = None, period: Optional[datetime.timedelta] = None)

Module contents

coxbuild.events.delay(duration: datetime.timedelta)

Delay occur.

coxbuild.events.forever(eventBuilder: Callable[[], AsyncIterator[coxbuild.services.EventContext | None]])

Repeat forever

async coxbuild.events.limit(event: AsyncIterator[coxbuild.services.EventContext | None], number: int = 1)

Limit event occur times.

number: limit times, 0 for no-occur, positive integer for finite occur, negative integer for infinite occur

coxbuild.events.now()

Occur immediately.

async coxbuild.events.occur(event: AsyncIterator[coxbuild.services.EventContext | None])

Create an awaitable for occuring event once.

coxbuild.events.once(event: AsyncIterator[coxbuild.services.EventContext | None])

Occur event once.

coxbuild.events.onceevent(f: Callable[[P], Awaitable[coxbuild.services.EventContext | None]]) → Callable[[P], AsyncIterator[coxbuild.services.EventContext | None]]

Wrap an awaitable function into event function.

coxbuild.events.periodic(period: datetime.timedelta)

Occur forever in an interval.

async coxbuild.events.repeat(eventBuilder: Callable[[], AsyncIterator[coxbuild.services.EventContext | None]], repeat: int = 0, continueOnError: bool = False)

Repeat event.

eventBuilder: a callable that return event repeat: repeat times, 0 for no-repeat, positive integer for finite repeat, negative integer for infinite repeat continueOnError: continue repeat when error