feat(scheduler): add task executors for common operations
Add three built-in task executors for various automation tasks. Executors: 1. example_executor - Simple test implementation with configurable message and delay 2. doc_sync_executor - Mirror documentation from upstream Git repos to Gitea 3. config_backup_executor - Backup Docker configs and data directories doc_sync_executor features: - Clones upstream repository (GitHub, GitLab, etc.) - Supports full repository mirroring or selective path syncing - Pushes to Gitea with authentication - Creates date-tagged snapshots (YYYY-MM-DD) - Generates .SYNC_INFO.md with sync metadata config_backup_executor features: - Backs up multiple source paths with exclusion patterns - Optional compression (tar.gz) - Retention policy (days-based cleanup) - Timestamped backups Pre-configured tasks: - backup_docker_configs_daily (priority 20, daily 03:05) - sync_fastapi_docs_monthly (priority 60, 11th @ 04:00) - sync_ollama_docs_monthly (priority 60, 12th @ 04:00) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.5
@@ -0,0 +1,23 @@
|
||||
"""
|
||||
Executor modules for The Scheduler.
|
||||
|
||||
Each executor must implement an async execute(config, settings) function.
|
||||
|
||||
Example:
|
||||
async def execute(config: dict, settings: Settings) -> str:
|
||||
'''
|
||||
Perform the task.
|
||||
|
||||
Args:
|
||||
config: Task-specific configuration from scheduled_tasks.config
|
||||
settings: Global scheduler settings
|
||||
|
||||
Returns:
|
||||
Output message (success)
|
||||
|
||||
Raises:
|
||||
Exception: On failure (will be logged as error)
|
||||
'''
|
||||
# Your task logic here
|
||||
return "Task completed successfully"
|
||||
"""
|
||||
@@ -0,0 +1,159 @@
|
||||
"""
|
||||
Config Backup Executor
|
||||
Backs up Docker container configs and host-based service configs.
|
||||
Replicates functionality of maintenance container's backup-configs.sh
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
import tarfile
|
||||
import tempfile
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any
|
||||
|
||||
from src.config import Settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def execute(config: dict, settings: Settings) -> str:
|
||||
"""
|
||||
Execute config backup task.
|
||||
|
||||
Config schema:
|
||||
{
|
||||
"sources": [
|
||||
{
|
||||
"path": "/data/docker-data",
|
||||
"name": "docker-data",
|
||||
"excludes": ["*/cache/*", "*/temp/*", "*.log"]
|
||||
}
|
||||
],
|
||||
"backup_dir": "/backups/docker-configs",
|
||||
"retention_days": 30,
|
||||
"compress": true
|
||||
}
|
||||
|
||||
Args:
|
||||
config: Backup configuration
|
||||
settings: Global scheduler settings
|
||||
|
||||
Returns:
|
||||
Summary of backup operation
|
||||
|
||||
Raises:
|
||||
Exception: On backup failure
|
||||
"""
|
||||
sources = config.get('sources', [])
|
||||
backup_dir = Path(config.get('backup_dir', '/backups/docker-configs'))
|
||||
retention_days = config.get('retention_days', 30)
|
||||
compress = config.get('compress', True)
|
||||
|
||||
if not sources:
|
||||
raise ValueError("No backup sources configured")
|
||||
|
||||
timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
|
||||
backup_filename = f"docker-configs-{timestamp}.tar.gz"
|
||||
backup_file = backup_dir / backup_filename
|
||||
|
||||
logger.info(f"Starting Docker configs backup: {backup_filename}")
|
||||
|
||||
# Create backup directory
|
||||
backup_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create temporary directory for staging
|
||||
with tempfile.TemporaryDirectory(prefix='backup-') as temp_dir:
|
||||
temp_path = Path(temp_dir)
|
||||
results = []
|
||||
|
||||
# Backup each source
|
||||
for source in sources:
|
||||
source_path = Path(source['path'])
|
||||
source_name = source['name']
|
||||
excludes = source.get('excludes', [])
|
||||
|
||||
if not source_path.exists():
|
||||
logger.warning(f"Source path does not exist: {source_path}")
|
||||
continue
|
||||
|
||||
logger.info(f"Backing up {source_name} from {source_path}")
|
||||
|
||||
# Create tar for this source
|
||||
source_tar = temp_path / f"{source_name}.tar.gz"
|
||||
|
||||
def tar_filter(tarinfo):
|
||||
"""Filter function to exclude patterns."""
|
||||
for pattern in excludes:
|
||||
# Simple pattern matching (could be enhanced with fnmatch)
|
||||
if pattern.replace('*/', '').replace('/*', '') in tarinfo.name:
|
||||
logger.debug(f"Excluding: {tarinfo.name}")
|
||||
return None
|
||||
return tarinfo
|
||||
|
||||
with tarfile.open(source_tar, 'w:gz') as tar:
|
||||
tar.add(
|
||||
source_path,
|
||||
arcname=source_name,
|
||||
filter=tar_filter,
|
||||
recursive=True
|
||||
)
|
||||
|
||||
source_size = source_tar.stat().st_size / (1024 * 1024) # MB
|
||||
results.append(f"{source_name}: {source_size:.2f}MB")
|
||||
logger.info(f"Backed up {source_name}: {source_size:.2f}MB")
|
||||
|
||||
# Combine all source backups into final archive
|
||||
logger.info("Creating combined backup archive...")
|
||||
with tarfile.open(backup_file, 'w:gz') as final_tar:
|
||||
for item in temp_path.glob('*.tar.gz'):
|
||||
final_tar.add(item, arcname=item.name)
|
||||
|
||||
# Verify backup created
|
||||
if not backup_file.exists():
|
||||
raise Exception("Backup file was not created")
|
||||
|
||||
backup_size = backup_file.stat().st_size / (1024 * 1024) # MB
|
||||
logger.info(f"Backup created successfully: {backup_size:.2f}MB")
|
||||
|
||||
# Clean up old backups
|
||||
await cleanup_old_backups(backup_dir, retention_days)
|
||||
|
||||
# Count remaining backups
|
||||
backup_count = len(list(backup_dir.glob('docker-configs-*.tar.gz')))
|
||||
total_size = sum(f.stat().st_size for f in backup_dir.glob('docker-configs-*.tar.gz'))
|
||||
total_size_mb = total_size / (1024 * 1024)
|
||||
|
||||
output = (
|
||||
f"Backup completed: {backup_filename} ({backup_size:.2f}MB). "
|
||||
f"Sources: {', '.join(results)}. "
|
||||
f"Retention: {backup_count} backups, {total_size_mb:.2f}MB total."
|
||||
)
|
||||
|
||||
logger.info(output)
|
||||
return output
|
||||
|
||||
|
||||
async def cleanup_old_backups(backup_dir: Path, retention_days: int):
|
||||
"""Remove backups older than retention period."""
|
||||
cutoff_date = datetime.now() - timedelta(days=retention_days)
|
||||
removed_count = 0
|
||||
removed_size = 0
|
||||
|
||||
logger.info(f"Cleaning up backups older than {retention_days} days...")
|
||||
|
||||
for backup_file in backup_dir.glob('docker-configs-*.tar.gz'):
|
||||
# Get file modification time
|
||||
file_mtime = datetime.fromtimestamp(backup_file.stat().st_mtime)
|
||||
|
||||
if file_mtime < cutoff_date:
|
||||
file_size = backup_file.stat().st_size
|
||||
logger.info(f"Removing old backup: {backup_file.name} (from {file_mtime:%Y-%m-%d})")
|
||||
backup_file.unlink()
|
||||
removed_count += 1
|
||||
removed_size += file_size
|
||||
|
||||
if removed_count > 0:
|
||||
removed_size_mb = removed_size / (1024 * 1024)
|
||||
logger.info(f"Removed {removed_count} old backups, freed {removed_size_mb:.2f}MB")
|
||||
else:
|
||||
logger.info("No old backups to remove")
|
||||
@@ -0,0 +1,259 @@
|
||||
"""
|
||||
Documentation Sync Executor
|
||||
Syncs documentation from upstream Git repositories to Gitea.
|
||||
Clones source repos, extracts docs directories, pushes to Gitea.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
from src.config import Settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def execute(config: dict, settings: Settings) -> str:
|
||||
"""
|
||||
Sync documentation from upstream Git repo to Gitea.
|
||||
|
||||
Config schema:
|
||||
{
|
||||
"project": "fastapi",
|
||||
"upstream_repo": "https://github.com/tiangolo/fastapi.git",
|
||||
"docs_paths": ["/docs", "/docs_src"],
|
||||
"gitea_repo": "library/docs-fastapi",
|
||||
"gitea_url": "http://gitea:3000",
|
||||
"branch": "main"
|
||||
}
|
||||
|
||||
Args:
|
||||
config: Documentation sync configuration
|
||||
settings: Global scheduler settings
|
||||
|
||||
Returns:
|
||||
Summary of sync operation
|
||||
|
||||
Raises:
|
||||
Exception: On sync failure
|
||||
"""
|
||||
project = config.get('project')
|
||||
upstream_repo = config.get('upstream_repo')
|
||||
docs_paths = config.get('docs_paths', ['/docs'])
|
||||
gitea_repo = config.get('gitea_repo')
|
||||
gitea_url = config.get('gitea_url', settings.gitea_url)
|
||||
branch = config.get('branch', 'main')
|
||||
|
||||
if not all([project, upstream_repo, gitea_repo]):
|
||||
raise ValueError("Missing required config: project, upstream_repo, or gitea_repo")
|
||||
|
||||
logger.info(f"Starting doc sync for {project}")
|
||||
logger.info(f"Upstream: {upstream_repo}")
|
||||
logger.info(f"Gitea: {gitea_repo}")
|
||||
|
||||
# Work directory
|
||||
work_dir = Path(f"/app/task-data/doc-sync/{project}")
|
||||
upstream_dir = work_dir / "upstream"
|
||||
gitea_dir = work_dir / "gitea"
|
||||
|
||||
try:
|
||||
# Clean work directory
|
||||
if work_dir.exists():
|
||||
logger.info(f"Cleaning work directory: {work_dir}")
|
||||
shutil.rmtree(work_dir)
|
||||
work_dir.mkdir(parents=True)
|
||||
|
||||
# Clone upstream repo (shallow clone for speed)
|
||||
logger.info(f"Cloning upstream repo...")
|
||||
await _run_command([
|
||||
'git', 'clone',
|
||||
'--depth', '1',
|
||||
'--branch', branch,
|
||||
upstream_repo,
|
||||
str(upstream_dir)
|
||||
])
|
||||
|
||||
# Get upstream version/commit
|
||||
upstream_commit = await _get_git_commit(upstream_dir)
|
||||
upstream_date = datetime.now().strftime('%Y-%m-%d')
|
||||
logger.info(f"Upstream commit: {upstream_commit[:8]}")
|
||||
|
||||
# Clone Gitea repo (or create if doesn't exist)
|
||||
# Build authenticated URL for Gitea
|
||||
gitea_url_clean = gitea_url.replace('http://', '').replace('https://', '')
|
||||
gitea_clone_url = f"http://{settings.gitea_user}:{settings.gitea_password}@{gitea_url_clean}/{gitea_repo}.git"
|
||||
# Log without credentials
|
||||
logger.info(f"Cloning Gitea repo: {gitea_url}/{gitea_repo}.git")
|
||||
|
||||
# Try to clone, if fails create new repo
|
||||
try:
|
||||
await _run_command([
|
||||
'git', 'clone',
|
||||
gitea_clone_url,
|
||||
str(gitea_dir)
|
||||
])
|
||||
except Exception as e:
|
||||
logger.warning(f"Gitea repo doesn't exist, will create: {e}")
|
||||
gitea_dir.mkdir(parents=True)
|
||||
await _run_command(['git', 'init'], cwd=gitea_dir)
|
||||
await _run_command(['git', 'checkout', '-b', branch], cwd=gitea_dir)
|
||||
|
||||
# Set remote
|
||||
await _run_command([
|
||||
'git', 'remote', 'add', 'origin',
|
||||
gitea_clone_url
|
||||
], cwd=gitea_dir)
|
||||
|
||||
# Clear existing content in Gitea repo (except .git)
|
||||
for item in gitea_dir.iterdir():
|
||||
if item.name != '.git':
|
||||
if item.is_dir():
|
||||
shutil.rmtree(item)
|
||||
else:
|
||||
item.unlink()
|
||||
|
||||
# Determine what to copy based on docs_paths
|
||||
copied_paths = []
|
||||
|
||||
# If docs_paths is empty or contains "." or "/", sync entire repo
|
||||
if not docs_paths or any(p in [".", "/", ""] for p in docs_paths):
|
||||
logger.info(f"Copying entire repository...")
|
||||
for item in upstream_dir.iterdir():
|
||||
if item.name != '.git':
|
||||
dest = gitea_dir / item.name
|
||||
if item.is_dir():
|
||||
shutil.copytree(item, dest)
|
||||
else:
|
||||
shutil.copy2(item, dest)
|
||||
copied_paths = ["entire repository"]
|
||||
else:
|
||||
# Copy only specified paths
|
||||
logger.info(f"Copying specific paths: {docs_paths}")
|
||||
for doc_path in docs_paths:
|
||||
source = upstream_dir / doc_path.lstrip('/')
|
||||
if source.exists():
|
||||
dest = gitea_dir / source.name
|
||||
logger.info(f"Copying {source.name}...")
|
||||
if source.is_dir():
|
||||
shutil.copytree(source, dest)
|
||||
else:
|
||||
shutil.copy2(source, dest)
|
||||
copied_paths.append(source.name)
|
||||
else:
|
||||
logger.warning(f"Path not found in upstream: {doc_path}")
|
||||
|
||||
if not copied_paths:
|
||||
raise Exception("No documentation paths were copied")
|
||||
|
||||
# Create .SYNC_INFO.md with metadata (don't overwrite README.md from upstream)
|
||||
sync_info_path = gitea_dir / ".SYNC_INFO.md"
|
||||
content_desc = "Complete repository mirror" if "entire repository" in copied_paths else f"Paths: {', '.join(copied_paths)}"
|
||||
sync_info_content = f"""# Sync Information
|
||||
|
||||
This is a mirror of the {project.title()} repository.
|
||||
|
||||
**Synced from:** {upstream_repo}
|
||||
**Branch:** {branch}
|
||||
**Commit:** {upstream_commit}
|
||||
**Sync Date:** {upstream_date}
|
||||
**Content:** {content_desc}
|
||||
|
||||
---
|
||||
|
||||
This repository is automatically synced monthly by The Scheduler.
|
||||
For the latest updates, visit the official repository.
|
||||
"""
|
||||
sync_info_path.write_text(sync_info_content)
|
||||
|
||||
# Git add, commit, push
|
||||
await _run_command(['git', 'add', '.'], cwd=gitea_dir)
|
||||
|
||||
# Check if there are changes
|
||||
status = await _run_command(
|
||||
['git', 'status', '--porcelain'],
|
||||
cwd=gitea_dir,
|
||||
capture=True
|
||||
)
|
||||
|
||||
if not status.strip():
|
||||
logger.info("No changes detected, skipping commit")
|
||||
return f"Documentation already up to date (commit: {upstream_commit[:8]})"
|
||||
|
||||
# Commit changes
|
||||
commit_msg = f"Sync {project} docs from {upstream_commit[:8]} on {upstream_date}"
|
||||
await _run_command([
|
||||
'git', 'commit',
|
||||
'-m', commit_msg
|
||||
], cwd=gitea_dir)
|
||||
|
||||
# Tag with date
|
||||
tag = f"sync-{upstream_date}"
|
||||
await _run_command([
|
||||
'git', 'tag', '-f', tag,
|
||||
'-m', f"Documentation snapshot {upstream_date}"
|
||||
], cwd=gitea_dir)
|
||||
|
||||
# Push to Gitea
|
||||
logger.info("Pushing to Gitea...")
|
||||
await _run_command([
|
||||
'git', 'push', 'origin', branch, '--force'
|
||||
], cwd=gitea_dir)
|
||||
|
||||
await _run_command([
|
||||
'git', 'push', 'origin', tag, '--force'
|
||||
], cwd=gitea_dir)
|
||||
|
||||
# Cleanup
|
||||
logger.info("Cleaning up work directory...")
|
||||
shutil.rmtree(work_dir)
|
||||
|
||||
result = (
|
||||
f"Successfully synced {project} documentation. "
|
||||
f"Copied: {', '.join(copied_paths)}. "
|
||||
f"Upstream commit: {upstream_commit[:8]}. "
|
||||
f"Tagged: {tag}"
|
||||
)
|
||||
logger.info(result)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
# Cleanup on error
|
||||
if work_dir.exists():
|
||||
shutil.rmtree(work_dir)
|
||||
raise
|
||||
|
||||
|
||||
async def _run_command(
|
||||
cmd: List[str],
|
||||
cwd: Optional[Path] = None,
|
||||
capture: bool = False
|
||||
) -> str:
|
||||
"""Run shell command asynchronously."""
|
||||
logger.debug(f"Running: {' '.join(cmd)} (cwd: {cwd})")
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
cwd=cwd,
|
||||
stdout=asyncio.subprocess.PIPE if capture else asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE
|
||||
)
|
||||
|
||||
stdout, stderr = await proc.communicate()
|
||||
|
||||
if proc.returncode != 0:
|
||||
error_msg = stderr.decode() if stderr else "Unknown error"
|
||||
raise Exception(f"Command failed: {' '.join(cmd)}\n{error_msg}")
|
||||
|
||||
return stdout.decode() if capture else ""
|
||||
|
||||
|
||||
async def _get_git_commit(repo_dir: Path) -> str:
|
||||
"""Get current git commit hash."""
|
||||
output = await _run_command(
|
||||
['git', 'rev-parse', 'HEAD'],
|
||||
cwd=repo_dir,
|
||||
capture=True
|
||||
)
|
||||
return output.strip()
|
||||
@@ -0,0 +1,42 @@
|
||||
"""
|
||||
Example executor demonstrating the pattern.
|
||||
Shows how to write task executors for The Scheduler.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from src.config import Settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def execute(config: dict, settings: Settings) -> str:
|
||||
"""
|
||||
Example task executor.
|
||||
|
||||
Args:
|
||||
config: Task configuration from scheduled_tasks.config JSONB field
|
||||
Example: {"message": "Hello", "delay_seconds": 2}
|
||||
settings: Global scheduler settings (database, API keys, etc.)
|
||||
|
||||
Returns:
|
||||
Output message describing what was done
|
||||
|
||||
Raises:
|
||||
Exception: On failure (will be caught and logged by executor framework)
|
||||
"""
|
||||
message = config.get('message', 'No message configured')
|
||||
delay = config.get('delay_seconds', 1)
|
||||
|
||||
logger.info(f"Example executor starting: {message}")
|
||||
|
||||
# Simulate some work
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
# You can access settings
|
||||
logger.info(f"Using database: {settings.postgres_db}")
|
||||
|
||||
# Return success message
|
||||
output = f"Executed example task: {message} (took {delay}s)"
|
||||
logger.info(output)
|
||||
|
||||
return output
|
||||
Reference in New Issue
Block a user