Get a scan job by ID. Args: scan_id: Unique scan identifier Returns: Dict with scan job data or None if not found
(self, scan_id: str)
| 1980 | options = COALESCE(excluded.options, scan_jobs.options), |
| 1981 | updated_at = CURRENT_TIMESTAMP |
| 1982 | """, (scan_id, scan_type, target, status, progress_percent, |
| 1983 | findings_count, current_check, started_at, completed_at, |
| 1984 | error_message, options_json)) |
| 1985 | |
| 1986 | conn.commit() |
| 1987 | return True |
| 1988 | except Exception as e: |
| 1989 | logger.error(f"Failed to save scan job {scan_id}: {e}") |
| 1990 | return False |
| 1991 | |
| 1992 | def get_scan_job(self, scan_id: str) -> Optional[Dict[str, Any]]: |
| 1993 | """ |
| 1994 | Get a scan job by ID. |
| 1995 | |
| 1996 | Args: |
| 1997 | scan_id: Unique scan identifier |
| 1998 | |
| 1999 | Returns: |
| 2000 | Dict with scan job data or None if not found |
| 2001 | """ |
| 2002 | try: |
| 2003 | with self.get_connection() as conn: |
| 2004 | cursor = conn.cursor() |
| 2005 | cursor.execute("SELECT * FROM scan_jobs WHERE scan_id = ?", (scan_id,)) |
| 2006 | row = cursor.fetchone() |
| 2007 | |
| 2008 | if row: |
| 2009 | result = dict(row) |
| 2010 | # Parse options JSON |
| 2011 | if result.get('options'): |
| 2012 | try: |
| 2013 | result['options'] = json.loads(result['options']) |
no test coverage detected