目次
仮想環境作成
venv パッケージがあるか確認（標準搭載のはず）
sudo apt update sudo apt install python3-venv -y
プロジェクトディレクトリ作成
mkdir myproject cd myproject
仮想環境作成（プロジェクトディレクトリにて作成）
python3 -m venv myenv
仮想環境の有効化
source myenv/bin/activate
仮想環境の無効化（終了）
作業が終わったら以下で無効化します。
deactivate
ライブラリインストール
PIP インストール
sudo apt update sudo apt install python3-pip
MySQL コネクター
sudo apt update pip install mysql-connector-python
Excel 読み取り pands openpyxl
sudo apt update pip install pandas openpyxl
Vs Codeで、仮想環境を認識する
VS Code で myproject フォルダを開く
Ctrl+Shift+P → 「Python: Interpreter を選択」→ ./myenv/bin/python を選択
python MySQL ソースコード
データベースの条件（必要に応じてソースコードを変更）
データベース名:eng001
ユーザー；root
パスワード:root
ホスト名:
ソースコード
# pip install mysql-connector-python
import mysql.connector
from mysql.connector import Error
import pandas as pd
import openpyxl
#
# DSN（Data Source Name）を使用して接続設定を取得
# dsn(host='locakhost', user='root', password="root", database='mysql', port=3306)
def dsn(host, user, password, database, port=3306):
config = {
'host': host,
'user': user,
'password': password,
'database': database,
'port': port
}
return config
def dsn_admin():
config = {
'host': 'localhost', # MySQL Server のホスト名または IP アドレス
'user': 'root', # MySQL のユーザー名
'password': 'root', # MySQL のパスワード
'database': 'eng001', # 接続するデータベース名
'port': 3306 # MySQL のポート番号（デフォルトは 3306）
}
return config
def dsn_user():
config = {
'host': 'localhost',
'user': 'db_user',
'password': 'db_user',
'database': 'eng001',
'port': 3306
}
return config
def test_connection():
try:
dsn_config = dsn_user()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
print("✅ MySQL に接続できました！")
server_info = connection.get_server_info()
print("MySQL Server バージョン:", server_info)
except Error as e:
print("❌ 接続エラー:", e)
def create_database():
try:
dsn_config = dsn_admin()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
strsql = """
CREATE DATABASE eng001p
DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci;
"""
cursor.execute(strsql)
connection.commit()
print("✅ データベースが作成されました。")
except Error as e:
print("❌ エラー:", e)
def drop_database():
try:
dsn_config = dsn_admin()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
strsql = "DROP DATABASE IF EXISTS eng001p;"
cursor.execute(strsql)
connection.commit()
print("✅ データベースが削除されました。")
except Error as e:
print("❌ エラー:", e)
def create_user_grant():
try:
dsn_config = dsn_admin()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
# ユーザーの作成
cursor.execute("CREATE USER 'db_user'@'%' IDENTIFIED BY 'db_user';")
# 権限の付与
cursor.execute("GRANT ALL PRIVILEGES ON eng001.* TO 'db_user'@'%';")
connection.commit()
print("✅ ユーザーと権限が作成されました。")
except Error as e:
print("❌ エラー:", e)
def create_table_user_master():
try:
dsn_config = dsn_user()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
strSql = """
CREATE TABLE IF NOT EXISTS user_master (
id CHAR(16) NOT NULL PRIMARY KEY,
name TEXT,
pass TEXT,
mail TEXT,
rights INT
);
"""
cursor.execute(strSql)
connection.commit()
print("✅ テーブルが作成されました。")
except Error as e:
print("❌ エラー:", e)
def drop_table_user_master():
try:
dsn_config = dsn_user()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
strSql = "DROP TABLE user_master;"
cursor.execute(strSql)
connection.commit()
print("✅ テーブルが削除されました。")
except Error as e:
print("❌ エラー:", e)
def add_user_master():
try:
dsn_config = dsn_user()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
data = [
("user000", "三田 さやか", "pas000", "user00@gmail.com", 0),
("user001", "須田 翼", "pas001", "mail01@gmail.com", 1),
("user002", "前田 直行", "pas002", "mail02@gmail.com", 1),
("user003", "須田 翼", "pas003", "mail03@gmail.com", 1),
("user004", "林 香奈子", "pas004", "mail04@gmail.com", 1),
("user005", "中川 修子", "pas005", "mail05@gmail.com", 1),
]
strSql = """
INSERT INTO user_master (id, name, pass, mail, rights)
VALUES (%s, %s, %s, %s, %s)
"""
cursor.executemany(strSql, data)
connection.commit()
print("✅ データが追加されました。")
except Error as e:
print("❌ エラー:", e)
def select_user_master():
try:
dsn_config = dsn_user()
with mysql.connector.connect(**dsn_config) as connection:
if connection.is_connected():
with connection.cursor() as cursor:
strSql = "SELECT * FROM user_master;"
cursor.execute(strSql)
result = cursor.fetchall()
for row in result:
print(row)
except Error as e:
print("❌ エラー:", e)
def insert_users_from_excel(file_path):
# Excelからデータを読み込み
df = pd.read_excel(file_path, engine='openpyxl')
# 欠損値チェック
if df.isnull().values.any():
print("⚠️ Excelに欠損データがあります。")
return
# 登録用のタプルリストに変換
records = list(df.itertuples(index=False, name=None))
dsn_config = dsn_user()
try:
with mysql.connector.connect(**dsn_config) as connection:
connection.autocommit = False
if connection.is_connected():
with connection.cursor() as cursor:
sql = """
INSERT INTO user_master (id, name, pass, mail, rights)
VALUES (%s, %s, %s, %s, %s)
"""
cursor.executemany(sql, records)
connection.commit()
print("✅ Excelからのデータ登録完了")
except mysql.connector.Error as e:
print("❌ エラー:", e)
# ロールバックは接続確認後に限定
try:
if 'connection' in locals() and connection.is_connected():
connection.rollback()
print("🔁 ロールバックしました。")
except:
pass
def test_transaction():
try:
# DSN（Data Source Name）を使用して接続設定を取得
dsn_config = dsn_user()
connection = mysql.connector.connect(**dsn_config)
if connection.is_connected():
print("✅ MySQL に接続できました！")
server_info = connection.get_server_info()
res = connection.cmd_query("SHOW TABLE STATUS WHERE Name = 'user_master';")
print("テーブル情報:", res)
print("MySQL Server バージョン:", server_info)
except Error as e:
print("❌ 接続エラー:", e)
finally:
if 'connection' in locals() and connection.is_connected():
connection.close()
print("🔒 接続を閉じました。")
# エントリーポイント
if __name__ == "__main__":
test_connection()
# create_database()
# create_user_grant()
# drop_database()
# drop_table_user_master()
create_table_user_master()
add_user_master()
select_user_master()
# 実行（ファイルパスを指定）
# insert_users_from_excel("book1.xlsx")
