Metadata-Version: 2.4
Name: pyboot-execute
Version: 0.1.8
Summary: Task scheduling and background execution plugin for PyBoot
Author: Siddharth Karn
License-Expression: MIT
Project-URL: Homepage, https://github.com/devSiddharthKarn/pyboot
Project-URL: Documentation, https://github.com/devSiddharthKarn/pyboot/tree/main/plugins/pyboot_execute
Project-URL: Repository, https://github.com/devSiddharthKarn/pyboot
Project-URL: Issues, https://github.com/devSiddharthKarn/pyboot/issues
Keywords: pyboot,scheduler,task-scheduling,background-jobs,apscheduler
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyboot-framework>=0.1.3
Requires-Dist: APScheduler<4,>=3.10
Dynamic: license-file

# pyboot-execute

`pyboot-execute` is a PyBoot plugin for registering background services and
running recurring synchronous jobs with APScheduler's
`BackgroundScheduler`.

## Features

- Register background service classes with `BackgroundService`.
- Run synchronous methods at a fixed interval.
- Run a method again after a delay following the previous completion.
- Apply an optional initial delay to interval and fixed-delay jobs.
- Run methods from five-field cron expressions with `CronJob`.
- Resolve constructor dependencies through PyBoot's IoC container.
- Start the scheduler when PyBoot reaches its ready lifecycle stage.
- Shut down the scheduler during normal interpreter exit.

## Requirements

- Python 3.10 or newer.
- `pyboot-framework` 0.1.3 or newer.
- APScheduler 3.10 through 3.x.

The plugin uses APScheduler's in-process `BackgroundScheduler`. It is not a
distributed or persistent job scheduler.

## Installation

Install the published distributions:

```bash
python -m pip install pyboot-framework pyboot-execute
```

For a source checkout from the repository root:

```bash
python -m pip install -e pyboot
python -m pip install -e plugins/pyboot_execute
```

The package declares the PyBoot and APScheduler dependencies. Using
`python -m pip` ensures they are installed into the interpreter used to run
the application.

Verify plugin discovery:

```bash
python -c "from importlib.metadata import entry_points; print([e.name for e in entry_points(group='pyboot.plugins')])"
```

The output should include `execute`.

## Quick start

Create a background service in the package passed to `bootstrap()`:

```python
from pyboot_execute import BackgroundService, Scheduled


@BackgroundService()
class NotificationService:
    @Scheduled(interval_ms=5_000)
    def send_notifications(self) -> None:
        print("Checking pending notifications")
```

Bootstrap PyBoot:

```python
from pyboot.core.application import PyBoot


pyboot = PyBoot(app_instance=None, debug=False)
pyboot.bootstrap(package="app")
```

When PyBoot loads the installed `execute` entry point, it discovers the
background service, registers it as an IoC bean, schedules its method, and
starts the scheduler during `on_ready`.

## Scheduling modes

### Fixed interval

`interval_ms` schedules executions at a recurring interval:

```python
@BackgroundService()
class CacheRefresher:
    @Scheduled(interval_ms=60_000, initial_delay_ms=5_000)
    def refresh(self) -> None:
        print("Refreshing cache")
```

The first execution occurs after `initial_delay_ms`; later executions use the
configured interval. If the interval job takes longer than its interval,
APScheduler's normal job execution rules apply and executions may overlap or
be skipped according to the scheduler configuration.

### Fixed delay

`fixed_delay_ms` schedules the next run only after the previous invocation
returns:

```python
@BackgroundService()
class DataSynchronizer:
    @Scheduled(fixed_delay_ms=30_000, initial_delay_ms=10_000)
    def synchronize(self) -> None:
        print("Synchronizing data")
```

If the method takes 8 seconds and the delay is 30 seconds, the next run is
scheduled approximately 30 seconds after completion, not 30 seconds after the
previous start.

Exactly one of `interval_ms` and `fixed_delay_ms` must be supplied. Both must
be positive. `initial_delay_ms` may be omitted or set to `None`; it must not be
negative.

### Cron schedules

`CronJob` schedules a synchronous method using APScheduler's
`CronTrigger.from_crontab()`:

```python
from pyboot_execute import BackgroundService, CronJob


@BackgroundService()
class CleanupService:
    @CronJob("0 0 * * *")
    def remove_expired_sessions(self) -> None:
        print("Removing expired sessions")
```

The expression uses five fields:

```text
minute hour day-of-month month day-of-week
```

Examples:

| Expression | Meaning |
| --- | --- |
| `* * * * *` | Every minute |
| `*/5 * * * *` | Every five minutes |
| `0 * * * *` | At the start of every hour |
| `0 0 * * *` | Every day at midnight |
| `0 9 * * 1-5` | At 09:00 on weekdays |
| `0 0 1 * *` | At midnight on the first day of each month |

Cron jobs use the scheduler's configured timezone. This plugin currently
creates `BackgroundScheduler()` without an explicit timezone, so deployments
should document the host or container timezone when relying on wall-clock
schedules.

## Dependency injection

Background services are regular PyBoot-managed beans. Declare dependencies in
the constructor with type annotations:

```python
from pyboot.core.builtins.component import Component
from pyboot_execute import BackgroundService, Scheduled


@Component()
class ReportRepository:
    def pending_reports(self) -> list[str]:
        return ["daily-summary"]


@BackgroundService()
class ReportWorker:
    def __init__(self, repository: ReportRepository) -> None:
        self.repository = repository

    @Scheduled(interval_ms=60_000)
    def generate_reports(self) -> None:
        for report in self.repository.pending_reports():
            print(f"Generating {report}")
```

Do not instantiate a `BackgroundService` manually. PyBoot creates the bean and
resolves its required constructor dependencies. All required constructor
parameters need type annotations that the IoC container can resolve.

## Public API

### `BackgroundService`

`@BackgroundService()` marks a class for registration by `ExecutePlugin`.
Scheduled methods are inspected on the resulting bean after PyBoot creates it.

### `Scheduled`

```python
@Scheduled(
    interval_ms: int | None = None,
    fixed_delay_ms: int | None = None,
    initial_delay_ms: int | None = None,
)
```

Exactly one primary schedule value is required. Invalid combinations and
non-positive values raise `ValueError` during decorator construction.

### `CronJob`

```python
@CronJob(trigger: str)
```

`CronJob` schedules the decorated method using APScheduler's
`CronTrigger.from_crontab()`. The trigger string must be a valid five-field
cron expression. Invalid expressions raise an APScheduler trigger error during
application loading.

### `ScheduleType`

`ScheduleType.INTERVAL` and `ScheduleType.FIXED_DELAY` describe the internal
schedule metadata used by `Scheduled`.

### `ExecutePlugin`

`ExecutePlugin` is registered automatically through the
`pyboot.plugins` entry-point group:

```toml
[project.entry-points."pyboot.plugins"]
execute = "pyboot_execute.execute:ExecutePlugin"
```

Most users should not instantiate it directly.

## Application layout

```text
service/
├── main.py
└── app/
    ├── __init__.py
    ├── components.py
    └── jobs.py
```

PyBoot scans the package passed to `bootstrap()` and its submodules. Keep
background services and their dependencies inside that package, and define
classes in the modules being scanned.

## Lifecycle and shutdown

The plugin registers background services during `on_register`, obtains their
beans during `after_load`, and starts its `BackgroundScheduler` during
`on_ready`. It registers an `atexit` callback and calls
`shutdown(wait=False)` when the scheduler is running.

Normal interpreter shutdown is covered; crashes and forced termination are not.
Jobs are recreated from decorators on every process start. The plugin does not
persist job state or dynamically created jobs.

## Production guidance

### Multiple workers

Every process creates its own scheduler. If an application runs with four
workers, the same job can execute four times. Use a single dedicated scheduler
process, a distributed scheduler, or an application-level distributed lock
when a job must run once globally.

### Reliability

- Make jobs idempotent where possible.
- Use explicit timeouts for external calls.
- Add structured logging and metrics around each job.
- Catch and handle expected business failures inside the job.
- Keep secrets out of log messages.
- Consider a persistent or distributed job system for critical workloads.
- Test startup and shutdown behavior in the deployment environment.

### Concurrency

Scheduled methods are synchronous and run in APScheduler worker threads.
CPU-heavy or long-running jobs can consume worker capacity. Do not assume the
main application thread and scheduled workers share transaction or request
context.

## Testing

Test the service logic independently from scheduling, then verify registration
with a PyBoot bootstrap:

```python
from pyboot.core.application import PyBoot


def test_application_bootstrap() -> None:
    pyboot = PyBoot(app_instance=None, debug=False)
    pyboot.bootstrap(package="app")
```

For deterministic job tests, call the service method directly or use a
controlled scheduler test fixture. Avoid waiting on real millisecond delays in
unit tests.

## License

This project is distributed under the MIT License. See [LICENSE](./LICENSE).
