import mysql.connector

config = {
    'user': 'root',
    'password': '123456',
    'host': 'localhost',
    'database': 'test_db'
}
conn = mysql.connector.connect(**config)
cursor = conn.cursor()

# 創(chuàng)建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    email VARCHAR(255)
)
""")

# 插入數(shù)據(jù)
cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ('Alice', 'alice@example.com'))
conn.commit()

try:
    cursor.execute("INSERT INTO users (name) VALUES ('Bob')")
    cursor.execute("UPDATE users SET email='error' WHERE id=999")  # 模擬錯(cuò)誤
    conn.commit()
except mysql.connector.Error as e:
    conn.rollback()

import cx_Oracle

dsn = cx_Oracle.makedsn("localhost", 1521, service_name="ORCLCDB")
conn = cx_Oracle.connect(user="scott", password="tiger", dsn=dsn)
cursor = conn.cursor()

# 使用序列插入數(shù)據(jù)
cursor.execute("CREATE SEQUENCE user_seq START WITH 1 INCREMENT BY 1")
cursor.execute("""
INSERT INTO users (id, name) 
VALUES (user_seq.NEXTVAL, :name)""", name="Charlie")
conn.commit()

p_name = cursor.var(str)
cursor.callproc("get_user", [1, p_name])
print(p_name.getvalue())  # 輸出結(jié)果

import pyodbc

conn = pyodbc.connect(
    Driver='{ODBC Driver 17 for SQL Server}',
    Server='localhost',
    Database='test_db',
    UID='sa',
    PWD='password'
)
cursor = conn.cursor()

# 分頁(yè)查詢(xún)（OFFSET FETCH）
cursor.execute("""
SELECT * FROM users
ORDER BY id
OFFSET 10 ROWS
FETCH NEXT 10 ROWS ONLY
""")
print(cursor.fetchall())

data = [('David', 'david@example.com'), ('Eva', 'eva@example.com')]
cursor.executemany("INSERT INTO users (name, email) VALUES (?, ?)", data)
conn.commit()

# 示例：MySQL 上下文管理器
with mysql.connector.connect(**config) as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT 1")

try:
    conn = cx_Oracle.connect("invalid_connection")
except cx_Oracle.DatabaseError as e:
    error = e.args[0]
    print(f"Code: {error.code}, Message: {error.message}")
finally:
    if conn:
        conn.close()

class DatabaseAutomator:
    """通用數(shù)據(jù)庫(kù)操作封裝"""
    def __init__(self, db_type, **config):
        self.db_type = db_type
        self.config = config
        self.conn = None

    def __enter__(self):
        if self.db_type == "mysql":
            self.conn = mysql.connector.connect(**self.config)
        elif self.db_type == "oracle":
            dsn = cx_Oracle.makedsn(**self.config)
            self.conn = cx_Oracle.connect(user=self.config['user'], password=self.config['password'], dsn=dsn)
        # 其他數(shù)據(jù)庫(kù)類(lèi)似
        return self.conn.cursor()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.conn:
            if exc_type: self.conn.rollback()
            else: self.conn.commit()
            self.conn.close()

# 使用示例
with DatabaseAutomator("mysql", user="root", password="123456", host="localhost", database="test_db") as cursor:
    cursor.execute("SELECT * FROM users")

from sqlalchemy import create_engine

# MySQL
engine = create_engine("mysql+pymysql://user:password@localhost/db")

# Oracle
engine = create_engine("oracle+cx_oracle://user:password@localhost:1521/ORCLCDB")

# SQL Server
engine = create_engine("mssql+pyodbc://user:password@localhost/db?driver=ODBC+Driver+17+for+SQL+Server")

with engine.connect() as conn:
    result = conn.execute("SELECT * FROM users")
    print(result.fetchall())