66d1eec875
Since we are working with an Azure SQL database, we need to fill the generated customer data in a fitting schema. The schema will be described in more detail in an updated README file later. The added script uses `pyodbc` to connect to the database and create the tables. This requires a connection string, which will not be checked out to this repo for security reasons and must be obtained separately. Additionally, a script `test_sql_connection.py` is added with this commit, which is a simple utility to test the `pyodbc` connection.
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""Script to test connection to Azure SQL Database."""
|
|
|
|
import os
|
|
|
|
import pyodbc
|
|
|
|
|
|
def test_connection() -> bool:
|
|
"""Test the connection to Azure SQL Database.
|
|
|
|
This function attempts to establish a connection to an Azure SQL Database
|
|
using a connection string stored in the environment variable
|
|
'AZURE_SQL_CONNECTION_STRING'. It executes a simple query to verify
|
|
the connection and prints the result of the connection attempt.
|
|
|
|
Returns
|
|
-------
|
|
bool
|
|
True if the connection was successful and closed properly,
|
|
False if there was an error connecting to the database.
|
|
|
|
Notes
|
|
-----
|
|
- The function requires the 'AZURE_SQL_CONNECTION_STRING' environment
|
|
variable to be set with a valid connection string.
|
|
- It uses pyodbc to establish the database connection.
|
|
- The function prints success or error messages to stdout.
|
|
- In case of a successful connection, it executes "SELECT @@version;"
|
|
as a test query.
|
|
- The function ensures that both the cursor and the connection are
|
|
closed after the operation, regardless of its success or failure.
|
|
"""
|
|
connection_string = os.environ.get("AZURE_SQL_CONNECTION_STRING")
|
|
try:
|
|
conn = pyodbc.connect(connection_string)
|
|
cursor = conn.cursor()
|
|
print("Connected to Azure SQL Database successfully!")
|
|
|
|
# Example query
|
|
cursor.execute("SELECT @@version;")
|
|
|
|
except pyodbc.Error as e:
|
|
print(f"Error connecting to Azure SQL Database: {e}")
|
|
return False
|
|
|
|
finally:
|
|
if "cursor" in locals():
|
|
cursor.close()
|
|
if "conn" in locals():
|
|
conn.close()
|
|
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_connection()
|