【備忘録】PythonでMySQLのデータベースにデータを作成する方法

備考

Python:3.9
MySQL:8.0.31
動作環境:MacOS
開発環境:PyCharm

事前準備

データベース作成済み

マイグレーション実行済み

1. PyCharmにて新規プロジェクトを作成する

2. プロジェクトを以下の構成になるようにファイルを作成する

フォルダ構成.

.
├── main.py
├── env.py
├── .env
├── requirements.txt
├── crud
│  ├── certification_crud.py
│  └── user_crud.py
├── database
│  ├── database.py
│  └── models.py
└── schema
   └── user_schema.py

3. requirements.txtファイルを以下の内容にする

requirements.txt

sqlalchemy
pydantic
passlib[bcrypt]
pytz
python-dotenv
mysqlclient

4. ターミナルにてコマンドを入力してモジュールをインストールする

ターミナル.

pip install -r requirements.txt

5. .envファイルを以下の内容にする

.env

# DB接続情報
DATABASE = 'mysql'
DB_USER = 'root'
DB_PASSWORD = 'MySQLのパスワード'
DB_HOST = 'localhost'
DB_PORT = '3306'
DB_NAME = '対象のDB名'

6. env.pyファイルを以下の内容にする

env.py

import os
from dotenv import load_dotenv
from os.path import join, dirname

dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)


class Env:
    # DB接続情報
    DATABASE = os.environ.get("DATABASE")
    DB_USER = os.environ.get("DB_USER")
    DB_PASSWORD = os.environ.get("DB_PASSWORD")
    DB_HOST = os.environ.get("DB_HOST")
    DB_PORT = os.environ.get("DB_PORT")
    DB_NAME = os.environ.get("DB_NAME")

7. database.pyファイルを以下の内容にする

database.py

from env import Env
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base

# データベースに接続するための情報
database_url = '{}://{}:{}@{}:{}/{}?charset=utf8'.format(Env.DATABASE, Env.DB_USER, Env.DB_PASSWORD, Env.DB_HOST, Env.DB_PORT, Env.DB_NAME)
engine = create_engine(database_url, connect_args={"connect_timeout": 15}, echo=False, pool_recycle=10, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# データベースに接続するための処理
def db_session():
    return scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

8. models.pyファイルを以下の内容にする

models.py

from database.database import Base
from datetime import datetime
from sqlalchemy import Column, VARCHAR, INT, TEXT, DATETIME
import pytz

jst = pytz.timezone('Asia/Tokyo')


# 仮にユーザーを用意
class User(Base):
    __tablename__ = 'users'
    __table_args__ = {"mysql_collate": "utf8_general_ci"}
    id = Column(INT, primary_key=True, autoincrement=True, unique=True, nullable=False, index=True)
    name = Column(VARCHAR(255), primary_key=False, autoincrement=False, unique=False, nullable=False, index=False)
    email = Column(VARCHAR(255), primary_key=False, autoincrement=False,  unique=True, nullable=False, index=True)
    password = Column(TEXT, primary_key=False, autoincrement=False, unique=False, nullable=False, index=False)
    create_time = Column(DATETIME, primary_key=False, autoincrement=False, unique=False, nullable=False, index=False, default=lambda: datetime.now(jst))

9. certification_crud.pyファイルを以下の内容にする

certification_crud.py

from passlib.context import CryptContext


# ハッシュ化などの処理を使用するために定義する
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


# ユーザーが入力したパスワードをハッシュ化したバスワードにする
def create_password_hash(user_password: str):
    return pwd_context.hash(user_password)

10. user_crud.pyファイルを以下の内容にする

user_crud.py

from database import models
from schema import user_schema
from . import certification_crud
from sqlalchemy.orm import Session


# ユーザーを作成する処理
def create_user(db: Session, user: user_schema.CreateUser):
    db_user = models.User(**user.dict())

    # パスワードをハッシュ化する
    db_user.password = certification_crud.create_password_hash(db_user.password)

    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    # 作成したユーザー情報を返す
    return db_user

11. user_schema.pyファイルを以下の内容にする

user_schema.py

from pydantic import BaseModel


# ユーザーを作成
class CreateUser(BaseModel):
    name: str
    email: str
    password: str

12. main.pyファイルを以下の内容にする

main.py

from database.database import SessionLocal
from crud import user_crud
from schema import user_schema

if __name__ == '__main__':

    print("データ作成を開始します")

    db = SessionLocal()

    # ユーザー作成
    user_data = [
        {
            "name": "TestUser1",
            "email": "[e-mail]",
            "password": "TestPassword1",
        },
        {
            "name": "TestUser2",
            "email": "[e-mail]",
            "password": "TestPassword2",
        },
        {
            "name": "TestUser3",
            "email": "[e-mail]",
            "password": "TestPassword3",
        },
    ]

    users = []

    for user in user_data:
        db_user = user_crud.create_user(
            db,
            user_schema.CreateUser(
                name=user["name"],
                email=user["email"],
                password=user["password"],
            )
        )
        users.append(db_user)

    print("データ作成が完了しました")

13. ターミナルにてコマンドを入力してをデータ作成を実施する

ターミナル.

Python main.py