Step 1: Install Azure SQL DB Drivers
Install the new version of SQL DB Drivers using official documentation:
Linux, MacOS, Windows
Major update to previous answers: use the last supported version of DB driver ODBC Driver 17 for SQL Server instead of outdated versions ODBC Driver 13 for SQL Server or versions without explicitly defined a version, e.g. SQL Server.
Just print in the terminal: pip install SQLAlchemy
Step 3: Wrapping specific DB logic in AzureDbConnection class
from dataclasses import dataclass
from typing import Dict, Any, Iterable
from pandas import DataFrame
from sqlalchemy import create_engine, inspect
import urllib
@dataclass(frozen=True)
class ConnectionSettings:
"""Connection Settings."""
server: str
database: str
username: str
password: str
driver: str = '{ODBC Driver 18 for SQL Server}'
timeout: int = 30
class AzureDbConnection:
"""
Azure SQL database connection.
"""
def __init__(self, conn_settings: ConnectionSettings, echo: bool = False) -> None:
conn_params = urllib.parse.quote_plus(
'Driver=%s;' % conn_settings.driver +
'Server=tcp:%s.database.windows.net,1433;' % conn_settings.server +
'Database=%s;' % conn_settings.database +
'Uid=%s;' % conn_settings.username +
'Pwd=%s;' % conn_settings.password +
'Encrypt=yes;' +
'TrustServerCertificate=no;' +
'Connection Timeout=%s;' % conn_settings.timeout
)
conn_string = f'mssql+pyodbc:///?odbc_connect={conn_params}'
self.db = create_engine(conn_string, echo=echo)
def connect(self) -> None:
"""Estimate connection."""
self.conn = self.db.connect()
def get_tables(self) -> Iterable[str]:
"""Get list of tables."""
inspector = inspect(self.db)
return [t for t in inspector.get_table_names()]
def dispose(self) -> None:
"""Dispose opened connections."""
self.conn.close()
self.db.dispose()
Major update to previous answers: do not forget to close the connection and dispose of the DB engine explicitly as soon as it stops being needed.
Enjoy!
Set connection settings and credentials using Azure DB blade on Azure Portal:
conn_settings = ConnectionSettings(
server='<db_server_name>',
database='<db_name>',
username='<user_name>',
password='***')
Open DB connection:
db_conn = AzureDbConnection(conn_settings)
db_conn.connect()
Test connection (for example, get available tables list), do other stuff, and close it finally:
try:
for t in db_conn.get_tables():
print(t)
# Do another DB-related stuff:
# ...
finally:
db_conn.dispose()
