Python Learning
1. Pre-Learning Highlights
Previously, we had an article titled “Python Learning [43]: How Python Programs Share the Same Variable” which discussed how to share variables in Python programs. At that time, we used examples of classes and students. As we delve deeper into the language, we find an interesting phenomenon: regardless of which computer language books you read, many examples are based on elements such as schools, classes, teachers, courses, and students to construct programs or design databases.
Why is this the case? It’s quite simple, because the elements of schools are the easiest for everyone to understand, and these elements can naturally be categorized and linked. For example, when designing a database (which is similar to classes), there are intricate connections between classes, between classes and students, and between classes and teachers. These are also the foundations of relational databases.
In actual engineering projects, the use of databases is more common than school elements. Therefore, this article uses database connections as an example to illustrate the benefits of sharing variables in programs.
Taking Python’s connection to a MySQL database as an example, we define the database connection in the class’s __init__ method and use this connection in various functional modules (such as creating tables, inserting tables, querying tables, etc.).
2. Application of Shared Variables in Database Connections
Below is a complete example of connecting to a MySQL database using Python, encapsulating database operations in a class, initializing the connection in __init__, and reusing this connection in various methods.
2.1 Install Dependencies
|
pip install mysql-connector-python
# or
pip install pymysql
|
2.2 Database Connection Class
# -*- coding: utf-8 -*-
import mysql.connector
from mysql.connector import Error
import logging
class MySQLManager:
"""MySQL Database Management Class"""
def __init__(self, host='localhost', user='root', password='root@123', database='test_db'):
"""
Initialize database connection
Args:
host: Database host
user: Username
password: Password
database: Database name
"""
self.host = host
self.user = user
self.password = password
self.database = database
self.connection = None
self.cursor = None
# Configure logging
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
# Initialize connection
self._connect()
def _connect(self):
"""Establish database connection"""
try:
self.connection = mysql.connector.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database,
charset='utf8mb4',
collation='utf8mb4_unicode_ci'
)
if self.connection.is_connected():
self.cursor = self.connection.cursor()
self.logger.info(f"Successfully connected to MySQL database: {self.database}")
except Error as e:
self.logger.error(f"Database connection failed: {e}")
raise
def _ensure_connection(self):
"""Ensure database connection is valid"""
try:
if self.connection is None or not self.connection.is_connected():
self.logger.warning("Database connection has been disconnected, attempting to reconnect...")
self._connect()
except Error:
self._connect()
def create_table(self, table_name, columns_definition):
"""
Create table
Args:
table_name: Table name
columns_definition: Column definition string
"""
self._ensure_connection()
try:
create_table_query = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
{columns_definition}
)
"""
self.cursor.execute(create_table_query)
self.connection.commit()
self.logger.info(f"Table {table_name} created successfully")
except Error as e:
self.logger.error(f"Failed to create table: {e}")
self.connection.rollback()
raise
def insert_data(self, table_name, data_dict):
"""
Insert data
Args:
table_name: Table name
data_dict: Data dictionary {column_name: value}
"""
self._ensure_connection()
try:
columns = ', '.join(data_dict.keys())
placeholders = ', '.join(['%s'] * len(data_dict))
insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
values = tuple(data_dict.values())
self.cursor.execute(insert_query, values)
self.connection.commit()
self.logger.info(f"Data inserted successfully, affected rows: {self.cursor.rowcount}")
return self.cursor.lastrowid
except Error as e:
self.logger.error(f"Failed to insert data: {e}")
self.connection.rollback()
raise
def query_data(self, table_name, columns="*", where_clause=None, params=None):
"""
Query data
Args:
table_name: Table name
columns: Columns to query
where_clause: WHERE condition
params: Parameter tuple
Returns:
Query result list
"""
self._ensure_connection()
try:
query = f"SELECT {columns} FROM {table_name}"
if where_clause:
query += f" WHERE {where_clause}"
self.cursor.execute(query, params or ())
results = self.cursor.fetchall()
# Get column names
column_names = [desc[0] for desc in self.cursor.description]
self.logger.info(f"Queried {len(results)} records")
return results, column_names
except Error as e:
self.logger.error(f"Failed to query data: {e}")
raise
def update_data(self, table_name, update_dict, where_clause, params=None):
"""
Update data
Args:
table_name: Table name
update_dict: Update field dictionary
where_clause: WHERE condition
params: Parameter tuple
"""
self._ensure_connection()
try:
set_clause = ', '.join([f"{key} = %s" for key in update_dict.keys()])
update_query = f"UPDATE {table_name} SET {set_clause} WHERE {where_clause}"
# Merge parameters: update values + WHERE condition parameters
all_params = tuple(update_dict.values()) + (params or ())
self.cursor.execute(update_query, all_params)
self.connection.commit()
self.logger.info(f"Data updated successfully, affected rows: {self.cursor.rowcount}")
return self.cursor.rowcount
except Error as e:
self.logger.error(f"Failed to update data: {e}")
self.connection.rollback()
raise
def delete_data(self, table_name, where_clause, params=None):
"""
Delete data
Args:
table_name: Table name
where_clause: WHERE condition
params: Parameter tuple
"""
self._ensure_connection()
try:
delete_query = f"DELETE FROM {table_name} WHERE {where_clause}"
self.cursor.execute(delete_query, params or ())
self.connection.commit()
self.logger.info(f"Data deleted successfully, affected rows: {self.cursor.rowcount}")
return self.cursor.rowcount
except Error as e:
self.logger.error(f"Failed to delete data: {e}")
self.connection.rollback()
raise
def close(self):
"""Close database connection"""
try:
if self.cursor:
self.cursor.close()
if self.connection and self.connection.is_connected():
self.connection.close()
self.logger.info("Database connection has been closed")
except Error as e:
self.logger.error(f"Error closing connection: {e}")
def __enter__(self):
"""Support context manager"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Automatically close connection when exiting context"""
self.close()
|
2.3 Usage Example
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from mysql_manager import MySQLManager
def main():
# Database configuration
db_config = {
'host': 'localhost',
'user': 'root',
'password': 'root@123',
'database': 'test_db'
}
# Use context manager to automatically manage connection
with MySQLManager(**db_config) as db:
# 1. Create users table
users_table_definition = """
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
age INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
"""
table_name='users1'
db.create_table(table_name, users_table_definition)
# 2. Insert data
users_data = [
{'name': 'Zhang San', 'email': '[email protected]', 'age': 25},
{'name': 'Li Si', 'email': '[email protected]', 'age': 30},
{'name': 'Wang Wu', 'email': '[email protected]', 'age': 28}
]
for user in users_data:
user_id = db.insert_data(table_name, user)
print(f"Inserted user: {user['name']}, ID: {user_id}")
# 3. Query data
print("\n=== All Users ===")
results, columns = db.query_data('user1')
for row in results:
print(dict(zip(columns, row)))
# 4. Conditional query
print("\n=== Users older than 26 ===")
results, columns = db.query_data(
'user1',
where_clause='age > %s',
params=(26,)
)
for row in results:
print(dict(zip(columns, row)))
# 5. Update data
print("\n=== Update User Age ===")
updated_count = db.update_data(
'user1',
{'age': 35},
'name = %s',
('Li Si',)
)
print(f"Updated {updated_count} records")
# 6. Delete data
print("\n=== Delete User ===")
deleted_count = db.delete_data(
'user1',
'name = %s',
('Wang Wu',)
)
print(f"Deleted {deleted_count} records")
# 7. Verify final results
print("\n=== Final User List ===")
results, columns = db.query_data('user1')
for row in results:
print(dict(zip(columns, row)))
if __name__ == "__main__":
main()
|
The output is as follows:
![Python Learning [44]: Application of Shared Variables in Database Connections](https://boardor.com/wp-content/uploads/2026/01/f8fb1a71-ec69-47fa-8831-d43c2e72f11a.png)
The SQL query results from the backend database for the users table are as follows:
![Python Learning [44]: Application of Shared Variables in Database Connections](https://boardor.com/wp-content/uploads/2026/01/5a79a7ed-d9ce-4111-b626-8ad0be33cd32.png)
Comparing the output results from Python with the SQL query results, there is a typical difference in style between Python and database table styles, but upon careful analysis, the content expressed by both is consistent. This is also what was mentioned earlier regarding JSON data and dictionary data, where key-value pairs can be converted into two-dimensional database table data.
In summary, using shared variables to implement database connections makes the program clearer. The main features are:
1. Connection Reuse: Initialize the connection in __init__, shared by all methods
2. Automatic Reconnection: The _ensure_connection() method ensures the connection is valid
3. Error Handling: Comprehensive exception handling and logging
4. Transaction Support: Automatic commit and rollback
5. Context Management: Supports with statement to automatically close connections
6. Connection Pool: Advanced versions support connection pool management
This design pattern ensures efficient management of database connections and maintainability of the code.
3. Conclusion
As we learn Python, we gradually delve deeper. Writing programs is a long-term process, and as our understanding deepens, we will write more and more complex programs. But fundamentally, we must master the basic knowledge.
Let us maintain our enthusiasm for learning and practice more. See you next time!
![Python Learning [44]: Application of Shared Variables in Database Connections](https://boardor.com/wp-content/uploads/2026/01/34026ddd-d17e-44d7-8b79-d72da434a485.png)
#python