tessl/pypi-tutor
The Docker-based Open edX distribution designed for peace of mind
task-execution.mddocs/
Task Execution
Abstract task runner framework supporting different deployment contexts (Local, Dev, Kubernetes) with Docker Compose integration. The task system provides a unified interface for executing commands across various deployment environments.
Capabilities
Base Task Runner
Abstract base class for all task execution contexts providing common functionality.
class BaseTaskRunner:
"""
Abstract base class for task execution in different environments.
"""
def __init__(self, root: str, config: Config):
"""
Initialize task runner with root directory and configuration.
Args:
root (str): Project root directory path
config (Config): Configuration dictionary
"""
def run_task_from_template(self, service: str, *path: str) -> None:
"""
Run a task command loaded from a template file.
Args:
service (str): Service name to run task in
*path (str): Path components to template file containing command
"""
def run_task_from_str(self, service: str, command: str) -> None:
"""
Run a task command from a string.
Args:
service (str): Service name to run task in
command (str): Command string to execute
"""
def render(self, *path: str) -> str:
"""
Render a template file with current configuration.
Args:
*path (str): Path components to template file
Returns:
str: Rendered template content
"""
def run_task(self, service: str, command: str) -> int:
"""
Execute a command in the specified service. Must be implemented by subclasses.
Args:
service (str): Service name to run command in
command (str): Command to execute
Returns:
int: Exit code of the command
"""Docker Compose Task Runner
Base class for task runners that use Docker Compose for service management.
class BaseComposeTaskRunner(BaseTaskRunner):
"""
Base class for Docker Compose-based task runners.
"""
def docker_compose(self, *command: str) -> int:
"""
Execute a docker-compose command. Must be implemented by subclasses.
Args:
*command (str): Docker Compose command arguments
Returns:
int: Exit code of the docker-compose command
"""
def run_task(self, service: str, command: str) -> int:
"""
Execute a command in a service using docker-compose exec.
Args:
service (str): Service name to run command in
command (str): Command to execute
Returns:
int: Exit code of the command
"""
def start(self) -> None:
"""
Start all services using docker-compose up.
"""
def stop(self) -> None:
"""
Stop all services using docker-compose stop.
"""
def restart(self) -> None:
"""
Restart all services.
"""
def logs(self, *services: str, follow: bool = False, tail: int = None) -> None:
"""
Display logs from services.
Args:
*services (str): Service names to show logs for (empty for all)
follow (bool): Follow log output
tail (int): Number of lines to show from end
"""Local Task Runner
Task runner for local development with Docker Compose.
class LocalTaskRunner(BaseComposeTaskRunner):
"""
Task runner for local development environment.
"""
def docker_compose(self, *command: str) -> int:
"""
Execute docker-compose command with local configuration.
Args:
*command (str): Docker Compose command arguments
Returns:
int: Exit code of the command
"""
def is_running(self) -> bool:
"""
Check if local services are currently running.
Returns:
bool: True if services are running
"""
def launch(self, non_interactive: bool = False, pullimages: bool = False,
skip_build: bool = False) -> None:
"""
Configure and launch the platform from scratch.
Args:
non_interactive (bool): Skip interactive configuration
pullimages (bool): Pull images before starting
skip_build (bool): Skip building custom images
"""Development Task Runner
Task runner for development mode with additional debugging capabilities.
class DevTaskRunner(BaseComposeTaskRunner):
"""
Task runner for development environment with debugging features.
"""
def docker_compose(self, *command: str) -> int:
"""
Execute docker-compose command with development configuration.
Args:
*command (str): Docker Compose command arguments
Returns:
int: Exit code of the command
"""
def hosts(self) -> None:
"""
Display status of all services with their URLs and access information.
"""
def launch(self, non_interactive: bool = False, pullimages: bool = False,
skip_build: bool = False) -> None:
"""
Configure and launch the platform in development mode.
Args:
non_interactive (bool): Skip interactive configuration
pullimages (bool): Pull images before starting
skip_build (bool): Skip building custom images
"""Kubernetes Task Runner
Task runner for Kubernetes deployments using kubectl.
class K8sTaskRunner(BaseTaskRunner):
"""
Task runner for Kubernetes deployment environment.
"""
def kubectl(self, *command: str) -> int:
"""
Execute a kubectl command.
Args:
*command (str): kubectl command arguments
Returns:
int: Exit code of the kubectl command
"""
def run_task(self, service: str, command: str) -> int:
"""
Execute a command in a Kubernetes pod.
Args:
service (str): Service/pod name to run command in
command (str): Command to execute
Returns:
int: Exit code of the command
"""
def start(self) -> None:
"""
Start all Kubernetes resources.
"""
def stop(self) -> None:
"""
Stop all Kubernetes resources.
"""
def launch(self, non_interactive: bool = False, pullimages: bool = False,
skip_build: bool = False) -> None:
"""
Configure and launch the platform on Kubernetes.
Args:
non_interactive (bool): Skip interactive configuration
pullimages (bool): Pull images before starting
skip_build (bool): Skip building custom images
"""
def logs(self, *services: str, follow: bool = False, tail: int = None) -> None:
"""
Display logs from Kubernetes pods.
Args:
*services (str): Service names to show logs for
follow (bool): Follow log output
tail (int): Number of lines to show from end
"""Task Context Management
Context classes that provide environment-specific settings and behavior.
class Context:
"""
Base context providing access to configuration and root directory.
"""
def __init__(self, root: str):
"""
Initialize context with project root.
Args:
root (str): Project root directory path
"""
@property
def root(self) -> str:
"""
Get the project root directory.
Returns:
str: Project root directory path
"""
class LocalContext(Context):
"""
Context for local development environment.
"""
pass
class DevContext(Context):
"""
Context for development environment with debugging features.
"""
pass
class K8sContext(Context):
"""
Context for Kubernetes deployment environment.
"""
passCommon Services
Standard Service Names
Services available across different deployment environments.
# Core Open edX services
"lms" # Learning Management System
"cms" # Content Management System (Studio)
"lms-worker" # LMS Celery worker
"cms-worker" # CMS Celery worker
# Supporting services
"mysql" # MySQL database
"redis" # Redis cache and message broker
"elasticsearch" # Search engine (if enabled)
"mongodb" # MongoDB (if enabled)
"caddy" # Reverse proxy (local/dev)
"nginx" # Web server (production)
# Development services
"lms-dev" # LMS development server
"cms-dev" # CMS development server
"openedx-dev" # Development image builderUsage Examples
Local Development
from tutor.commands.local import LocalTaskRunner, LocalContext
from tutor import config
# Initialize local environment
context = LocalContext("/path/to/tutor/root")
config_data = config.load(context.root)
runner = LocalTaskRunner(context.root, config_data)
# Launch platform
runner.launch(non_interactive=True)
# Check if running
if runner.is_running():
print("Platform is running")
# Run a command in LMS
runner.run_task("lms", "./manage.py lms shell")
# View logs
runner.logs("lms", "cms", follow=True)Development Mode
from tutor.commands.dev import DevTaskRunner, DevContext
from tutor import config
# Initialize development environment
context = DevContext("/path/to/tutor/root")
config_data = config.load(context.root)
runner = DevTaskRunner(context.root, config_data)
# Launch in development mode
runner.launch()
# Show service URLs
runner.hosts()
# Run development tasks
runner.run_task("lms-dev", "python manage.py createsuperuser")Kubernetes Deployment
from tutor.commands.k8s import K8sTaskRunner, K8sContext
from tutor import config
# Initialize Kubernetes environment
context = K8sContext("/path/to/tutor/root")
config_data = config.load(context.root)
runner = K8sTaskRunner(context.root, config_data)
# Deploy to Kubernetes
runner.launch(non_interactive=True)
# Check pod status
runner.kubectl("get", "pods")
# Execute command in pod
runner.run_task("lms", "python manage.py migrate")
# View pod logs
runner.logs("lms", follow=True, tail=100)Custom Task Execution
from tutor.tasks import BaseTaskRunner
from tutor import config
class CustomTaskRunner(BaseTaskRunner):
def run_task(self, service: str, command: str) -> int:
# Custom implementation for task execution
print(f"Running {command} in {service}")
return 0
# Use custom runner
config_data = config.load("/path/to/tutor/root")
runner = CustomTaskRunner("/path/to/tutor/root", config_data)
# Run template-based task
runner.run_task_from_template("lms", "tasks", "migrate.sh")
# Run string command
runner.run_task_from_str("cms", "python manage.py collectstatic")Template-based Commands
from tutor.commands.local import LocalTaskRunner, LocalContext
from tutor import config
context = LocalContext("/path/to/tutor/root")
config_data = config.load(context.root)
runner = LocalTaskRunner(context.root, config_data)
# Create template file at templates/tasks/custom-setup.sh
# Content: python manage.py migrate && python manage.py collectstatic
# Run task from template
runner.run_task_from_template("lms", "tasks", "custom-setup.sh")
# Template will be rendered with configuration variablesInstall with Tessl CLI
npx tessl i tessl/pypi-tutor