API Storage

src.api.storage

delete_analysis_by_id

delete_analysis_by_id(rec_id)

Delete a single analysis by id. Returns number of rows deleted (0 or 1).

Source code in src/api/storage.py

def delete_analysis_by_id(rec_id: str) -> int:
    """Delete a single analysis by id. Returns number of rows deleted (0 or 1)."""
    if not rec_id:
        return 0
    init_db()
    con = sqlite3.connect(str(db_path()))
    try:
        cur = con.execute("DELETE FROM analyses WHERE id = ?", (rec_id,))
        con.commit()
        return cur.rowcount or 0
    finally:
        con.close()

get_analysis_by_id

get_analysis_by_id(rec_id)

Return stored analysis result JSON for a given record id, if present.

Source code in src/api/storage.py

def get_analysis_by_id(rec_id: str) -> Optional[Dict[str, Any]]:
    """Return stored analysis result JSON for a given record id, if present."""
    if not rec_id:
        return None
    init_db()
    con = sqlite3.connect(str(db_path()))
    try:
        cur = con.execute(
            "SELECT result_json FROM analyses WHERE id = ? LIMIT 1",
            (rec_id,),
        )
        row = cur.fetchone()
        if not row:
            return None
        try:
            return json.loads(row[0])
        except Exception:
            return None
    finally:
        con.close()

get_analysis_by_sha256

get_analysis_by_sha256(sha256)

Return the most recent analysis result JSON for a given sha256, if present.

Source code in src/api/storage.py

def get_analysis_by_sha256(sha256: str) -> Optional[Dict[str, Any]]:
    """Return the most recent analysis result JSON for a given sha256, if present."""
    if not sha256:
        return None
    init_db()
    con = sqlite3.connect(str(db_path()))
    try:
        cur = con.execute(
            "SELECT result_json FROM analyses WHERE sha256 = ? ORDER BY created_at DESC LIMIT 1",
            (sha256,),
        )
        row = cur.fetchone()
        if not row:
            return None
        try:
            return json.loads(row[0])
        except Exception:
            return None
    finally:
        con.close()

list_analyses

list_analyses(page=1, page_size=20, sha256=None, sha1=None, md5=None, date_from=None, date_to=None)

List analyses with pagination and optional hash/date filtering.

Returns a dict: {"page", "page_size", "total", "items": [...]} Items exclude the heavy result JSON to keep responses light.

Source code in src/api/storage.py

def list_analyses(
    page: int = 1,
    page_size: int = 20,
    sha256: Optional[str] = None,
    sha1: Optional[str] = None,
    md5: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
) -> Dict[str, Any]:
    """List analyses with pagination and optional hash/date filtering.

    Returns a dict: {"page", "page_size", "total", "items": [...]}
    Items exclude the heavy result JSON to keep responses light.
    """
    init_db()
    page = max(1, int(page))
    page_size = min(max(1, int(page_size)), 200)

    where: List[str] = []
    args: List[Any] = []
    if sha256:
        where.append("sha256 = ?")
        args.append(sha256)
    if sha1:
        where.append("sha1 = ?")
        args.append(sha1)
    if md5:
        where.append("md5 = ?")
        args.append(md5)
    if date_from:
        where.append("created_at >= ?")
        args.append(date_from)
    if date_to:
        where.append("created_at <= ?")
        args.append(date_to)

    where_sql = (" WHERE " + " AND ".join(where)) if where else ""

    con = sqlite3.connect(str(db_path()))
    try:
        cur = con.execute(f"SELECT COUNT(*) FROM analyses{where_sql}", tuple(args))
        total = int(cur.fetchone()[0])
        offset = (page - 1) * page_size
        cur = con.execute(
            f"""
            SELECT id, created_at, file_name, size_bytes, md5, sha1, sha256, hint, model
            FROM analyses
            {where_sql}
            ORDER BY created_at DESC
            LIMIT ? OFFSET ?
            """,
            tuple(args) + (page_size, offset),
        )
        cols = [c[0] for c in cur.description]
        items = [dict(zip(cols, row)) for row in cur.fetchall()]

        return {"page": page, "page_size": page_size, "total": total, "items": items}
    finally:
        con.close()

purge_analyses_by_sha256

purge_analyses_by_sha256(sha256)

Delete all analyses for a given sha256. Returns number of rows deleted.

Source code in src/api/storage.py

def purge_analyses_by_sha256(sha256: str) -> int:
    """Delete all analyses for a given sha256. Returns number of rows deleted."""
    if not sha256:
        return 0
    init_db()
    con = sqlite3.connect(str(db_path()))
    try:
        cur = con.execute("DELETE FROM analyses WHERE sha256 = ?", (sha256,))
        con.commit()
        return cur.rowcount or 0
    finally:
        con.close()