drop_Table_SynapseNotebooks_python

drop_Table_SynapseNotebooks_Python.py

"""
db=DedicatedSQLAuth()
res=db.ddlStatements("drop table my_schema.A")
"""
class DedicatedSQLAuth():
    def __init__(self):
        self.dbName= "myDb"
        self.serverName="myServer"
        self.token= mssparkutils.credentials.getToken("DW")

    def use_token(self):
        connection_string = 'Driver={ODBC Driver 17 for SQL Server};server='+self.serverName+';database='+self.dbName+';'
        SQL_COPT_SS_ACCESS_TOKEN = 1256
        
        exptoken = b''
        for i in bytes(self.token, "UTF-8"):
            exptoken += bytes({i})
            exptoken += bytes(1)
        tokenstruct = struct.pack("=i", len(exptoken)) + exptoken
        cnxn = pyodbc.connect(connection_string, attrs_before = { SQL_COPT_SS_ACCESS_TOKEN:tokenstruct })
        cursor = cnxn.cursor()
        return cursor, cnxn
    
    def ddlStatements(self,query:str):
        try:
            cursor, cnxn=self.use_token()
            cnxn.autocommit=True
            cursor.execute(query)
            cnxn.close()
        except Exception as e:
             raise Exception(f"ddlStatements : Failed to execute - {query}")
             defaultLogger.critical(e)

    def dmlStatements(self,query:str):
        res=[]
        try:
            cursor, cnxn=self.use_token()
            cnxn.autocommit=True
            cursor.execute(query)
            for i in cursor.fetchall():
                res.append(i)
            cnxn.close()
            return res
        except Exception as e:
            raise Exception(f"dmlStatements : Failed to execute - {query}")
            defaultLogger.critical(e)