The first application using Python.
My application hangs on pymysql calls, is this library stable?
Just trying to connect to a MySQL database here, but every Python package seems to be difficult to use? In any other language it's very simple and has no weird problems.
migrateFresh.py:
import pymysql.cursors import os import glob from bootstrap import * reset_database() migrate_and_seed()
Then I have 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()
P粉3842444732024-04-02 09:58:03
I recommend that you split the migration and seed functions and enter the connection context manager separately in each function:
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