How to fine tune the complex python function for creating merge command to run in bigquery

Syntax errors:

• You have a try with nothing else.
• merge_query is trying to concatenate a int into a str.
• for key in key_list: col_no_key.remove(key) is trying to remove items from a different list
• get_key_list_type is not defined (neither is run_sys_command, but that's not really part of your question). I put in some stubs to get this to run.

This is wide open to an SQL injection (eg https://xkcd.com/327/). If anyone other than you is putting anything into your database at any point, be very careful. If anyone other than you is calling this function, start over from scratch.

Other items:

• I got rid of the bare try.
• f-strings are really nice when you are combining lots of literal strings with variables.
• You have if dc!=-1:...else:...if dc!=-1:.... Just combine the ifs.
• for i in range(len(key_list)) is always a red flag. You generally want to say for item in listofitems. You can't in this case, because you need the index to reassign to. This means that you would need the enumerate function. But you can avoid all of that by using map or a list comprehension.
• split takes a second argument that tells how many splits to do. This means you can combine your splits into one line.
• Try to use variable += instead of variable = variable +. It shortens things up, and makes it clear that you're appending.
• The only point of b, c, and d is to count where you are in the loop. I would recommend reusing them instead of declaring a new one. I would recommend enumerate. I would recommend a list comprehension with join.
• Your function is a bit long. We should break it up into smaller pieces. The merge_query, ON, UPDATE, and INSERT make it clear what those smaller pieces should be. (This is a bit easier after the previous point.)
• I'm ok with abreviating command as cmd. I'm not ok with abreviateing code as cd. Spell out return_code.
• If you exit the last for loop without breaking, it's because return_code is not zero. Let's not test something we already know to be true, and just raise the exception. We should also use err to pass along useful information.
• You shouldn't print(merge_cmd). If you end up calling this function a large number of times, that will flood your console. Use the Python debugger if you're not sure about the content of the variable.
• The Python recommendations are that ALL_CAPS should be constant variables. But SQL doesn't really go by this, so it makes some sense for your code to not go by that as well.

import time

def on_piece(key_type:str) -> str:
    #key,type_ = key_type.split(':') # if you know there's only one :
    key,type_ = key_type.split(':',1)
    #key,type_,_ = key_type.split(':',2) # if more than that could happen
    return f" TARGET.{key} = CAST(SOURCE.{key} as {type_})"

def update_piece(col_list_type:str) -> str:
    col,type_ = col_list_type.split(':',1)
    if col == 'dt':
        return f" TARGET.{col} =  CAST(SOURCE.calendar_date as date)"
    else:
        return f" TARGET.{col} = CAST(SOURCE.{col} as {type_})"

def insert_piece(col_list_type:str) -> str:
    col,type_ = col_list_type.split(':',1)
    if col == 'dt':
        return " CAST(SOURCE.calendar_date as date)"
    else:
        return f" CAST(SOURCE.{col} as {type_})"

def merge_cmd(database:str, table:str, dc:int, keys:str, columns:str, filter_column:str, new_main_column_type_list:str) -> None:
    if dc != '-1':
        merge_query = f"MERGE INTO {database}.{table}_{dc} TARGET USING {database}_stg.{table}_{dc}_stg SOURCE "
        keys += ',dc'
    else:
        merge_query = f"MERGE INTO {database}.{table} TARGET USING {database}_stg.{table}_stg SOURCE "

    key_list = [ key.lower() for key in keys.split(',') ]
    col_no_key = [ col.lower() for col in columns.split(',') ]

    #for key in key_list: # ValueError: list.remove(x): x not in list
    #    col_no_key.remove(key)
    cols_list = columns.split(',')

    key_list_type = get_key_list_type(key_list, new_main_column_type_list)
    col_no_key_list_type = get_key_list_type(col_no_key, new_main_column_type_list)
    cols_list_type = get_key_list_type(cols_list, new_main_column_type_list)

    ON = " ON " + " AND ".join( on_piece(key_type) for key_type in key_list_type )

    UPDATE = " WHEN MATCHED THEN UPDATE SET "
    UPDATE += ",".join( update_piece(col_list_type)
                        for col_list_type in col_no_key_list_type )

    INSERT = " WHEN NOT MATCHED BY TARGET THEN INSERT (" + columns + ") VALUES ("
    INSERT += ",".join( insert_piece(col_list_type)
                        for col_list_type in cols_list_type )
    INSERT += ")"

    merge_query += ON + UPDATE + INSERT
    merge_query = "'" + merge_query + "'"
    merge_cmd = "bq --location=us query --use_legacy_sql=false " + merge_query

    return_code, out, err = run_sys_command(merge_cmd)

    if return_code != 0:
        for i in range(20):
            time.sleep(30)
            return_code, out, err = run_sys_command(merge_cmd)
            if return_code == 0:
                break
        raise Exception('Merge Failed: '+err) # if err is a string
        #raise Exception('Merge Failed') from err # if err is an Exception

def run_sys_command(merge_cmd):
    return 0,'out','err'

def get_key_list_type(some_list, new_main_column_type_list):
    return [ 'STUB:STUB' for _ in some_list ]

if __name__ == "__main__":
    merge_cmd('testdb', 'test', 8134, 'load_id | test_load_id', 'invoice_id,dt', 'last_change_ts', 'invoice_id:INT64')