Benjamin_Boenisch/breakpilot-compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix SQLAlchemy 2.x compatibility: wrap raw SQL in text()

Browse Source 
SQLAlchemy 2.x requires raw SQL strings to be explicitly wrapped
in text(). Fixed 16 instances across 5 route files.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
2026-03-10 13:56:38 +01:00
parent 033fa52e5b
commit 86588aff09
3 changed files with 28 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
backend-compliance/compliance/api/screening_routes.py
View File

@@ -17,6 +17,7 @@ from typing import Optional
import httpx
from fastapi import APIRouter, File, Form, UploadFile, HTTPException
from pydantic import BaseModel
from sqlalchemy import text
from database import SessionLocal
@@ -366,13 +367,13 @@ async def scan_dependencies(
db = SessionLocal()
try:
db.execute(
"""INSERT INTO compliance_screenings
text("""INSERT INTO compliance_screenings
(id, tenant_id, status, sbom_format, sbom_version,
total_components, total_issues, critical_issues, high_issues, medium_issues, low_issues,
sbom_data, started_at, completed_at)
VALUES (:id, :tenant_id, 'completed', 'CycloneDX', '1.5',
:total_components, :total_issues, :critical, :high, :medium, :low,
:sbom_data::jsonb, :started_at, :completed_at)""",
:sbom_data::jsonb, :started_at, :completed_at)"""),
{
"id": screening_id,
"tenant_id": tenant_id,
@@ -391,11 +392,11 @@ async def scan_dependencies(
# Persist security issues
for issue in issues:
db.execute(
"""INSERT INTO compliance_security_issues
text("""INSERT INTO compliance_security_issues
(id, screening_id, severity, title, description, cve, cvss,
affected_component, affected_version, fixed_in, remediation, status)
VALUES (:id, :screening_id, :severity, :title, :description, :cve, :cvss,
:component, :version, :fixed_in, :remediation, :status)""",
:component, :version, :fixed_in, :remediation, :status)"""),
{
"id": issue["id"],
"screening_id": screening_id,
@@ -486,10 +487,10 @@ async def get_screening(screening_id: str):
db = SessionLocal()
try:
result = db.execute(
"""SELECT id, status, sbom_format, sbom_version,
text("""SELECT id, status, sbom_format, sbom_version,
total_components, total_issues, critical_issues, high_issues,
medium_issues, low_issues, sbom_data, started_at, completed_at
FROM compliance_screenings WHERE id = :id""",
FROM compliance_screenings WHERE id = :id"""),
{"id": screening_id},
)
row = result.fetchone()
@@ -498,9 +499,9 @@ async def get_screening(screening_id: str):
# Fetch issues
issues_result = db.execute(
"""SELECT id, severity, title, description, cve, cvss,
text("""SELECT id, severity, title, description, cve, cvss,
affected_component, affected_version, fixed_in, remediation, status
FROM compliance_security_issues WHERE screening_id = :id""",
FROM compliance_security_issues WHERE screening_id = :id"""),
{"id": screening_id},
)
issues_rows = issues_result.fetchall()
@@ -566,12 +567,12 @@ async def list_screenings(tenant_id: str = "default"):
db = SessionLocal()
try:
result = db.execute(
"""SELECT id, status, total_components, total_issues,
text("""SELECT id, status, total_components, total_issues,
critical_issues, high_issues, medium_issues, low_issues,
started_at, completed_at, created_at
FROM compliance_screenings
WHERE tenant_id = :tenant_id
ORDER BY created_at DESC""",
ORDER BY created_at DESC"""),
{"tenant_id": tenant_id},
)
rows = result.fetchall()