はじめに
業務の中で、ChaliceとMySQLをローカル環境で検証する必要がありました。
今回はSQLAlchemy+PyMySQLを利用したデモ用コードを用いて、その際の手順をご共有出来ればと思います。
目次
検証環境
- macOS：M2
- Docker：27.4.0
- Docker Compose：v2.31.0-desktop.2
- Chalice：1.31.3
- Python：3.12.8
今回のファイル構成
ファイル構成は以下のとおりです。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
. ├── docker-compose.yaml ├── local-test │ ├── .chalice │ │ └── config.json │ ├── app.py │ ├── database.py │ ├── models │ │ └── user.py │ ├── repositories │ │ └── user_repository.py │ ├── requirements.txt │ ├── services │ │ └── user_service.py │ └── util.py └── mysql ├── DB │ └── init.sql └── my.cnf |
Chaliceとは？
AWSが提供しているサーバーレスアプリケーションを作成するためのフレームワークです。
今回はChaliceを用いて、ローカル環境でAPIを構築します。
ローカル環境にMySQLコンテナを立てる
Docker Composeを使用して、MySQLコンテナを立てます。
MySQL初期データを配置
データベースの作成とテーブルの作成を行います。
Path：./mysql/DB/init.sql
|
1 2 3 4 5 6 7 8 9 10 11 12 |
-- データベース作成 CREATE DATABASE IF NOT EXISTS local_test; USE local_test; -- users テーブルの作成 CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, email VARCHAR(100) NOT NULL UNIQUE, password_hash VARCHAR(255) NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); |
MySQL設定ファイルを配置
Path：./mysql/my.cnf
|
1 2 3 4 5 6 7 8 9 10 |
[mysqld] character-set-server=utf8mb4 collation_server = utf8mb4_general_ci transaction-isolation = READ-COMMITTED [mysql] default-character-set=utf8mb4 [client] default-character-set=utf8mb4 |
MySQLコンテナを作成
docker-compose.yamlを作成
以下の設定で、コンテナを作成します。
Path：./docker-compose.yaml
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
volumes: mysql_data: services: mysql: image: mysql:8.0.41 container_name: mysql_container environment: MYSQL_ROOT_PASSWORD: root MYSQL_DATABASE: local_test MYSQL_USER: local_test_user MYSQL_PASSWORD: local_test_password ports: - "3306:3306" volumes: - mysql_data:/var/lib/mysql - ./mysql/my.cnf:/etc/mysql/conf.d/my.cnf - ./mysql/DB:/docker-entrypoint-initdb.d |
MySQLコンテナの立ち上げ
|
1 |
docker-compose up -d |
Chalice側の実装
Chaliceとその他パッケージのインストール
requirements.txtを作成
今回作成するrequirements.txtは以下になります。
Path：./local-test/requirements.txt
|
1 2 3 4 5 |
chalice==1.31.3 sqlalchemy==2.0.37 pymysql==1.1.1 cryptography==44.0.0 werkzeug==3.1.3 |
パッケージのインストール
requirements.txtからパッケージをインストールします。
|
1 |
pip install -r requirements.txt |
Chaliceプロジェクトを作成
新規のChaliceプロジェクトを作成します。
今回はlocal-testという名前でプロジェクトを作成します。
|
1 |
chalice new-project local-test |
tips
Chaliceのコード中にboto3を使用する記述があると、Chaliceをローカル環境で使用する際にも、
chalice local実行時に~/.aws/credentialsを確認して、AWSの認証情報を取得するようです。
今回はローカル環境だけでChaliceを使用するので、認証情報の設定は不要になります。
config.jsonを設定
以下のようにconfig.jsonを書き換えます。
環境変数として、データベースの情報を記述しています。
Path：./local-test/.chalice/config.json
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
{ "version": "2.0", "app_name": "local-test", "stages": { "local": { "environment_variables": { "DB_HOST": "localhost", "DB_DATABASE": "local_test", "DB_USERNAME": "local_test_user", "DB_PASSWORD": "local_test_password" } } } } |
検証用のコードを作成
今回の検証では、ユーザー登録などを想定したデモ用コードを以下のように作成しました。
データベースの接続
Path：./local-test/database.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import scoped_session, sessionmaker # データベースの情報をchaliceのconfig.jsonから取得 DB_USERNAME = os.environ["DB_USERNAME"] DB_PASSWORD = os.environ["DB_PASSWORD"] DB_HOST = os.environ["DB_HOST"] DB_DATABASE = os.environ["DB_DATABASE"] # 接続先DBの設定 DATABASE = f"mysql+pymysql://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOST}/{DB_DATABASE}?charset=utf8mb4" Engine = create_engine(DATABASE, pool_pre_ping=True, echo=True, pool_recycle=28500) Base = declarative_base() # Sessionの作成 session = scoped_session( sessionmaker( autoflush=False, bind=Engine, expire_on_commit=False ) ) |
モデル定義
Userモデルを定義しています。
Path：./local-test/models/user.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
from sqlalchemy import Column, Integer, String, DateTime from database import Base from datetime import datetime from werkzeug.security import generate_password_hash, check_password_hash class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(50), nullable=False, unique=True) email = Column(String(100), nullable=False, unique=True) password_hash = Column(String(255), nullable=False) created_at = Column(DateTime, default=datetime.utcnow) def set_password(self, password): #パスワードをハッシュ化 self.password_hash = generate_password_hash(password) def check_password(self, password): #入力されたパスワードがハッシュと一致するか確認 return check_password_hash(self.password_hash, password) |
データベース情報の取得・更新処理
データベースとのやりとりについて、記述しています。
Path：./local-test/repositories/user_repository.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
from database import session from models.user import User def get_user_by_email(email: str): #メールアドレスでユーザーを検索 user_by_email = session.query( User ).filter( User.email == email ).first() return user_by_email def create_user(username: str, email: str, password: str): #新しいユーザーを作成（commit はしない) new_user = User(username=username, email=email) new_user.set_password(password) session.add(new_user) session.flush() session.refresh(new_user) return new_user def get_user_by_id(user_id: int): #ID でユーザーを検索 user_by_id = session.query( User ).filter( User.id == user_id ).first() return user_by_id |
ロジック
ビジネスロジックを記述しています。
Path：./local-test/services/user_service.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
from repositories.user_repository import get_user_by_email, create_user, get_user_by_id from util import transactional, session_managed from chalice import Response import json @transactional def register_user(username: str, email: str, password: str): if not username or not email or not password: return Response( body=json.dumps({"error": "すべてのフィールドを入力してください"}, ensure_ascii=False), status_code=400, headers={'Content-Type': 'application/json; charset=utf-8'} ) if get_user_by_email(email): return Response( body=json.dumps({"error": "このメールアドレスは既に登録されています"}, ensure_ascii=False), status_code=400, headers={'Content-Type': 'application/json; charset=utf-8'} ) new_user = create_user(username, email, password) return Response( body=json.dumps({"message": "ユーザー登録が完了しました", "user_id": new_user.id}, ensure_ascii=False), status_code=201, headers={'Content-Type': 'application/json; charset=utf-8'} ) @session_managed def login_user(email: str, password: str): if not email or not password: return Response( body=json.dumps({"error": "メールアドレスとパスワードを入力してください"}, ensure_ascii=False), status_code=400, headers={'Content-Type': 'application/json; charset=utf-8'} ) user = get_user_by_email(email) if user is None or not user.check_password(password): return Response( body=json.dumps({"error": "メールアドレスまたはパスワードが正しくありません"}, ensure_ascii=False), status_code=401, headers={'Content-Type': 'application/json; charset=utf-8'} ) return Response( body=json.dumps({"message": "ログイン成功", "user_id": user.id}, ensure_ascii=False), status_code=200, headers={'Content-Type': 'application/json; charset=utf-8'} ) @session_managed def get_user_details(user_id: int): user = get_user_by_id(user_id) if not user: return Response( body=json.dumps({"error": "ユーザーが見つかりません"}, ensure_ascii=False), status_code=404, headers={'Content-Type': 'application/json; charset=utf-8'} ) return Response( body=json.dumps({ "id": user.id, "username": user.username, "email": user.email, "created_at": user.created_at.isoformat() }, ensure_ascii=False), status_code=200, headers={'Content-Type': 'application/json; charset=utf-8'} ) |
セッション管理
データベースのセッションを管理するための、デコレーターを記述しています。
ロジックの部分で使用します。
Path：./local-test/util.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
from database import session from functools import wraps def transactional(func): """トランザクションと session を自動的に管理するデコレーター""" @wraps(func) def wrapper(*args, **kwargs): try: result = func(*args, **kwargs) # 関数の実行 session.commit() # 正常終了時にトランザクションをコミット return result except Exception as e: session.rollback() # 例外発生時にロールバック raise e finally: session.remove() # セッションをクリーンアップ return wrapper def session_managed(func): """読み取り専用操作のための session 管理デコレーター""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) finally: session.remove() # 読み取り操作でもセッションを解放 return wrapper |
ルーティング処理
どのようなリクエストに対して、どのようなレスポンスを返すかを記述しています。
Chaliceで実行されるメインとなるファイルです。
Path：./local-test/app.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
from chalice import Chalice from services.user_service import register_user, login_user, get_user_details app = Chalice(app_name="local-test") @app.route('/register', methods=['POST']) def register(): #リクエスト情報取得 request = app.current_request.json_body username = request.get('username') email = request.get('email') password = request.get('password') #レスポンス取得 response_object = register_user(username, email, password) return response_object @app.route('/login', methods=['POST']) def login(): #リクエスト情報取得 request = app.current_request.json_body email = request.get('email') password = request.get('password') #レスポンス取得 response_object = login_user(email, password) return response_object @app.route('/user/{user_id}', methods=['GET']) def get_user(user_id): #ユーザーID取得 user_id = int(user_id) #レスポンス取得 response_object = get_user_details(user_id) return response_object |
動作検証
ローカル環境でChaliceを起動
Chaliceを起動します。ステージを指定せずにchalice localだけで実行すると、
config.jsonで指定した環境変数が読み込まれないので注意が必要です。
デフォルトで、http://127.0.0.1:8000でホストされます。
|
1 |
chalice local --stage local |
curlコマンドでリクエストを送ってみる
ユーザーの登録
リクエスト
|
1 2 3 4 5 6 7 |
curl -X POST http://localhost:8000/register \ -H "Content-Type: application/json" \ -d '{ "username": "johndoe", "email": "johndoe@example.com", "password": "securepassword" }' |
レスポンス
|
1 |
{"message": "ユーザー登録が完了しました", "user_id": 1} |
ログイン
リクエスト
|
1 2 3 4 5 6 |
curl -X POST http://localhost:8000/login \ -H "Content-Type: application/json" \ -d '{ "email": "johndoe@example.com", "password": "securepassword" }' |
レスポンス
|
1 |
{"message": "ログイン成功", "user_id": 1} |
ユーザー情報取得
リクエスト
|
1 |
curl -X GET http://localhost:8000/user/1 |
レスポンス
|
1 |
{"id": 1, "username": "johndoe", "email": "johndoe@example.com", "created_at": "2025-02-02T04:07:30"} |
まとめ
以上、ローカル環境でChaliceとMySQLを連携するデモをご紹介しました。
カジュアルにローカル環境でAPIを構築して、試行錯誤できる点はとても便利だと思います。
みなさまの開発アイデアを形にする際に、本記事が少しでもお役に立てれば幸いです。
投稿者プロフィール
-
2024年11月にスカイアーチネットワークスに中途入社しました。
AWSとCDKを勉強中です。
好きなもの：アニメ、読書、テニス
最新の投稿
AWS2025年2月10日ChaliceとMySQLをローカル環境で動かしてみた
