如何避免 python 函数中的硬编码字符串值

How to avoid the hard coded string values in python function

我已经编写了下面的 python 函数来创建 sql 我想在 google 的 BigQuery 平台中 运行 的语句。当我查看代码时,我发现许多字符串文字都是多余的,例如“_stg”的用法。我怎样才能避免这些硬编码的字符串值和冗余代码?此外,在 python 代码中硬编码 sql 这样的查询是否正常,或者是否有更好的 pythonic 方法来执行此操作?

def create_main_bq_table(database, table, dc, filter_column, cmt):
    try:
        create_statement = "create table if not exists " + database + "." + table
        select_statement = " partition by dt as select *,"
        from_statement = "from " + database + "_stg." + table
        if filter_column != 'none':
            if dc == '-1':
                query = create_statement + select_statement + "date(" + filter_column + ") as dt " + from_statement + "_stg"
            else:
                query = create_statement + "_" + dc + select_statement + "date(" + filter_column + ") as dt " + from_statement + "_" + dc + "_stg"

        elif filter_column == 'none' and cmt != 'snapshot':
            if dc == '-1':
                query = create_statement + select_statement + "date(last_run_time) as dt " + from_statement + "_stg"
            else:
                query = create_statement + "_" + dc + select_statement + "date(last_run_time) as dt " + from_statement + "_" + dc + "_stg"

        else:
            if dc == '-1':
                query = "create or replace table  " + database + "." + table + " as select * from " + database + "_stg." + table + "_stg"
            else:
                query = "create or replace table  " + database + "." + table + "_" + dc + " as select * from " + database + "_stg." + table + "_" + dc + "_stg"


    except Exception as e:
        print(e)
        raise

我不相信您可以完全删除硬编码元素 -- 但这里有一种方法可以减少您拥有它们的次数。

我不会说这个解决方案更像 pythonic,但它至少遵循 DRY 和其他最佳编码实践的原则来减少大量字符串。看起来你基本上是在做类似的陈述,只是做了一些调整。

def sql_query(create_statement, dc, select_statement, filter_statement, from_statement):
    if dc == '-1':
        return f"{create_statement}{select_statement}{filter_statement}{from_statement}_stg"
    else:
        return f"{create_statement}_{dc}{select_statement}{filter_statement}{from_statement}_{dc}_stg"


def create_main_bq_table(database, table, dc, filter_column, cmt):
    create_statement = f"create table if not exists {database}.{table}"
    select_statement = " partition by dt as select *,"
    from_statement = f"from {database}_stg.{table}"
    
    if filter_column != 'none':
        query = sql_query(create_statement, dc, select_statement, f'date({filter_column}) as dt ', from_statement)
    elif filter_column == 'none' and cmt != 'snapshot':
        query = sql_query(create_statement, dc, select_statement, f'date(last_run_time) as dt ', from_statement)
    else:
        query = sql_query(f"create or replace table  {database}.{table}", dc, ' as select * ', '', from_statement)

删除了 try catch 块,因为我不明白为什么它会导致错误,因为它是字符串创建(您可能必须最后尝试一下)

另一件事是添加类型提示(做不到,因为不确定函数变量的所有类型)