### Installing APScheduler via Pip (Shell)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

This command installs the base APScheduler library using Python's package installer, pip. It is the standard method for getting the library. Requires pip to be installed.

```shell
$ pip install apscheduler
```

--------------------------------

### Installing APScheduler with Extras (Shell)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

This command installs APScheduler along with optional dependencies for specific features, such as using Psycopg for event brokering and SQLAlchemy for data storage. Multiple extras can be specified comma-separated within brackets. Requires pip and Python.

```shell
pip install apscheduler[psycopg,sqlalchemy]
```

--------------------------------

### Starting All Docker Services with Docker Compose

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/examples/README.rst

Starts all required background services defined in the Docker Compose configuration in detached mode (-d). This command is used to set up the environment needed by various APScheduler examples.

```bash
docker compose up -d
```

--------------------------------

### Starting Specific Docker Service with Docker Compose

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/examples/README.rst

Starts only a specified background service (e.g., 'postgresql') defined in the Docker Compose configuration in detached mode (-d). Allows selective setup of required services for specific examples.

```bash
docker compose up -d postgresql
```

--------------------------------

### Install APScheduler Core Library Shell

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/userguide.rst

This command installs the core APScheduler library using pip. It requires pip to be installed on the system. This is the minimum required installation to use the basic features.

```Shell
$ pip install apscheduler
```

--------------------------------

### Install APScheduler with Extras Shell

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/userguide.rst

This command installs APScheduler along with specified extra dependencies for additional features like different data stores or event brokers. The extras are provided as a comma-separated list in brackets after the package name, ensuring compatible versions.

```Shell
pip install apscheduler[psycopg,sqlalchemy]
```

--------------------------------

### Setting up Scheduler and Jobs - Raw WSGI - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

Configures and starts the APScheduler within the raw WSGI application context. It creates a synchronous SQLAlchemy database engine (using a MySQL driver in this specific example), configures an SQLAlchemyDataStore, sets up a RedisEventBroker, creates a synchronous Scheduler instance, adds the 'tick' schedule to run every second, and immediately starts the scheduler in the background.

```Python
engine = create_engine("mysql+pymysql://root:secret@localhost/testdb")
data_store = SQLAlchemyDataStore(engine)
event_broker = RedisEventBroker("redis://localhost")
scheduler = Scheduler(data_store, event_broker)
scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")
scheduler.start_in_background()
```

--------------------------------

### Running ASGI Example with Hypercorn or Uvicorn

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/integrations.rst

These bash commands show how to run the provided ASGI application example using either the Hypercorn or Uvicorn ASGI servers. Both commands point the server to the `scheduler_middleware` entry point in the `example` module, allowing the middleware to handle the application's lifespan and the integrated scheduler.

```Bash
hypercorn example:scheduler_middleware
```

```Bash
uvicorn example:scheduler_middleware
```

--------------------------------

### Starting Test Services with Docker Compose (Bash)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/contributing.rst

This command uses Docker Compose to start the necessary background services (like databases) required by the APScheduler test suite in detached mode. Ensure Docker and Docker Compose (v2 plugin) are installed.

```bash
docker compose up -d
```

--------------------------------

### Initializing AsyncScheduler with SQLAlchemy PostgreSQL Store (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (12).txt

Initializes the APScheduler `AsyncScheduler` configured to use a persistent SQLAlchemy data store connected to a PostgreSQL database. This example requires the 'postgresql' service to be running and the `sqlalchemy` and `asyncpg` prerequisites installed. It sets up an async engine using `create_async_engine` with a PostgreSQL connection string, creates an `SQLAlchemyDataStore` instance, and passes it to the scheduler, demonstrating database persistence by scheduling the `tick` function with a specific ID.

```python
from __future__ import annotations

from asyncio import run
from datetime import datetime

from sqlalchemy.ext.asyncio import create_async_engine

from apscheduler import AsyncScheduler
from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.triggers.interval import IntervalTrigger


def tick():
    print("Hello, the time is", datetime.now())


async def main():
    engine = create_async_engine(
        "postgresql+asyncpg://postgres:secret@localhost/testdb"
    )
    data_store = SQLAlchemyDataStore(engine)
    async with AsyncScheduler(data_store) as scheduler:
        await scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")
        await scheduler.run_until_stopped()


run(main())
```

--------------------------------

### Initializing Qt App and Scheduling Job with APScheduler - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (10).txt

This snippet initializes the Qt application and the `MainWindow`. It then sets up an APScheduler instance, configures it to use the `QtJobExecutor` by adding it to the executor map, and schedules the `window.update_time` method to run every second using an `IntervalTrigger`. The scheduler is started in the background, and the Qt event loop (`app.exec()`) is started to keep the application running.

```Python
app = QApplication(sys.argv)
window = MainWindow()
window.show()
with Scheduler() as scheduler:
    scheduler.job_executors["qt"] = QtJobExecutor()
    scheduler.add_schedule(
        window.update_time, IntervalTrigger(seconds=1), job_executor="qt"
    )
    scheduler.start_in_background()
    app.exec()
```

--------------------------------

### Starting Asyncpg Event Broker - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (6).txt

Initializes the AsyncpgEventBroker instance when the scheduler starts. It calls the base class `start`, creates a memory stream for communicating notifications internally, registers the stream sender with the exit stack for cleanup, and starts the background task responsible for listening to PostgreSQL notifications and sending them.

```Python
async def start(self, exit_stack: AsyncExitStack, logger: Logger) -> None:
        await super().start(exit_stack, logger)
        self._send, receive = create_memory_object_stream[str](100)
        await exit_stack.enter_async_context(self._send)
        await self._task_group.start(self._listen_notifications, receive)
```

--------------------------------

### Integrating Synchronous APScheduler with WSGI

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/integrations.rst

This Python snippet demonstrates a basic WSGI application and shows how to initialize and start a synchronous APScheduler instance in the background when the module is imported. This setup is suitable for WSGI environments where the scheduler runs in the same process as the application.

```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()
```

--------------------------------

### Setup Sync SQLAlchemy DataStore and Asyncpg EventBroker

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Configures the necessary components for synchronous APScheduler operation, including creating an async SQLAlchemy engine (used by the AsyncpgEventBroker) connecting to a PostgreSQL database and initializing a SQLAlchemyDataStore and an AsyncpgEventBroker. These components facilitate communication between the scheduler and worker processes.

```python
import logging

from sqlalchemy.ext.asyncio import create_async_engine

from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker

logging.basicConfig(level=logging.INFO)
engine = create_async_engine("postgresql+asyncpg://postgres:secret@localhost/testdb")
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)

# Uncomment the next two lines to use the MQTT event broker instead
# from apscheduler.eventbrokers.mqtt import MQTTEventBroker
# event_broker = MQTTEventBroker()
```

--------------------------------

### Running WSGI Example with uWSGI

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/integrations.rst

This bash command shows how to run the provided WSGI application example using the uWSGI server. It highlights the crucial `--enable-threads` flag required for APScheduler to function correctly within uWSGI by preventing it from disabling Python threads.

```Bash
uwsgi --enable-threads --http :8080 --wsgi-file example.py
```

--------------------------------

### Initializing AsyncScheduler with SQLAlchemy MySQL Store (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (12).txt

Initializes the APScheduler `AsyncScheduler` configured to use a persistent SQLAlchemy data store connected to a MySQL or MariaDB database. It requires the 'mysql' service to be running and the `sqlalchemy` and `asyncmy` prerequisites installed. The code uses `create_async_engine` with a MySQL connection string to set up the engine, creates an `SQLAlchemyDataStore`, and passes it to the scheduler, scheduling the `tick` function every second with a specific ID.

```python
from __future__ import annotations

from asyncio import run
from datetime import datetime

from sqlalchemy.ext.asyncio import create_async_engine

from apscheduler import AsyncScheduler
from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.triggers.interval import IntervalTrigger


def tick():
    print("Hello, the time is", datetime.now())


async def main():
    engine = create_async_engine(
        "mysql+asyncmy://root:secret@localhost/testdb?charset=utf8mb4"
    )
    data_store = SQLAlchemyDataStore(engine)
    async with AsyncScheduler(data_store) as scheduler:
        await scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")
        await scheduler.run_until_stopped()


run(main())
```

--------------------------------

### Setup Async SQLAlchemy DataStore and Asyncpg EventBroker

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Configures the necessary components for asynchronous APScheduler operation, including creating an async SQLAlchemy engine connecting to a PostgreSQL database and initializing a SQLAlchemyDataStore and an AsyncpgEventBroker from the engine. These components are shared between the scheduler and worker processes via the database.

```python
from sqlalchemy.ext.asyncio import create_async_engine

from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker

engine = create_async_engine(
    "postgresql+asyncpg://postgres:secret@localhost/testdb"
)
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)

# Uncomment the next two lines to use the Redis event broker instead
# from apscheduler.eventbrokers.redis import RedisEventBroker
# event_broker = RedisEventBroker.from_url("redis://localhost")
```

--------------------------------

### Configuring APScheduler Tasks (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

Demonstrates how APScheduler parameters are inherited from TaskDefaults, task decorators, and direct configuration via configure_task. Shows defining a task function, setting global defaults, initializing a scheduler, and configuring a specific task, highlighting the metadata merging logic and parameter override order.

```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}
    )
```

--------------------------------

### Subscribing to APScheduler Events (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

Shows how to subscribe a listener function to specific APScheduler events, such as `JobAcquired` and `JobReleased`. The listener function receives an event object containing details about the event, and the example prints the name of the event class received.

```Python
from apscheduler import Event, JobAcquired, JobReleased

def listener(event: Event) -> None:
    print(f"Received {event.__class__.__name__}")

scheduler.subscribe(listener, {JobAcquired, JobReleased})
```

--------------------------------

### Setting up Scheduler and Jobs - Flask - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

Configures and starts the APScheduler within the Flask application context. It creates a synchronous SQLAlchemy database engine, configures an SQLAlchemyDataStore, sets up a RedisEventBroker, creates a synchronous Scheduler instance, adds the 'tick' schedule to run every second, and immediately starts the scheduler in the background.

```Python
@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"


engine = create_engine("postgresql+psycopg://postgres:secret@localhost/testdb")
data_store = SQLAlchemyDataStore(engine)
event_broker = RedisEventBroker("redis://localhost")
scheduler = Scheduler(data_store, event_broker)
scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")
scheduler.start_in_background()
```

--------------------------------

### Initializing SyncScheduler with Memory Store (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (12).txt

Initializes the synchronous APScheduler `Scheduler` using the default in-memory data store. The code uses a context manager (`with Scheduler()`) which handles starting and stopping the scheduler automatically. It adds the `tick` function to be scheduled every 1 second using an `IntervalTrigger` and then calls `run_until_stopped` to keep the scheduler running in the main thread until it receives a stop signal.

```python
from __future__ import annotations

from datetime import datetime

from apscheduler import Scheduler
from apscheduler.triggers.interval import IntervalTrigger


def tick():
    print("Hello, the time is", datetime.now())


with Scheduler() as scheduler:
    scheduler.add_schedule(tick, IntervalTrigger(seconds=1))
    scheduler.run_until_stopped()
```

--------------------------------

### Starting Scheduler in Background (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

Initializes and starts the main scheduler process within a background task group (`_services_task_group`). It calls `run_until_stopped` to manage the scheduler's lifecycle. Requires the scheduler to be initialized but not yet started.

```python
async def start_in_background(self) -> None:
        self._check_initialized()
        await self._services_task_group.start(
            self.run_until_stopped,
            name=f"Scheduler {self.identity!r} main task",
        )
```

--------------------------------

### Getting an Async MongoDB Session

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (5).txt

This asynchronous context manager provides a MongoDB client session. It uses `anyio.to_thread.run_sync` to run the synchronous session start and end operations without blocking the event loop, ensuring proper session management.

```python
@asynccontextmanager
async def _get_session(self) -> AsyncGenerator[ClientSession, None]:
 session = await to_thread.run_sync(self._client.start_session)
 try:
 yield session
 finally:
 await to_thread.run_sync(session.end_session)
```

--------------------------------

### Initializing AsyncScheduler with Memory Store (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (12).txt

Initializes the APScheduler `AsyncScheduler` using the default in-memory data store within an asyncio application context. It defines an asynchronous `main` function that sets up the scheduler, adds a job (`tick`) to run every 1 second using `IntervalTrigger`, and then calls `run_until_stopped` to keep the scheduler running indefinitely. The `asyncio.run` function starts the event loop and executes the `main` coroutine.

```python
from __future__ import annotations

from asyncio import run
from datetime import datetime

from apscheduler import AsyncScheduler
from apscheduler.triggers.interval import IntervalTrigger


def tick():
    print("Hello, the time is", datetime.now())


async def main():
    async with AsyncScheduler() as scheduler:
        await scheduler.add_schedule(tick, IntervalTrigger(seconds=1))
        await scheduler.run_until_stopped()


run(main())
```

--------------------------------

### Running ASGI Application with Uvicorn

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/examples/README.rst

Runs an ASGI (Asynchronous Server Gateway Interface) application using the uvicorn server. The command requires the filename of the application module without the '.py' extension, followed by ':app' to reference the application instance.

```bash
uvicorn <filename_without_py_extension>:app
```

--------------------------------

### Internal Scheduler Main Loop (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

The main asynchronous loop that runs the scheduler. It initializes services, sets context, manages task groups, starts cleanup, schedule processing, and job processing loops based on the scheduler's role. It handles state transitions (starting, started, stopping, stopped) and publishes events. Requires the scheduler to be in the `stopped` state initially.

```python
async def run_until_stopped(
        self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED
    ) -> None:
        """Run the scheduler until explicitly stopped."""
        if self._state is not RunState.stopped:
            raise RuntimeError(
                f'Cannot start the scheduler when it is in the "{self._state}" state'
            )

        self._state = RunState.starting
        async with AsyncExitStack() as exit_stack:
            await self._ensure_services_initialized(exit_stack)

            # Set this scheduler as the current scheduler
            token = current_async_scheduler.set(self)
            exit_stack.callback(current_async_scheduler.reset, token)

            exception: BaseException | None = None
            try:
                async with create_task_group() as task_group:
                    self._scheduler_cancel_scope = task_group.cancel_scope
                    exit_stack.callback(setattr, self, "_scheduler_cancel_scope", None)

                    # Start periodic cleanups
                    if self.cleanup_interval:
                        task_group.start_soon(
                            self._cleanup_loop,
                            name=f"Scheduler {self.identity!r} clean-up loop",
                        )
                        self.logger.debug(
                            "Started internal cleanup loop with interval: %s",
                            self.cleanup_interval,
                        )

                    # Start processing due schedules, if configured to do so
                    if self.role in (SchedulerRole.scheduler, SchedulerRole.both):
                        await task_group.start(
                            self._process_schedules,
                            name=f"Scheduler {self.identity!r} schedule processing loop",
                        )

                    # Start processing due jobs, if configured to do so
                    if self.role in (SchedulerRole.worker, SchedulerRole.both):
                        await task_group.start(
                            self._process_jobs,
                            name=f"Scheduler {self.identity!r} job processing loop",
                        )

                    # Signal that the scheduler has started
                    self._state = RunState.started
                    self.logger.info("Scheduler started")
                    task_status.started()
                    await self.event_broker.publish_local(SchedulerStarted())
            except BaseException as exc:
                exception = exc
                raise
            finally:
                self._state = RunState.stopped

                if not exception or isinstance(exception, get_cancelled_exc_class()):
                    self.logger.info("Scheduler stopped")
                elif isinstance(exception, Exception):
                    self.logger.exception("Scheduler crashed")
                elif exception:
                    self.logger.info(
                        "Scheduler stopped due to %s", exception.__class__.__name__
                    )

                with move_on_after(3, shield=True):
                    await self.event_broker.publish_local(
                        SchedulerStopped(exception=exception)
                    )
```

--------------------------------

### Async Scheduler/Worker Entry Point Configuration

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Configures basic logging to see scheduler output and runs the main asynchronous function. This serves as the entry point for both the asynchronous scheduler and worker scripts.

```python
import asyncio
import logging

# Assuming main() function is defined and data_store/event_broker are configured

logging.basicConfig(level=logging.INFO)
asyncio.run(main())
```

--------------------------------

### Starting Base Event Broker - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (6).txt

Initializes the base event broker's runtime resources when the scheduler starts up. It sets the logger instance, creates an AnyIO task group which manages the execution of asynchronous tasks like event delivery, and initializes a capacity limiter designed to restrict the number of threads used for synchronous operations or callbacks, preventing thread exhaustion.

```Python
async def start(self, exit_stack: AsyncExitStack, logger: Logger) -> None:
        self._logger = logger
        self._task_group = await exit_stack.enter_async_context(create_task_group())
        self._thread_limiter = CapacityLimiter(1)
```

--------------------------------

### Running Synchronous APScheduler in Foreground (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

This Python snippet demonstrates how to initialize and run the synchronous APScheduler in the foreground using a context manager. The run_until_stopped() method blocks execution until the scheduler is stopped, typically used when the scheduler is the main part of the application. Requires the apscheduler library.

```python
from apscheduler import Scheduler

with Scheduler() as scheduler:
    # Add schedules, configure tasks here
    scheduler.run_until_stopped()
```

--------------------------------

### Configuring APScheduler Logging Level - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

This Python snippet sets up basic logging and increases the logging level specifically for the 'apscheduler' logger to DEBUG. This is useful for troubleshooting and gaining detailed insights into the scheduler's internal operations and potential issues. It requires the built-in `logging` module.

```python
import logging

logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
```

--------------------------------

### Running APScheduler Separate Scheduler Script (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/examples/README.rst

Executes the Python script responsible for running the APScheduler scheduler process in the separate worker/scheduler example. This script is configured to add and update scheduled jobs which workers will then pick up.

```bash
python sync_scheduler.py
```

--------------------------------

### Getting All Schedules in APScheduler Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

Retrieves a list of all schedules currently managed by the scheduler. It ensures the necessary services are initialized and then delegates the call to the asynchronous scheduler's `get_schedules` method.

```Python
def get_schedules(self) -> list[Schedule]:
    portal = self._ensure_services_ready()
    return portal.call(self._async_scheduler.get_schedules)
```

--------------------------------

### Starting Scheduler in Background Thread Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

Launches the APScheduler instance in a separate background thread. This method includes checks for environments like uWSGI that might disable threads and registers atexit hooks for clean shutdown.

```Python
def start_in_background(self) -> None:
    """
    Launch the scheduler in a new thread.

    This method registers :mod:`atexit` hooks to shut down the scheduler and wait
    for the thread to finish.

    :raises RuntimeError: if the scheduler is not in the ``stopped`` state

    """
    # Check if we're running under uWSGI with threads disabled
    uwsgi_module = sys.modules.get("uwsgi")
    if not getattr(uwsgi_module, "has_threads", True):
        raise RuntimeError(
            "The scheduler seems to be running under uWSGI, but threads have "
            "been disabled. You must run uWSGI with the --enable-threads "
            "option for the scheduler to work."
        )

    portal = self._ensure_services_ready()
    portal.call(self._async_scheduler.start_in_background)
```

--------------------------------

### Running WSGI Application with uWSGI

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/examples/README.rst

Runs a WSGI (Web Server Gateway Interface) application using the uWSGI server. It specifies the transport type (-T), HTTP port (:8000), and the filename of the WSGI application module using '--wsgi-file'.

```bash
uwsgi -T --http :8000 --wsgi-file <filename>
```

--------------------------------

### Getting All Jobs APScheduler Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

This asynchronous method retrieves all one-off jobs currently stored in the data store. It queries the data store for all jobs and returns them as a sequence. The order of the jobs in the returned sequence is not specified.

```python
async def get_jobs(self) -> Sequence[Job]:
 """Retrieve all jobs from the data store."""
 self._check_initialized()
 return await self.data_store.get_jobs()
```

--------------------------------

### Getting All Jobs in APScheduler Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

Retrieves a sequence of all currently running or queued jobs from the scheduler. It ensures services are ready and then calls the underlying asynchronous scheduler's `get_jobs` method via the portal.

```Python
def get_jobs(self) -> Sequence[Job]:
    portal = self._ensure_services_ready()
    return portal.call(self._async_scheduler.get_jobs)
```

--------------------------------

### Setting up FastAPI Application - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

Configures and initializes the FastAPI application. It includes a basic root endpoint `root` and sets up the FastAPI application instance, integrating the `lifespan` context manager to handle scheduler startup and shutdown.

```Python
async def root() -> Response:
    return PlainTextResponse("Hello, world!")


app = FastAPI(lifespan=lifespan)
app.add_api_route("/", root)
```

--------------------------------

### Initializing APScheduler SQLAlchemy Data Store - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (5).txt

Handles the startup logic for the data store. It verifies the async library, sets up resource disposal on exit, ensures the database schema is created and at the correct version, and detects dialect support for features like `UPDATE...RETURNING`.

```python
    async def start(
        self, exit_stack: AsyncExitStack, event_broker: EventBroker, logger: Logger
    ) -> None:
        asynclib = sniffio.current_async_library() or "(unknown)"
        if asynclib != "asyncio":
            raise RuntimeError(
                f"This data store requires asyncio; currently running: {asynclib}"
            )

        if self._close_on_exit:
            if isinstance(self._engine, AsyncEngine):
                exit_stack.push_async_callback(self._engine.dispose)
            else:
                exit_stack.callback(self._engine.dispose)

        await super().start(exit_stack, event_broker, logger)

        # Verify that the schema is in place
        async for attempt in self._retry():
            with attempt:
                async with self._begin_transaction() as conn:
                    # Create the schema first if it doesn't exist yet
                    if self.schema:
                        await self._execute(
                            conn, CreateSchema(name=self.schema, if_not_exists=True)
                        )

                    if self.start_from_scratch:
                        for table in self._metadata.sorted_tables:
                            await self._execute(conn, DropTable(table, if_exists=True))

                    await self._create_metadata(conn)
                    query = select(self._t_metadata.c.schema_version)
                    result = await self._execute(conn, query)
                    version = result.scalar()
                    if version is None:
                        await self._execute(
                            conn, self._t_metadata.insert(), {"schema_version": 1}
                        )
                    elif version > 1:
                        raise RuntimeError(
                            f"Unexpected schema version ({version}); "
                            f"only version 1 is supported by this version of "
                            f"APScheduler"
                        )

        # Find out if the dialect supports UPDATE...RETURNING
        async for attempt in self._retry():
            with attempt:
                update = (
                    self._t_metadata.update()
                    .values(schema_version=self._t_metadata.c.schema_version)
                    .returning(self._t_metadata.c.schema_version)
                )
                async with self._begin_transaction() as conn:
                    try:
                        await self._execute(conn, update)
                    except (CompileError, ProgrammingError):
                        pass  # the support flag is False by default
                    else:
                        self._supports_update_returning = True
```

--------------------------------

### Using APScheduler AndTrigger (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

Illustrates combining triggers using `AndTrigger`. This example creates an `AndTrigger` that fires every two months at 10:00 but only on weekdays, by combining a `CalendarIntervalTrigger` and a `CronTrigger` to find common fire times.

```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),
)
```

--------------------------------

### Setting up Flask Application - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

Initializes the Flask web application instance `app`. This is the standard way to create a Flask application, which will be exposed as a WSGI callable.

```Python
app = Flask(__name__)
```

--------------------------------

### Using APScheduler OrTrigger (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

Shows how to combine multiple simple triggers using `OrTrigger` to create complex scheduling logic. This example creates an `OrTrigger` that fires on weekdays at 10:00 or weekends at 11:00 by combining two `CronTrigger` instances.

```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),
)
```

--------------------------------

### Building Documentation with Tox (Bash)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/contributing.rst

This command instructs Tox to run the documentation build environment (docs). It uses Sphinx to generate HTML documentation from the source files, placing the output in the build/sphinx/html directory.

```bash
tox -e docs
```

--------------------------------

### Starting the Data Store

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (5).txt

This asynchronous method initializes the data store upon application startup. It pushes a cleanup callback to close the MongoDB client if it was created internally, performs superclass startup, checks the MongoDB server version, and initializes the database (creating indexes) with retry logic.

```python
async def start(
 self, exit_stack: AsyncExitStack, event_broker: EventBroker, logger: Logger
) -> None:
 if self._close_on_exit:
 exit_stack.push_async_callback(to_thread.run_sync, self._client.close)

 await super().start(exit_stack, event_broker, logger)
 server_info = await to_thread.run_sync(self._client.server_info)
 if server_info["versionArray"] < [4, 0]:
 raise RuntimeError(
 f"MongoDB server must be at least v4.0; current version = "
 f"{server_info['version']}"
 )

 async for attempt in self._retry():
 with attempt:
 await to_thread.run_sync(self._initialize)
```

--------------------------------

### Configuring APScheduler Debug Logging - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/userguide.rst

This snippet shows how to enable debug-level logging for the APScheduler library using Python's built-in `logging` module. It configures basic logging and specifically sets the 'apscheduler' logger to DEBUG level to get detailed internal information.

```python
import logging

logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
```

--------------------------------

### Combining Triggers with OrTrigger in APScheduler (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/userguide.rst

This example demonstrates the use of OrTrigger to schedule a task based on the conditions of multiple underlying triggers. It shows how to create a trigger that fires on specific times on weekdays (10:00) and different times on weekends (11:00) by combining two CronTrigger instances.

```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),
)
```

--------------------------------

### Run AsyncScheduler as Worker (Async Worker Role)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Initializes an `AsyncScheduler` instance using the shared data store and event broker. It then calls `run_until_stopped`, causing the scheduler to start its internal execution loop. This instance will pick up schedules added by other scheduler instances and execute the corresponding jobs.

```python
import asyncio

from apscheduler import AsyncScheduler

# Assuming data_store and event_broker are already configured (e.g., see Setup snippet)

async def main():
    async with AsyncScheduler(data_store, event_broker) as scheduler:
        await scheduler.run_until_stopped()
```

--------------------------------

### Defining Trivial WSGI Application - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

A minimal, framework-agnostic WSGI application function `application`. It takes the standard WSGI `environ` and `start_response` arguments, prepares a simple 'Hello, World!' response body and headers, calls `start_response` with the status and headers, and returns an iterable containing the response body. This function serves as the basic WSGI application logic.

```Python
def application(environ, start_response):
    response_body = b"Hello, World!"
    response_headers = [
        ("Content-Type", "text/plain"),
        ("Content-Length", str(len(response_body))),
    ]
    start_response("200 OK", response_headers)
    return [response_body]
```

--------------------------------

### Add Schedule with AsyncScheduler (Async Scheduler Role)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Initializes an `AsyncScheduler` instance using the shared data store and event broker. It then adds a single schedule for the `tick` function to run every second using an `IntervalTrigger`. The scheduler exits after adding the schedule without starting its internal execution loop.

```python
import asyncio

from example_tasks import tick

from apscheduler import AsyncScheduler
from apscheduler.triggers.interval import IntervalTrigger

# Assuming data_store and event_broker are already configured (e.g., see Setup snippet)

async def main():
    async with AsyncScheduler(data_store, event_broker) as scheduler:
        await scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")
        # Note: we don't actually start the scheduler here!
```

--------------------------------

### Run Scheduler as Worker (Sync Worker Role)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Initializes a synchronous `Scheduler` instance using the shared data store and event broker. It then calls `run_until_stopped`, causing the scheduler to start its internal execution loop. This instance acts as the worker, picking up schedules added by other scheduler instances and executing the corresponding jobs.

```python
from apscheduler import Scheduler

# Assuming data_store and event_broker are already configured (e.g., see Setup snippet)

with Scheduler(data_store, event_broker) as scheduler:
    scheduler.run_until_stopped()
```

--------------------------------

### Add Schedule with Scheduler (Sync Scheduler Role)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (11).txt

Initializes a synchronous `Scheduler` instance using the shared data store and event broker. It then adds a single schedule for the `tick` function to run every second using an `IntervalTrigger`. The scheduler exits after adding the schedule without starting its internal execution loop.

```python
from example_tasks import tick

from apscheduler import Scheduler
from apscheduler.triggers.interval import IntervalTrigger

# Assuming data_store and event_broker are already configured (e.g., see Setup snippet)

with Scheduler(data_store, event_broker) as scheduler:
    scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")
    # Note: we don't actually start the scheduler here!
```

--------------------------------

### Setting up Starlette Application with Middleware - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

Initializes the necessary components for a Starlette application with APScheduler integration. It creates a database engine, datastore, event broker, and the AsyncScheduler instance. It defines application routes (`root` endpoint) and sets up the `SchedulerMiddleware` using the created scheduler instance. Finally, it instantiates the Starlette application with the defined routes and middleware.

```Python
async def root(request: Request) -> Response:
    return PlainTextResponse("Hello, world!")


engine = create_async_engine("postgresql+asyncpg://postgres:secret@localhost/testdb")
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)
scheduler = AsyncScheduler(data_store, event_broker)
routes = [Route("/", root)]
middleware = [Middleware(SchedulerMiddleware, scheduler=scheduler)]
app = Starlette(routes=routes, middleware=middleware)
```

--------------------------------

### Setting Up Dev Env for Direct Pytest (Bash)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/contributing.rst

These commands create a Python virtual environment named venv, activate it (for Linux/macOS), and install the current project in editable mode (-e .) along with its required test dependencies (--group test). This allows running pytest directly.

```bash
python -m venv venv
source venv/bin/activate
pip install --group test -e .
```

--------------------------------

### Implementing Custom Data Store Interface - APScheduler - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/extending.rst

This snippet provides the basic structure for creating a custom APScheduler data store by inheriting from `apscheduler.abc.DataStore`. It shows how to implement the mandatory `start` method, which is used for initialization and receiving the `EventBroker` instance needed for sending events related to store operations. A custom data store must implement numerous other abstract methods defined in the `DataStore` interface.

```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
```

--------------------------------

### Initializing MongoDB Client and Collections

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (5).txt

This `attrs` post-initialization hook sets up the MongoDB client, either by creating a new one from a URI or using a provided client instance. It also configures custom type encoding/decoding for PyMongo and gets references to the necessary database collections ('tasks', 'schedules', 'jobs', 'job_results').

```python
def __attrs_post_init__(self) -> None:
 if isinstance(self.client_or_uri, str):
 self._client = MongoClient(self.client_or_uri)
 self._close_on_exit = True
 else:
 self._client = self.client_or_uri

 type_registry = TypeRegistry(
 [
 CustomEncoder(timedelta, timedelta.total_seconds),
 CustomEncoder(ConflictPolicy, operator.attrgetter("name")),
 CustomEncoder(CoalescePolicy, operator.attrgetter("name")),
 CustomEncoder(JobOutcome, operator.attrgetter("name")),
 ]
 )
 codec_options: CodecOptions = CodecOptions(
 type_registry=type_registry,
 uuid_representation=UuidRepresentation.STANDARD,
 )
 database = self._client.get_database(self.database, codec_options=codec_options)
 self._tasks = database["tasks"]
 self._schedules = database["schedules"]
 self._jobs = database["jobs"]
 self._jobs_results = database["job_results"]
```

--------------------------------

### Accessing the Scheduler State (AsyncScheduler, Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

A property that provides read-only access to the current running state of the `AsyncScheduler`. The state is represented by the `RunState` enumeration, indicating whether the scheduler is stopped, starting, or started.

```Python
@property
    def state(self) -> RunState:
        """The current running state of the scheduler."""
        return self._state
```

--------------------------------

### Defining Trivial ASGI Application - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

A minimal, framework-agnostic ASGI application function `original_app`. It handles both 'http' requests (returning 'Hello, world!') and 'lifespan' events (responding to startup and shutdown completion). This serves as the core application logic before middleware is applied.

```Python
async def original_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
```

--------------------------------

### Running APScheduler Separate Worker Script (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/examples/README.rst

Executes the Python script responsible for running the APScheduler worker process in the separate worker/scheduler example. This script is configured to listen for and execute scheduled tasks from the data store.

```bash
python sync_worker.py
```

--------------------------------

### Setting up Scheduler Middleware Class - Starlette - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (13).txt

Defines an ASGI middleware class `SchedulerMiddleware` for Starlette. It takes the ASGI application and an APScheduler instance during initialization. In its `__call__` method, it checks for 'lifespan' scope; if detected, it starts the scheduler (using an async context manager) before delegating to the next application in the chain. For other scopes, it simply calls the next application.

```Python
class SchedulerMiddleware:
    def __init__(
        self,
        app: ASGIApp,
        scheduler: AsyncScheduler,
    ) -> None:
        self.app = app
        self.scheduler = scheduler

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "lifespan":
            async with self.scheduler:
                await self.scheduler.add_schedule(
                    tick, IntervalTrigger(seconds=1), id="tick"
                )
                await self.scheduler.start_in_background()
                await self.app(scope, receive, send)
        else:
            await self.app(scope, receive, send)
```

--------------------------------

### Post-Initialization Logic for AsyncScheduler (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

Handles post-initialization setup for the `AsyncScheduler`. It sets a default identity if none is provided, populates `job_executors` with standard executors if empty, and validates or sets default values for `task_defaults` such as the default job executor and maximum running jobs per task. Raises `ValueError` if the default job executor is not found among the provided executors.

```Python
def __attrs_post_init__(self) -> None:
        if not self.identity:
            self.identity = f"{platform.node()}-{os.getpid()}-{id(self)}"

        if not self.job_executors:
            self.job_executors = {
                "async": AsyncJobExecutor(),
                "threadpool": ThreadPoolJobExecutor(),
                "processpool": ProcessPoolJobExecutor(),
            }

        if self.task_defaults.job_executor is unset:
            self.task_defaults.job_executor = next(iter(self.job_executors))
        elif self.task_defaults.job_executor not in self.job_executors:
            valid_executors = ", ".join(self.job_executors)
            raise ValueError(
                f"the default job executor must be one of the given job executors "
                f"({valid_executors})"
            )

        if self.task_defaults.max_running_jobs is unset:
            self.task_defaults.max_running_jobs = 1

        if self.task_defaults.misfire_grace_time is unset:
            self.task_defaults.misfire_grace_time = None
```

--------------------------------

### Accessing Current Job Context in APScheduler Task (Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/src/docs/User guide — APScheduler documentation.md

Demonstrates how to access information about the currently running job within a task function using the `current_job` context variable provided by APScheduler. It shows how to retrieve the job ID and the ID of the schedule that spawned the job.

```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}"
    )
```

--------------------------------

### Implementing the Background Cleanup Loop (AsyncScheduler, Python)

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (9).txt

An internal coroutine responsible for periodically triggering the scheduler's cleanup process. It runs in a loop while the scheduler's state is 'starting' or 'started', calling the `cleanup()` method and then pausing for the duration specified by `cleanup_interval`. This automates the removal of expired job results and finished schedules.

```Python
async def _cleanup_loop(self) -> None:
        delay = self.cleanup_interval.total_seconds()
        assert delay > 0
        while self._state in (RunState.starting, RunState.started):
            await self.cleanup()
            await sleep(delay)
```

--------------------------------

### Representing Memory DataStore Instance - Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (5).txt

Implements the `__repr__` method for the `MemoryDataStore` class using a helper function `create_repr`. This provides a developer-friendly string representation of the data store instance.

```Python
from .._utils import create_repr
from .base import BaseDataStore

@attrs.define(eq=False, repr=False)
class MemoryDataStore(BaseDataStore):
    # ... (other attributes)

    def __repr__(self) -> str:
        return create_repr(self)
```

--------------------------------

### Initializing MongoDB Indexes

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (5).txt

This internal method is responsible for creating the necessary indexes on the MongoDB collections used by the data store. It optionally clears existing data if `start_from_scratch` is enabled and creates indexes on fields like task_id, next_fire_time, acquired_by, created_at, and expires_at.

```python
def _initialize(self) -> None:
 with self._client.start_session() as session:
 if self.start_from_scratch:
 self._tasks.delete_many({}, session=session)
 self._schedules.delete_many({}, session=session)
 self._jobs.delete_many({}, session=session)
 self._jobs_results.delete_many({}, session=session)

 self._schedules.create_index("task_id", session=session)
 self._schedules.create_index("next_fire_time", session=session)
 self._schedules.create_index("acquired_by", session=session)
 self._jobs.create_index("task_id", session=session)
 self._jobs.create_index("schedule_id", session=session)
 self._jobs.create_index("created_at", session=session)
 self._jobs.create_index("acquired_by", session=session)
 self._jobs_results.create_index("expires_at", session=session)
```

--------------------------------

### Calculating Next Trigger Time in Python

Source: https://github.com/shak-shat/apscheduler4context7/blob/main/snippets/agronholm-apscheduler (1).txt

The core logic method for finding the next datetime when the trigger should fire. It starts from the last fire time (or start time) and iterates through the fields (second, minute, hour, etc.), incrementing them as needed to find the earliest future time that satisfies all constraints. Returns the next fire time or `None` if the end time is reached.

```Python
def next(self) -> datetime | None:
        if self._last_fire_time:
            next_time = datetime.fromtimestamp(
                self._last_fire_time.timestamp() + 1, self.timezone
            )
        else:
            next_time = self.start_time

        fieldnum = 0
        while 0 <= fieldnum < len(self._fields):
            field = self._fields[fieldnum]
            curr_value = field.get_value(next_time)
            next_value = field.get_next_value(next_time)

            if next_value is None:
                # No valid value was found
                next_time, fieldnum = self._increment_field_value(
                    next_time, fieldnum - 1
                )
            elif next_value > curr_value:
                # A valid, but higher than the starting value, was found
                if field.real:
                    next_time = self._set_field_value(next_time, fieldnum, next_value)
                    fieldnum += 1
                else:
                    next_time, fieldnum = self._increment_field_value(
                        next_time, fieldnum
                    )
            else:
                # A valid value was found, no changes necessary
                fieldnum += 1

            # Return if the date has rolled past the end date
            if self.end_time and next_time > self.end_time:
                return None

        if fieldnum >= 0:
            self._last_fire_time = next_time
            return next_time

        return None
```