### Implement a Custom Data Store Source: https://github.com/agronholm/apscheduler/blob/master/docs/extending.md Inherit from DataStore and implement the start method to initialize the store and store the event broker. ```python from contextlib import AsyncExitStack from apscheduler.abc import DataStore, EventBroker class MyCustomDataStore(DataStore): _event_broker: EventBroker async def start(self, exit_stack: AsyncExitStack, event_broker: EventBroker) -> None: # Save the event broker in a member attribute and initialize the store self._event_broker = event_broker # See the interface class for the rest of the abstract methods ``` -------------------------------- ### Install APScheduler with extras Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Command to install APScheduler along with specific optional dependencies. ```bash pip install apscheduler[psycopg,sqlalchemy] ``` -------------------------------- ### Install APScheduler via pip Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Standard installation command for the APScheduler package. ```bash $ pip install apscheduler ``` -------------------------------- ### Start Docker Compose Services for Testing Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Starts the necessary external services (like databases) for running the APScheduler test suite using Docker Compose. Ensure Docker and Docker Compose are installed. ```bash docker compose up -d ``` -------------------------------- ### Run WSGI application with uWSGI Source: https://github.com/agronholm/apscheduler/blob/master/docs/integrations.md Start the application with uWSGI, ensuring the --enable-threads flag is set to allow the scheduler to function. ```bash uwsgi --enable-threads --http :8080 --wsgi-file example.py ``` -------------------------------- ### Integrate APScheduler with WSGI Source: https://github.com/agronholm/apscheduler/blob/master/docs/integrations.md Use the synchronous scheduler and start it as a side effect of importing the module. Ensure the server is configured to support threads. ```python from apscheduler import Scheduler def app(environ, start_response): """Trivial example of a WSGI application.""" response_body = b"Hello, World!" response_headers = [ ("Content-Type", "text/plain"), ("Content-Length", str(len(response_body))), ] start_response(200, response_headers) return [response_body] scheduler = Scheduler() scheduler.start_in_background() ``` -------------------------------- ### Run Scheduler Synchronously (Background Thread - WSGI Alternative) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md An alternative method for running the scheduler in a background thread, often used in WSGI applications. This approach initializes the scheduler separately before starting it in the background. ```python from apscheduler import Scheduler scheduler = Scheduler() # Add schedules, configure tasks here scheduler.start_in_background() ``` -------------------------------- ### Set up Development Environment and Run Pytest Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Configures a local development environment for APScheduler by creating a virtual environment, activating it, installing the project in editable mode with test dependencies, and then running Pytest directly. Activation command varies by OS. ```bash python -m venv venv source venv/bin/activate pip install --group test -e . pytest ``` -------------------------------- ### Install Pre-commit Hooks Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Installs pre-commit hooks for the APScheduler repository. These hooks perform code style and quality checks locally before commits, ensuring consistency with GitHub's checks. ```bash pre-commit install ``` -------------------------------- ### Implement a Basic Thread Job Executor Source: https://github.com/agronholm/apscheduler/blob/master/docs/extending.md Create a JobExecutor subclass to handle job execution using a thread pool. The run_job() method must call the provided function with the job's arguments and keyword arguments, returning the result. This example uses anyio.to_thread.run_sync for running synchronous callables in a separate thread. ```python from contextlib import AsyncExitStack from functools import partial from anyio import to_thread from apscheduler import Job from apscheduler.abc import JobExecutor class ThreadJobExecutor(JobExecutor): async def run_job(self, func: Callable[..., Any], job: Job) -> Any: wrapped = partial(func, *job.args, **job.kwargs) return await to_thread.run_sync(wrapped) ``` -------------------------------- ### Implement a Thread Job Executor with Capacity Limiter Source: https://github.com/agronholm/apscheduler/blob/master/docs/extending.md Extend the ThreadJobExecutor by overriding the start() method to initialize an AnyIO CapacityLimiter. This allows controlling the maximum number of concurrent threads used for job execution. The run_job() method then uses this limiter when running jobs. ```python from contextlib import AsyncExitStack from functools import partial from anyio import CapacityLimiter, to_thread from apscheduler import Job from apscheduler.abc import JobExecutor class ThreadJobExecutor(JobExecutor): _limiter: CapacityLimiter def __init__(self, max_threads: int): self.max_threads = max_threads async def start(self, exit_stack: AsyncExitStack) -> None: self._limiter = CapacityLimiter(self.max_workers) async def run_job(self, func: Callable[..., Any], job: Job) -> Any: wrapped = partial(func, *job.args, **job.kwargs) return await to_thread.run_sync(wrapped, limiter=self._limiter) ``` -------------------------------- ### Enable DEBUG logging for APScheduler Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Increase the logging level for the 'apscheduler' logger to DEBUG to get more detailed information. This is useful for troubleshooting. Ensure basic logging is configured first. ```python import logging logging.basicConfig() logging.getLogger('apscheduler').setLevel(logging.DEBUG) ``` -------------------------------- ### Run ASGI application with Hypercorn or Uvicorn Source: https://github.com/agronholm/apscheduler/blob/master/docs/integrations.md Execute the ASGI application using Hypercorn or Uvicorn servers. ```bash hypercorn example:scheduler_middleware ``` ```bash uvicorn example:scheduler_middleware ``` -------------------------------- ### Configure Task Inheritance Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Illustrates the lookup order for task parameters, showing how settings from the scheduler, decorator, and configure_task method are merged. ```python from apscheduler import Scheduler, TaskDefaults, task @task(max_running_jobs=3, metadata={"foo": ["taskfunc"]}) def mytaskfunc(): print("running stuff") task_defaults = TaskDefaults( misfire_grace_time=15, job_executor="processpool", metadata={"global": 3, "foo": ["bar"]} ) with Scheduler(task_defaults=task_defaults) as scheduler: scheduler.configure_task( "sometask", func=mytaskfunc, job_executor="threadpool", metadata={"direct": True} ) ``` -------------------------------- ### Build APScheduler Documentation Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Builds the project's documentation using Tox with the 'docs' environment. The generated HTML documentation will be located in the 'build/sphinx/html' directory. ```bash tox -e docs ``` -------------------------------- ### Integrate APScheduler with ASGI Source: https://github.com/agronholm/apscheduler/blob/master/docs/integrations.md Use the asynchronous scheduler and tie its lifespan to the application by wrapping it in middleware. ```python from apscheduler import AsyncScheduler async def app(scope, receive, send): """Trivial example of an ASGI application.""" if scope["type"] == "http": await receive() await send( { "type": "http.response.start", "status": 200, "headers": [ [b"content-type", b"text/plain"], ], } ) await send( { "type": "http.response.body", "body": b"Hello, world!", "more_body": False, } ) elif scope["type"] == "lifespan": while True: message = await receive() if message["type"] == "lifespan.startup": await send({"type": "lifespan.startup.complete"}) elif message["type"] == "lifespan.shutdown": await send({"type": "lifespan.shutdown.complete"}) return async def scheduler_middleware(scope, receive, send): if scope['type'] == 'lifespan': async with AsyncScheduler() as scheduler: await app(scope, receive, send) else: await app(scope, receive, send) ``` -------------------------------- ### Support Classes Source: https://github.com/agronholm/apscheduler/blob/master/docs/api.md Support classes for handling unset options and sentinel values. ```APIDOC ## Support classes for unset options ### `apscheduler.unset` * **Description**: Sentinel value for unset option values. ``` -------------------------------- ### Run Scheduler Synchronously (Foreground) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Use this method to run the scheduler in the foreground. The script will block until the scheduler is explicitly stopped. ```python from apscheduler import Scheduler with Scheduler() as scheduler: # Add schedules, configure tasks here scheduler.run_until_stopped() ``` -------------------------------- ### Subscribe to Scheduler Events (Synchronous) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Listen for specific scheduler events like job acquisition and release using a synchronous callback function. Ensure the scheduler supports synchronous callbacks. ```python from apscheduler import Event, JobAcquired, JobReleased def listener(event: Event) -> None: print(f"Received {event.__class__.__name__}") scheduler.subscribe(listener, {JobAcquired, JobReleased}) ``` -------------------------------- ### Subscribe to Scheduler Events (Asynchronous) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Listen for scheduler events using an asynchronous callback function. Asynchronous schedulers support both sync and async callbacks. ```python from apscheduler import Event, JobAcquired, JobReleased async def listener(event: Event) -> None: print(f"Received {event.__class__.__name__}") scheduler.subscribe(listener, {JobAcquired, JobReleased}) ``` -------------------------------- ### Context Variables Source: https://github.com/agronholm/apscheduler/blob/master/docs/api.md Context variables provide access to scheduler and job information within the execution context. ```APIDOC ## Context Variables ### `apscheduler.current_scheduler` * **Type**: `ContextVar[Scheduler]` * **Description**: The current scheduler instance. ### `apscheduler.current_async_scheduler` * **Type**: `ContextVar[AsyncScheduler]` * **Description**: The current asynchronous scheduler instance. ### `apscheduler.current_job` * **Type**: `ContextVar[Job]` * **Description**: The job being currently run (available when running the job’s target callable). ``` -------------------------------- ### Run Scheduler Synchronously (Background Thread) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md This is the preferred method for running the scheduler in a background thread. It allows the main script to continue execution while the scheduler runs concurrently. ```python from apscheduler import Scheduler with Scheduler() as scheduler: # Add schedules, configure tasks here scheduler.start_in_background() ``` -------------------------------- ### Implement a Custom Trigger Class Source: https://github.com/agronholm/apscheduler/blob/master/docs/extending.md Create a class inheriting from apscheduler.abc.Trigger to implement custom scheduling logic. Ensure the next() method returns a timezone-aware datetime or None, and never returns the same or an earlier datetime. The __getstate__() and __setstate__() methods must handle serialization and deserialization of the trigger's state. ```python from __future__ import annotations from apscheduler.abc import Trigger class MyCustomTrigger(Trigger): def next() -> datetime | None: ... # Your custom logic here def __getstate__(): ... # Return the serializable state here def __setstate__(state): ... # Restore the state from the return value of __getstate__() ``` -------------------------------- ### Run Async Scheduler Asynchronously (Foreground) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Use this method to run an asynchronous scheduler in the foreground within an asyncio event loop. The `run_until_stopped()` coroutine will block until the scheduler is stopped. ```python import asyncio from apscheduler import AsyncScheduler async def main(): async with AsyncScheduler() as scheduler: # Add schedules, configure tasks here await scheduler.run_until_stopped() asyncio.run(main()) ``` -------------------------------- ### Create and Checkout a New Git Branch Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Creates a new Git branch for your feature or fix and checks it out, allowing you to make isolated changes. Replace 'myfixname' with a descriptive branch name. ```bash git checkout -b myfixname ``` -------------------------------- ### Run Async Scheduler Asynchronously (Background Task) Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md This method runs an asynchronous scheduler as a background task within an asyncio event loop. The `start_in_background()` coroutine allows the main asyncio application to continue running. ```python import asyncio from apscheduler import AsyncScheduler async def main(): async with AsyncScheduler() as scheduler: # Add schedules, configure tasks here await scheduler.start_in_background() asyncio.run(main()) ``` -------------------------------- ### Clone APScheduler Repository Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Clones a forked APScheduler repository to your local machine using Git. Replace 'yourusername' with your actual GitHub username. ```bash git clone git@github.com/yourusername/apscheduler ``` -------------------------------- ### Run APScheduler Tests with Tox Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Executes the APScheduler test suite using Tox, which manages dependencies and virtual environments for different Python versions. Tox can pass arguments to pytest using '--'. ```bash tox tox -- -k somekeyword ``` -------------------------------- ### Combine triggers with AndTrigger Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Use AndTrigger to ensure a task only runs when multiple triggers coincide. It iterates until a matching run time is found or a trigger is exhausted. ```python from apscheduler.triggers.calendarinterval import CalendarIntervalTrigger from apscheduler.triggers.combining import AndTrigger from apscheduler.triggers.cron import CronTrigger trigger = AndTrigger( CalendarIntervalTrigger(months=2, hour=10), CronTrigger(day_of_week="mon-fri", hour=10), ) ``` -------------------------------- ### Combine triggers with OrTrigger Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Use OrTrigger to combine multiple triggers when a task needs to run under different conditions. It returns the earliest run time from the provided triggers. ```python from apscheduler.triggers.combining import OrTrigger from apscheduler.triggers.cron import CronTrigger trigger = OrTrigger( CronTrigger(day_of_week="mon-fri", hour=10), CronTrigger(day_of_week="sat-sun", hour=11), ) ``` -------------------------------- ### Access Current Job Information Source: https://github.com/agronholm/apscheduler/blob/master/docs/userguide.md Retrieve the ID and schedule ID of the currently executing job within a task function. Ensure the job is spawned from a schedule to access `schedule_id`. ```python from apscheduler import current_job def my_task_function(): job_info = current_job.get().id print( f"This is job {job_info.id} and was spawned from schedule " f"{job_info.schedule_id}" ) ``` -------------------------------- ### Push Changes to Forked Repository Source: https://github.com/agronholm/apscheduler/blob/master/docs/contributing.md Pushes your committed local changes to your forked APScheduler repository on GitHub. This makes your changes available for creating a pull request. ```bash git push ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.