import prefect
from sqlserver import SqlServerExecute

class prefect.tasks.sql_server.sql_server.SqlServerExecute(db_name="OTI_Djoliba", user="sa", host="EQ-EQ6288793\SQLEXPRESS", port=1433, driver="ODBC Driver 17 for SQL Server", query= None, data=None, commit=False, **kwargs)

from prefect import Flow, Parameter, task
from prefect.tasks.sql_server import SqlServerExecute
from prefect.tasks.secrets import PrefectSecret


execute_query = SqlServerExecute(
    db_name="xxx", user="yyy", host="zzz", port=42, commit=True
)


@task
def define_query_from_param(table_name: str):
    return f"""
    INSERT INTO {table_name} ("index","id","first_name","last_name","email","gender","city","country")
    VALUES
    (1000,1001,'Anna','Geller', '<mailto:anna@example.com|anna@example.com>','Female','Berlin','Germany');
    """


with Flow("postgres_example") as flow:
    postgres_pwd = PrefectSecret("SQL_SERVER_DB_PASSWORD")
    table_name = Parameter("table_name", default="stage.customers")
    query = define_query_from_param(table_name)
    execute_query(password=postgres_pwd, query=query)

if __name__ == "__main__":
    flow.run()

from prefect import Flow, task
from prefect.tasks.sql_server import SqlServerExecute
from prefect.tasks.secrets import PrefectSecret


execute_query = SqlServerExecute(
    db_name="xxx", user="yyy", host="zzz", port=42, commit=True
)


@task
def define_query():
    return """
    INSERT INTO xxx ("index","id","first_name","last_name","email","gender","city","country")
    VALUES
    (1000,1001,'Anna','Geller', '<mailto:anna@example.com|anna@example.com>','Female','Berlin','Germany');
    """


with Flow("postgres_example") as flow:
    postgres_pwd = PrefectSecret("SQL_SERVER_DB_PASSWORD")
    query = define_query()
    execute_query(password=postgres_pwd, query=query)

if __name__ == "__main__":
    flow.run()