Automate Posting Hugo Blog to Social Sites (with a db) Part 2
How To automate posting to social sites
3 Minutes, 32 Seconds
2024-06-15 00:00 +0000
Background
Previous posts in this series
Expand a the mysql class
I create a repo at https://github.com/justin-napolitano/mysql-utility-class.git to enable importing as a submodule the class that i have been workign on.
Set up the db
In another part in this series, I detailed setting up the mysql db via the command line. I am going to furher that workflow by modifying the files in that repo and then running thm to generat tables within my instance of mysql.
Setup you dev environment…again
Copy the .env file
Copy over the .env files from the previous few steps.
Import the Config repo
git submodule add https://github.com/justin-napolitano/mysql-config.git mysql-config
Import the utility class repo
git submodule add https://github.com/justin-napolitano/mysql-utility-class.git mysql-utility-class
Setup the package as a module
From root drop an empty _init.py file
touch __init__.py
From the utility class directory drop another init.py
This one however will contain a relative import to enable access to the class
Touch
cd {to the utility class directory} && touch __init__.py
Echo to file
echo "from .MySQLConnector import MySQLConnector" > __init__.py
Check the module hierarchy
We should be looking like this
your_project/
|-- __init__.py
|-- main.py
|-- .env
`-- MySQLConnector/
|-- __init__.py
`-- MySQLConnector.py
Create the main file
Touch main.py
touch main.,py
Modify main.py
My file currently looks like this to test the connect
from MySQLConnector import MySQLConnector
from dotenv import load_dotenv
import os
if __name__ == "__main__":
load_dotenv() # Load environment variables from .env file
connection = MySQLConnector()
connection.connect()
connection.disconnect()
Test the conection
python -m main.py
Modify the class for more features.
I want to be able to
- create a db
- drop a db
- use a db
- execute a script fro file
Create db
def create_database(self, database_name):
try:
self.cursor.execute(f"CREATE DATABASE {database_name}")
self.cursor.execute(f"SHOW DATABASES LIKE '{database_name}'")
result = self.cursor.fetchone()
if result:
print(f"Database {database_name} created successfully")
return True
else:
print(f"Database {database_name} was not created")
return False
except Error as e:
print(f"Error while creating database: {e}")
return False
Drop db
def drop_database(self, database_name):
try:
self.cursor.execute(f"DROP DATABASE {database_name}")
self.cursor.execute(f"SHOW DATABASES LIKE '{database_name}'")
result = self.cursor.fetchone()
if not result:
print(f"Database {database_name} dropped successfully")
return True
else:
print(f"Database {database_name} was not dropped")
return False
except Error as e:
print(f"Error while dropping database: {e}")
return False
use db
def use_database(self,database_name):
try:
self.cursor.execute(f"USE {database_name}")
print(f"Using database {database_name}")
except Error as e:
print(f"Error while selecting database: {e}")
Execute Script
def execute_script_from_file(self, file_path):
try:
with open(file_path, 'r') as file:
sql_script = file.read()
sql_commands = sql_script.split(';')
for command in sql_commands:
if command.strip():
self.cursor.execute(command)
print(f"Executed: {command}")
self.connection.commit()
print("SQL script executed successfully")
except Error as e:
print(f"Error while executing SQL script: {e}")
Test the module
With the added logic my main.py file looks like
def execute_script_from_file(self, file_path):
try:
with open(file_path, 'r') as file:
sql_script = file.read()
sql_commands = sql_script.split(';')
for command in sql_commands:
if command.strip():
self.cursor.execute(command)
print(f"Executed: {command}")
self.connection.commit()
print("SQL script executed successfully")
except Error as e:
print(f"Error while executing SQL script: {e}")
I execute with the following from the project root.
python -m main.py
The current MySQLConnector Class
Below is as it stand.. The most up to date file can always be found at https://github.com/justin-napolitano/mysql-utility-class.git
import mysql.connector
from mysql.connector import Error
from dotenv import load_dotenv
import os
class MySQLConnector:
def __init__(self):
self.user = os.getenv('DB_USER')
self.password = os.getenv('DB_PASSWORD')
self.host = os.getenv('DB_HOST')
self.database = os.getenv('DB_NAME')
self.connection = None
self.cursor = None
def connect(self):
try:
self.connection = mysql.connector.connect(
user=self.user,
password=self.password,
host=self.host
# Do not specify database here
)
if self.connection.is_connected():
self.cursor = self.connection.cursor()
print("Connected to MySQL server")
except Error as e:
print(f"Error while connecting to MySQL: {e}")
def disconnect(self):
if self.cursor:
self.cursor.close()
if self.connection.is_connected():
self.connection.close()
print("MySQL connection is closed")
def create_database(self, database_name):
try:
self.cursor.execute(f"CREATE DATABASE {database_name}")
self.cursor.execute(f"SHOW DATABASES LIKE '{database_name}'")
result = self.cursor.fetchone()
if result:
print(f"Database {database_name} created successfully")
return True
else:
print(f"Database {database_name} was not created")
return False
except Error as e:
print(f"Error while creating database: {e}")
return False
def drop_database(self, database_name):
try:
self.cursor.execute(f"DROP DATABASE {database_name}")
self.cursor.execute(f"SHOW DATABASES LIKE '{database_name}'")
result = self.cursor.fetchone()
if not result:
print(f"Database {database_name} dropped successfully")
return True
else:
print(f"Database {database_name} was not dropped")
return False
except Error as e:
print(f"Error while dropping database: {e}")
return False
def use_database(self,database_name):
try:
self.cursor.execute(f"USE {database_name}")
print(f"Using database {database_name}")
except Error as e:
print(f"Error while selecting database: {e}")
def execute_script_from_file(self, file_path):
try:
with open(file_path, 'r') as file:
sql_script = file.read()
sql_commands = sql_script.split(';')
for command in sql_commands:
if command.strip():
self.cursor.execute(command)
print(f"Executed: {command}")
self.connection.commit()
print("SQL script executed successfully")
except Error as e:
print(f"Error while executing SQL script: {e}")
# Usage example
if __name__ == "__main__":
load_dotenv() # Load environment variables from .env file
db_name = 'testing_db'
db = MySQLConnector()
db.connect()
db.create_database(db_name) # Replace 'new_database' with the desired database name
db.use_database(db_name) # Use the specified database from .env
db.drop_database(db_name)
db.disconnect()