Database Utilities¶

from gigq.db_utils import get_connection

# Get a connection (will be reused in this thread)
conn = get_connection("jobs.db")

from gigq.db_utils import close_connection

# Close the connection to a specific database
close_connection("jobs.db")

from gigq import close_connections

# Close all connections for this thread
close_connections()

import threading
from gigq import JobQueue, close_connections

def worker_thread():
    queue = JobQueue("jobs.db")

    # Process jobs...

    # Clean up before thread exits
    queue.close()  # This closes the specific connection used by the queue
    close_connections()  # This ensures all thread-local connections are closed

thread = threading.Thread(target=worker_thread)
thread.start()
thread.join()

# Thread-local storage
_thread_local = threading.local()

# Initialize thread's connection dictionary if needed
if not hasattr(_thread_local, 'connections'):
    _thread_local.connections = {}

# Get or create connection for this database
if db_path not in _thread_local.connections:
    conn = sqlite3.connect(db_path, timeout=30.0)
    conn.row_factory = sqlite3.Row
    _thread_local.connections[db_path] = conn

import threading
import signal
import time
from gigq import Worker, JobQueue, close_connections

def worker_thread(db_path, worker_id, stop_event):
    """Worker function to run in a separate thread."""
    worker = Worker(db_path, worker_id=worker_id, polling_interval=1)
    queue = JobQueue(db_path, initialize=False)

    print(f"Worker {worker_id} started")

    try:
        # Process jobs until stop_event is set
        while not stop_event.is_set():
            processed = worker.process_one()
            if not processed:
                # No job available, sleep briefly
                time.sleep(1)

            # Periodically check for jobs that might be stuck
            if worker_id == "worker-1" and random.random() < 0.05:  # 5% chance
                timed_out = worker._check_for_timeouts()
                if timed_out:
                    print(f"Worker {worker_id} detected {timed_out} timed out jobs")

    except Exception as e:
        print(f"Worker {worker_id} error: {str(e)}")
    finally:
        # Always clean up connections
        worker.close()
        queue.close()
        close_connections()
        print(f"Worker {worker_id} stopped")

def run_worker_pool(db_path, num_workers=4):
    """Run a pool of worker threads with proper shutdown handling."""
    stop_event = threading.Event()
    threads = []

    # Start worker threads
    for i in range(num_workers):
        worker_id = f"worker-{i+1}"
        thread = threading.Thread(
            target=worker_thread,
            args=(db_path, worker_id, stop_event)
        )
        thread.daemon = True
        thread.start()
        threads.append(thread)

    print(f"Started {num_workers} worker threads")

    # Set up signal handling for graceful shutdown
    def handle_signal(sig, frame):
        print(f"Received signal {sig}, shutting down workers...")
        stop_event.set()

    signal.signal(signal.SIGINT, handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)

    try:
        # Keep the main thread alive until interrupted
        while not stop_event.is_set():
            time.sleep(1)
    except KeyboardInterrupt:
        stop_event.set()
    finally:
        # Wait for all threads to complete
        print("Waiting for worker threads to stop...")
        for thread in threads:
            thread.join(timeout=5)
        print("All worker threads stopped")

        # Clean up main thread connections
        close_connections()

# Usage:
# run_worker_pool("jobs.db", num_workers=4)

from flask import Flask, g
from gigq import JobQueue, close_connections

app = Flask(__name__)

def get_queue():
    """Get or create a thread-local queue."""
    if not hasattr(g, 'job_queue'):
        g.job_queue = JobQueue("jobs.db")
    return g.job_queue

@app.teardown_appcontext
def close_queue(e=None):
    """Close the queue when the request ends."""
    queue = g.pop('job_queue', None)
    if queue is not None:
        queue.close()
    close_connections()

@app.route('/submit')
def submit_job():
    queue = get_queue()
    # Submit job...
    return "Job submitted"

import schedule
import time
from gigq import JobQueue, close_connections

def daily_task():
    queue = JobQueue("jobs.db")
    try:
        # Submit daily jobs...
        print("Daily task complete")
    finally:
        queue.close()
        close_connections()

# Schedule the task
schedule.every().day.at("01:00").do(daily_task)

# Run the scheduler
while True:
    schedule.run_pending()
    time.sleep(60)