blog_api/internal/database.py

import pymysql
from fastapi import HTTPException, status
DB_CONFIG = {
    "host": "111.229.38.129",
    "user": "root",
    "password": "zl981023",
    "database": "blogapi",
    "charset": "utf8mb4",
    "cursorclass": pymysql.cursors.DictCursor,
}

# 创建数据库连接


def create_connection():
    return pymysql.connect(**DB_CONFIG)

# 执行 SQL 查询


def execute_query(query, params=None, fetchall=False):
    conn = create_connection()
    with conn.cursor() as cursor:
        cursor.execute(query, params)
        if fetchall:
            result = cursor.fetchall()
        else:
            result = cursor.fetchone()
    conn.commit()
    conn.close()
    return result

# 查询单个数据


def fetch_one(query, params=None):
    result = execute_query(query, params)
    return result if result else None

# 查询所有数据


def fetch_all(query, params=None, fetchall=True):
    return execute_query(query, params, fetchall)


# 验证数据是否存在，不存在抛出异常
def raise_if_not_found(item, message="Item not found"):
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=message
        )

# 验证唯一，非唯一抛出异常


def raise_if_exists(item, message="Item already exists"):
    if item:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=message
        )

# 成功响应数据
def response_success(data={}, message="data do success"):
    if not data:
        return {
            "status": status.HTTP_200_OK,
            "message": message,
        }
    else:
        return {
            "status": status.HTTP_200_OK,
            "message": message,
            "data": data
        }