LoginSignup
2
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@kewton

Pythonプロジェクトの時間節約術: PydanticとPysonDBでデータ管理を効率化

Posted at

はじめに

Python開発で時間を節約し、データの整合性を確保する方法をお探しですか？PydanticとPysonDBを組み合わせることで、その両方を実現できます。この記事では、型安全性を提供するPydanticと、JSONベースのデータベース操作を簡単にするPysonDBの使用方法を解説し、これらを利用してデータ管理プロセスを効率化する手法を紹介します。

Pydanticによる型安全性とデータバリデーション

Pythonの開発プロセスにおいて、データの型安全性とバリデーションは、バグを減らし、アプリケーションの信頼性を高めるために不可欠です。Pydanticは、Pythonの強力な型ヒントシステムを活用して、データバリデーションと設定管理を簡単にし、開発者がより安全で読みやすいコードを書くことを支援します。このセクションでは、Pydanticを使用して型安全性を強化し、データモデルを通じてアプリケーションデータの整合性を保つ方法について説明します。

型ヒントとPydanticの基本

Pydanticの中心的な機能は、Pythonの型ヒントを使用して、モデル定義時にデータの形状と型を明示することです。これにより、開発者は入力データに対して自動的な型チェックとバリデーションを実行できます。例えば、特定のフィールドが文字列であるべきか、または特定の範囲内の整数であるべきかを定義することができます。Pydanticはこの情報を使用して、実行時にデータが指定された型と一致するかどうかを検証します。

データバリデーションの実装

Pydanticを使用すると、複雑なデータバリデーションロジックを簡潔に実装できます。例えば、電子メールアドレスが有効な形式であるか、URLが実際に存在するかなど、さまざまな条件をチェックするカスタムバリデータを簡単に追加できます。Pydanticはこれらのバリデーションを自動的に処理し、不正なデータに対しては明確なエラーメッセージを提供します。

実践的な利用例

次に、Pydanticを使用してユーザー入力のバリデーションを行うシンプルな例を示します。この例では、ユーザーがフォームを介して送信したデータを検証し、特定の基準を満たす場合にのみ処理を続行します。

from pydantic import BaseModel, EmailStr, ValidationError

class User(BaseModel):
    name: str
    email: EmailStr
    age: int

try:
    user = User(name="John Doe", email="johndoe@example.com", age=30)
    print("User is valid:", user)
except ValidationError as e:
    print(e.json())

PysonDBによるシンプルなデータベース操作

ythonでのデータ管理と言えば、多くの場合、複雑なSQLデータベースや外部のデータベースサービスが頭に浮かびます。しかし、すべてのプロジェクトがそのような複雑なシステムを必要とするわけではありません。特に、プロトタイピングや小規模アプリケーションでは、設定や管理の手間を最小限に抑えたいものです。ここで、PysonDBの出番です。PysonDBは、Python向けのシンプルなJSONベースのデータベースで、迅速な開発とテストに最適です。このセクションでは、PysonDBを使用してデータベース操作を行う基本的な方法について解説します。

PysonDBの概要

PysonDBは、外部サーバーを必要としないファイルベースの軽量データベースです。データはすべてJSON形式で保存され、Pythonの辞書型として簡単に扱うことができます。これにより、SQLクエリを学ぶ必要がなく、Pythonの基本的な知識だけでデータベース操作が可能になります。

基本的なCRUD操作

PysonDBを使用したCRUD（作成、読み取り、更新、削除）操作は非常に直感的です。以下に、基本的な操作の例を示します。

from pysondb import db

# データベースファイルを生成または開く
a_db = db.getDb("myDatabase.json")

# データの追加
a_db.add({"name": "John Doe", "age": 30})

# データの読み取り
users = a_db.getAll()
print(users)

# データの更新
a_db.updateById(1, {"age": 31})

# データの削除
a_db.deleteById(1)

この例では、まずデータベースファイルを生成（または既存のファイルを開く）ことから始めます。その後、新しいデータの追加、保存されているデータの読み取り、特定のレコードの更新、そしてレコードの削除という一連の操作を行います。

PysonDBの利点

PysonDBの主な利点はそのシンプルさにあります。外部の依存関係がなく、軽量でありながら、データの永続化と基本的なクエリ機能を提供します。これにより、小規模なプロジェクトやデータのプロトタイピングが容易になり、開発プロセスをスピードアップすることができます。

PydanticとPysonDBの組み合わせでデータ管理を革新

Pythonでの効率的なデータ管理は、多くのプロジェクトにおいて中心的な課題です。型安全性を提供するPydanticと、シンプルなJSONベースのデータベース操作を実現するPysonDBの組み合わせは、この課題に対する革新的な解決策を提供します。このセクションでは、これら二つのツールを組み合わせることで、どのようにデータ管理プロセスを効率化し、プロジェクトの開発を加速させることができるのかを探ります。

  • データバリデーションとストレージの統合
    Pydanticによる厳密なデータバリデーションと、PysonDBによる効率的なデータストレージとの統合は、開発プロセスにおいてシームレスな体験を提供します。Pydanticを使用して定義されたモデルは、データがデータベースに保存される前にその整合性を保証します。これにより、アプリケーションのデータ層全体で一貫したデータ構造と品質を維持することができます。

  • 開発速度の向上
    PydanticとPysonDBを組み合わせることで、開発者はデータのバリデーションとストレージの実装にかかる時間を大幅に短縮できます。これらのライブラリが提供するシンプルで直感的なAPIにより、コードの量が減り、読みやすく、保守しやすいコードベースを維持することが可能になります。

  • エラーの削減
    型安全性と自動データバリデーションにより、データ不整合や型エラーによるバグの発生リスクが低下します。Pydanticでバリデーションされたデータは、PysonDBによって安全に保存され、アプリケーションの他の部分で使用される際にもその整合性が保証されます。これにより、ランタイムエラーやデータ関連の問題を事前に排除し、アプリケーションの全体的な信頼性を高めることができます。

  • 柔軟なデータ管理戦略
    PydanticとPysonDBを使用することで、開発者はデータ管理戦略を柔軟に調整することができます。Pydanticのモデルはロジックの変更に迅速に対応できるように設計されており、PysonDBは新しいデータ構造に簡単に適応できます。この柔軟性により、開発者はビジネス要件の変化に素早く対応し、プロジェクトの進化に合わせてデータ管理システムを容易に更新できます。

組み合わせたコード例

PydanticとPysonDBの組み合わせて、下記課題の解決を狙ったパッケージを紹介します。
（下記コードの詳細はこちらをご確認ください。）

  • データ整合性の保証
  • 開発の効率化
  • コードの再利用性とメンテナンスの容易さ
  • 型安全性によるエラーの削減
  • データ操作の柔軟性
from pydantic import BaseModel, ValidationError, Field
from pysondb import db
from pysondb.db import JsonDatabase
from datetime import datetime
import os

DEFAULT_DB_PATH = "./myjsondb/"


class DoFactory:
    def to_dict(self):
        """
        * インスタンスのプロパティを辞書に変換します。
        * __dict__を利用して、インスタンスの属性とその値を取得します。
        * この方法は、インスタンスの直接的な公開プロパティのみを扱います。
        * プライベート属性やプロパティメソッドは含まれません。
        """
        return {key: value for key, value in self.__dict__.items() if not key.startswith('_')}

    def to_query_dict(self):
        """
        * インスタンスのプロパティを辞書に変換します。
        * __dict__を利用して、インスタンスの属性とその値を取得します。
        * この方法は、インスタンスの直接的な公開プロパティのみを扱います。
        * プライベート属性やプロパティメソッドは含まれません。
        """
        return {key: value for key, value in self.__dict__.items() if not key.startswith('_') and len(value) > 0}

    @classmethod
    def from_json_dict(cls, json_dict):
        """
        * 辞書からインスタンスを作成します
        """
        instance = cls()
        for key, value in json_dict.items():
            setattr(instance, key, value)
        return instance


def _base_validate_and_upsertbyprimarykey_to_jsondatabase(_MyDataModel: BaseModel, dataset_db: JsonDatabase, datainstance: DoFactory) -> bool:
    try:
        data = datainstance.to_dict()
        validated_data = _MyDataModel(**data)
        _query = {}
        for _key in _MyDataModel.model_json_schema()["required"]:
            _query[_key] = data[_key]
        _data = dataset_db.getByQuery(query=_query)
        if len(_data) == 0:
            dataset_db.add(validated_data.model_dump(mode="json"))
        else:
            _id = _data[0]["id"]
            dataset_db.updateById(_id, validated_data.model_dump(mode="json"))
    except ValidationError as e:
        print("A validation error has occurred.errormessage=", e.json())
        return False
    return True


def _base_validate_and_upsertsbyprimarykey_to_jsondatabase(_MyDataModel: BaseModel, dataset_db: JsonDatabase, dataList: list[DoFactory]) -> bool:
    try:
        _dataList = []
        for data in dataList:
            validated_data = _MyDataModel(**data.to_dict())
            _dataList.append(validated_data.model_dump(mode="json"))

        for data in _dataList:
            _query = {}
            for _key in _MyDataModel.model_json_schema()["required"]:
                _query[_key] = data[_key]
            _data = dataset_db.getByQuery(query=_query)
            if len(_data) == 0:
                # データベースにデータを挿入
                dataset_db.add(data)
            else:
                _id = _data[0]["id"]
                dataset_db.updateById(_id, data)
    except ValidationError as e:
        print("A validation error has occurred.errormessage=", e.json())
        return False
    return True


def _base_validate_and_update_all_to_jsondatabase(_MyDataModel: BaseModel, dataset_db: JsonDatabase, queryinstance: DoFactory, datainstance: DoFactory) -> bool:
    try:
        data = datainstance.to_dict()
        _query = queryinstance.to_query_dict()
        _data = dataset_db.getByQuery(query=_query)
        for _rec in _data:
            _data = _rec
            _id = _rec["id"]
            _data.update(data)
            validated_data = _MyDataModel(**_data)
            dataset_db.updateById(_id, validated_data.model_dump())
    except ValidationError as e:
        print("A validation error has occurred.errormessage=", e.json())
        return False
    return True


def _base_delete_to_jsondatabase(dataset_db: JsonDatabase, queryinstance: DoFactory) -> bool:
    try:
        _query = queryinstance.to_query_dict()
        _data = dataset_db.getByQuery(query=_query)
        for _rec in _data:
            _id = _rec["id"]
            dataset_db.deleteById(_id)
    except ValidationError as e:
        print("A validation error has occurred.errormessage=", e.json())
        return False
    return True


def _gettimestamp():
    current_time = datetime.now()
    timestamp_str = current_time.strftime('%Y%m%d%H%M%S%f')
    return timestamp_str


class ValidatedSchemaFactory(BaseModel):
    registration_date: str = Field(default_factory=_gettimestamp)


class BaseJsonDbORM:
    _instances = {}
    dbpath: str = None
    dbname: str = None
    schema: ValidatedSchemaFactory = None
    jsondb: JsonDatabase = None
    issingleton: bool = False

    def __new__(cls, *args, **kwargs):
        if cls.issingleton:
            if cls not in cls._instances:
                instance = super().__new__(cls)
                cls._instances[cls] = instance
            return cls._instances[cls]
        return super().__new__(cls)

    def __init__(self):
        if self.dbpath is None:
            os.makedirs(DEFAULT_DB_PATH, exist_ok=True)
            dataset_path = os.path.join(DEFAULT_DB_PATH, f"{self.dbname}.json")
        else:
            os.makedirs(self.dbpath, exist_ok=True)
            dataset_path = os.path.join(self.dbpath, f"{self.dbname}.json")
        self.jsondb = db.getDb(dataset_path)

    def upsertByprimaryKey(self, datainstance: DoFactory) -> bool:
        return _base_validate_and_upsertbyprimarykey_to_jsondatabase(self.schema, self.jsondb, datainstance)

    def upsertsByprimaryKey(self, datainstanceList: list[DoFactory]) -> bool:
        return _base_validate_and_upsertsbyprimarykey_to_jsondatabase(self.schema, self.jsondb, datainstanceList)

    def update_all(self, queryinstance: DoFactory, datainstance: DoFactory) -> bool:
        return _base_validate_and_update_all_to_jsondatabase(self.schema, self.jsondb, queryinstance, datainstance)

    def delete(self, queryinstance: DoFactory) -> bool:
        return _base_delete_to_jsondatabase(self.jsondb, queryinstance)

使用イメージ

前述のパッケージの使用方法を紹介します。

  1. "myStreamlit.py"というファイルを作成します. 
    from localjsondb.jsonDB import ValidatedSchemaFactory, BaseJsonDbORM, DoFactory


class _MyStreamlitProp:
    formname: str
    keyname: str
    value: dict

class _MyStreamlitSchema(_MyStreamlitProp, ValidatedSchemaFactory):
    pass

class MyStreamlitDo(_MyStreamlitProp, DoFactory):
    pass

class MyStreamlitOrm(BaseJsonDbORM):
    dbpath = "./mydb/system"
    schema = _MyStreamlitSchema 

    def __init__(self, _dbname):
        self.dbname = _dbname
        super().__init__()
    
MyStremalit = MyStreamlitOrm("myStreamlit")
  2. "datamanipulate.py"というファイルを作成します. 
    from myStreamlit import MyStreamlitDo, MyStremalit


if __name__ == '__main__':
    myStreamlitDo = MyStreamlitDo()
    myStreamlitDo.formname = "main"
    myStreamlitDo.keyname = "form_menu"
    myStreamlitDo.value = {
        "form_menu":[
            "chat",
            "Document registration",
            "Document editing"
        ]
    }

    MyStremalit.upsertByprimaryKey(myStreamlitDo)
  3. "datamanipulate.py"を実行します. 
    python datamanipulate.py
  4. "./mydb/system/myStreamlit.json"を確認します 
    {
"data": [
    {
        "registration_date": "20240308223121194049",
        "formname": "main",
        "keyname": "form_menu",
        "value": {
            "form_menu": [
            "chat",
            "Document registration",
            "Document editing"
            ]
        },
        "id": <一意のID>
    }
]
}

おわりに

この記事では、Pythonプロジェクトでのデータ管理プロセスを効率化し、開発の生産性を向上させるための強力なツールであるPydanticとPysonDBの組み合わせについて探求しました。型安全性を提供するPydanticと、シンプルながら強力なJSONベースのデータベースであるPysonDBを組み合わせることで、データ整合性を確保しつつ、迅速かつ柔軟にデータベース操作を行うことが可能になります。

詳しい使用方法はこちらをご確認ください。

2
4
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
2
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?