feat(add-data): Add script for SQL schema initialization
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.
This commit is contained in:
56
data_preparation/test_sql_connection.py
Normal file
56
data_preparation/test_sql_connection.py
Normal file
@@ -0,0 +1,56 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user