shipany.bot.contrib.apschedulers

shipany.bot.contrib.apschedulers package

Submodules

shipany.bot.contrib.apschedulers.bindings module

shipany.bot.contrib.apschedulers.bindings.default_bindings(binder)

Bind the default dependencies.

• Parameters: binder (Binder) — The dependency injector binder.
• Return type: None

shipany.bot.contrib.apschedulers.server module

class shipany.bot.contrib.apschedulers.server.ApschedulerScheduleServer

Bases: ScheduleServer

Creates a default AsyncIOScheduler instance.

Once created, you can use the scheduler to add jobs. The instance will be automatically injected into the function when using the @inject.autoparams() decorator.

@inject.autoparams()
async def run_task(scheduler: ScheduleServer) -> None:
  scheduler.add_job("shipany.bot.jobs.webhook:triggered", args=[webhook, body])

• Parameters:
  • default_jobstore (Mapping *[*str , t.Any ] , optional) — Description of the default jobstore. If not provided, the local sqlite database will be used. Defaults to None.
  • default_executor (Mapping *[*str , t.Any ] , optional) — Description of default executor. If not provided, asyncio will be used. Defaults to None.
  • job_defaults (Mapping *[*str , t.Any ] , optional) — Job default settings. Defaults to None.

__init__(*, default_jobstore=None, default_executor=None, job_defaults=None)

• Parameters:
  • self (Self)
  • default_jobstore (Mapping *[*str , Any ] | None)
  • default_executor (Mapping *[*str , Any ] | None)
  • job_defaults (Mapping *[*str , Any ] | None)
• Return type: None

async serve(flow, config=None)

Start the scheduler.

• Parameters:
  • flow (Flow) — the flow to be used by the server
  • config (ApplicationConfig , optional) — the application configuration. Defaults to None.
  • self (Self)
• Return type: None

add_job(*args, **kwargs)

Add a job to the scheduler.

• Parameters:
  • args — positional arguments to be passed to the job
  • kwargs — keyword arguments to be passed to the job
  • self (Self)
• Return type: None

Module contents