pip install mysql-connector-python pymysql psycopg2

# MySQL連接示例
import mysql.connector

mysql_conn = mysql.connector.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='your_database'
)
print("MySQL連接成功！")

# PostgreSQL連接示例
import psycopg2

pg_conn = psycopg2.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)
print("PostgreSQL連接成功！")

# SQLite連接示例
import sqlite3

sqlite_conn = sqlite3.connect('example.db')
print("SQLite連接成功！")

import configparser
import mysql.connector

# 讀取配置文件
config = configparser.ConfigParser()
config.read('db_config.ini')

# 獲取數(shù)據(jù)庫(kù)連接參數(shù)
db_params = {
    'host': config['mysql']['host'],
    'user': config['mysql']['user'],
    'password': config['mysql']['password'],
    'database': config['mysql']['database']
}

# 建立連接
try:
    conn = mysql.connector.connect(**db_params)
    if conn.is_connected():
        print("數(shù)據(jù)庫(kù)連接成功！")
except Exception as e:
    print(f"連接失敗: {e}")

# 創(chuàng)建游標(biāo)對(duì)象
cursor = conn.cursor()

# 執(zhí)行簡(jiǎn)單查詢
cursor.execute("SELECT * FROM users")

# 獲取所有結(jié)果
results = cursor.fetchall()

# 遍歷結(jié)果
for row in results:
    print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")

# 關(guān)閉游標(biāo)
cursor.close()

# 參數(shù)化查詢示例
user_id = 5
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

# 插入數(shù)據(jù)時(shí)的參數(shù)化查詢
new_user = ('John Doe', 'john@example.com', 30)
cursor.execute("INSERT INTO users (name, email, age) VALUES (%s, %s, %s)", new_user)
conn.commit()

import sqlite3

# 創(chuàng)建連接
conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 創(chuàng)建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    age INTEGER
)
''')

# 插入數(shù)據(jù)
def create_user(name, email, age):
    try:
        cursor.execute(
            "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
            (name, email, age)
        )
        conn.commit()
        print("用戶添加成功！")
    except sqlite3.IntegrityError:
        print("郵箱已存在！")

# 查詢數(shù)據(jù)
def read_users():
    cursor.execute("SELECT * FROM users")
    return cursor.fetchall()

# 更新數(shù)據(jù)
def update_user(user_id, name=None, email=None, age=None):
    updates = []
    params = []
    
    if name:
        updates.append("name = ?")
        params.append(name)
    if email:
        updates.append("email = ?")
        params.append(email)
    if age is not None:
        updates.append("age = ?")
        params.append(age)
    
    params.append(user_id)
    cursor.execute(
        f"UPDATE users SET {', '.join(updates)} WHERE id = ?",
        params
    )
    conn.commit()

# 刪除數(shù)據(jù)
def delete_user(user_id):
    cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
    conn.commit()

# 使用示例
create_user('Alice', 'alice@example.com', 25)
create_user('Bob', 'bob@example.com', 30)

print("所有用戶:")
users = read_users()
for user in users:
    print(user)

update_user(1, age=26)

print("更新后的用戶:")
users = read_users()
for user in users:
    print(user)

# 關(guān)閉連接
conn.close()

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

# 定義基類
Base = declarative_base()

# 定義模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100), unique=True)
    age = Column(Integer)

# 創(chuàng)建引擎和會(huì)話
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()

# 創(chuàng)建表
Base.metadata.create_all(engine)

# 插入數(shù)據(jù)
new_user = User(name='Charlie', email='charlie@example.com', age=35)
session.add(new_user)
session.commit()

# 查詢數(shù)據(jù)
users = session.query(User).filter(User.age > 25).all()
for user in users:
    print(f"{user.name}: {user.email}")

# 更新數(shù)據(jù)
user = session.query(User).filter_by(name='Charlie').first()
if user:
    user.age = 36
    session.commit()

# 刪除數(shù)據(jù)
user = session.query(User).filter_by(name='Charlie').first()
if user:
    session.delete(user)
    session.commit()

# 關(guān)閉會(huì)話
session.close()

from sqlalchemy import or_, and_

# 復(fù)雜查詢示例
users = session.query(User).filter(
    or_(
        User.age < 25,
        User.age > 35
    ),
    User.email.like('%@example.com')
).order_by(User.name.desc()).limit(10).all()

# 聚合查詢
from sqlalchemy import func

age_stats = session.query(
    func.count(User.id),
    func.avg(User.age),
    func.min(User.age),
    func.max(User.age)
).first()

# 連接查詢
class Address(Base):
    __tablename__ = 'addresses'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    street = Column(String(100))
    city = Column(String(50))
    
    user = relationship("User", back_populates="addresses")

User.addresses = relationship("Address", order_by=Address.id, back_populates="user")

# 執(zhí)行連接查詢
users_with_addresses = session.query(User).join(Address).filter(Address.city == 'New York').all()

# Django模型定義
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=50)
    email = models.EmailField(unique=True)
    age = models.IntegerField()
    created_at = models.DateTimeField(auto_now_add=True)
    
    def __str__(self):
        return self.name

# 使用Django ORM進(jìn)行查詢
from myapp.models import User

# 創(chuàng)建用戶
User.objects.create(name='Alice', email='alice@example.com', age=25)

# 查詢用戶
users = User.objects.filter(age__gt=20).order_by('-created_at')

# 更新用戶
User.objects.filter(email='alice@example.com').update(age=26)

# 刪除用戶
User.objects.filter(email='alice@example.com').delete()

import mysql.connector
from mysql.connector import Error

try:
    conn = mysql.connector.connect(
        host='localhost',
        database='your_database',
        user='your_username',
        password='your_password'
    )
    
    # 開啟事務(wù)
    conn.start_transaction()
    
    cursor = conn.cursor()
    
    # 執(zhí)行多個(gè)操作
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    
    # 提交事務(wù)
    conn.commit()
    print("事務(wù)執(zhí)行成功！")
    
except Error as e:
    # 回滾事務(wù)
    conn.rollback()
    print(f"事務(wù)執(zhí)行失敗: {e}")
    
finally:
    if conn.is_connected():
        cursor.close()
        conn.close()

from mysql.connector import pooling
import threading

# 創(chuàng)建連接池
dbconfig = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "your_database"
}

connection_pool = pooling.MySQLConnectionPool(
    pool_name="my_pool",
    pool_size=5,
    pool_reset_session=True,
    **dbconfig
)

def query_with_pool(user_id):
    try:
        # 從連接池獲取連接
        conn = connection_pool.get_connection()
        cursor = conn.cursor()
        
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        result = cursor.fetchone()
        
        print(f"用戶查詢結(jié)果: {result}")
        
        # 將連接返回給連接池
        cursor.close()
        conn.close()
        
    except Error as e:
        print(f"查詢失敗: {e}")

# 多線程使用連接池示例
threads = []
for i in range(10):
    thread = threading.Thread(target=query_with_pool, args=(i % 5 + 1,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

from contextlib import contextmanager

@contextmanager
def get_db_connection():
    conn = None
    try:
        conn = connection_pool.get_connection()
        yield conn
    except Error as e:
        print(f"數(shù)據(jù)庫(kù)錯(cuò)誤: {e}")
        if conn:
            conn.rollback()
        raise
    finally:
        if conn:
            conn.close()

@contextmanager
def get_db_cursor():
    with get_db_connection() as conn:
        cursor = conn.cursor()
        try:
            yield cursor
            conn.commit()
        except:
            conn.rollback()
            raise
        finally:
            cursor.close()

# 使用上下文管理器
with get_db_cursor() as cursor:
    cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", 
                  ("John", "john@example.com"))
    print("數(shù)據(jù)插入成功！")

# 批量插入示例
def bulk_insert_users(users_data):
    try:
        cursor = conn.cursor()
        
        # 批量插入
        cursor.executemany(
            "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
            users_data
        )
        
        conn.commit()
        print(f"成功插入 {len(users_data)} 條記錄")
        
    except Error as e:
        conn.rollback()
        print(f"批量插入失敗: {e}")

# 使用示例
users_data = [
    ('User1', 'user1@example.com', 25),
    ('User2', 'user2@example.com', 30),
    ('User3', 'user3@example.com', 35)
]
bulk_insert_users(users_data)

# 數(shù)據(jù)庫(kù)工具類示例
class DatabaseManager:
    def __init__(self, **config):
        self.config = config
        self.connection_pool = None
        
    def create_pool(self, pool_size=5):
        try:
            self.connection_pool = pooling.MySQLConnectionPool(
                pool_name="app_pool",
                pool_size=pool_size,
                **self.config
            )
            print("連接池創(chuàng)建成功！")
        except Error as e:
            print(f"連接池創(chuàng)建失敗: {e}")
            raise
    
    @contextmanager
    def get_connection(self):
        if not self.connection_pool:
            self.create_pool()
        
        conn = None
        try:
            conn = self.connection_pool.get_connection()
            yield conn
        finally:
            if conn:
                conn.close()
    
    @contextmanager
    def get_cursor(self, dictionary=False):
        with self.get_connection() as conn:
            cursor = conn.cursor(dictionary=dictionary)
            try:
                yield cursor
                conn.commit()
            except:
                conn.rollback()
                raise
            finally:
                cursor.close()
    
    def execute_query(self, query, params=None):
        with self.get_cursor() as cursor:
            cursor.execute(query, params or ())
            return cursor.fetchall()
    
    def execute_command(self, query, params=None):
        with self.get_cursor() as cursor:
            cursor.execute(query, params or ())
            return cursor.rowcount

# 使用示例
db_config = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "your_database"
}

db_manager = DatabaseManager(**db_config)

# 執(zhí)行查詢
users = db_manager.execute_query("SELECT * FROM users WHERE age > %s", (25,))
for user in users:
    print(user)

# 執(zhí)行命令
row_count = db_manager.execute_command(
    "UPDATE users SET age = %s WHERE id = %s",
    (26, 1)
)
print(f"更新了 {row_count} 行數(shù)據(jù)")

import time
from functools import wraps

def query_logger(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        
        print(f"查詢執(zhí)行時(shí)間: {end_time - start_time:.4f}秒")
        return result
    return wrapper

# 使用裝飾器記錄查詢時(shí)間
class MonitoredDatabaseManager(DatabaseManager):
    @query_logger
    def execute_query(self, query, params=None):
        return super().execute_query(query, params)
    
    @query_logger
    def execute_command(self, query, params=None):
        return super().execute_command(query, params)

# 使用監(jiān)控版數(shù)據(jù)庫(kù)管理器
monitored_db = MonitoredDatabaseManager(**db_config)
users = monitored_db.execute_query("SELECT * FROM users WHERE age > %s", (25,))