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()
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