Python-peewee connection method

Method 1

To connect to a MySQL database, we will use MySQLDatabase. After the database name, you can specify arbitrary connection parameters that will be passed back to the driver (either MySQLdb or pymysql).

# http://docs.peewee-orm.com/en/latest/peewee/database.html#using-mysql
from peewee import (
    MySQLDatabase,
    Model,
    DateTimeField,
)
#MySQL connection
db = MySQLDatabase('my_database')

class BaseModel(Model):
    """
    DB-Base
    """
    class Meta:
        database = db

    created_at = DateTimeField(default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)

    def save(self, *args, **kwargs):
        self.updated_at = datetime.datetime.now()
        return super(BaseModel, self).save(*args, **kwargs)

reference

http://docs.peewee-orm.com/en/latest/index.html

Method 2

Connection pooling is provided by the pool module, included in the playhouse extensions library.

The pool supports:

• Timeout after which connections will be recycled.
• Upper bound on the number of open connections.

#http://docs.peewee-orm.com/en/latest/peewee/database.html#connection-pooling

from playhouse.pool import PooledMySQLDatabase
from peewee import (
    Model,
    UUIDField,
    CharField,
)
#MySQL connection
db = PooledMySQLDatabase('my_database', max_connections=my['max_connections'])

reference

https://stackoverflow.com/questions/34764266/peewee-pooledmysqldatabase-has-cache-or-buffer Q：peewee PooledMySQLDatabase has cache or buffer? A： No cache, no buffer. It has to do with transaction management. Just be sure that you're running your statements in transactions and you should be fine.

Multi-threaded apps DO NOT REQUIRE A POOL with peewee. Peewee correctly manages per-thread connections even without a pool.

Method 3

peewee-async is a library providing asynchronous interface powered by asyncio for peewee ORM.

#https://peewee-async.readthedocs.io/en/latest/
import peewee_async

from peewee import (
    Model,
    UUIDField,
    CharField,
)

#Self-determined connection
class AsyncMySQLConnection(peewee_async.AsyncMySQLConnection):
    """
    Asynchronous database connection pool.
    """

    def __init__(self, *, database=None, loop=None, timeout=None, **kwargs):
        self.pool = None
        self.loop = loop
        self.database = database
        self.timeout = timeout
        kwargs.setdefault('pool_recycle', 3600)
        self.connect_params = kwargs

#MySQL connection
db = peewee_async.PooledMySQLDatabase('my_database'
    max_connections=my['max_connections'],
    async_conn=AsyncMySQLConnection)

reference

https://www.w3cschool.cn/hack_tutorial/hack_tutorial-qay32dz7.html