Spaces:
Build error
Build error
File size: 1,655 Bytes
7293b6f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
import pyodbc
import logging
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
def connect_to_database() -> pyodbc.Connection:
"""Connect to the Azure SQL Database."""
try:
server = os.getenv('AZURE_SQL_SERVER')
database = os.getenv('AZURE_SQL_DATABASE')
username = os.getenv('AZURE_SQL_USERNAME')
password = os.getenv('AZURE_SQL_PASSWORD')
driver = '{ODBC Driver 17 for SQL Server}'
conn_str = f'DRIVER={driver};SERVER={server};DATABASE={database};UID={username};PWD={password}'
conn = pyodbc.connect(conn_str)
logging.info(f"Connected to the Azure SQL Database at {server}")
return conn
except pyodbc.Error as e:
logging.error(f"Error connecting to the Azure SQL Database: {e}")
return None
def close_database_connection(conn: pyodbc.Connection) -> None:
"""Close the database connection."""
try:
if conn:
conn.close()
logging.info("Database connection closed.")
except pyodbc.Error as e:
logging.error(f"Error closing the database connection: {e}")
# Example usage with context management
class DatabaseConnection:
def __enter__(self) -> pyodbc.Connection:
self.conn = connect_to_database()
return self.conn
def __exit__(self, exc_type, exc_value, traceback) -> None:
close_database_connection(self.conn)
# Example usage
if __name__ == "__main__":
with DatabaseConnection() as conn:
if conn:
# Perform database operations
pass
|