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