Python如何连接MySQL数据库？PyMySQL详细使用教程

pip install pymysql

import pymysql

# 数据库连接信息
host = 'your_host'
port = 3306
user = 'your_user'
password = 'your_password'
database = 'your_database'

# 建立连接
try:
    connection = pymysql.connect(host=host, port=port, user=user, password=password, database=database, cursorclass=pymysql.cursors.DictCursor)

    # 使用cursor()方法获取操作游标
    with connection.cursor() as cursor:
        # SQL 查询语句
        sql = "SELECT * FROM your_table"
        cursor.execute(sql)
        results = cursor.fetchall()

        for row in results:
            print(row)

    # 提交事务
    connection.commit()

except Exception as e:
    print(f"数据库连接或操作出错：{e}")
finally:
    # 关闭连接
    if connection:
        connection.close()

from dbutils.pooled_db import PooledDB

# 连接池配置
pool = PooledDB(
    creator=pymysql,  # 使用的连接器
    maxconnections=5,  # 连接池允许的最大连接数
    mincached=2,       # 初始化时，连接池至少创建的空闲连接数
    maxcached=5,       # 连接池中最多闲置的连接数
    maxshared=3,       # 连接池中最多共享的连接数
    blocking=True,     # 连接池中如果没有可用连接后，是否阻塞等待
    host='your_host',
    port=3306,
    user='your_user',
    password='your_password',
    database='your_database',
    cursorclass=pymysql.cursors.DictCursor
)

# 使用连接池获取连接
connection = pool.connection()
try:
    with connection.cursor() as cursor:
        sql = "SELECT * FROM your_table WHERE id = %s"
        cursor.execute(sql, (1,)) # 注意这里使用参数化查询，防止SQL注入
        result = cursor.fetchone()
        print(result)
except Exception as e:
    print(f"数据库操作出错：{e}")
finally:
    connection.close() # 连接用完后，放回连接池

import pymysql

# 数据库连接信息
host = 'your_host'
port = 3306
user = 'your_user'
password = 'your_password'
database = 'your_database'

connection = pymysql.connect(host=host, port=port, user=user, password=password, database=database)

try:
    with connection.cursor() as cursor:
        # 开始事务
        connection.begin()

        # 执行SQL语句
        sql1 = "UPDATE accounts SET balance = balance - 100 WHERE id = 1"
        cursor.execute(sql1)

        sql2 = "UPDATE accounts SET balance = balance + 100 WHERE id = 2"
        cursor.execute(sql2)

        # 提交事务
        connection.commit()
        print("事务执行成功")

except Exception as e:
    # 回滚事务
    connection.rollback()
    print(f"事务执行失败：{e}")
finally:
    connection.close()

import pymysql

# 数据库连接信息
host = 'your_host'
port = 3306
user = 'your_user'
password = 'your_password'
database = 'your_database'

connection = pymysql.connect(host=host, port=port, user=user, password=password, database=database)

try:
    with connection.cursor() as cursor:
        # 使用参数化查询
        sql = "SELECT * FROM users WHERE username = %s AND password = %s"
        username = "test_user"
        password = "test_password"
        cursor.execute(sql, (username, password)) # 注意这里是tuple
        result = cursor.fetchone()

        if result:
            print("登录成功")
        else:
            print("登录失败")

except Exception as e:
    print(f"数据库操作出错：{e}")
finally:
    connection.close()

import pymysql

# 数据库连接信息
host = 'your_host'
port = 3306
user = 'your_user'
password = 'your_password'
database = 'your_database'
charset = 'utf8mb4' # 推荐使用utf8mb4，支持更多字符

connection = pymysql.connect(host=host, port=port, user=user, password=password, database=database, charset=charset)

try:
    with connection.cursor() as cursor:
        sql = "SELECT name FROM your_table WHERE id = %s"
        cursor.execute(sql, (1,))
        result = cursor.fetchone()
        print(result['name']) # 假设name字段是中文
except Exception as e:
    print(f"数据库操作出错：{e}")
finally:
    connection.close()

import pymysql
import datetime

# 数据库连接信息
host = 'your_host'
port = 3306
user = 'your_user'
password = 'your_password'
database = 'your_database'

connection = pymysql.connect(host=host, port=port, user=user, password=password, database=database, cursorclass=pymysql.cursors.DictCursor)

try:
    with connection.cursor() as cursor:
        # 插入日期
        now = datetime.datetime.now()
        sql = "INSERT INTO your_table (created_at) VALUES (%s)"
        cursor.execute(sql, (now,))
        connection.commit()

        # 查询日期
        sql = "SELECT created_at FROM your_table WHERE id = %s"
        cursor.execute(sql, (1,))
        result = cursor.fetchone()
        print(result['created_at']) # 返回的是datetime对象

except Exception as e:
    print(f"数据库操作出错：{e}")
finally:
    connection.close()

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime

# 数据库连接信息
host = 'your_host'
port = 3306
user = 'your_user'
password = 'your_password'
database = 'your_database'

# 创建数据库引擎
engine = create_engine(f'mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4')

# 创建基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    email = Column(String(255))
    created_at = Column(DateTime, default=datetime.datetime.now)

# 创建表
Base.metadata.create_all(engine)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 添加数据
new_user = User(name='Alice', email='alice@example.com')
session.add(new_user)
session.commit()

# 查询数据
user = session.query(User).filter_by(name='Alice').first()
print(user.email)

# 关闭会话
session.close()