Is using quoted_name safe way for parametrizing table name and fields in python's SQL Alchemy?

Question

I spent a lot of time looking for solution to parametrize table names and field names in SQL Alchemy plain textual SQL queries for SQL Server. I stumbled upon several stackoverflow questions and other resources like:

SQL Alchemy Parametrized Query , binding table name as parameter gives error

Answer to above which I don't like as it is just building query from string which is proun to SQL Injection attacks

I know it is possible (I was doing it that way in the past) to do by creating table objects from sqlalchemy.ext.declarative.declarative_base but it requires to declare whole schema of your database which is a lot of unscalable code.

Without much luck with SQL Server I found solution in Postgres psycopg2 using psycopg2.sql.Identifier. So from here I started looking for equivalent in SQL Alchemy. I found quoted_name. Which I understand works as identifier preventing from SQL Injections. But is it really? Could somebody confirm that it is safe to use?

Code example below which returns number of rows in the passed in table:

    def count_rows(self, table_name: str) -> int:

        query_base = "SELECT COUNT(*) FROM {}"
        query_params = [quoted_name(table_name, True)]
        query = text((query_base).format(*query_params))

        with self.engine.connect() as con:
            result = con.execute(query).fetchone()

        return result[0]

Answer 1

I don't get the impression from the documentation this is the purpose for which quoted_name is intended. My reading was that it's for cases where non-standard naming conventions for column or table names are in use, requiring quotation for them to work.

I think there are two possible solutions:

1. exercise total control over the allowed table names

f"SELECT COUNT(*) FROM {table_name}" is fine if you don't allow table_name to be provided by the user without filtering.

For example, you could simply have

    ...
    allowed = ["table_1", ..., "table_N"]
    if table_name not in allowed:
        raise ValueError(f"Table name must be one of {allowed}. Received {table_name}.")

There are, of course, plenty of other ways to do this. But the idea is to either map user input to allowed values, reject disallowed values, or a mixture of both.

2. reflect the schema

You mentioned that

I know it is possible (I was doing it that way in the past) to do by creating table objects from sqlalchemy.ext.declarative.declarative_base but it requires to declare whole schema of your database which is a lot of unscalable code.

This is not true. You can 'reflect' the schema of an existing database as follows:

from sqlalchemy import create_engine, func, select, MetaData


class YourClass:
    def __init__(self, db_connection_string: str):
        """
        __init__ for YourClass

        (for example)

        """
        self.engine = create_engine(db_connection_string)
        self.metadata = MetaData(bind=self.engine)

        MetaData.reflect(self.metadata)

    def count_rows(self, table_name: str) -> int:
        """
        count_rows

        Returns the COUNT of the rows for a given table

        """
        table = self.metadata.tables[table_name]

        result = select([func.count()]).select_from(table).scalar()

        return result

Worth noting that this approach will also throw an exception if table_name doesn't exist in the database.

Alternative syntax - for full ORM-goodness, use a sessionmaker:

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker


class YourClass:
    def __init__(self, db_connection_string: str):
        self.engine = create_engine(db_connection_string)
        self.Session = sessionmaker(bind=self.engine)
        self.metadata = MetaData(bind=self.engine)

        MetaData.reflect(self.metadata)

    def count_rows(self, table_name: str) -> int:
        table = self.metadata.tables[table_name]
        # if you want a new session every call:
        with self.Session.begin() as session: 
            return session.query(table).count()