Python：Pymysql 静静地挂起，没有任何响应？

Question

第一个使用 Python 的应用程序。

我的应用程序挂在 pymysql 调用上，这个库稳定吗？

只是想在这里连接到 MySQL 数据库，但每个 Python 包似乎都很难使用？任何其他语言都非常简单，没有奇怪的问题。

migrateFresh.py：

import pymysql.cursors
import os
import glob
from bootstrap import *

reset_database()
migrate_and_seed()

然后我有 bootstrap.py：

import pymysql.cursors
from dotenv import load_dotenv
import os
import glob

load_dotenv()

def get_connection (database = ''):
    return pymysql.connect(
        host = os.environ.get("DB_HOST"),
        user = os.environ.get("DB_USER"),
        password = os.environ.get("DB_PASS"),
        database = database,
        cursorclass=pymysql.cursors.DictCursor)

def reset_database():
    connection = get_connection()

    with connection:
        with connection.cursor() as cursor:
            cursor.execute(f'DROP DATABASE IF EXISTS {os.environ.get("DB_NAME")};')
            cursor.execute(f'CREATE DATABASE {os.environ.get("DB_NAME")};')

        connection.commit()

    print('Database has been reset')

def migrate_and_seed():
    connection = get_connection(os.environ.get("DB_NAME"))

    with connection:
        with connection.cursor() as cursor:
            for f in sorted(glob.glob("migrations/*.sql")):
                print(f)
                with open(f, "r") as infile:
                    query = infile.read()
                    cursor.execute(query)

        connection.commit()

        with connection.cursor() as cursor:
            for f in sorted(glob.glob("seeders/*.sql")):
                print(f)
                with open(f, "r") as infile:
                    query = infile.read()
                    cursor.execute(query)

        connection.commit()

Answer 1

我建议您拆分迁移和种子函数，并在每个函数中分别输入连接上下文管理器：

import pymysql.cursors
from dotenv import load_dotenv
import os
import glob

load_dotenv()


def get_connection(database=''):
    return pymysql.connect(
        host=os.environ.get("DB_HOST"),
        user=os.environ.get("DB_USER"),
        password=os.environ.get("DB_PASS"),
        database=database,
        cursorclass=pymysql.cursors.DictCursor)


def reset_database():
    connection = get_connection()

    with connection:
        with connection.cursor() as cursor:
            cursor.execute(f'DROP DATABASE IF EXISTS {os.environ.get("DB_NAME")};')
            cursor.execute(f'CREATE DATABASE {os.environ.get("DB_NAME")};')

        connection.commit()

    print('Database has been reset')


def migrate_and_seed():
    ok = migrate()
    if not ok:
        print("migration failed. not seeding...")
        return
    ok = seed()
    if not ok:
        print("seeding failed.")


def migrate():
    try:
        with get_connection(os.environ.get("DB_NAME")) as connection:
            with connection.cursor() as cursor:
                for f in sorted(glob.glob("migrations/*.sql")):
                    print(f)
                    with open(f, "r") as infile:
                        query = infile.read()
                        cursor.execute(query)
    
            connection.commit()
        return True
    except Exception as e:
        print("Migrate Exception:", e)
        return False
    
    
def seed():
    try:
        with get_connection(os.environ.get("DB_NAME")) as connection:
            with connection.cursor() as cursor:
                for f in sorted(glob.glob("seeders/*.sql")):
                    print(f)
                    with open(f, "r") as infile:
                        query = infile.read()
                        cursor.execute(query)
        
            connection.commit()
    except Exception as e:
        print("Seed Exception:", e)
        return False