
Research
Security News
The Growing Risk of Malicious Browser Extensions
Socket researchers uncover how browser extensions in trusted stores are used to hijack sessions, redirect traffic, and manipulate user behavior.
Lightweight SQLite Job Queue
GigQ is a lightweight job queue system with SQLite as its backend. It provides a reliable way to manage and execute small jobs ("gigs") locally with atomicity guarantees, particularly suited for processing tasks like data transformations, API calls, or batch operations.
Zero External Dependencies
Simple Job Definition & Management
SQLite State Storage
Lightweight Concurrency
Basic Recovery
CLI Interface
The GigQ library is organized as follows:
gigq/
āāā docs/ # Documentation
ā āāā advanced/ # Advanced topics
ā āāā api/ # API reference
ā āāā examples/ # Documentation examples
ā āāā getting-started/ # Getting started guides
ā āāā user-guide/ # User guides
ā
āāā examples/ # Example applications
ā āāā __init__.py
ā āāā github_archive.py # GitHub Archive processing example
ā
āāā gigq/ # Main package code
ā āāā __init__.py # Package initialization and exports
ā āāā job.py # Job class implementation
ā āāā job_status.py # JobStatus enum implementation
ā āāā job_queue.py # JobQueue class implementation
ā āāā worker.py # Worker class implementation
ā āāā workflow.py # Workflow class implementation
ā āāā utils.py # Utility functions
ā āāā cli.py # Command-line interface
ā āāā table_formatter.py # Table formatting utilities
ā
āāā tests/ # Test directory
ā āāā __init__.py # Test package initialization
ā āāā README.md # Test documentation
ā āāā job_functions.py # Shared test functions
ā ā
ā āāā unit/ # Unit tests
ā ā āāā __init__.py
ā ā āāā run_all.py # Run all unit tests
ā ā āāā test_cli.py # CLI unit tests
ā ā āāā test_cli_formatter.py # CLI formatter tests
ā ā āāā test_job.py # Job class tests
ā ā āāā test_job_queue.py # JobQueue class tests
ā ā āāā test_table_formatter.py # Table formatter tests
ā ā āāā test_worker.py # Worker class tests
ā ā āāā test_workflow.py # Workflow class tests
ā ā āāā test_refactoring.py # Tests for refactored modules
ā ā
ā āāā integration/ # Integration tests
ā āāā __init__.py
ā āāā base.py # Base class for integration tests
ā āāā run_all.py # Run all integration tests
ā āāā test_basic.py # Basic job processing tests
ā āāā test_basic_workflow.py # Simple workflow tests
ā āāā test_cli.py # CLI integration tests
ā āāā test_concurrent_workers.py # Multiple workers tests
ā āāā test_error_handling.py # Error handling tests
ā āāā test_persistence.py # Persistence tests
ā āāā test_timeout_handling.py # Timeout handling tests
ā āāā test_workflow_dependencies.py # Workflow dependencies tests
ā
āāā .github/ # GitHub configuration
ā āāā workflows/ # GitHub Actions workflows
ā āāā ci.yml # Continuous integration workflow
ā āāā docs.yml # Documentation deployment workflow
ā
āāā LICENSE # MIT License
āāā README.md # Project readme
āāā README_REFACTORING.md # Refactoring documentation
āāā REFACTORING_SUMMARY.md # Summary of refactoring changes
āāā update_test_imports.py # Script to update test imports
āāā test_refactoring.py # Script to test refactored modules
āāā pyproject.toml # Project configuration
āāā setup.py # Package setup script
āāā py.typed # Type hint marker
Install GigQ from PyPI:
pip install gigq
This installs the core package with minimal dependencies.
For contributors and developers:
Clone the repository:
git clone https://github.com/kpouianou/gigq.git
cd gigq
Install in development mode with all dependencies:
# Install core package in development mode
pip install -e .
# For running examples
pip install -e ".[examples]"
# For building documentation
pip install -e ".[docs]"
# For development (linting, testing)
pip install -e ".[dev]"
# Or install everything at once
pip install -e ".[examples,docs,dev]"
Note: If you're only interested in using the CLI or basic functionality, the standard installation is sufficient.
from gigq import Job, JobQueue, Worker
# Define a job function
def process_data(filename, threshold=0.5):
# Process some data
print(f"Processing {filename} with threshold {threshold}")
return {"processed": True, "count": 42}
# Define a job
job = Job(
name="process_data_job",
function=process_data,
params={"filename": "data.csv", "threshold": 0.7},
max_attempts=3,
timeout=300
)
# Create or connect to a job queue
queue = JobQueue("jobs.db")
job_id = queue.submit(job)
print(f"Submitted job with ID: {job_id}")
# Start a worker
worker = Worker("jobs.db")
worker.start() # This blocks until the worker is stopped
Or use the CLI:
# Start a worker
gigq --db jobs.db worker
# Process just one job
gigq --db jobs.db worker --once
# Check job status
status = queue.get_status(job_id)
print(f"Job status: {status['status']}")
Or use the CLI:
gigq --db jobs.db status your-job-id
GigQ allows you to create workflows of dependent jobs:
from gigq import Workflow
# Create a workflow
workflow = Workflow("data_processing")
# Add jobs with dependencies
job1 = Job(name="download", function=download_data, params={"url": "https://example.com/data.csv"})
job2 = Job(name="process", function=process_data, params={"filename": "data.csv"})
job3 = Job(name="analyze", function=analyze_data, params={"processed_file": "processed.csv"})
# Add jobs to workflow with dependencies
workflow.add_job(job1)
workflow.add_job(job2, depends_on=[job1])
workflow.add_job(job3, depends_on=[job2])
# Submit all jobs in the workflow
job_ids = workflow.submit_all(queue)
GigQ comes with a command-line interface for common operations:
# Submit a job
gigq submit my_module.my_function --name "My Job" --param "filename=data.csv" --param "threshold=0.7"
# List jobs
gigq list
gigq list --status pending
# Check job status
gigq status your-job-id --show-result
# Cancel a job
gigq cancel your-job-id
# Requeue a failed job
gigq requeue your-job-id
# Start a worker
gigq worker
# Clear completed jobs
gigq clear
gigq clear --before 7 # Clear jobs completed more than 7 days ago
See the examples/github_archive.py
script for a complete example of using GigQ to process GitHub Archive data.
GigQ uses a simple SQLite schema with two main tables:
jobs
- Stores job definitions and current statejob_executions
- Tracks individual execution attemptsThe schema is designed for simplicity and efficiency with appropriate indexes for common operations.
GigQ uses SQLite's built-in locking mechanisms to ensure safety when multiple workers are running. Each worker claims jobs using an exclusive transaction, preventing duplicate execution.
Failed jobs can be automatically retried up to a configurable number of times. Detailed error information is stored in the database for debugging. Jobs that exceed their timeout are automatically detected and marked as failed or requeued.
For local development:
pip install setuptools wheel
pip install -e .
python -m unittest discover tests
This project is licensed under the MIT License - see the LICENSE file for details.
FAQs
A lightweight job queue system with SQLite backend and zero dependencies
We found that gigq demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago.Ā It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
Security News
Socket researchers uncover how browser extensions in trusted stores are used to hijack sessions, redirect traffic, and manipulate user behavior.
Research
Security News
An in-depth analysis of credential stealers, crypto drainers, cryptojackers, and clipboard hijackers abusing open source package registries to compromise Web3 development environments.
Security News
pnpm 10.12.1 introduces a global virtual store for faster installs and new options for managing dependencies with version catalogs.