How could this python function be mocked to be tested in another module? note: sqla and db_session is a separate function that manages database transactions, as specified here: https://stackoverflow.com/a/33648514
from mymodule.models.sqla import db_session
from airflow.hooks.mssql_hook import MsSqlHook
def insert_data(conn_id, truncate):
id = 0
query_storeproc = "EXEC [dbo].insert_my_data @truncate = :truncate, @id = :id"
storeproc_params = {'truncate' : truncate, 'load_cycle_id' : load_cycle_id}
with db_session(MsSqlHook(conn_id).get_uri()) as db:
db.execute(query_storeproc, storeproc_params)
Aucun commentaire:
Enregistrer un commentaire