Python 连接 MySQL：参数化查询、事务与资源管理

python -m pip install mysql-connector-python
export DB_HOST=127.0.0.1
export DB_USER=app_user
export DB_PASSWORD='replace-me'
export DB_NAME=engineering_notes

import os

import mysql.connector
from mysql.connector import MySQLConnection


def connect() -> MySQLConnection:
    return mysql.connector.connect(
        host=os.environ.get("DB_HOST", "127.0.0.1"),
        user=os.environ["DB_USER"],
        password=os.environ["DB_PASSWORD"],
        database=os.environ.get("DB_NAME", "engineering_notes"),
        charset="utf8mb4",
    )


def create_task(project_id: int, title: str) -> int:
    connection = connect()
    cursor = connection.cursor()
    try:
        cursor.execute(
            "INSERT INTO tasks (project_id, title) VALUES (%s, %s)",
            (project_id, title),
        )
        task_id = cursor.lastrowid
        connection.commit()
        return int(task_id)
    except Exception:
        connection.rollback()
        raise
    finally:
        cursor.close()
        connection.close()


def list_tasks(project_id: int) -> list[tuple[int, str, str]]:
    connection = connect()
    cursor = connection.cursor()
    try:
        cursor.execute(
            "SELECT id, title, status FROM tasks WHERE project_id = %s ORDER BY id",
            (project_id,),
        )
        return [(int(row[0]), str(row[1]), str(row[2])) for row in cursor.fetchall()]
    finally:
        cursor.close()
        connection.close()


if __name__ == "__main__":
    new_id = create_task(1, "verify parameterized query")
    print("created:", new_id)
    print(list_tasks(1))

def create_many(project_id: int, titles: list[str]) -> int:
    connection = connect()
    cursor = connection.cursor()
    try:
        rows = [(project_id, title) for title in titles]
        cursor.executemany(
            "INSERT INTO tasks (project_id, title) VALUES (%s, %s)",
            rows,
        )
        connection.commit()
        return cursor.rowcount
    except Exception:
        connection.rollback()
        raise
    finally:
        cursor.close()
        connection.close()