Get record_id of a task stored in SQL database file

luigi_sql_record_id.py

import sqlite3
from luigi import configuration
config = configuration.get_config()

def get_record_id(task_id):
    '''
    every task has an associated `self.task_id`
    call `get_record_id(self.task_id)` to obtain that
    '''
    conn= sqlite3.connect(config['task_history']['db_connection'].split('sqlite:///')[1])

    cur = conn.cursor()
    cur.execute(f"SELECT id FROM tasks WHERE task_id='{task_id}'")
    record_id = cur.fetchone()[0]

    conn.close()

    return record_id

If needed, you can use my luigi.cfg (place it in the same directory).