Metadata-Version: 2.4
Name: APSchedulerServer
Version: 0.1.2
Summary: 基于 APScheduler 的任务调度服务器，支持多种 JobStore 后端
Author-email: fly <fzkf117@163.com>
License: MIT
Keywords: apscheduler,scheduler,task,job,cron,redis,mysql,django
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.6
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: APScheduler>=3.10.0
Provides-Extra: redis
Requires-Dist: redis>=4.0.0; extra == "redis"
Provides-Extra: mysql
Requires-Dist: mysql-connector-python>=8.0.0; extra == "mysql"
Provides-Extra: django
Requires-Dist: django>=3.2.0; extra == "django"
Provides-Extra: sqlalchemy
Requires-Dist: sqlalchemy>=1.4.0; extra == "sqlalchemy"
Provides-Extra: all
Requires-Dist: APSchedulerServer[django,mysql,redis,sqlalchemy]; extra == "all"
Dynamic: license-file

# APSchedulerServer 使用教程

## 项目简介

APSchedulerServer 是一个基于 APScheduler 的任务调度服务器，支持多种 JobStore 后端（Redis、MySQL、Django 等），提供了灵活的任务配置和注册机制。

## 项目结构

```
APSchedulerServer/
├── src/                     # 源代码目录
│   └── APSchedulerServer/   # 核心代码包
│       ├── __init__.py
│       ├── base_config.py   # 基础配置类
│       ├── loader.py        # 配置加载器
│       └── main.py          # 主程序入口
├── APSchedulerConfig/       # 配置目录（用户需要创建）
│   ├── config.py           # 调度器配置
│   └── registry_tasks.py   # 任务注册
├── pyproject.toml          # 项目配置文件
├── setup.py               # 安装脚本
└── README.md              # 项目文档
```

## 安装

### 基础安装

```bash
pip install APSchedulerServer
```

### 安装可选依赖

根据所选的 JobStore，安装额外的依赖：

```bash
# Redis JobStore
pip install APSchedulerServer[redis]

# MySQL JobStore
pip install APSchedulerServer[mysql]

# Django JobStore (需要在 Django 项目中使用)
pip install APSchedulerServer[django]

# SQLAlchemy JobStore
pip install APSchedulerServer[sqlalchemy]

# 安装所有可选依赖
pip install APSchedulerServer[all]
```

### 从源码安装

```bash
git clone https://github.com/yourusername/APSchedulerServer.git
cd APSchedulerServer
pip install -e .
```

## 快速开始

### 1. 基础配置

编辑 `APSchedulerConfig/config.py`，配置调度器和 JobStore：

```python
from APSchedulerServer.base_config import BaseConfig
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.jobstores.django import DjangoJobStore


class ApsConfig(BaseConfig):

   def __init__(self):
      print("配置加载成功")

   def set_job_stores(self):
      """配置 JobStore"""
      return {
         # Redis JobStore
         'redis': RedisJobStore(
            jobs_key='apscheduler.jobs',
            run_times_key='apscheduler.run_times',
            host='localhost',
            port=6379,
            db=0
         ),
         # MySQL JobStore
         'mysql': SQLAlchemyJobStore(
            url='mysql+mysqlconnector://user:password@localhost/scheduler_db'
         ),
         # Django JobStore
         'django': DjangoJobStore(),
         # Memory JobStore (默认)
         'default': MemoryJobStore()
      }

   def set_scheduler(self):
      """配置调度器"""
      scheduler = super().set_scheduler()
      # 可以在这里添加自定义配置
      return scheduler
```

### 2. 任务注册

编辑 `APSchedulerConfig/registry_tasks.py`，注册你的定时任务：

```python
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.date import DateTrigger
import datetime

def set_registry_tasks(scheduler):
    """注册所有定时任务"""

    # 示例1: 使用 Cron 表达式（每天 8:30 执行）
    scheduler.add_job(
        func=daily_task,
        trigger=CronTrigger(hour=8, minute=30),
        id='daily_task',
        name='每日任务',
        replace_existing=True
    )

    # 示例2: 使用间隔触发器（每 5 分钟执行一次）
    scheduler.add_job(
        func=interval_task,
        trigger=IntervalTrigger(minutes=5),
        id='interval_task',
        name='间隔任务',
        replace_existing=True
    )

    # 示例3: 使用日期触发器（在指定时间执行一次）
    scheduler.add_job(
        func=one_time_task,
        trigger=DateTrigger(run_date=datetime.datetime(2024, 1, 1, 12, 0)),
        id='one_time_task',
        name='一次性任务',
        replace_existing=True
    )

    print("所有任务注册完成")

# 定义任务函数
def daily_task():
    """每日任务"""
    print("执行每日任务")
    # 在这里添加你的业务逻辑

def interval_task():
    """间隔任务"""
    print("执行间隔任务")
    # 在这里添加你的业务逻辑

def one_time_task():
    """一次性任务"""
    print("执行一次性任务")
    # 在这里添加你的业务逻辑
```

### 3. 运行调度器

#### 方法1：使用命令行工具（推荐）

```bash
apscheduler-server
```

#### 方法2：使用 Python 模块方式

在项目根目录下执行：

```bash
python -m APSchedulerServer.main
```

#### 方法3：直接运行

```python
from APSchedulerServer.main import main

if __name__ == '__main__':
    main()
```

## JobStore 配置详解

### Redis JobStore

```python
from apscheduler.jobstores.redis import RedisJobStore

def set_job_stores(self):
    return {
        'redis': RedisJobStore(
            jobs_key='apscheduler.jobs',      # 存储任务的键名
            run_times_key='apscheduler.run_times',  # 存储运行时间的键名
            host='localhost',                 # Redis 主机
            port=6379,                        # Redis 端口
            db=0,                             # Redis 数据库
            password='your_password'          # Redis 密码（可选）
        )
    }
```

### MySQL JobStore

```python
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

def set_job_stores(self):
    return {
        'mysql': SQLAlchemyJobStore(
            url='mysql+mysqlconnector://user:password@localhost/scheduler_db'
        )
    }
```

支持的数据库 URL 格式：
- MySQL: `mysql+mysqlconnector://user:password@host/dbname`
- PostgreSQL: `postgresql://user:password@host/dbname`
- SQLite: `sqlite:///jobs.sqlite`

### Django JobStore

```python
from apscheduler.jobstores.django import DjangoJobStore

def set_job_stores(self):
    return {
        'django': DjangoJobStore()
    }
```

注意：使用 Django JobStore 需要将 `apscheduler` 添加到 Django 的 `INSTALLED_APPS` 中。

## 高级配置

### 修改执行器配置

编辑 `APSchedulerServer/base_config.py` 中的 `SCHEDULER_EXECUTORS`：

```python
SCHEDULER_EXECUTORS = {
    "default": {"type": "threadpool", "max_workers": 20},
    "cpu": {"type": "processpool", "max_workers": 8}
}
```

### 修改任务默认参数

编辑 `APSchedulerServer/base_config.py` 中的 `SCHEDULER_JOB_DEFAULTS`：

```python
SCHEDULER_JOB_DEFAULTS = {
    "coalesce": True,              # 合并错过的执行
    "max_instances": 5,            # 最大并发实例数
    "misfire_grace_time": 300      # 错过执行的宽限期（秒）
}
```

### 时区配置

编辑 `APSchedulerServer/base_config.py` 中的 `TIME_ZONE`：

```python
TIME_ZONE = "Asia/Shanghai"
```

## 常见问题

### 1. 找不到模块错误

确保在项目根目录下运行：

```bash
python -m APSchedulerServer.main
```

### 2. Redis 连接失败

检查 Redis 服务是否启动，以及连接配置是否正确。

### 3. MySQL 连接失败

确保 MySQL 服务已启动，并且数据库已创建。

### 4. 任务没有执行

检查任务的触发器配置是否正确，以及调度器是否正常运行。

## 示例项目

### 示例 1: 使用 Redis JobStore

```python
# APSchedulerConfig/config.py
from APSchedulerServer.base_config import BaseConfig
from apscheduler.jobstores.redis import RedisJobStore


class ApsConfig(BaseConfig):
   def __init__(self):
      print("Redis JobStore 配置加载成功")

   def set_job_stores(self):
      return {
         'default': RedisJobStore(
            host='localhost',
            port=6379,
            db=0
         )
      }
```

### 示例 2: 使用 MySQL JobStore

```python
# APSchedulerConfig/config.py
from APSchedulerServer.base_config import BaseConfig
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore


class ApsConfig(BaseConfig):
   def __init__(self):
      print("MySQL JobStore 配置加载成功")

   def set_job_stores(self):
      return {
         'default': SQLAlchemyJobStore(
            url='mysql+mysqlconnector://user:password@localhost/scheduler_db'
         )
      }
```

### 示例 3: 使用 Django JobStore

```python
# APSchedulerConfig/config.py
from APSchedulerServer.base_config import BaseConfig
from apscheduler.jobstores.django import DjangoJobStore


class ApsConfig(BaseConfig):
   def __init__(self):
      print("Django JobStore 配置加载成功")

   def set_job_stores(self):
      return {
         'default': DjangoJobStore()
      }
```

### 示例 4: 使用 SQLAlchemy JobStore

```python
# APSchedulerConfig/config.py
from APSchedulerServer.base_config import BaseConfig
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore


class ApsConfig(BaseConfig):
   def __init__(self):
      print("SQLAlchemy JobStore 配置加载成功")

   def set_job_stores(self):
      return {
         'default': SQLAlchemyJobStore(
            url='postgresql://user:password@localhost/scheduler_db'
         )
      }
```

## 最佳实践

1. **选择合适的 JobStore**：
   - 开发环境：使用 MemoryJobStore
   - 生产环境：使用 Redis 或 SQLAlchemy JobStore
   - Django 项目：使用 Django JobStore
   - 需要跨平台支持：使用 SQLAlchemy JobStore（支持 PostgreSQL、MySQL、SQLite 等）

2. **任务设计**：
   - 任务函数应该是幂等的（可以安全地多次执行）
   - 避免长时间运行的任务
   - 合理设置任务的并发实例数

3. **监控和日志**：
   - 添加任务执行日志
   - 监控任务执行状态
   - 设置任务失败告警

4. **错误处理**：
   - 在任务函数中添加适当的异常处理
   - 设置合理的重试策略

## 许可证

本项目采用 MIT 许可证。
