Decorators

@walnats.decorators.filter_time(year: int | Iterable[int] | None = None, month: int | Iterable[int] | None = None, day: int | Iterable[int] | None = None, hour: int | Iterable[int] | None = None, minute: int | Iterable[int] | None = None) → None

Run the handler only if the input time matches the pattern.

The decorator is a companion for walnats.Clock in situations when an actor doesn’t need to run every minute but you don’t have enough actors to create a separate clock just for them.

The idea is similar to how cron patterns work. The arguments form the pattern to match. If an argument is not specified, any value is accepted. If one or multiple values specified, any of the given values should match.

Run the handler once a day at midnight:

@walnats.decorators.filter_time(hour=0, minute=0)
async def create_backup(dt):
    ...

Run the handler every 5 minutes:

@walnats.decorators.filter_time(minute=range(0, 60, 5))
async def send_pulse(dt):
    ...
@walnats.decorators.rate_limit(max_jobs: int, period: float) → None

Limit how many jobs at most can be started per time interval.

The count starts as soon as the handler starts. For example, if you limit 10 jobs per 60 seconds, 11th job will wait while 60 seconds pass since the first jobs has been started before starting itself.

@walnats.decorators.rate_limit(32, 60)
async def send_sms(phone_number: int) -> None:
    ...
Parameters:

  • max_jobs – the jobs limit in the time interval.

  • period – the time interval (in seconds).

@walnats.decorators.require(predicate: Callable[[], bool], pause: float = 0) → None

Require the predicate to be true before the handler can be called.

If the predicate is not true, the handler invocation will be delayed. Useful for waiting for dependencies. For example, to make sure the handler does not start before a connection to the database is established.

@walnats.decorators.require(db.is_connected)
async def write_audit_log(record: LogRecord) -> None:
    await db.save_record(record)
Parameters:

  • predicate – callable that indicates if the handler can be called.

  • pause – delay (in seconds) between each check of the predicate.

@walnats.decorators.suppress(exc: type[BaseException], *excs: type[BaseException], logger: logging.Logger | logging.LoggerAdapter | None = <Logger walnats.decorators (WARNING)>) → None

Ignore specified exceptions.

Useful for avoiding retries for errors that cannot be retried. Avoid using it for flow control.

@walnats.decorators.suppress(NotFoundError)
async def update_parcel_status(event: ParcelDiff) -> None:
    ...
Parameters:

  • excs – exception types to suppress.

  • logger – logger to use to log suppressed exceptions. If None, no logs will be written, the exception will be just silently discarded.