Re: MYSQL - Python Connector - No Response
Posted by:
Tomasz Sas
Date: January 16, 2025 07:47AM
```
import mysql.connector
from mysql.connector import Error
from typing import Any
class MySQLDatabaseConnector:
def __init__(self, host: str, user: str, password: str, database: str, port: int = 3306):
"""
Initialize the MySQLDatabaseConnector with connection parameters.
:param host: Host where the MySQL database server is running.
:param user: MySQL username.
:param password: MySQL password.
:param database: Name of the MySQL database to connect to.
:param port: Port number (default is 3306).
"""
self.host = host
self.user = user
self.password = password
self.database = database
self.port = port
self.connection = None
def connect(self) -> None:
"""
Establish a connection to the MySQL database.
"""
try:
self.connection = mysql.connector.connect(
host=self.host,
port=self.port,
user=self.user,
password=self.password,
database=self.database
)
if self.connection.is_connected():
print("MySQL database connection successful.")
except mysql.connector.Error as err:
print(f"Error: {err}")
raise
except Exception as e:
print(f"Unexpected error: {e}")
raise
def execute_query(self, query: str, parameters: tuple = ()) -> Any:
"""
Execute an SQL query on the connected MySQL database.
:param query: The SQL query string.
:param parameters: Parameters to use in the SQL query (default is an empty tuple).
:return: Query results for SELECT queries; otherwise, None.
"""
if not self.connection or not self.connection.is_connected():
raise ValueError("Connection is not established. Call 'connect()' first!")
try:
cursor = self.connection.cursor()
cursor.execute(query, parameters)
if query.strip().upper().startswith("SELECT"):
results = cursor.fetchall()
return results
else:
self.connection.commit()
print("Query executed successfully.")
except Error as e:
print(f"Error executing query: {e}")
raise
finally:
cursor.close()
def close_connection(self) -> None:
"""
Closes the database connection.
"""
if self.connection:
self.connection.close()
print("MySQL database connection closed.")
else:
print("No connection to close.")
# Example usage
if __name__ == "__main__":
# Replace with your MySQL credentials and database information
db_connector = MySQLDatabaseConnector(
host="127.0.0.1",
user="root",
password="my-secret-password",
database="mydb",
port=9000
)
db_connector.connect()
# Example: Fetch records
select_query = "SELECT * FROM mydb.users LIMIT 5;"
results = db_connector.execute_query(select_query)
print("Query results:", results)
db_connector.close_connection()
```
I try execute the code with poetry environment:
```
wheel 0.45.1
mysql-connector-python 9.1.0
setuptools 75.8.0
```
and in venv with pip packages:
```
mysql-connector-python 9.1.0
pip 24.3.1
setuptools 75.8.0
wheel 0.45.1
```
I run it on Windows 11 with python 3.12.2. I try also on python 3.13 and 3.9. Only one output I got using Terminal in PyCharm was:
"Process finished with exit code -1073741819 (0xC0000005)"
I try the same on linux and everything works fine.