Python: Pymysql digantung senyap tanpa tindak balas?

Question

Aplikasi pertama menggunakan Python.

Aplikasi saya tergantung pada panggilan pymysql, adakah perpustakaan ini stabil?

Hanya cuba menyambung ke pangkalan data MySQL di sini tetapi setiap pakej Python kelihatan sukar untuk digunakan? Dalam mana-mana bahasa lain ia sangat mudah dan tidak mempunyai masalah pelik.

migrateFresh.py:

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

reset_database()
migrate_and_seed()

Kemudian saya mempunyai 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

Saya mengesyorkan anda membahagikan fungsi migrasi dan seed dan masukkan pengurus konteks sambungan secara berasingan dalam setiap fungsi:

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