Sanicチュートリアル:5.データベース使用{{でーたべーす:しよう}}

8509 ワード

説明では、Sanicはasync/await構文を使用してプロジェクトを記述できる非同期非ブロックフレームワークであり、非同期フレームワークである以上、使用中に使用されるサードパーティ製パッケージも非同期であることが望ましい.例えばhttpリクエストでは、aihttpではなくrequestsを使用することが望ましい.データベースへの接続についても同様である.次に、Sanicでデータベースに接続する方法をコード形式で説明します.
操作Mysql
mysqlデータベースの非同期操作については、いくつかのスクリプトでしか使用したことがありません.aiomysqlを使用しています.公式ドキュメントでは、sqlalchemyを組み合わせてORMを作成し、aiomysqlが独自に作成した非同期エンジンを提供しています.

from aiomysql.sa import create_engine	
#

次に、非同期文でデータベースを操作する具体的な例を作成します.まず、次のディレクトリを作成します.

aio_mysql	
├── demo.py	
├── model.py	
└── requirements.txt

テーブルの作成:

create database test_mysql;	
CREATE TABLE user	
(	
  id        INT AUTO_INCREMENT	
    PRIMARY KEY,	
  user_name VARCHAR(16) NOT NULL,	
  pwd       VARCHAR(32) NOT NULL,	
  real_name VARCHAR(6)  NOT NULL	
);

準備が整いました.次はコードを書きます.

# script: model.py	
import sqlalchemy as sa	
metadata = sa.MetaData()	
user = sa.Table(	
    'user',	
    metadata,	
    sa.Column('id', sa.Integer, autoincrement=True, primary_key=True),	
    sa.Column('user_name', sa.String(16), nullable=False),	
    sa.Column('pwd', sa.String(32), nullable=False),	
    sa.Column('real_name', sa.String(6), nullable=False),	
)	
# script: demo.py	
import asyncio	
from aiomysql.sa import create_engine	
from model import user,metadata	
async def go(loop):	
    """	
    aiomysql    ：https://github.com/aio-libs/aiomysql	
    :param loop:	
    :return:	
    """	
    engine = await create_engine(user='root', db='test_mysql',	
                                 host='127.0.0.1', password='123456', loop=loop)	
    async with engine.acquire() as conn:	
        await conn.execute(user.insert().values(user_name='user_name01', pwd='123456', real_name='real_name01'))	
        await conn.execute('commit')	
        async for row in conn.execute(user.select()):	
            print(row.user_name, row.pwd)	
    engine.close()	
    await engine.wait_closed()	
loop = asyncio.get_event_loop()	
loop.run_until_complete(go(loop))

python demoを実行します.py、次の出力が表示されます.

user_name01 123456

簡単ですね.具体的な例はaio_を参照してください.mysql、SQLAlchemyのような操作が好きな場合は、非同期ORM、ginoをお勧めします.
操作MongoDB
私のアマチュアの书く1つのプロジェクト、基本的にMongoDBを使ってデータを贮蔵して、非同期の操作MongoDBに対して、现在Pythonは主にmotorを使って、使うのは依然としてとても简単で、しかし具体的な机能を结び付けて、异なる需要があって、最后に各种の接続方案を形成して、ここで私は主に自分がどのように使うのかを分かち合って、ディレクトリは次のとおりです.

aio_mongo	
├── demo.py	
└── requirements.txt

MongoDBは、リレーショナル・データベースと非リレーショナル・データベースの間にある分散ファイル・ベースのデータベースであるため、使用も柔軟であり、demo.pyを開きます.

#!/usr/bin/env python	
import os	
from functools import wraps	
from motor.motor_asyncio import AsyncIOMotorClient	
MONGODB = dict(	
    MONGO_HOST=os.getenv('MONGO_HOST', ""),	
    MONGO_PORT=os.getenv('MONGO_PORT', 27017),	
    MONGO_USERNAME=os.getenv('MONGO_USERNAME', ""),	
    MONGO_PASSWORD=os.getenv('MONGO_PASSWORD', ""),	
    DATABASE='test_mongodb',	
)	
class MotorBaseOld:	
    """	
           db     ，         	
    """	
    _db = None	
    MONGODB = MONGODB	
    def client(self, db):	
        # motor	
        self.motor_uri = 'mongodb://{account}{host}:{port}/{database}'.format(	
            account='{username}:{password}@'.format(	
                username=self.MONGODB['MONGO_USERNAME'],	
                password=self.MONGODB['MONGO_PASSWORD']) if self.MONGODB['MONGO_USERNAME'] else '',	
            host=self.MONGODB['MONGO_HOST'] if self.MONGODB['MONGO_HOST'] else 'localhost',	
            port=self.MONGODB['MONGO_PORT'] if self.MONGODB['MONGO_PORT'] else 27017,	
            database=db)	
        return AsyncIOMotorClient(self.motor_uri)	
    @property	
    def db(self):	
        if self._db is None:	
            self._db = self.client(self.MONGODB['DATABASE'])[self.MONGODB['DATABASE']]	
        return self._db

私が最初に使用したのは、MongoDBを接続するために使用され、上のコードは集合中のdbが_dbメンテナンスは、一度しか作成されないことを保証します.プロジェクトで勝手にコレクションを変更しない場合は、大きな問題はありません.そうでない場合は、次のような接続方法で、コレクションとdbを自由に交換することをお勧めします.

def singleton(cls):	
    """	
                            ：https://github.com/howie6879/Sanic-For-Pythoneer/blob/master/docs/part2/%E9%99%84%E5%BD%95%EF%BC%9A%E5%85%B3%E4%BA%8E%E8%A3%85%E9%A5%B0%E5%99%A8.md	
    :param cls: cls	
    :return: instance	
    """	
    _instances = {}	
    @wraps(cls)	
    def instance(*args, **kw):	
        if cls not in _instances:	
            _instances[cls] = cls(*args, **kw)	
        return _instances[cls]	
    return instance	
@singleton	
class MotorBase:	
    """	
      mongodb                	
    About motor's doc: https://github.com/mongodb/motor	
    """	
    _db = {}	
    _collection = {}	
    MONGODB = MONGODB	
    def __init__(self):	
        self.motor_uri = ''	
    def client(self, db):	
        # motor	
        self.motor_uri = 'mongodb://{account}{host}:{port}/{database}'.format(	
            account='{username}:{password}@'.format(	
                username=self.MONGODB['MONGO_USERNAME'],	
                password=self.MONGODB['MONGO_PASSWORD']) if self.MONGODB['MONGO_USERNAME'] else '',	
            host=self.MONGODB['MONGO_HOST'] if self.MONGODB['MONGO_HOST'] else 'localhost',	
            port=self.MONGODB['MONGO_PORT'] if self.MONGODB['MONGO_PORT'] else 27017,	
            database=db)	
        return AsyncIOMotorClient(self.motor_uri)	
    def get_db(self, db=MONGODB['DATABASE']):	
        """	
            db  	
        :param db: database name	
        :return: the motor db instance	
        """	
        if db not in self._db:	
            self._db[db] = self.client(db)[db]	
        return self._db[db]	
    def get_collection(self, db_name, collection):	
        """	
                	
        :param db_name: database name	
        :param collection: collection name	
        :return: the motor collection instance	
        """	
        collection_key = db_name + collection	
        if collection_key not in self._collection:	
            self._collection[collection_key] = self.get_db(db_name)[collection]	
        return self._collection[collection_key]

MotorBaseインスタンスの重複作成を回避するために、リソースの有効利用を保証するための単一のモードを実現することができます.具体的なコードとdemoの実行はaio_を参照してください.mongo
操作Redis
Redisの非同期操作については、asyncio_redisを選択しました.これを使用する必要はありません.他のライブラリがより良いかもしれません.私はこの例を挙げて、次のディレクトリを作成します.

aio_redis	
├── demo.py	
└── requirements.txt

redis接続プールを作成するには、次の手順に従います.

#!/usr/bin/env python	
import os	
import asyncio_redis	
REDIS_DICT = dict(	
    IS_CACHE=True,	
    REDIS_ENDPOINT=os.getenv('REDIS_ENDPOINT', "localhost"),	
    REDIS_PORT=os.getenv('REDIS_PORT', 6379),	
    REDIS_PASSWORD=os.getenv('REDIS_PASSWORD', None),	
    DB=0,	
    POOLSIZE=10,	
)	
class RedisSession:	
    """	
      redis   	
    """	
    _pool = None	
    async def get_redis_pool(self):	
        if not self._pool:	
            self._pool = await asyncio_redis.Pool.create(	
                host=str(REDIS_DICT.get('REDIS_ENDPOINT', "localhost")), port=int(REDIS_DICT.get('REDIS_PORT', 6379)),	
                poolsize=int(REDIS_DICT.get('POOLSIZE', 10)), password=REDIS_DICT.get('REDIS_PASSWORD', None),	
                db=REDIS_DICT.get('DB', None)	
            )	
        return self._pool

具体的にはaio_を参照redis、使うのは簡単で、多くの叙述をしません.
説明
他のタイプのデータベースを使用する場合も、実際には同じように使用されます.この章のコードアドレスはdemo 05を参照してください

2進数の中で0あるいは1の個数を求めます

Mybatis-Plus