Go to Qiita Advent Calendar 2024 Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@kazzito

SQLAlchemyとfactory_boy

Last updated at Posted at 2024-09-04

SQLAlchemyで各テーブルをモデルクラス化し、factoryboyを使う想定でサンプルになるコードを示してください

  • 単純なテーブルを(id,text,int,datetime)モデルクラスとして作成
  • 上記のモデルクラスを使って登録、更新、取得処理
  • 上記の処理のunit testをfactoryboyで作成し、モデルクラスのfactoryを作って対応

以下は、AWS Lambda環境でPythonを使用してSQLAlchemyを導入し、単純なテーブルをモデルクラス化し、factory_boyを用いてユニットテストを行うサンプルコードです。以下の手順に従って実装を進めてください。

目次

  1. 環境のセットアップ
  2. モデルクラスの定義
  3. CRUD操作の実装
  4. Factoryの定義
  5. ユニットテストの作成
  6. AWS Lambdaへのデプロイ
  7. 補足: Lambda Layerの作成

1. 環境のセットアップ

まず、必要なPythonパッケージをインストールします。SQLAlchemy、factory_boy、およびpytestを使用します。

pip install sqlalchemy factory_boy pytest psycopg2-binary

注: PostgreSQLを使用する場合はpsycopg2-binaryを、他のデータベースを使用する場合は適切なDBドライバーをインストールしてください。

2. モデルクラスの定義

SQLAlchemyを使用して、単純なテーブルをモデルクラスとして定義します。以下では、id（主キー）、text（文字列）、int_field（整数）、created_at（日時）を持つテーブルを例にします。

# models.py

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class MyModel(Base):
    __tablename__ = 'my_table'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    text = Column(String, nullable=False)
    int_field = Column(Integer, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    def __repr__(self):
        return f"<MyModel(id={self.id}, text='{self.text}', int_field={self.int_field}, created_at={self.created_at})>"

説明

  • Base: SQLAlchemyのデクラレーティブベース。すべてのモデルクラスはこれを継承します。
  • MyModel: my_tableという名前のテーブルを表すモデルクラス。
  • 各カラムは適切なデータ型と制約を持ちます。

3. CRUD操作の実装

モデルを使用して、データベースに対する基本的なCRUD（Create, Read, Update, Delete）操作を実装します。ここでは、create, update, getの操作を示します。

# operations.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, MyModel
from datetime import datetime

# データベースURLを環境変数などから取得するのが望ましいですが、ここでは例としてSQLiteを使用
DATABASE_URL = "sqlite:///./test.db"  # PostgreSQLの場合: "postgresql://user:password@host:port/dbname"

# エンジンの作成
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {})

# セッションメーカーの作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# テーブルの作成（初回のみ必要）
Base.metadata.create_all(bind=engine)

def create_record(session, text, int_field):
    """新しいレコードを作成し、データベースに追加します。"""
    new_record = MyModel(text=text, int_field=int_field)
    session.add(new_record)
    session.commit()
    session.refresh(new_record)
    return new_record

def update_record(session, record_id, new_text=None, new_int_field=None):
    """既存のレコードを更新します。"""
    record = session.query(MyModel).filter(MyModel.id == record_id).first()
    if not record:
        return None
    if new_text is not None:
        record.text = new_text
    if new_int_field is not None:
        record.int_field = new_int_field
    session.commit()
    session.refresh(record)
    return record

def get_record(session, record_id):
    """指定したIDのレコードを取得します。"""
    return session.query(MyModel).filter(MyModel.id == record_id).first()

説明

  • create_record: 新しいレコードを作成してデータベースに保存します。
  • update_record: 指定したIDのレコードを更新します。
  • get_record: 指定したIDのレコードを取得します。

4. Factoryの定義

factory_boyを使用して、テストデータを生成するためのファクトリークラスを定義します。

# factories.py

import factory
from factory import alchemy
from models import MyModel
from datetime import datetime

class MyModelFactory(alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = MyModel
        sqlalchemy_session = None  # テスト時にセッションを指定
        sqlalchemy_session_persistence = 'commit'
    
    id = factory.Sequence(lambda n: n + 1)
    text = factory.Faker('sentence')
    int_field = factory.Faker('random_int', min=1, max=100)
    created_at = factory.LazyFunction(datetime.utcnow)

説明

  • MyModelFactory: MyModelのインスタンスを生成するファクトリークラス。
  • factory.Sequence: 連番のIDを生成します。
  • factory.Faker: ランダムなデータを生成します。
  • factory.LazyFunction: ランダムな日時を生成します。

5. ユニットテストの作成

pytestとfactory_boyを使用して、CRUD操作のユニットテストを作成します。

# test_operations.py

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, MyModel
from operations import create_record, update_record, get_record
from factories import MyModelFactory

# テスト用のデータベースURL（インメモリSQLiteを使用）
TEST_DATABASE_URL = "sqlite:///:memory:"

@pytest.fixture(scope='module')
def engine():
    """テスト用のデータベースエンジンを作成します。"""
    engine = create_engine(TEST_DATABASE_URL)
    Base.metadata.create_all(bind=engine)
    return engine

@pytest.fixture(scope='function')
def session(engine):
    """各テスト関数に対して新しいセッションを提供します。トランザクションはテスト後にロールバックされます。"""
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()
    
    # ファクトリーのセッションを設定
    MyModelFactory._meta.sqlalchemy_session = session
    
    yield session
    
    session.close()
    transaction.rollback()
    connection.close()

def test_create_record(session):
    """レコードの作成をテストします。"""
    record = create_record(session, text="Test text", int_field=42)
    assert record.id is not None
    assert record.text == "Test text"
    assert record.int_field == 42
    assert record.created_at is not None

def test_update_record(session):
    """レコードの更新をテストします。"""
    # ファクトリーでレコードを作成
    record = MyModelFactory.create()
    updated_record = update_record(session, record.id, new_text="Updated text", new_int_field=100)
    assert updated_record.text == "Updated text"
    assert updated_record.int_field == 100

def test_get_record(session):
    """レコードの取得をテストします。"""
    # ファクトリーでレコードを作成
    record = MyModelFactory.create()
    fetched_record = get_record(session, record.id)
    assert fetched_record.id == record.id
    assert fetched_record.text == record.text
    assert fetched_record.int_field == record.int_field
    assert fetched_record.created_at == record.created_at

説明

  • engineフィクスチャ: テスト用のインメモリSQLiteデータベースエンジンを作成します。
  • sessionフィクスチャ: 各テスト関数に対して新しいセッションを提供し、テスト後にトランザクションをロールバックします。
  • test_create_record: レコードの作成機能をテストします。
  • test_update_record: レコードの更新機能をテストします。
  • test_get_record: レコードの取得機能をテストします。

6. AWS Lambdaへのデプロイ

ここまでのコードをAWS Lambdaで使用するには、以下の手順に従います。

  1. コードのパッケージング:

    • models.py, operations.py, および依存パッケージ（SQLAlchemy, psycopg2-binaryなど）を含むパッケージを作成します。
    • 依存パッケージはLambda Layerとして別途パッケージングすることも可能です。

  2. Lambda関数の作成:

    • AWSコンソールまたはAWS CLIを使用して新しいLambda関数を作成します。
    • 作成したコードとLayerをアップロードします。

  3. 環境変数の設定:

    • データベースURLや認証情報などの機密情報は、Lambda関数の環境変数として設定します。

  4. VPC設定:

    • データベースがVPC内にある場合、Lambda関数を適切なVPCに接続し、セキュリティグループを設定します。

  5. ハンドラーの実装:

    • Lambda関数のエントリーポイントを実装し、operations.pyの関数を呼び出します。

例: Lambdaハンドラーの実装

# lambda_handler.py

import json
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from operations import create_record, update_record, get_record
from models import Base

# 環境変数からデータベースURLを取得
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///./test.db')

# エンジンとセッションの作成
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# テーブルの作成（デプロイ時に一度だけ実行することが望ましい）
Base.metadata.create_all(bind=engine)

def lambda_handler(event, context):
    """
    AWS Lambdaのエントリーポイント。
    eventの内容に応じてCRUD操作を実行します。
    """
    session = SessionLocal()
    
    try:
        operation = event.get('operation')
        data = event.get('data', {})
        
        if operation == 'create':
            record = create_record(session, text=data.get('text'), int_field=data.get('int_field'))
            response = {'id': record.id, 'text': record.text, 'int_field': record.int_field, 'created_at': record.created_at.isoformat()}
        
        elif operation == 'update':
            record = update_record(session, record_id=data.get('id'), new_text=data.get('text'), new_int_field=data.get('int_field'))
            if record:
                response = {'id': record.id, 'text': record.text, 'int_field': record.int_field, 'created_at': record.created_at.isoformat()}
            else:
                response = {'error': 'Record not found'}
        
        elif operation == 'get':
            record = get_record(session, record_id=data.get('id'))
            if record:
                response = {'id': record.id, 'text': record.text, 'int_field': record.int_field, 'created_at': record.created_at.isoformat()}
            else:
                response = {'error': 'Record not found'}
        
        else:
            response = {'error': 'Invalid operation'}
        
        return {
            'statusCode': 200,
            'body': json.dumps(response)
        }
    
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
    
    finally:
        session.close()

説明

  • lambda_handler: Lambda関数のエントリーポイント。eventに基づいて適切なCRUD操作を実行します。
  • 環境変数からデータベースURLを取得し、接続を設定します。

補足: Lambda Layerの作成

AWS Lambdaでは、依存ライブラリをLayerとしてパッケージングすることが推奨されます。以下にLayerの作成手順を示します。

  1. ディレクトリの作成:

    mkdir python

  2. 依存パッケージのインストール:

    pip install --target=python sqlalchemy psycopg2-binary factory_boy

  3. ZIPファイルの作成:

    zip -r layer.zip python

  4. Layerのアップロード:

    • AWSコンソールのLambdaセクションに移動し、「レイヤー」から新しいレイヤーを作成します。
    • 作成したlayer.zipをアップロードし、適切なランタイム（例: Python 3.9）を選択します。

  5. Lambda関数にLayerを追加:

    • 作成したLambda関数の設定画面で、先ほど作成したLayerを追加します。

まとめ

以上で、AWS Lambda環境でPythonを使用してSQLAlchemyを導入し、単純なテーブルをモデルクラス化し、factory_boyを用いたユニットテストを作成する手順を示しました。以下に、プロジェクトのディレクトリ構造の例を示します。

project/
│
├── models.py
├── operations.py
├── factories.py
├── test_operations.py
├── lambda_handler.py
├── requirements.txt
└── layer.zip  # Lambda Layer

コマンド例

  • テストの実行:

    pytest

  • Layerの作成:

    mkdir python
pip install --target=python sqlalchemy psycopg2-binary factory_boy
zip -r layer.zip python

注意点

  • セキュリティ: データベースの認証情報は環境変数やAWS Secrets Managerなどを使用して安全に管理してください。
  • パフォーマンス: Lambda関数内でのデータベース接続はコールドスタート時に遅延を引き起こす可能性があるため、コネクションプールの使用や接続の再利用を検討してください。
  • 依存関係の管理: 依存ライブラリのバージョン管理を適切に行い、互換性を維持してください。

これで、AWS LambdaでPythonを使用してSQLAlchemyを導入し、モデルクラスの作成、CRUD操作の実装、およびfactory_boyを用いたユニットテストの作成方法についての基本的なガイドを提供しました。質問やさらなる詳細が必要な場合は、お気軽にお知らせください。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?