Die erste Anwendung mit Python.
Meine Anwendung bleibt bei PyMySQL-Aufrufen hängen. Ist diese Bibliothek stabil?
Ich versuche hier nur eine Verbindung zu einer MySQL-Datenbank herzustellen, aber jedes Python-Paket scheint schwierig zu verwenden zu sein? In jeder anderen Sprache ist es sehr einfach und weist keine seltsamen Probleme auf.
migrateFresh.py:
import pymysql.cursors import os import glob from bootstrap import * reset_database() migrate_and_seed()
Dann habe ich 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()
我建议您拆分迁移和种子函数,并在每个函数中分别输入连接上下文管理器:
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