Python中数据库操作与异常处理详解

import sqlite3

try:
    conn = sqlite3.connect('production.db')
except sqlite3.OperationalError as e:
    print(f"数据库连接失败: {e}")
    # 可能的原因：文件打不开、权限不足、网络不通

# 表不存在
cursor.execute("SELECT * FROM users")  
# sqlite3.OperationalError: no such table: users

# 列不存在
cursor.execute("SELECT phone FROM user")  
# sqlite3.OperationalError: no such column: phone

# SQL语法错误
cursor.execute("CREATE TABLE users id INTEGER")  
# sqlite3.OperationalError: near "id": syntax error

# 违反了唯一约束
cursor.execute("INSERT INTO users (phone) VALUES ('13800138000')")
# sqlite3.IntegrityError: UNIQUE constraint failed: users.phone

import sqlite3

def register_user(phone, name):
    try:
        conn.execute(
            "INSERT INTO users (phone, name) VALUES (?, ?)",
            (phone, name)
        )
        conn.commit()
        return {"status": "success", "msg": "注册成功"}
    except sqlite3.IntegrityError as e:
        if "UNIQUE constraint failed: users.phone" in str(e):
            # 手机号已注册，应该返回"用户已存在"，而不是"服务器错误"
            return {"status": "exists", "msg": "该手机号已注册"}
        else:
            # 其他完整性错误，记录详细日志
            logger.error(f"IntegrityError: {e}")
            raise
    except sqlite3.OperationalError as e:
        # 连接问题，可以重试
        logger.warning(f"数据库操作失败，准备重试: {e}")
        time.sleep(1)
        return register_user(phone, name)  # 递归重试（要有上限）

def get_user(user_id):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
    result = cursor.fetchone()
    cursor.close()
    conn.close()
    return result

def get_user(user_id):
    with sqlite3.connect('app.db') as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
            return cursor.fetchone()
    # 退出with块时，cursor和conn会自动关闭，即使发生了异常

class UserRegistrationError(Exception):
    """用户注册相关的所有异常的基类"""
    pass

class UserAlreadyExistsError(UserRegistrationError):
    pass

class DatabaseConnectionError(UserRegistrationError):
    pass

def register_user(phone, name):
    try:
        # 数据库操作...
    except sqlite3.IntegrityError as e:
        if "UNIQUE constraint" in str(e):
            raise UserAlreadyExistsError(f"手机号{phone}已注册") from e
    except sqlite3.OperationalError as e:
        raise DatabaseConnectionError("数据库暂时不可用") from e

try:
    result = register_user(phone, name)
except UserAlreadyExistsError:
    # 前端显示"用户已存在"
except DatabaseConnectionError:
    # 前端显示"系统繁忙，请稍后重试"

import sqlite3
import time
import logging

logger = logging.getLogger(__name__)

class DatabaseError(Exception):
    pass

class DuplicateRecordError(DatabaseError):
    pass

def with_retry(max_retries=3, delay=1):
    """重试装饰器，用于处理临时性数据库错误"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except sqlite3.OperationalError as e:
                    if attempt == max_retries - 1:
                        raise DatabaseError(f"操作失败，已重试{max_retries}次") from e
                    logger.warning(f"第{attempt+1}次重试，错误: {e}")
                    time.sleep(delay * (attempt + 1))  # 指数退避
        return wrapper
    return decorator

@with_retry(max_retries=3)
def create_user(phone, name):
    try:
        with sqlite3.connect('app.db') as conn:
            conn.execute(
                "INSERT INTO users (phone, name) VALUES (?, ?)",
                (phone, name)
            )
            conn.commit()
            return {"id": conn.lastrowid, "phone": phone}
    except sqlite3.IntegrityError as e:
        if "UNIQUE constraint" in str(e):
            raise DuplicateRecordError(f"手机号{phone}已存在") from e
        raise DatabaseError(f"数据完整性错误: {e}") from e