### Set up local development environment for Invenio-Jobs Source: https://github.com/inveniosoftware/invenio-jobs/blob/master/CONTRIBUTING.rst Commands to clone the repository, create a virtual environment, and install dependencies for local development. ```bash git clone git@github.com:your_name_here/invenio-jobs.git mkvirtualenv invenio-jobs cd invenio-jobs/ pip install -e .[all] ``` -------------------------------- ### Create a New Job via REST API (cURL) Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Demonstrates how to create a new job configuration using a cURL command. This example shows creating a job with an interval schedule (every 4 hours), specifying a default queue, notification emails, and notification statuses. ```bash # Create a new job with interval schedule (every 4 hours) curl -X POST "https://your-instance.com/api/jobs" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{ "title": "Sync External Records", "task": "sync_records", "description": "Synchronize records from external database", "active": true, "default_queue": "low", "schedule": { "type": "interval", "hours": 4 }, "notification_emails": ["admin@example.com"], "notification_statuses": ["FAILED", "PARTIAL_SUCCESS"] }' # Response: # { ``` -------------------------------- ### CLI Commands for Invenio Jobs Management (Bash) Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Provides examples of command-line interface (CLI) commands for managing jobs within the Invenio Jobs module. These commands allow for job orchestration without direct interaction with the web interface. The specific commands are not detailed in the provided text, but the section indicates their availability. ```bash # The invenio-jobs module provides CLI commands for job management. # Example usage is not provided in the source text. # Refer to the official documentation for specific commands. ``` -------------------------------- ### Start Custom Celery Beat Scheduler - Bash Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Starts the Celery Beat scheduler using a custom `RunScheduler` that reads job schedules directly from the database. This allows for dynamic scheduling without static configuration files. ```bash celery -A invenio_app.celery beat \ -l ERROR \ --scheduler invenio_jobs.services.scheduler:RunScheduler \ -s /var/run/celery-schedule \ --pidfile /var/run/celerybeat.pid ``` -------------------------------- ### Define Custom Job Type with Celery Task and Schema Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Illustrates how to define a custom job type by extending `JobType`. This includes defining a Celery task, a Marshmallow schema for task arguments, and a method to build those arguments at runtime. The example shows a `SyncRecordsJob` with a `sync_records_task` and a `SyncRecordsArgsSchema`. ```python from datetime import datetime, timezone from marshmallow import Schema, fields from marshmallow_utils.fields import TZDateTime from invenio_jobs.jobs import JobType, PredefinedArgsSchema from celery import shared_task # Define a Celery task @shared_task def sync_records_task(since=None, batch_size=100): """Synchronize records modified since a given date.""" from flask import current_app current_app.logger.info(f"Syncing records since {since} with batch size {batch_size}") # Your sync logic here return {"processed": 150, "errors": 0} # Define custom arguments schema class SyncRecordsArgsSchema(PredefinedArgsSchema): """Schema for sync records arguments.""" job_arg_schema = fields.String( metadata={"type": "hidden"}, dump_default="SyncRecordsArgsSchema", load_default="SyncRecordsArgsSchema", ) since = TZDateTime( timezone=timezone.utc, format="iso", metadata={ "description": "Start date for sync (ISO 8601 format). Leave empty to continue from last successful run." }, allow_none=True, ) batch_size = fields.Integer( load_default=100, metadata={"description": "Number of records to process per batch."} ) # Define the job type class SyncRecordsJob(JobType): """Job type for synchronizing records.""" id = "sync_records" title = "Sync Records" description = "Synchronizes records from external sources" task = sync_records_task arguments_schema = SyncRecordsArgsSchema @classmethod def build_task_arguments(cls, job_obj, since=None, batch_size=100, **kwargs): """Build arguments passed to the task at execution time.""" return { "since": since, "batch_size": batch_size, } # Register in setup.cfg entry points: # [options.entry_points] # invenio_jobs.jobs = # sync_records = mymodule.jobs:SyncRecordsJob ``` -------------------------------- ### GET /api/logs/jobs Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Retrieve execution logs for jobs, searchable by run ID. ```APIDOC ## GET /api/logs/jobs ### Description Fetches indexed logs for job executions. Supports search queries via the 'q' parameter. ### Method GET ### Endpoint /api/logs/jobs ### Parameters #### Query Parameters - **q** (string) - Required - Search query, typically the run ID ### Response #### Success Response (200) - **hits** (object) - Contains total count and array of log entries ``` -------------------------------- ### Initialize InvenioJobs Extension with Flask Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Demonstrates how to initialize the InvenioJobs extension within a Flask application. It configures job queues, logging settings, and provides access to extension functionalities like registered jobs and queues via a proxy. ```python from flask import Flask from invenio_jobs import InvenioJobs app = Flask(__name__) app.config.update( JOBS_QUEUES={ "celery": { "name": "celery", "title": "Default", "description": "Default queue", }, "low": { "name": "low", "title": "Low", "description": "Low priority queue", }, }, JOBS_DEFAULT_QUEUE=None, # Uses Celery's default if not set JOBS_LOGGING=True, JOBS_LOGGING_LEVEL="DEBUG", JOBS_LOGGING_INDEX="job-logs", JOBS_LOGGING_RETENTION_DAYS=90, ) # Initialize the extension jobs_ext = InvenioJobs(app) # Access registered jobs with app.app_context(): from invenio_jobs.proxies import current_jobs # Get all registered job types all_jobs = current_jobs.jobs # Get available queues queues = current_jobs.queues # Get default queue default_queue = current_jobs.default_queue ``` -------------------------------- ### Programmatic Job Management with Invenio Jobs Service (Python) Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Demonstrates how to interact with the Invenio Jobs service programmatically using Python. This includes creating, searching, reading, updating, and deleting jobs, as well as managing runs and retrieving logs. It requires the invenio_jobs and invenio_access libraries. ```python from flask import Flask from invenio_access.permissions import system_identity from invenio_jobs.proxies import ( current_jobs_service, current_runs_service, current_tasks_service, current_jobs_logs_service, ) app = Flask(__name__) with app.app_context(): # Create a job job_data = { "title": "Data Export Job", "task": "export_records", "description": "Export records to external system", "active": True, "default_queue": "low", "schedule": {"type": "interval", "hours": 6}, } job_result = current_jobs_service.create(system_identity, job_data) job_id = job_result._obj.id print(f"Created job: {job_id}") # Search for jobs search_results = current_jobs_service.search( system_identity, params={"q": "export", "size": 10} ) for hit in search_results.to_dict()["hits"]["hits"]: print(f"Found job: {hit['title']}") # Read a specific job job = current_jobs_service.read(system_identity, job_id) print(f"Job details: {job.to_dict()}") # Update job updated_job = current_jobs_service.update( system_identity, job_id, { "title": "Data Export Job", "task": "export_records", "active": False, "default_queue": "celery", "args": {}, "custom_args": "{}", } ) # Trigger a run run_data = { "title": "Manual export run", "queue": "celery", "args": {}, } run_result = current_runs_service.create(system_identity, job_id, run_data) run_id = run_result._obj.id print(f"Started run: {run_id}") # Get run status run = current_runs_service.read(system_identity, job_id, run_id) print(f"Run status: {run.to_dict()['status']}") # Stop a run stopped_run = current_runs_service.stop(system_identity, job_id, run_id) # Search for logs logs = current_jobs_logs_service.search( system_identity, params={"q": str(run_id)} ) for log in logs.to_dict()["hits"]["hits"]: print(f"[{log['level']}] {log['message']}") # Delete job (will cascade delete runs) current_jobs_service.delete(system_identity, job_id) ``` -------------------------------- ### Run Celery Scheduler for Invenio-Jobs Source: https://github.com/inveniosoftware/invenio-jobs/blob/master/README.rst This command initiates the Celery beat process using the custom Invenio-Jobs scheduler. It requires specifying the application instance and paths for the schedule file and PID file to ensure proper job execution. ```console celery -A invenio_app.celery beat -l ERROR --scheduler invenio_jobs.services.scheduler:RunScheduler -s /var/run/celery-schedule --pidfile /var/run/celerybeat.pid ``` -------------------------------- ### POST /api/jobs Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Create a new background job with a specific task and schedule configuration. ```APIDOC ## POST /api/jobs ### Description Creates a new job definition. Supports both interval and crontab scheduling. ### Method POST ### Endpoint /api/jobs ### Request Body - **title** (string) - Required - Name of the job - **task** (string) - Required - The task identifier to execute - **active** (boolean) - Optional - Whether the job is enabled - **schedule** (object) - Required - Scheduling configuration (type: interval or crontab) ### Request Example { "title": "Daily Embargo Check", "task": "update_expired_embargos", "active": true, "schedule": { "type": "crontab", "minute": "0", "hour": "0" } } ``` -------------------------------- ### Subtask Management for Batch Processing (Python) Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Illustrates how to manage subtasks for batch processing within a job using the invenio_jobs service. This involves adding total entries, creating subtask runs, simulating processing, and finalizing subtasks with their results. It requires the invenio_access and invenio_jobs libraries. ```python from invenio_access.permissions import system_identity from invenio_jobs.proxies import current_runs_service def process_batch_job(run_id, job_id, items): """Example of a job that spawns subtasks for batch processing.""" # Add total entries to track current_runs_service.add_total_entries( system_identity, run_id=run_id, job_id=job_id, total_entries=len(items) ) # Create subtasks for each batch batch_size = 100 for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] # Create a subtask run subtask = current_runs_service.create_subtask_run( system_identity, parent_run_id=run_id, job_id=job_id, args={"batch_start": i, "batch_size": len(batch)} ) subtask_id = subtask._obj.id # Process subtask (in a real scenario, this would be async) current_runs_service.start_processing_subtask( system_identity, run_id=subtask_id, job_id=job_id ) # Simulate processing with some errors errors = 2 if i == 0 else 0 # Finalize subtask with results current_runs_service.finalize_subtask( system_identity, run_id=subtask_id, job_id=job_id, success=(errors == 0), errored_entries_count=errors, inserted_entries_count=len(batch) - errors, updated_entries_count=0 ) # Parent run status is automatically updated when all subtasks complete ``` -------------------------------- ### Retrieve Job Logs Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Query indexed job logs from OpenSearch. Logs are searchable by run ID and support pagination for large datasets. ```bash curl "https://your-instance.com/api/logs/jobs?q=660e8400-e29b-41d4-a716-446655440001" -H "Authorization: Bearer $TOKEN" ``` -------------------------------- ### List All Jobs - Invenio CLI Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Displays a list of all currently registered jobs. It shows details such as Job ID, Title, Queue, Task name, and whether the job is active. ```bash invenio jobs list ``` -------------------------------- ### Configure Invenio-Jobs Logging Settings Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Defines the configuration variables for enabling job logging, setting log levels, specifying the OpenSearch index, and managing log retention and batching policies. ```python JOBS_LOGGING = True JOBS_LOGGING_LEVEL = "DEBUG" JOBS_LOGGING_INDEX = "job-logs" JOBS_LOGGING_RETENTION_DAYS = 90 JOBS_LOGS_MAX_RESULTS = 2000 JOBS_LOGS_BATCH_SIZE = 500 ``` -------------------------------- ### Manage development branches and testing Source: https://github.com/inveniosoftware/invenio-jobs/blob/master/CONTRIBUTING.rst Commands to create a new feature branch, execute the test suite, and commit changes to the repository. ```bash git checkout -b name-of-your-bugfix-or-feature ./run-tests.sh git add . git commit -s -m "component: title without verbs" -m "* NEW Adds your new feature." -m "* FIX Fixes an existing issue." -m "* BETTER Improves and existing feature." -m "* Changes something that should not be visible in release notes." git push origin name-of-your-bugfix-or-feature ``` -------------------------------- ### Create a New Job - Invenio CLI Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Creates a new job with a specified title and task name. This is the primary command for adding new automated tasks to the system. ```bash invenio jobs create --title "My New Job" --task "sync_records" ``` -------------------------------- ### Invenio Jobs Configuration Reference - Python Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Provides a reference for configuring the Invenio Jobs module through Flask's application configuration. This includes setting permission policies, defining queues, and configuring search options. ```python # config.py # Permission policies JOBS_TASKS_PERMISSION_POLICY = TasksPermissionPolicy # Who can list/read tasks JOBS_PERMISSION_POLICY = JobPermissionPolicy # Who can CRUD jobs JOBS_RUNS_PERMISSION_POLICY = RunPermissionPolicy # Who can manage runs APP_LOGS_PERMISSION_POLICY = JobLogsPermissionPolicy # Who can search logs # Queue configuration JOBS_QUEUES = { "celery": {"name": "celery", "title": "Default", "description": "Default queue"}, "low": {"name": "low", "title": "Low Priority", "description": "Low priority tasks"}, "high": {"name": "high", "title": "High Priority", "description": "Urgent tasks"}, } JOBS_DEFAULT_QUEUE = None # Falls back to Celery's task_default_queue # Search/sort options JOBS_SORT_OPTIONS = { "jobs": {"title": "Jobs", "fields": ["jobs"]}, "last_run_start_time": {"title": "Last run", "fields": ["last_run_start_time"]}, } JOBS_SEARCH = { "facets": [], "sort": ["jobs", "last_run_start_time", "user", "next_run"], } ``` -------------------------------- ### POST /api/jobs/{id}/runs Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Trigger a manual execution of a specific job with custom arguments. ```APIDOC ## POST /api/jobs/{id}/runs ### Description Manually triggers a new execution run for an existing job. ### Method POST ### Endpoint /api/jobs/{id}/runs ### Parameters #### Path Parameters - **id** (string) - Required - The unique identifier of the job ### Request Body - **title** (string) - Optional - Name for the run - **args** (object) - Optional - Execution arguments ### Response #### Success Response (200) - **id** (string) - The unique run ID - **status** (string) - Current status (e.g., QUEUED) ``` -------------------------------- ### Job Logging with Context - Python Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Demonstrates how to automatically enrich logs with job context (job_id, run_id, etc.) within Celery tasks. It also shows how to manually set context for non-task code. ```python from flask import current_app from invenio_jobs.logging.jobs import set_job_context, job_context from celery import shared_task # Within a Celery task, context is automatically set @shared_task(bind=True) def my_task(self, **kwargs): """Task with automatic logging context.""" # Logs are automatically enriched with job_id, run_id, task_id current_app.logger.info("Starting processing") current_app.logger.warning("Found duplicate records") current_app.logger.error("Failed to process item X") # Manually set context for testing or non-task code with set_job_context({ "job_id": "550e8400-e29b-41d4-a716-446655440000", "run_id": "660e8400-e29b-41d4-a716-446655440001", "identity_id": "1", "task_id": "abc123", "parent_task_id": None, }): current_app.logger.info("This log will be indexed with full context") # Check current context ctx = job_context.get() if ctx is not job_context.default: print(f"Currently in job: {ctx['job_id']}") ``` -------------------------------- ### List Job Types - Invenio CLI Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Lists all registered job types available in the Invenio system. This command helps users understand the available tasks that can be scheduled or run. ```bash invenio jobs types ``` -------------------------------- ### Manage Job Runs - Invenio CLI Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Commands to list runs for a specific job, trigger a new run, and view job logs. These commands are essential for monitoring job execution and debugging. ```bash invenio jobs runs 550e8400-e29b-41d4-a716-446655440000 invenio jobs run 550e8400-e29b-41d4-a716-446655440000 invenio jobs log 660e8400-e29b-41d4-a716-446655440001 invenio jobs log 660e8400-e29b-41d4-a716-446655440001 -f ``` -------------------------------- ### Manage Jobs via REST API Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Perform CRUD operations on background jobs. This includes creating jobs with specific schedules, listing existing jobs, updating configurations, and deleting jobs. ```bash curl -X POST "https://your-instance.com/api/jobs" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{ "title": "Daily Embargo Check", "task": "update_expired_embargos", "active": true, "default_queue": "celery", "schedule": { "type": "crontab", "minute": "0", "hour": "0", "day_of_week": "*", "day_of_month": "*", "month_of_year": "*" } }' curl "https://your-instance.com/api/jobs?q=sync" -H "Authorization: Bearer $TOKEN" curl "https://your-instance.com/api/jobs/550e8400-e29b-41d4-a716-446655440000" -H "Authorization: Bearer $TOKEN" curl -X PUT "https://your-instance.com/api/jobs/550e8400-e29b-41d4-a716-446655440000" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"title": "Sync External Records", "task": "sync_records", "active": true}' curl -X DELETE "https://your-instance.com/api/jobs/550e8400-e29b-41d4-a716-446655440000" -H "Authorization: Bearer $TOKEN" ``` -------------------------------- ### Manage Job Runs via REST API Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Control the execution of jobs by triggering new runs, listing historical runs, and stopping active processes. Supports optional subtask inclusion for complex workflows. ```bash curl -X POST "https://your-instance.com/api/jobs/550e8400-e29b-41d4-a716-446655440000/runs" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"title": "Manual sync run", "queue": "celery", "args": {"batch_size": 50}}' curl "https://your-instance.com/api/jobs/550e8400-e29b-41d4-a716-446655440000/runs?include_subtasks=true" -H "Authorization: Bearer $TOKEN" curl -X POST "https://your-instance.com/api/jobs/550e8400-e29b-41d4-a716-446655440000/runs/660e8400-e29b-41d4-a716-446655440001/actions/stop" -H "Authorization: Bearer $TOKEN" ``` -------------------------------- ### Schedule Jobs - Invenio CLI Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Schedules a job to run at a specific interval using crontab syntax. It also supports specifying a timezone for the schedule. ```bash invenio jobs schedule 550e8400-e29b-41d4-a716-446655440000 --schedule "*/5 * * * *" invenio jobs schedule 550e8400-e29b-41d4-a716-446655440000 --schedule "0 8 * * *" --tz "Europe/Zurich" ``` -------------------------------- ### Display Run Notification Plain Text Email Template Source: https://github.com/inveniosoftware/invenio-jobs/blob/master/invenio_jobs/templates/semantic-ui/invenio_jobs/emails/run_notification.txt This Jinja2 template generates a plain text email notification for Invenio Jobs runs. It uses context variables to display the run status, user messages, action items, and detailed technical information. Conditional logic is included to show summary information for partial success or failure. ```jinja2 {# Run notification plain text email template Context variables: - job: Job object - run: Run object - status_info: Dict with title, user_message, action, color - run_url: URL to view run details - success_count: (optional) Number of successfully processed items #} {{ status_info.user_message }} {{ status_info.action }} View full details: {{ run_url }} {% if run.status.name == "PARTIAL_SUCCESS" and run.errored_entries and run.total_entries %} Summary: - Successfully processed: {{ success_count }} items - Failed to process: {{ run.errored_entries }} items {% endif %} --- Technical Details: Status: {{ run.status.name }} Run ID: {{ run.id }} Job ID: {{ job.id }} {% if run.started_at %}Started at: {{ run.started_at }} {% endif %}{% if run.finished_at %}Finished at: {{ run.finished_at }} {% endif %}{% if (run.status.name == "FAILED" or run.status.name == "PARTIAL_SUCCESS") and run.failed_subtasks %}Failed subtasks: {{ run.failed_subtasks }}/{{ run.total_subtasks }} {% endif %} {% if run.message %} Error details: {{ run.message }} {% endif %} ``` -------------------------------- ### Jobs REST API Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt The Jobs REST API provides CRUD operations for managing job configurations. Jobs can be scheduled using crontab or interval patterns and configured with default arguments. ```APIDOC ## POST /api/jobs ### Description Creates a new job configuration. Jobs can be scheduled using crontab or interval patterns and configured with default arguments. ### Method POST ### Endpoint /api/jobs ### Parameters #### Request Body - **title** (string) - Required - The title of the job. - **task** (string) - Required - The identifier of the Celery task to run. - **description** (string) - Optional - A description of the job. - **active** (boolean) - Optional - Whether the job is active (defaults to true). - **default_queue** (string) - Optional - The default Celery queue to use for the job. - **schedule** (object) - Optional - The scheduling configuration for the job. - **type** (string) - Required - The type of schedule ('interval' or 'crontab'). - **hours** (integer) - Required if type is 'interval' - The interval in hours. - **minutes** (integer) - Required if type is 'interval' - The interval in minutes. - **day_of_month** (string) - Required if type is 'crontab' - Day of the month (e.g., '*/5'). - **day_of_week** (string) - Required if type is 'crontab' - Day of the week (e.g., '*'). - **hour** (string) - Required if type is 'crontab' - Hour of the day (e.g., '0'). - **minute** (string) - Required if type is 'crontab' - Minute of the hour (e.g., '0'). - **notification_emails** (array of strings) - Optional - Email addresses to notify upon job completion. - **notification_statuses** (array of strings) - Optional - Job statuses that trigger notifications (e.g., 'FAILED', 'PARTIAL_SUCCESS'). ### Request Example ```json { "title": "Sync External Records", "task": "sync_records", "description": "Synchronize records from external database", "active": true, "default_queue": "low", "schedule": { "type": "interval", "hours": 4 }, "notification_emails": ["admin@example.com"], "notification_statuses": ["FAILED", "PARTIAL_SUCCESS"] } ``` ### Response #### Success Response (201 Created) - **id** (string) - The unique identifier of the created job. - **title** (string) - The title of the job. - **task** (string) - The identifier of the Celery task. - **description** (string) - The description of the job. - **active** (boolean) - Whether the job is active. - **default_queue** (string) - The default Celery queue. - **schedule** (object) - The scheduling configuration. - **notification_emails** (array of strings) - Email addresses for notifications. - **notification_statuses** (array of strings) - Job statuses triggering notifications. #### Response Example ```json { "id": "a1b2c3d4-e5f6-7890-1234-567890abcdef", "title": "Sync External Records", "task": "sync_records", "description": "Synchronize records from external database", "active": true, "default_queue": "low", "schedule": { "type": "interval", "hours": 4 }, "notification_emails": ["admin@example.com"], "notification_statuses": ["FAILED", "PARTIAL_SUCCESS"] } ``` ``` -------------------------------- ### Delete a Job - Invenio CLI Source: https://context7.com/inveniosoftware/invenio-jobs/llms.txt Deletes a specified job by its ID. It includes an option to bypass the confirmation prompt for automated deletion. ```bash invenio jobs delete 550e8400-e29b-41d4-a716-446655440000 invenio jobs delete 550e8400-e29b-41d4-a716-446655440000 --yes ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.