MySQL Forums
Forum List  »  Connector/Python

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.

Options: Reply• Quote


Subject
Written By
Posted
Re: MYSQL - Python Connector - No Response
January 16, 2025 07:47AM


Sorry, only registered users may post in this forum.

Content reproduced on this site is the property of the respective copyright holders. It is not reviewed in advance by Oracle and does not necessarily represent the opinion of Oracle or any other party.