Parameterizing Table Names

[skwzrd]

Hello,

I'm not sure if I should pose this question here, or in the SQLAlchemy community, but I was wondering if it is possible to parameterize database objects such as table names using asyncmy. An example of a library that can perform this is psycopg2. This documentation demonstrating this feature is here: https://www.psycopg.org/docs/sql.html#psycopg2.sql.Identifier.

I have tried the following:

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

database = {
    'host': "127.0.0.1",
    'port': 3306,
    'db': "db_name",
    'user': "user",
    'password': "password",
    'charset': "utf8mb4",
}

# https://docs.sqlalchemy.org/en/20/dialects/mysql.html#asyncmy
url = (
    f'mysql+asyncmy://'
    f'{database["user"]}:{database["password"]}'
    f'@{database["host"]}:{database["port"]}'
    f'/{database["db"]}?charset={database["charset"]}'
)

MY_TABLE = 'table_name'

async def run():
    engine = create_async_engine(url)
    async with engine.connect() as conn:
        # 1.
        # result = await conn.execute(text("""select * from :table_name ;"""), [{'table_name': MY_TABLE}])
        # sqlalchemy.exc.ProgrammingError: (asyncmy.errors.ProgrammingError) (1064, "You have an error in your SQL syntax;

        # 2.
        # result = await conn.execute(text("""select * from :table_name ;"""), {'table_name': MY_TABLE})
        # sqlalchemy.exc.ProgrammingError: (asyncmy.errors.ProgrammingError) (1064, "You have an error in your SQL syntax;

        # 3.
        # result = await conn.execute(text("""select * from :table_name ;"""), table_name=MY_TABLE)
        # TypeError: AsyncConnection.execute() got an unexpected keyword argument 'table_name'

        # 4.
        # result = await conn.execute(text("""select * from :table_name ;""").bindparams(table_name=MY_TABLE))
        # sqlalchemy.exc.ProgrammingError: (asyncmy.errors.ProgrammingError) (1064, "You have an error in your SQL syntax;

        # 5. The only method that works, but it requires validating MY_TABLE strings. 
        result = await conn.execute(text(f"""select * from {MY_TABLE} ;"""))

        for row in result:
            print(row)

if __name__ == '__main__':
    asyncio.run(run())

Thank you, Michael