refactor(storage/sql): use SQLAlchemy Core for count_by_column

Replace the f-string + text() construction with select/func.count over
a reflected Table and named column.  Eliminates the raw SQL build
entirely so static analysis no longer has a string-interpolated SQL
statement to flag, and the allowlist check on `column` remains in
place as defence-in-depth against future callers passing arbitrary
identifiers.

Author: fgogolli

src/orb/infrastructure/storage/sql/strategy.py

@@ -3,7 +3,9 @@
 from contextlib import contextmanager
 from typing import Any, Optional
 
-from sqlalchemy import text
+from sqlalchemy import MetaData, Table
+from sqlalchemy import column as sa_column
+from sqlalchemy import func, select, text
 
 from orb.infrastructure.logging.logger import get_logger
 from orb.infrastructure.storage.base.strategy import BaseStorageStrategy
@@ -411,24 +413,26 @@ def count_by_column(self, column: str) -> dict[str, int]:
         gracefully to the list-and-count slow path if needed.
         """
         # Validate the column against the strategy's registered columns dict
-        # before interpolation.  SQL identifiers cannot be bound parameters, so
-        # the only safe path is allowlist-then-interpolate.  Anything else is
-        # rejected.
+        # before building the query.  This is the only place untrusted strings
+        # could ever enter the SQL build; rejecting unknown columns keeps the
+        # query construction below restricted to identifiers we registered at
+        # construction time.
         if column not in self.columns:
             raise StorageError(
                 f"count_by_column: column {column!r} is not in the registered "
                 f"schema for table {self.table_name!r}"
             )
-        # table_name and columns keys are constructor-time constants set by the
-        # repository layer; both have just been validated against the in-memory
-        # schema map above.  No request-time string crosses this boundary.
-        sql_statement = text(
-            f"SELECT {column} AS bucket, COUNT(*) AS cnt FROM {self.table_name} GROUP BY {column}"
-        )
+        # Build the SELECT via SQLAlchemy Core constructs — no raw SQL string
+        # interpolation.  The column object is created by name (validated
+        # above) and the Table is reflected from MetaData by name (also
+        # validated since self.table_name is a constructor-time constant).
+        bucket = sa_column(column)
+        table = Table(self.table_name, MetaData())
+        stmt = select(bucket.label("bucket"), func.count().label("cnt")).select_from(table).group_by(bucket)
         with self.lock_manager.read_lock():
             try:
                 with self.connection_manager.get_session() as session:
-                    result = session.execute(sql_statement)
+                    result = session.execute(stmt)
                     rows = result.fetchall()
                 counts: dict[str, int] = {}
                 for row in rows: